| |
| """Regression for live source handover; no real TikTok/YouTube requests. |
| |
| Starting a replacement pump while swap_source=True made the new thread exit before |
| reading a single byte. FFmpeg was kept alive but YouTube received no replacement media. |
| """ |
| import io |
| import http.server |
| import json |
| import shutil |
| import subprocess |
| import tempfile |
| import os |
| import pathlib |
| import sys |
| import threading |
| import time |
| import unittest |
| from unittest.mock import Mock, patch |
|
|
| ROOT = pathlib.Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else pathlib.Path(__file__).parent |
| sys.path.insert(0, str(ROOT)) |
| os.environ.setdefault('DATA_DIR', tempfile.mkdtemp(prefix='handover-test-data-')) |
| from relay import Sender |
| from youtube_manager import ManagedYouTube |
|
|
|
|
| class InlineThread: |
| """Deterministic scheduler: the new pump runs immediately when start() is called.""" |
| def __init__(self, target, args=(), daemon=False): |
| self.target = target |
| self.args = args |
| self.daemon = daemon |
| self.started = False |
|
|
| def start(self): |
| self.started = True |
| self.target(*self.args) |
|
|
| def join(self, timeout=None): |
| pass |
|
|
| def is_alive(self): |
| return False |
|
|
|
|
| class Response: |
| def __init__(self, chunks=(b'replacement media',)): |
| self.chunks = chunks |
| self.closed = False |
| self.reads = 0 |
|
|
| def raise_for_status(self): |
| pass |
|
|
| def iter_content(self, chunk_size=None): |
| for chunk in self.chunks: |
| self.reads += 1 |
| yield chunk |
|
|
| def close(self): |
| self.closed = True |
|
|
|
|
| class HandoverTests(unittest.TestCase): |
| def setUp(self): |
| self.env = patch.dict(os.environ, {'SOURCE_STALL_TIMEOUT': '0'}) |
| self.env.start() |
| self.addCleanup(self.env.stop) |
| self.sender = Sender('/bin/true', lambda *a: None) |
| self.p = Mock() |
| self.p.poll.return_value = None |
| self.p.stdin = io.BytesIO() |
| self.old_pump = Mock() |
| self.old_pump.is_alive.return_value = False |
| self.old_response = Response() |
| self.new_response = Response() |
| self.stop = threading.Event() |
| self.sess = {'last': time.monotonic() - 4, 'max_seconds': 180, |
| 'force_stop': False, 'timeouts': 0, 'error': '', 'ended': True} |
| self.bridge = Mock() |
| self.bridge.feed.side_effect = lambda chunk: chunk |
| self.refresh = Mock(return_value={'url': 'https://example.invalid/source.flv', 'codec': 'h264', 'label': 'hd'}) |
|
|
| def attach(self, refresh=None): |
| with patch('relay.requests.get', return_value=self.new_response), patch('relay.threading.Thread', InlineThread): |
| return self.sender._try_reattach('alice', self.p, self.bridge, self.sess, |
| self.stop, refresh or self.refresh, 10, 3, self.old_response, self.old_pump) |
|
|
| def test_new_pump_reads_media_even_when_scheduled_immediately(self): |
| attached = self.attach() |
| self.assertIsNotNone(attached) |
| self.assertEqual(self.new_response.reads, 1) |
| self.assertEqual(self.p.stdin.getvalue(), b'replacement media') |
| self.assertEqual(self.sender.snapshot()['source_bytes'], len(b'replacement media')) |
| self.assertFalse(self.sess['swap_source']) |
|
|
| def test_old_pump_must_stop_before_bridge_is_reused(self): |
| self.old_pump.is_alive.return_value = True |
| self.assertIsNone(self.attach()) |
| self.bridge.reconnect.assert_not_called() |
| self.refresh.assert_not_called() |
| self.assertEqual(self.new_response.reads, 0) |
|
|
| def test_stop_during_resolution_does_not_resurrect_the_reader(self): |
| def stop_then_refresh(): |
| self.stop.set() |
| return self.refresh() |
| self.assertIsNone(self.attach(stop_then_refresh)) |
| self.bridge.reconnect.assert_not_called() |
| self.assertEqual(self.new_response.reads, 0) |
|
|
| def test_refresh_does_not_fabricate_source_heartbeat(self): |
| self.new_response = Response(chunks=()) |
| previous = self.sess['last'] |
| self.assertIsNotNone(self.attach()) |
| self.assertEqual(self.sess['last'], previous) |
| self.assertEqual(self.sender.snapshot()['source_bytes'], 0) |
|
|
| def test_opened_source_has_bounded_first_byte_grace(self): |
| previous = self.sess['last'] |
| now = time.monotonic() |
| self.new_response = Response(chunks=()) |
| self.attach() |
| self.assertGreater(self.sess['reattach_not_before'], now) |
| self.assertLessEqual(self.sess['reattach_not_before'], time.monotonic() + 5) |
| self.assertEqual(self.sess['last'], previous) |
|
|
| def test_cancelled_reader_cannot_deliver_a_late_chunk(self): |
| cancel = threading.Event() |
| response = Response() |
| def late_chunk(_size=None): |
| cancel.set() |
| yield b'old reader late packet' |
| response.iter_content = late_chunk |
| with patch('relay.threading.Thread', InlineThread): |
| self.sender._pump_source(response, self.p, self.bridge, self.stop, |
| self.sess, 10, 3, cancel) |
| self.assertEqual(self.p.stdin.getvalue(), b'') |
| self.assertEqual(self.sender.snapshot()['source_bytes'], 0) |
|
|
| def test_no_source_is_a_failed_handover_not_a_fake_success(self): |
| self.refresh.return_value = None |
| self.assertIsNone(self.attach()) |
| self.bridge.reconnect.assert_not_called() |
|
|
|
|
| class CanvasTests(unittest.TestCase): |
| def test_placeholder_16x16_does_not_pin_the_live_canvas(self): |
| sender = Sender('/bin/true', lambda *a: None) |
| self.assertEqual(sender.lock_gold({'width': 16, 'height': 16}), (0, 0)) |
| self.assertEqual(sender.lock_gold({'width': 720, 'height': 1440}), (720, 1280)) |
|
|
| def test_unknown_or_placeholder_matchgop_uses_safe_constant_canvas(self): |
| for candidate in ({}, {'width': 16, 'height': 16}): |
| sender = Sender('/bin/true', lambda *a: None) |
| sender.mode_override = 'matchgop' |
| args = sender.command('test.flv', candidate) |
| vf = args[args.index('-vf') + 1] |
| self.assertIn('scale=w=720:h=1280:', vf) |
| |
| sender.lock_gold({'width': 960, 'height': 1920}) |
| self.assertEqual((sender._gop_w, sender._gop_h), (720, 1280)) |
|
|
| def test_normal_and_real_1080_source_dimensions_remain_supported(self): |
| for size in ((160, 120), (320, 240), (1080, 1920)): |
| sender = Sender('/bin/true', lambda *a: None) |
| self.assertEqual(sender.lock_gold({'width': size[0], 'height': size[1]}), size) |
|
|
|
|
| class CachedSourceTests(unittest.TestCase): |
| def setUp(self): |
| self.tmp = tempfile.TemporaryDirectory() |
| self.addCleanup(self.tmp.cleanup) |
| self.manager = ManagedYouTube('/bin/true', lambda u, force=False: None, |
| lambda *a: None, lambda: ['alice'], pathlib.Path(self.tmp.name) / 'journal.json') |
| self.manager.video_sources = lambda sources: list(sources) |
| self.hd = {'url': 'https://example.invalid/hd.flv', 'media': 'flv', 'codec': 'h264', |
| 'quality_name': 'hd', 'width': 720, 'height': 1280, 'label': 'hd'} |
| self.sd = dict(self.hd, url='https://example.invalid/sd.flv', quality_name='sd', |
| width=480, height=854, label='sd') |
| self.obs = {'status': 'live', 'room_id': 'same-room', 'sources': [self.hd, self.sd]} |
| self.manager._observe = Mock(return_value=self.obs) |
|
|
| def test_recent_compatible_source_is_reused_without_blocking_metadata_lookup(self): |
| self.manager._held['alice'] = (time.monotonic(), self.obs) |
| chosen = self.manager._refresh_flv('alice', self.hd) |
| self.assertEqual(chosen['url'], self.sd['url']) |
| self.manager._observe.assert_not_called() |
|
|
| def test_stale_source_snapshot_requires_fresh_lookup(self): |
| self.manager._held['alice'] = (time.monotonic() - 60, self.obs) |
| self.assertEqual(self.manager._refresh_flv('alice', self.hd)['url'], self.sd['url']) |
| self.manager._observe.assert_called_once_with('alice', force=True, bypass_throttle=True, allow_hold=False) |
|
|
| def test_cache_with_wrong_codec_cannot_be_spliced_into_open_publisher(self): |
| self.manager._held['alice'] = (time.monotonic(), dict(self.obs, sources=[dict(self.sd, codec='h265')])) |
| chosen = self.manager._refresh_flv('alice', self.hd) |
| self.assertEqual(chosen['codec'], 'h264') |
| self.manager._observe.assert_called_once() |
|
|
|
|
| @unittest.skipUnless(shutil.which('ffmpeg') and shutil.which('ffprobe'), 'FFmpeg required') |
| class RealHTTPContinuityTests(unittest.TestCase): |
| def run_continuity(self, mode): |
| with tempfile.TemporaryDirectory(prefix='source-handover-media-') as tmp: |
| root = pathlib.Path(tmp) |
| source = root / 'generated.flv' |
| output = root / 'output.flv' |
| subprocess.run([shutil.which('ffmpeg'), '-v', 'error', '-y', '-f', 'lavfi', |
| '-i', 'testsrc2=size=160x120:rate=20', '-f', 'lavfi', '-i', 'sine=sample_rate=48000', |
| '-t', '2', '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', |
| '-g', '20', '-c:a', 'aac', '-f', 'flv', str(source)], check=True, capture_output=True) |
| media = source.read_bytes() |
| calls = [] |
| class Handler(http.server.BaseHTTPRequestHandler): |
| def log_message(self, *a): |
| pass |
| def do_GET(self): |
| calls.append(time.monotonic()) |
| self.send_response(200) |
| self.send_header('Content-Type', 'video/x-flv') |
| self.send_header('Content-Length', str(len(media))) |
| self.end_headers() |
| |
| |
| if len(calls) > 1: |
| time.sleep(1) |
| try: |
| self.wfile.write(media) |
| self.wfile.flush() |
| except (OSError, BrokenPipeError): |
| pass |
| server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) |
| server.daemon_threads = True |
| server_thread = threading.Thread(target=server.serve_forever, daemon=True) |
| server_thread.start() |
| stop = threading.Event() |
| errors = [] |
| candidate = {'url': 'http://127.0.0.1:%d/source.flv' % server.server_port, |
| 'media': 'flv', 'codec': 'h264', 'label': 'fixture', 'width': 160, 'height': 120} |
| sender = Sender(shutil.which('ffmpeg'), lambda *a: None) |
| sender.mode_override = mode |
| spawned = [] |
| from relay import subprocess as relay_subprocess |
| real_popen = relay_subprocess.Popen |
| def track_process(*a, **kw): |
| child = real_popen(*a, **kw) |
| spawned.append(child) |
| return child |
| def send(): |
| try: |
| sender.send('fixture', candidate, str(output), stop, refresh=lambda: dict(candidate)) |
| except Exception as exc: |
| errors.append(exc) |
| env = {'SOURCE_STALL_TIMEOUT': '3', 'YOUTUBE_REATTACH_GRACE_SECONDS': '2', |
| 'YOUTUBE_SOURCE_READ_TIMEOUT': '5', 'YOUTUBE_PUMP_JOIN_SECONDS': '2', |
| 'YOUTUBE_SEND_MODE': mode, 'YOUTUBE_VIDEO_MODE': 'original', |
| 'YOUTUBE_AUTO_TRANSCODE': '0', 'YOUTUBE_X264_PRESET': 'ultrafast'} |
| thread = None |
| try: |
| with patch.dict(os.environ, env), patch('relay.subprocess.Popen', side_effect=track_process): |
| thread = threading.Thread(target=send, daemon=True) |
| thread.start() |
| time.sleep(7.5) |
| alive = thread.is_alive() and sender.alive() |
| snapshot = sender.snapshot() |
| stop.set() |
| thread.join(6) |
| self.assertTrue(alive, 'Publisher ended instead of reusing the delayed replacement source: %r' % errors) |
| self.assertFalse(errors, repr(errors)) |
| self.assertEqual(len(spawned), 1, 'Source reconnect must keep the same FFmpeg process') |
| self.assertGreaterEqual(len(calls), 3) |
| self.assertGreater(snapshot.get('output_seconds', 0), 4) |
| self.assertFalse(thread.is_alive(), 'Sender did not stop cleanly') |
| probe = subprocess.run([shutil.which('ffprobe'), '-v', 'error', '-show_entries', |
| 'format=duration', '-of', 'json', str(output)], check=True, capture_output=True, text=True) |
| self.assertGreater(float(json.loads(probe.stdout)['format']['duration']), 4) |
| finally: |
| stop.set() |
| sender.terminate() |
| if thread: |
| thread.join(5) |
| server.shutdown() |
| server.server_close() |
|
|
| def test_eof_and_delayed_replacements_keep_same_copy_publisher(self): |
| self.run_continuity('copy') |
|
|
| def test_eof_and_delayed_replacements_keep_same_matchgop_publisher(self): |
| self.run_continuity('matchgop') |
|
|
|
|
| if __name__ == '__main__': |
| unittest.main(argv=[sys.argv[0]], verbosity=2) |
|
|
