Spaces:

Reoo88
/

Prayer-Time-Bot

Running

v2.41 dual simultaneous YouTube ingest (primary+backup tee fan-out)
fe77a7a verified 19 days ago
26.1 kB
#!/usr/bin/env python3
"""Lifecycle tests for TikTok Live Monitor → Direct Relay → YouTube.
No recording. No archive. Tests that can run without a live TikTok session
use mocks; watchdog/FFmpeg tests use a local HTTP source when ffmpeg exists.
"""
import json, os, sys, tempfile, threading, time, types, pathlib, subprocess, shutil

ROOT = pathlib.Path(__file__).resolve().parent
TMP = pathlib.Path(tempfile.mkdtemp(prefix='relay-test-'))
REAL_FF = shutil.which('ffmpeg')
_ff = os.environ.get('FFMPEG_PATH') or REAL_FF
if not _ff:
    _ff = str(TMP / 'dummy-ffmpeg')
    pathlib.Path(_ff).write_text('#!/bin/sh\necho dummy\nexit 0\n')
    os.chmod(_ff, 0o755)
os.environ.update(
    PASSWORD='test-pass-not-for-prod',
    DATA_DIR=str(TMP / 'data'),
    CHECK_INTERVAL='8',
    SOURCE_STALL_TIMEOUT='3',
    YOUTUBE_SOURCE_STALL_SECONDS='3',
    YOUTUBE_SOURCE_READ_TIMEOUT='1',
    YOUTUBE_PROGRESS_STALL_SECONDS='20',
    YOUTUBE_SEND_MODE='copy',
    YOUTUBE_TITLE_TEMPLATE='{broadcaster} | {date} {time}',
    FFMPEG_PATH=_ff,
)
sys.path.insert(0, str(ROOT))

passed = failed = 0
def check(name, cond, detail=''):
    global passed, failed
    if cond:
        passed += 1
        print('PASS  %s'%name)
    else:
        failed += 1
        print('FAIL  %s  %s'%(name, detail))

# ---------- syntax / import ----------
import ast
for f in ('app.py','relay.py','youtube_manager.py','youtube_api.py','flv_bridge.py','tiktok_auth.py'):
    ast.parse((ROOT/f).read_text())
check('syntax of core modules', True)

import app
from app import rank_sources, source_brief
from relay import Sender, SourceStalled, RelayStartError
from youtube_api import DEFAULT_SETTINGS
from datetime import datetime
from zoneinfo import ZoneInfo

check('import app (PASSWORD required, no mkvmerge)', True)
check('no recordings dir required', not hasattr(app, 'FLVWriter') and not hasattr(app, 'finalize') and not hasattr(app, 'record'))
check('ffmpeg path resolved or later skipped', True)

# ---------- source selection ----------
payload = {
    'blocks': [{
        'pull_data': {
            'stream_data': json.dumps({'data': {
                'origin': {'main': {
                    'flv': 'https://pull.example/origin.flv',
                    'hls': 'https://pull.example/origin.m3u8',
                    'sdk_params': json.dumps({'VCodec':'h265','resolution':'1080x1920','vbitrate':3200}),
                }},
                'hd': {'main': {
                    'flv': 'https://pull.example/hd.flv',
                    'sdk_params': json.dumps({'VCodec':'h264','resolution':'720x1280','vbitrate':1100}),
                }},
                'sd': {'main': {
                    'flv': 'https://pull.example/sd.flv',
                    'sdk_params': json.dumps({'VCodec':'h264','resolution':'480x854','vbitrate':600}),
                }},
            }})
        }
    }]
}
ranked = rank_sources(payload)
check('keeps every rendition including lower quality', len(ranked) >= 3, 'got %d'%len(ranked))
briefs = [source_brief(c) for c in ranked]
check('brief has no url', all('http' not in b and 'example' not in b for b in briefs), str(briefs))
# H264 FLV should beat HEVC origin because copy-compat + FLV
top = ranked[0]
check('prefers H264 over HEVC origin', '264' in (top.get('codec') or '').lower(), 'top=%s codec=%s'%(top.get('quality_name'), top.get('codec')))
check('still lists HEVC as fallback', any('265' in (c.get('codec') or '').lower() or 'hevc' in (c.get('codec') or '').lower() for c in ranked))
check('sd not dropped', any(c.get('quality_name')=='sd' for c in ranked))
# TikTok webcast/room/info returns hls_pull_url as one string on current live rooms.
# Calling .items() on it used to throw, so the fallback disappeared exactly when needed.
string_hls=rank_sources({'hls_pull_url':'https://pull.example/live.m3u8'})
check('single-string HLS fallback is accepted',
      len(string_hls)==1 and string_hls[0].get('media')=='hls',str(string_hls))

# ---------- title template ----------
tpl = os.environ['YOUTUBE_TITLE_TEMPLATE']
dt = datetime(2026,9,23,14,35,tzinfo=ZoneInfo('Asia/Baghdad'))
title = tpl.format(broadcaster='StrongArmor', date=dt.strftime('%Y-%m-%d'), time=dt.strftime('%H:%M'),
                   name='StrongArmor', title='x', username='u', room_id='1', nickname='StrongArmor', live_title='x', hashtags='')
check('youtube title has Baghdad date/time', title=='StrongArmor | 2026-09-23 14:35', title)
check('default settings title uses template vars', '{broadcaster}' in DEFAULT_SETTINGS['title'] or '{date}' in DEFAULT_SETTINGS['title'])

# ---------- stall timeout env ----------
s = Sender('ffmpeg', lambda *a: None)
check('SOURCE_STALL_TIMEOUT honored', abs(s.stall_limit-3)<0.01, str(s.stall_limit))

# ---------- recording remnants ----------
src = (ROOT/'app.py').read_text() + (ROOT/'index.html').read_text()
check('UI has no recordings panel', 'التسجيلات المحفوظة' not in (ROOT/'index.html').read_text())
check('UI has no history panel', 'سجل البثوث' not in (ROOT/'index.html').read_text())
check('no /api/files endpoint', '/api/files' not in (ROOT/'app.py').read_text())
check('Dockerfile has no mkvtoolnix', 'mkvtoolnix' not in (ROOT/'Dockerfile').read_text())
check('requirements has no huggingface_hub', 'huggingface_hub' not in (ROOT/'requirements.txt').read_text())

# ---------- health + monitor isolation ----------
h = app.health_public()
check('TEST public /health', h.get('application')=='running', str(h))
check('public health is minimal (no account/relay details)',
      set(h)<= {'status','application','monitor','version'}, str(h))
# TEST 1: offline account does not start ffmpeg
app.tasks['offlineuser'] = app._blank_task()
app.tasks['offlineuser']['enabled'] = True
orig = app.observe_session
app.observe_session = lambda user, force=False: {'status':'offline','room_id':'','sources':[],'nickname':'','live_title':''}
try:
    app._check_account('offlineuser')
    check('TEST 1 offline → no LIVE', app.tasks['offlineuser']['state']=='OFFLINE')
    check('TEST 1 no ffmpeg handle', not app.youtube.senders)
finally:
    app.observe_session = orig

# TEST 10: two accounts, one error does not kill the other
app.tasks['good'] = app._blank_task(); app.tasks['good']['enabled']=True
app.tasks['bad'] = app._blank_task(); app.tasks['bad']['enabled']=True
def obs(user, force=False):
    if user=='bad': raise RuntimeError('تيك توك حدّ الطلبات مؤقتًا (429)')
    return {'status':'offline','room_id':'','sources':[],'nickname':'','live_title':''}
app.observe_session = obs
try:
    app._check_account('good'); app._check_account('bad')
    check('TEST 10 good account still OFFLINE not dead', app.tasks['good']['state']=='OFFLINE')
    check('TEST 10 bad account isolated ERROR/BACKOFF', app.tasks['bad']['state'] in ('ERROR','BACKOFF'))
finally:
    app.observe_session = orig
    app.tasks.pop('good',None); app.tasks.pop('bad',None); app.tasks.pop('offlineuser',None)

# TEST 2 shape: LIVE detected updates state, still no ffmpeg without YouTube enable
flv = {'url':'https://pull.example/hd.flv','media':'flv','codec':'h264','quality_name':'hd',
       'resolution':'720x1280','brief':'hd 1280p H264','label':'hd · 720x1280 · h264','score':(1,1,1,1,1,1)}
app.tasks['liveuser'] = app._blank_task(); app.tasks['liveuser']['enabled']=True
app.observe_session = lambda user, force=False: {'status':'live','room_id':'123','sources':[flv],'nickname':'StrongArmor','live_title':'x'}
try:
    app._check_account('liveuser')
    check('TEST 2 LIVE detected', app.tasks['liveuser']['state']=='LIVE', app.tasks['liveuser']['state'])
    check('TEST 2 source brief stored', 'H264' in (app.tasks['liveuser'].get('source') or ''), str(app.tasks['liveuser']))
    check('TEST 2 no relay without YouTube enable', not app.youtube.senders)
finally:
    app.observe_session = orig
    app.tasks.pop('liveuser',None)

# ---------- watchdog / ffmpeg (optional) ----------
FF = REAL_FF
if not FF:
    print('SKIP TEST 3-6/8 ffmpeg lifecycle (no ffmpeg in this environment; Docker image has it)')
    check('TEST 3-8 noted as docker-only when ffmpeg missing', True)
else:
    import http.server, socketserver
    from relay import Sender
    src = TMP/'src.flv'
    subprocess.run([FF,'-hide_banner','-loglevel','error','-y','-f','lavfi','-i','testsrc2=size=160x120:rate=20',
                    '-f','lavfi','-i','sine=frequency=440:sample_rate=48000','-t','20',
                    '-c:v','libx264','-preset','ultrafast','-pix_fmt','yuv420p','-c:a','aac','-ar','48000','-ac','1',
                    '-f','flv',str(src)], check=True, capture_output=True)
    DATA = src.read_bytes()
    state = {'mode':'flow'}
    class H(http.server.BaseHTTPRequestHandler):
        protocol_version='HTTP/1.1'
        def log_message(self,*a): pass
        def do_GET(self):
            self.send_response(200); self.send_header('Content-Type','video/flv'); self.end_headers()
            pos=0
            try:
                while True:
                    if state['mode']=='dead': return
                    if state['mode']=='stall':
                        time.sleep(0.2); continue
                    self.wfile.write(DATA[pos:pos+4096]); self.wfile.flush()
                    time.sleep(0.02); pos=(pos+4096)%max(4096,len(DATA)-4096)
            except (BrokenPipeError,ConnectionResetError,OSError):
                pass
    class S(socketserver.ThreadingTCPServer):
        allow_reuse_address=True; daemon_threads=True
    srv=S(('127.0.0.1',0),H)
    threading.Thread(target=srv.serve_forever,daemon=True).start()
    port=srv.server_address[1]
    dest=str(TMP/'out.flv')
    cand={'url':'http://127.0.0.1:%d/live.flv'%port,'media':'flv','codec':'h264','label':'hd 720p H264','brief':'hd 720p H264'}
    logs=[]
    sender=Sender(FF, lambda u,m: logs.append(m))
    stop=threading.Event()

    # TEST 3: continuous data → relay stays alive a few seconds
    t0=time.monotonic(); err=None
    def run():
        global err
        try: sender.send('u',cand,dest,stop)
        except Exception as e: err=e
    th=threading.Thread(target=run,daemon=True); th.start()
    time.sleep(4)
    alive=sender.alive()
    check('TEST 3 continuous data keeps FFmpeg alive', alive, 'alive=%s err=%s'%(alive,err))
    snap=sender.snapshot()
    check('TEST 3 bytes received', (snap.get('source_bytes') or 0)>0, str(snap.get('source_bytes')))

    # TEST 5: stall → watchdog
    state['mode']='stall'
    stalled=False
    t_wait=time.monotonic()
    while time.monotonic()-t_wait<12:
        if any('WATCHDOG' in m or 'stall' in m.lower() or 'صمت' in m for m in logs) or isinstance(err,SourceStalled) or not sender.alive():
            stalled=True; break
        time.sleep(0.3)
    check('TEST 5 stall watchdog fires', stalled or not sender.alive(), 'err=%s logs=%s'%(err, logs[-6:]))
    stop.set(); sender.terminate(); th.join(timeout=8)
    check('TEST 4/6 cleanup leaves no process', sender.process is None or sender.process.poll() is not None)

    # TEST 8-ish: stop after live
    check('TEST 8 session ends, sender not alive', not sender.alive())
    srv.shutdown()

# ---------- backoff ----------
from youtube_manager import ManagedYouTube
logs=[]
yt=ManagedYouTube('ffmpeg', lambda u,force=False: {'status':'offline','room_id':'','sources':[]},
                  lambda u,m: logs.append((u,m)), lambda: ['a'], TMP/'yt.json')
yt.retry['a']=1
yt.retry['a']=yt._backoff('a')*2
check('TEST 11 backoff increases', yt._backoff('a')>=2, str(yt._backoff('a')))
yt.retry['a']=9999
check('TEST 11 backoff capped', yt._backoff('a')<=yt.retry_max)
check('reliability selection keeps lower H264 after preferred 720-class source',
      any(c.get('quality_name')=='sd' for c in yt.video_sources(ranked)),
      str([c.get('quality_name') for c in yt.video_sources(ranked)]))

# Same-codec lower rendition is spliced without reopening RTMP; quality is restored only on a
# later natural refresh after the stable hold interval.
high={'url':'https://x/high.flv','media':'flv','codec':'h264','quality_name':'hd',
      'resolution':'720x1280','width':720,'height':1280,'label':'hd','brief':'hd 720p H264'}
low={'url':'https://x/low.flv','media':'flv','codec':'h264','quality_name':'sd',
     'resolution':'480x854','width':480,'height':854,'label':'sd','brief':'sd 480p H264'}
_saved_observe,_saved_select=yt._observe,yt._select_source
yt._observe=lambda *a,**k: {'status':'live','room_id':'1','sources':[high,low]}
yt._select_source=lambda user,obs: [dict(high,user=user),dict(low,user=user)]
yt._rendition_pin['adaptive']=yt._source_pin(high)
chosen=yt._refresh_flv('adaptive',high)
check('stalled high FLV falls to lower same-codec source in the same publisher',
      chosen and chosen.get('quality_name')=='sd' and yt._quality_fallback.get('adaptive'),str(chosen))
os.environ['YOUTUBE_SOURCE_UPGRADE_SECONDS']='60'
yt._quality_fallback['adaptive']['since']=time.monotonic()-61
yt._rendition_pin['adaptive']=yt._source_pin(low)
restored=yt._refresh_flv('adaptive',low)
check('stable lower source restores preferred quality on a later natural refresh',
      restored and restored.get('quality_name')=='hd' and 'adaptive' not in yt._quality_fallback,str(restored))
yt._observe,yt._select_source=_saved_observe,_saved_select
os.environ.pop('YOUTUBE_SOURCE_UPGRADE_SECONDS',None)

# ---------- unstable source fallback bookkeeping ----------
yt.mark_unstable('bob', {'quality_name':'origin','media':'flv','codec':'h265','resolution':'1080x1920','brief':'origin 1080p HEVC'}, 'stalled')
check('TEST 7 unstable source recorded', 'origin|flv|h265|1080x1920' in yt._unstable.get('bob',set()))
obs={'status':'live','room_id':'1','sources':[
    {'url':'https://x/a.flv','media':'flv','codec':'h265','quality_name':'origin','resolution':'1080x1920','label':'origin','brief':'origin 1080p HEVC'},
    {'url':'https://x/b.flv','media':'flv','codec':'h264','quality_name':'hd','resolution':'720x1280','label':'hd','brief':'hd 720p H264'},
]}
# video_sources may filter; monkeypatch probe to skip network
yt.probe=lambda user,c: True
yt.video_sources=staticmethod(lambda sources: sources)
sel=yt._select_source('bob', obs)
check('TEST 7 fallback prefers non-unstable', sel and 'h264' in (sel[0].get('codec') or '').lower(), str([(c.get('quality_name'),c.get('codec')) for c in sel]))
hls_obs={'status':'live','room_id':'1','sources':[
    {'url':'https://x/a.flv','media':'flv','codec':'h264','quality_name':'hd','resolution':'720x1280','label':'hd flv','brief':'hd 720p H264'},
    {'url':'https://x/a.m3u8','media':'hls','codec':'h264','quality_name':'hd','resolution':'720x1280','label':'hd hls','brief':'hd 720p H264 HLS'},
]}
yt.starve['bob']=1
yt._unstable.pop('bob', None)
sel=yt._select_source('bob', hls_obs)
check('first FLV stall prefers HLS', sel and sel[0].get('media')=='hls', str([(c.get('media'),c.get('quality_name')) for c in (sel or [])]))
yt.starve.pop('bob', None)

# ---------- hub / env RTMPS without OAuth ----------
from ingest import hub, env_rtmp_destination, format_session_title, rtmp_configured
os.environ['YOUTUBE_RTMP_URL']='rtmps://a.rtmps.youtube.com:443/live2'
os.environ['YOUTUBE_STREAM_KEY']='unit-test-key'
check('env RTMPS dest built without OAuth', env_rtmp_destination().endswith('/unit-test-key'))
check('rtmp_configured true', rtmp_configured())
title,_,_=format_session_title('StrongArmor', datetime(2026,9,23,14,35,tzinfo=ZoneInfo('Asia/Baghdad')))
check('format_session_title Baghdad', title=='StrongArmor | 2026-09-23 14:35', title)
class _Alive:
    def is_alive(self):return True
check('hub occupy first session', hub.occupy('alice','direct', thread=_Alive()))
check('TEST 14 duplicate relay blocked', not hub.occupy('alice','oauth', thread=_Alive()))
check('other broadcaster independent', hub.occupy('bob','oauth', thread=_Alive()))
check('single env key busy', hub.env_key_busy())
hub.release('alice'); hub.release('bob')
check('hub released', not hub.busy('alice') and not hub.busy('bob'))
class _PushSnd:
    def alive(self):return True
class _IdleSnd:
    def alive(self):return False
class _PushTh:
    def is_alive(self):return True
check('env occupy first', hub.occupy('e1','direct', sender=_PushSnd(), thread=_PushTh(), dest='env'))
check('second env publisher blocked', hub.env_key_busy())
check('oauth dest independent of env', hub.occupy('e2','oauth', sender=_PushSnd(), thread=_PushTh(), dest='oauth'))
check('active_count is ffmpeg not idle threads', hub.active_count()==2)
hub.release('e1'); hub.release('e2')
check('env free after release', not hub.env_key_busy())
h=app.health_public()
check('public health hides last_error', 'last_error' not in h, str(h.keys()))
check('public health hides last_error_user', 'last_error_user' not in h, str(h.keys()))
check('health has version', bool(h.get('version')))

# Null TikTok payloads must be Offline, not AttributeError → sticky last_error
orig_gj = app.get_json
app.get_json = lambda url, params, user=None: {'data': {'liveRoom': None, 'user': None}}
try:
    obs = app._observe_session('nullroom')
    check('null liveRoom/user → offline not crash', obs.get('status')=='offline', str(obs))
finally:
    app.get_json = orig_gj
app.get_json = lambda url, params, user=None: {'data': None}
try:
    obs = app._observe_session('nulldata')
    check('null data → offline not crash', obs.get('status')=='offline', str(obs))
finally:
    app.get_json = orig_gj

app.tasks['staleerr'] = app._blank_task()
app.tasks['staleerr']['enabled'] = True
app.tasks['staleerr']['last_error'] = 'فشل طلب بيانات البث'
app.tasks['staleerr']['state'] = 'BACKOFF'
orig = app.observe_session
app.observe_session = lambda user, force=False: {'status':'offline','room_id':'','sources':[],'nickname':'','live_title':''}
try:
    app._check_account('staleerr')
    check('offline clears last_error', app.tasks['staleerr'].get('last_error')=='', str(app.tasks['staleerr']))
    check('offline after error is OFFLINE', app.tasks['staleerr']['state']=='OFFLINE')
    h=app.health_detail()
    check('authenticated detail last_error empty after offline', not h.get('last_error'), str(h))
    h=app.health_public()
    check('public health does not list broadcasters', 'broadcasters' not in h)
    check('public health hides metadata automation', 'metadata_automation' not in h)
finally:
    app.observe_session = orig
    app.tasks.pop('staleerr', None)

# OAuth off + env dest → DirectRelay occupies (video path does not need OAuth)
app.youtube.journal.change(lambda d: d.update(enabled=False))
app.tasks['diruser']=app._blank_task(); app.tasks['diruser']['enabled']=True
app.observe_session = lambda user, force=False: {'status':'live','room_id':'99','sources':[flv],'nickname':'N','live_title':'x'}
try:
    app._maybe_start_direct('diruser', {'status':'live','room_id':'99','sources':[flv],'nickname':'N','live_title':'x'})
    check('DirectRelay starts without OAuth', 'diruser' in app._direct_handles or hub.busy('diruser'))
finally:
    app._stop_direct('diruser','test')
    app.observe_session = orig
    app.tasks.pop('diruser', None)
    check('DirectRelay cleaned up', not hub.busy('diruser') and 'diruser' not in app._direct_handles)

# dead sender restart without waiting a full scan
class _DeadTh:
    def is_alive(self):return False
class _LiveTh:
    def is_alive(self):return True
class _DeadSnd:
    def alive(self):return False
    def terminate(self):pass
called=[]
yt.journal.change(lambda d: d['events'].update({'bob:1':{
    'username':'bob','room_id':'1','closing':False,'offline_since':None,
    'bound':True,'broadcast_id':'B','youtube_status':'live','title':'t'}}))
yt.senders['bob:1']={'thread':_DeadTh(),'stop':threading.Event(),'sender':_DeadSnd()}
yt.start_sender=lambda key: called.append(key)
n=yt.restart_dead_senders()
check('dead sender restart without full scan', n==1 and called==['bob:1'], 'n=%s called=%s'%(n,called))
yt.senders['bob:1']={'thread':_LiveTh(),'stop':threading.Event(),'sender':_DeadSnd()}
called.clear()
check('live sender not restarted', yt.restart_dead_senders()==0 and not called)
yt.journal.change(lambda d: d['events'].update({'bob:1': dict(d['events']['bob:1'], offline_since=time.time(), offline_count=1, youtube_status='live')}))
yt.senders['bob:1']={'thread':_DeadTh(),'stop':threading.Event(),'sender':_DeadSnd()}
called.clear()
check('first offline blip still restarts dead sender', yt.restart_dead_senders()==1 and called==['bob:1'], 'n=%s called=%s'%(yt.restart_dead_senders() if False else 1, called))
yt.journal.change(lambda d: d['events'].update({'bob:1': dict(d['events']['bob:1'], offline_since=time.time(), offline_count=2, youtube_status='live')}))
yt.senders['bob:1']={'thread':_DeadTh(),'stop':threading.Event(),'sender':_DeadSnd()}
called.clear()
check('confirmed offline skips restart', yt.restart_dead_senders()==0 and not called)
stall_obs={'status':'live','room_id':'1','sources':[
    {'url':'https://x/a.flv','media':'flv','codec':'h264','quality_name':'hd','resolution':'720x1280','label':'hd','brief':'hd 720p H264'},
]}
yt.video_sources=staticmethod(lambda sources: sources)
yt._skip_probe_once['bob']=True
probed=[]
yt.probe=lambda user,c: probed.append(c) or True
sel=yt._select_source('bob', stall_obs)
check('stall skip probe uses ranked sources', bool(sel) and not probed, 'sel=%s probed=%s'%(bool(sel), len(probed)))
yt._skip_hold.add('bob')
yt._held['bob']=(time.monotonic(), {'status':'live','room_id':'9','sources':[{'url':'https://x/old.flv'}]})
yt.observe=lambda user,force=False: {'status':'offline','room_id':'','sources':[]}
fresh=yt._observe('bob', allow_hold=True)
check('skip hold after stall is honest offline', fresh.get('status')=='offline', str(fresh))
yt.senders.pop('bob:1', None)
yt.journal.change(lambda d: d['events'].pop('bob:1', None))

df=(ROOT/'Dockerfile').read_text()
check('Dockerfile stall default 12', 'SOURCE_STALL_TIMEOUT=12' in df and 'YOUTUBE_SOURCE_STALL_SECONDS=12' in df)
check('Dockerfile concurrent cap 4', 'MAX_CONCURRENT_RELAYS=4' in df and 'YOUTUBE_MAX_EVENTS=4' in df)
check('Dockerfile defaults to byte-copy AVC/Enhanced-HEVC without forced x264',
      'YOUTUBE_LIGHT_MAX=0' in df and
      'YOUTUBE_AUTO_TRANSCODE=0' in df and
      'YOUTUBE_FORCE_TRANSCODE_USERS=' not in df and
      'YOUTUBE_ENHANCED_RTMP=auto' in df and
      'stream_monitor.py' in df and 'YOUTUBE_HEALTH_INTERVAL=20' in df and
      'STREAM_EVENT_MAX=10000' in df)
mgr_src=(ROOT/'youtube_manager.py').read_text()
check('run_manager calls restart_dead_senders', 'restart_dead_senders' in mgr_src.split('def run_manager')[1])
pump=(ROOT/'relay.py').read_text()
empty_i=pump.find("if not chunk:")
last_i=pump.find("sess['last'] = time.monotonic()", pump.find('def _pump_source'))
check('empty FLV chunks do not refresh last', empty_i!=-1 and last_i!=-1 and empty_i<last_i, 'empty=%s last=%s'%(empty_i,last_i))
td=pump.split('def _teardown')[1].split('def send')[0]
check('teardown does not stop supervisor Event', 'stop.set()' not in td and "sess['force_stop']" in td)
check('send keeps FFmpeg via refresh reattach', 'def _try_reattach' in pump and 'def send(self, user, candidate, destination, stop, refresh=None, extra_destinations=None)' in pump)
check('sender_loop pins codec family during refresh',
      "refresh=lambda: self._refresh_flv(user,getattr(sender,'current_candidate',None))" in mgr_src and
      'def _refresh_flv(self,user,current=None)' in mgr_src)
check('no runtime caller restarts output on FormatChanged',
      all('FormatChanged' not in (ROOT/name).read_text() for name in ('app.py','relay.py','youtube_manager.py')))
from flv_bridge import FLVBridge
def _flv_avc(ts=0, seq=False, payload=b'\x00'):
    if seq:
        body=bytes([0x17, 0, 0, 0, 0])+b'cfgX'+payload
    else:
        body=bytes([0x17, 1, 0, 0, 0])+payload
    size=len(body)
    hdr=bytes([9])+size.to_bytes(3,'big')+bytes([(ts>>16)&255,(ts>>8)&255,ts&255,(ts>>24)&255,0,0,0])
    return hdr+body+(11+size).to_bytes(4,'big')
def _file(tags):
    return b'FLV\x01\x05'+(9).to_bytes(4,'big')+b'\x00\x00\x00\x00'+tags
br=FLVBridge(detect_format_change=True)
out1=br.feed(_file(_flv_avc(0, seq=True)+_flv_avc(40, seq=False)))
check('first FLV emits file header', out1.startswith(b'FLV'))
last_before=br.last_out_ts
br.reconnect()
out2=br.feed(_file(_flv_avc(0, seq=True)+_flv_avc(40, seq=False, payload=b'\x01')))
check('reconnect drops second FLV header', bool(out2) and not out2.startswith(b'FLV'), repr(out2[:8]))
check('reconnect DTS stays monotonic', br.last_out_ts is not None and last_before is not None and br.last_out_ts>last_before, 'before=%s after=%s'%(last_before, br.last_out_ts))
check('health version is stall12', any(x in str(h.get('version') or '') for x in ('multi','guard','v2.')), str(h.get('version')))

# first TikTok offline blip must not kill FFmpeg (YouTube spinner root cause)
class _KeepTh:
    def is_alive(self):return True
    def join(self,timeout=None):pass
class _KeepSnd:
    def alive(self):return True
    def terminate(self):self.killed=True
    def snapshot(self):return {'source_age':1}
    killed=False
yt.journal.change(lambda d: d['events'].update({'keep:1':{
    'username':'keep','room_id':'1','closing':False,'offline_since':None,'offline_count':0,
    'bound':True,'broadcast_id':'B','youtube_status':'live','title':'t','was_live':True,
    'marker':'m','created_at':time.time(),'note':''}}))
yt.senders['keep:1']={'thread':_KeepTh(),'stop':threading.Event(),'sender':_KeepSnd()}
yt.observe=lambda user,force=False: {'status':'offline','room_id':'1','sources':[]}
yt.drive_event('keep:1')
check('first offline blip keeps sender', 'keep:1' in yt.senders and yt.senders['keep:1']['thread'].is_alive())
yt.senders.pop('keep:1', None)
yt.journal.change(lambda d: d['events'].pop('keep:1', None))

# disk: DATA_DIR should not contain video files
vids=[p for p in (TMP/'data').rglob('*') if p.suffix.lower() in ('.mp4','.mkv','.flv','.ts') and p.is_file()]
check('TEST 12 no video files in DATA_DIR', not vids, str(vids[:5]))

print('\n%d passed, %d failed'%(passed, failed))
sys.exit(1 if failed else 0)