cap MAX_CONCURRENT_RELAYS=2 (env, not a promise); health youtube fields; relay-only verify

test_completion.py

@@ -0,0 +1,297 @@
+"""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('default MAX_CONCURRENT_RELAYS is 2 (cap, not a guarantee)', ingest.max_concurrent_relays() == 2, 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('typo 99 is clamped to 8', ingest.max_concurrent_relays() == 8)
+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)