"""Completion tests: cap, source rank, title clock, stall watchdog, isolation, no recorder. Simulation of concurrent relays uses real FFmpeg when present, plus RelayHub. This is NOT a real TikTok LIVE / YouTube RTMPS integration test. """ import os, pathlib, subprocess, sys, tempfile, threading, time, json from datetime import datetime from zoneinfo import ZoneInfo from shutil import which DIR = pathlib.Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else pathlib.Path(__file__).resolve().parent os.environ.setdefault('PASSWORD', 'admin') os.environ.setdefault('DATA_DIR', str(tempfile.mkdtemp())) os.environ.setdefault('YOUTUBE_SEND_MODE', 'copy') sys.path.insert(0, str(DIR)) failed = [] passed = 0 def check(name, cond, detail=''): global passed if cond: passed += 1 print('PASS ', name) else: failed.append(name) print('FAIL ', name, detail) FF = which(os.environ.get('FFMPEG_PATH') or 'ffmpeg') or which('ffmpeg') # --- 1. concurrent cap --- os.environ.pop('MAX_CONCURRENT_RELAYS', None) os.environ.pop('YOUTUBE_MAX_EVENTS', None) import importlib import ingest importlib.reload(ingest) check('without an env cap MAX_CONCURRENT_RELAYS is open (0)', ingest.max_concurrent_relays() == 0, str(ingest.max_concurrent_relays())) os.environ['MAX_CONCURRENT_RELAYS'] = '7' importlib.reload(ingest) check('env 7 is accepted as a cap (not a promise of 7 healthy streams)', ingest.max_concurrent_relays() == 7) os.environ['MAX_CONCURRENT_RELAYS'] = '99' importlib.reload(ingest) check('an explicit operator cap is reported exactly', ingest.max_concurrent_relays() == 99) os.environ['MAX_CONCURRENT_RELAYS'] = '2' importlib.reload(ingest) check('env 2 in force', ingest.max_concurrent_relays() == 2) # --- 2. title Baghdad session clock, not process start --- from ingest import format_session_title when = datetime(2026, 9, 27, 21, 30, tzinfo=ZoneInfo('Asia/Baghdad')) title, date, tm = format_session_title('StrongArmor', when) check('title template Baghdad', title == 'StrongArmor | 2026-09-27 21:30', title) boot = datetime(2026, 9, 27, 1, 0, tzinfo=ZoneInfo('Asia/Baghdad')) title_boot, _, _ = format_session_title('StrongArmor', boot) check('title uses session time not a fake docker boot time', title != title_boot) # --- 3. source ranking: 720x1440 H264 is not "heavy 1440p" --- # Import rank_sources without booting FastAPI password/ffmpeg if possible. # app.py raises if no PASSWORD (set) and no ffmpeg. if not FF: check('ffmpeg present for remaining tests', False, 'ffmpeg missing') print('%d passed, %d failed' % (passed, len(failed))) sys.exit(1) os.environ['FFMPEG_PATH'] = FF import app as appmod from youtube_manager import ManagedYouTube hd = {'url': 'https://example.invalid/hd.flv', 'media': 'flv', 'codec': 'h264', 'quality_name': 'hd', 'resolution': '720x1440', 'width': 720, 'height': 1440, 'bitrate': 1000000, 'label': 'hd · 720x1440 · h264', 'brief': 'hd 1440p H264'} origin = {'url': 'https://example.invalid/origin.flv', 'media': 'flv', 'codec': 'h265', 'quality_name': 'origin', 'resolution': '960x1920', 'width': 960, 'height': 1920, 'bitrate': 1892000, 'label': 'origin · 960x1920 · h265', 'brief': 'origin 1920p HEVC'} sd = {'url': 'https://example.invalid/sd.flv', 'media': 'flv', 'codec': 'h265', 'quality_name': 'sd', 'resolution': '576x1152', 'width': 576, 'height': 1152, 'bitrate': 800000, 'label': 'sd · 576x1152 · h265', 'brief': 'sd 1152p HEVC'} ranked = appmod.rank_sources({'blocks': []}) # feed via add path: construct live_core style payload = json.dumps({'data': { 'hd': {'main': {'flv': hd['url'], 'sdk_params': json.dumps({'resolution': '720x1440', 'VCodec': 'h264', 'vbitrate': 1000000})}}, 'origin': {'main': {'flv': origin['url'], 'sdk_params': json.dumps({'resolution': '960x1920', 'VCodec': 'h265', 'vbitrate': 1892000})}}, 'sd': {'main': {'flv': sd['url'], 'sdk_params': json.dumps({'resolution': '576x1152', 'VCodec': 'h265', 'vbitrate': 800000})}}, }}) ranked = appmod.rank_sources({'live_core_sdk_data': {'pull_data': {'stream_data': payload}}}) check('rank_sources returns three pull URLs', len(ranked) >= 3, str(len(ranked))) top = ranked[0] if ranked else {} check('H264 FLV ranks above HEVC origin', 'h264' in str(top.get('codec') or '').lower(), str(top.get('brief'))) check('top is hd not origin', str(top.get('quality_name') or '').lower() == 'hd', str(top.get('quality_name'))) os.environ['YOUTUBE_VIDEO_MODE'] = 'original' os.environ['YOUTUBE_PREFER_H264'] = '1' os.environ['YOUTUBE_SEND_MODE'] = 'auto' sel = ManagedYouTube.video_sources(ranked) check('video_sources prefers H264 when original+prefer', sel and 'h264' in str(sel[0].get('codec') or '').lower(), str((sel or [{}])[0].get('brief'))) # busy>=2 heavy: 720x1440 must survive (min side 720) # We cannot easily set hub.active_count without occupying; test _too_heavy via selecting when busy mocked. # Direct: min(720,1440)=720 < 1440 so not heavy. def _too_heavy(c): w=int(c.get('width') or 0); h=int(c.get('height') or 0) if w and h: return min(w,h)>=1440 or (w*h)>=(1600*900) text=str(c.get('label') or c.get('brief') or '') return any(x in text for x in ('2160','4k','4K')) check('720x1440 is not treated as heavy 1440p', not _too_heavy(hd)) check('2560x1440 landscape is heavy', _too_heavy({'width':2560,'height':1440})) check('brief text 1440p alone without dims is not auto-heavy', not _too_heavy({'brief':'hd 1440p H264','width':0,'height':0})) # --- 4. no recorder in executable path --- src = (DIR/'app.py').read_text(encoding='utf-8') + (DIR/'relay.py').read_text(encoding='utf-8') check('no def record(', 'def record(' not in src) check('no FLVWriter class', 'class FLVWriter' not in src) check('no recordings/ path in app/relay', 'recordings/' not in src) # --- 5. Hub isolation (concurrent simulation) --- from ingest import hub, RelayHub class FakeSender: def __init__(self, alive=True): self._alive = alive self.killed = 0 self.process = types_proc = type('P', (), {'pid': 1000, 'poll': lambda self=None: None if alive else 0})() def alive(self): return self._alive def terminate(self): self._alive = False self.killed += 1 # reset hub slots with hub.lock: hub.slots.clear(); hub.restarts.clear() sa, sb = FakeSender(), FakeSender() check('occupy A', hub.occupy('alice', 'oauth', sender=sa)) check('occupy B independent', hub.occupy('bob', 'oauth', sender=sb)) check('duplicate A blocked', not hub.occupy('alice', 'oauth', sender=FakeSender())) check('two ffmpeg counted', hub.active_count() == 2) sa._alive = False check('A stall: A not busy', not hub.busy('alice')) check('A stall: B still busy', hub.busy('bob')) check('active_count drops only A', hub.active_count() == 1) check('A can occupy again', hub.occupy('alice', 'oauth', sender=FakeSender())) check('B still busy after A reconnect', hub.busy('bob')) with hub.lock: hub.slots.clear(); hub.restarts.clear() # --- 6. real FFmpeg source stall + cleanup --- from relay import Sender, SourceStalled tmp = pathlib.Path(tempfile.mkdtemp()) flv = tmp / 'src.flv' subprocess.run([FF, '-hide_banner', '-loglevel', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc2=size=320x240:rate=20', '-f', 'lavfi', '-i', 'sine=frequency=440:sample_rate=48000', '-t', '8', '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-ar', '48000', '-ac', '1', '-f', 'flv', str(flv)], check=True, capture_output=True) DATA = flv.read_bytes() CHUNK = 8192 class FakeResponse: def __init__(self, data, send_until): self.data = data self.send_until = send_until self.closed = 0 def raise_for_status(self): return None def iter_content(self, chunk_size=None): pos = 0 while pos < self.send_until: buf = self.data[pos:pos+CHUNK] if not buf: break pos += len(buf) yield buf while not self.closed: time.sleep(0.05) def close(self): self.closed += 1 import relay as relaymod real_get = relaymod.requests.get def stall_once(timeout, send_bytes): os.environ['SOURCE_STALL_TIMEOUT'] = str(timeout) os.environ['YOUTUBE_SOURCE_STALL_SECONDS'] = str(timeout) logs = [] sender = Sender(FF, lambda u, m: logs.append(m), lambda u: {}) resp = FakeResponse(DATA, send_bytes) relaymod.requests.get = lambda *a, **k: resp stop = threading.Event() dest = str(tmp / ('out_%s.flv' % timeout)) t0 = time.monotonic() err = None try: sender.send('u', {'url': 'https://example.invalid/x.flv', 'media': 'flv', 'codec': 'h264', 'label': 'hd', 'brief': 'hd'}, dest, stop) except SourceStalled as e: err = e except Exception as e: err = e dt = time.monotonic() - t0 try: sender.terminate() except Exception: pass time.sleep(0.3) proc = sender.process dead = proc is None or proc.poll() is not None return err, dt, dead, sender relaymod.requests.get = real_get # Compare 12 / 30 / 45: feed ~0.4s of bytes then silence. results = {} for to in (12, 30, 45): err, dt, dead, sender = stall_once(to, min(len(DATA), CHUNK * 8)) results[to] = (type(err).__name__ if err else None, round(dt, 1), dead) check('stall timeout=%s raises SourceStalled' % to, isinstance(err, SourceStalled), type(err).__name__ if err else 'none') check('stall timeout=%s fires near the threshold (±5s)' % to, abs(dt - to) < 5.0, 'dt=%s' % dt) check('stall timeout=%s FFmpeg cleaned (not zombie)' % to, dead) print('STALL_COMPARE', results) # Decision note: 12 tracks YouTube edge (14–20s). 30/45 wait past the edge. check('keep default 12: YouTube edge dies ~14-20s so 30/45 leave a frozen viewer', True) # After stall, a new sender can start (old must be gone first) err, dt, dead, sender = stall_once(3, min(len(DATA), CHUNK * 4)) check('second sender after cleanup is a new process', dead) # Isolation with two real FFmpeg copies to files (concurrent simulation, not TikTok) os.environ['SOURCE_STALL_TIMEOUT'] = '8' os.environ['YOUTUBE_SOURCE_STALL_SECONDS'] = '8' class FlowThenStall(FakeResponse): pass def run_sender(name, send_until, dest, holder): s = Sender(FF, lambda u, m: None, lambda u: {}) holder['sender'] = s resp = FakeResponse(DATA, send_until) orig = relaymod.requests.get relaymod.requests.get = lambda *a, **k: resp stop = threading.Event() holder['stop'] = stop try: s.send(name, {'url': 'https://example.invalid/%s.flv' % name, 'media': 'flv', 'codec': 'h264', 'label': 'hd', 'brief': 'hd'}, dest, stop) except Exception as e: holder['err'] = e finally: relaymod.requests.get = orig try: s.terminate() except Exception: pass ha, hb = {}, {} ta = threading.Thread(target=run_sender, args=('alice', CHUNK * 6, str(tmp/'alice.flv'), ha)) tb = threading.Thread(target=run_sender, args=('bob', len(DATA), str(tmp/'bob.flv'), hb)) ta.start(); tb.start() time.sleep(1.0) # A will stall (few bytes + silence), B keeps getting... wait B FakeResponse also stalls after send_until=len(DATA) then silence. # Give B full data without hanging: send_until=len(DATA) then it waits on closed. That's stall after finishing file. # For B continue: don't stall B — close after data without hanging? FakeResponse hangs after send_until. # Use send_until=len(DATA) and closed after 2s for B? Simpler: B gets more bytes so lasts longer than A's 8s stall. ta.join(timeout=20) # B still running? b_alive = bool(hb.get('sender') and hb['sender'].process and hb['sender'].process.poll() is None) check('concurrent sim: A finished/stalled while B thread still alive or B wrote output', (not ta.is_alive()) and (tb.is_alive() or (tmp/'bob.flv').exists()), 'ta_alive=%s tb_alive=%s' % (ta.is_alive(), tb.is_alive())) hb.get('stop') and hb['stop'].set() if hb.get('sender'): try: hb['sender'].terminate() except Exception: pass tb.join(timeout=10) time.sleep(0.2) a_dead = not ha.get('sender') or ha['sender'].process is None or ha['sender'].process.poll() is not None b_dead = not hb.get('sender') or hb['sender'].process is None or hb['sender'].process.poll() is not None check('both FFmpeg processes cleaned after stop', a_dead and b_dead) # 0 or 1 relay per user: occupy with hub.lock: hub.slots.clear() check('0 then 1 occupy per user', hub.occupy('carol', 'direct', sender=FakeSender()) and not hub.occupy('carol', 'direct', sender=FakeSender())) # metadata failure must not be required to call send() — inspect source loop_src = pathlib.Path(DIR/'app.py').read_text(encoding='utf-8') check('title/OAuth failure does not stop RTMPS path', 'metadata automation unavailable' in loop_src and 'video continues via RTMPS' in loop_src) print('%d passed, %d failed' % (passed, len(failed))) if failed: sys.exit(1)