Masters

Running

Masters / test_completion.py

Exact copy of Reoo88/On @ 826bfe0 (relay-v2.32-gap-filler)

f847423 verified 26 days ago

13.2 kB

"""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)

	"""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 (wh)>=(1600900)
	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)