Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Smart mode-governor contract. | |
| The production failures behind these tests: a health-triggered matchgop | |
| stayed latched for a whole event (journal re-applied it to every FFmpeg), | |
| HEVC sources were excluded from copy recovery, a pure data-starved verdict | |
| prescribed CPU-burning transcoding, and a 960x1920 transcode saturated the | |
| host. The governor must copy what YouTube can ingest, transcode only what | |
| copy cannot fix, cap the fallback canvas, and earn its way back to copy. | |
| """ | |
| import os | |
| import pathlib | |
| import sys | |
| import tempfile | |
| import threading | |
| import time | |
| import unittest | |
| from unittest.mock import patch | |
| ROOT = pathlib.Path(__file__).resolve().parent | |
| sys.path.insert(0, str(ROOT)) | |
| from relay import Sender | |
| from youtube_manager import ManagedYouTube | |
| class CopyRecoveryTests(unittest.TestCase): | |
| def make_sender(self, candidate): | |
| sender = Sender('/bin/true', lambda *a: None) | |
| sender.sticky_mode = 'matchgop' | |
| sender.sticky_since = time.monotonic() - 600 | |
| sender._last_light_output = 120.0 | |
| sender.current_candidate = candidate | |
| return sender | |
| def env(self): | |
| return patch.dict(os.environ, { | |
| 'YOUTUBE_SEND_MODE': 'auto', 'YOUTUBE_COPY_RECOVERY_SECONDS': '60', | |
| 'YOUTUBE_COPY_RECOVERY_MAX_FAILS': '2', | |
| }) | |
| def test_hevc_source_may_probe_copy(self): | |
| with self.env(): | |
| sender = self.make_sender({'codec': 'h265', 'label': 'origin · 960x1920 · HEVC'}) | |
| self.assertEqual(sender.send_mode(), 'copy') | |
| self.assertTrue(sender._recovering_copy) | |
| def test_h264_source_may_probe_copy(self): | |
| with self.env(): | |
| sender = self.make_sender({'codec': 'h264', 'label': 'hd · 720x1280 · H264'}) | |
| self.assertEqual(sender.send_mode(), 'copy') | |
| def test_unknown_codec_never_probes_copy(self): | |
| with self.env(): | |
| sender = self.make_sender({'codec': 'vp9', 'label': 'vp9'}) | |
| self.assertEqual(sender.send_mode(), 'matchgop') | |
| def test_operator_locked_mode_never_probes_copy(self): | |
| with self.env(): | |
| sender = self.make_sender({'codec': 'h264', 'label': 'h264'}) | |
| sender.sticky_operator_locked = True | |
| self.assertEqual(sender.send_mode(), 'matchgop') | |
| def test_recent_or_unproven_light_run_does_not_probe(self): | |
| with self.env(): | |
| sender = self.make_sender({'codec': 'h264', 'label': 'h264'}) | |
| sender.sticky_since = time.monotonic() | |
| self.assertEqual(sender.send_mode(), 'matchgop') | |
| sender = self.make_sender({'codec': 'h264', 'label': 'h264'}) | |
| sender._last_light_output = 3.0 | |
| self.assertEqual(sender.send_mode(), 'matchgop') | |
| class CanvasCapTests(unittest.TestCase): | |
| def test_matchgop_canvas_is_capped_at_720_class(self): | |
| # 960x1920 is capped to a 640x1280 canvas, never re-encoded at the | |
| # full 1920 height. Since v2.43 this tall 1:2 picture fills that canvas | |
| # and center-crops the excess height instead of adding pillarbox bars. | |
| sender = Sender('/bin/true', lambda *a: None) | |
| sender.mode_override = 'matchgop' | |
| args = sender.command('out.flv', {'codec': 'h265', 'width': 960, 'height': 1920}) | |
| vf = args[args.index('-vf') + 1] | |
| self.assertIn('crop=640:1280', vf) | |
| self.assertNotIn('pad=640:1280', vf) | |
| self.assertNotIn('1920', vf) | |
| def test_safe_size_mapping_still_applies(self): | |
| # Pre-existing rule: a 640x1280 picture is normalized onto the | |
| # YouTube-healthy 720x1280 canvas (640-wide copy measured Studio-black). | |
| sender = Sender('/bin/true', lambda *a: None) | |
| sender.mode_override = 'matchgop' | |
| args = sender.command('out.flv', {'codec': 'h264', 'width': 640, 'height': 1280}) | |
| vf = args[args.index('-vf') + 1] | |
| self.assertIn('scale=w=720:h=1280', vf) | |
| class FakeThread: | |
| def is_alive(self): | |
| return True | |
| class FakeSender: | |
| def __init__(self, candidate): | |
| self.sticky_mode = None | |
| self.current_candidate = candidate | |
| self.sticky_requests = [] | |
| self.telemetry_key = None | |
| def snapshot(self): | |
| return {'output_seconds': 60.0, 'source_bytes': 500000, 'send_mode': 'copy'} | |
| def send_mode(self): | |
| return 'copy' | |
| def request_sticky(self, mode, user, reason): | |
| self.sticky_requests.append((mode, user, reason)) | |
| self.sticky_mode = mode | |
| return True | |
| class HealthGovernorTests(unittest.TestCase): | |
| def setUp(self): | |
| self.tmp = tempfile.TemporaryDirectory(prefix='smart-health-') | |
| self.addCleanup(self.tmp.cleanup) | |
| self.manager = ManagedYouTube('/bin/true', lambda u: None, lambda *a: None, | |
| lambda: ['alice'], pathlib.Path(self.tmp.name) / 'journal.json') | |
| self.key = 'alice:room' | |
| self.manager.journal.change(lambda d: d['streams'].update({'alice': {'id': 'STREAM'}})) | |
| def run_health(self, status, issues, candidate): | |
| sender = FakeSender(candidate) | |
| self.manager.senders[self.key] = {'thread': FakeThread(), 'sender': sender} | |
| self.manager.remote = lambda resource, rid: { | |
| 'status': {'healthStatus': {'status': status, 'configurationIssues': issues}}} | |
| self.manager.stream_health(self.key, 'alice', {'last_health': 0}) | |
| return sender | |
| def test_pure_starved_does_not_prescribe_transcode(self): | |
| sender = self.run_health('bad', [ | |
| {'type': 'videoIngestionStarved', 'severity': 'error', 'reason': 'Video output low'}], | |
| {'codec': 'h264', 'width': 640, 'height': 1280, 'bitrate': 1200}) | |
| self.assertEqual(sender.sticky_requests, []) | |
| def test_oversized_signalled_copy_may_transcode(self): | |
| sender = self.run_health('bad', [ | |
| {'type': 'videoIngestionStarved', 'severity': 'error', 'reason': 'Video output low'}], | |
| {'codec': 'h264', 'width': 720, 'height': 1440, 'bitrate': 1000}) | |
| self.assertEqual(len(sender.sticky_requests), 1) | |
| def test_gop_fault_still_transcodes(self): | |
| sender = self.run_health('bad', [ | |
| {'type': 'gopSizeLong', 'severity': 'error', 'reason': 'GOP too long'}], | |
| {'codec': 'h264', 'width': 640, 'height': 1280, 'bitrate': 1200}) | |
| self.assertEqual(len(sender.sticky_requests), 1) | |
| class ManagerClearTests(unittest.TestCase): | |
| def test_verified_copy_clears_journaled_mode(self): | |
| tmp = tempfile.TemporaryDirectory(prefix='smart-clear-') | |
| self.addCleanup(tmp.cleanup) | |
| manager = ManagedYouTube('/bin/true', lambda u: None, lambda *a: None, | |
| lambda: ['alice'], pathlib.Path(tmp.name) / 'journal.json') | |
| key = 'alice:room' | |
| event = {'username': 'alice', 'session_key': key, 'room_id': 'room', | |
| 'broadcast_id': 'VIDEO', 'title': 't', 'nickname': 'Alice', | |
| 'phase': 'live', 'youtube_status': 'live', 'closing': False} | |
| manager.journal.change(lambda d: d['events'].update({key: event})) | |
| manager._remember_sender_mode('alice', 'matchgop', 960, 1920) | |
| self.assertEqual(manager.journal.read()['events'][key]['sender_mode'], 'matchgop') | |
| manager._clear_sender_mode('alice') | |
| ev = manager.journal.read()['events'][key] | |
| self.assertEqual(ev['sender_mode'], '') | |
| self.assertNotIn('alice', manager._sender_sticky) | |
| self.assertNotIn('alice', manager._sender_gop) | |
| class _BlockedStdin: | |
| """أنبوب لا يقبل شيئًا أبدًا (FFmpeg لا يبتلع).""" | |
| def __init__(self): | |
| self.r, self.w = os.pipe() | |
| def fileno(self): return self.w | |
| def write(self, view): return 0 | |
| def close(self): | |
| os.close(self.r); os.close(self.w) | |
| class WriteDeadlineRenewalTests(unittest.TestCase): | |
| """أثناء حشوٍ نشط تتجدّد مهلة الكتابة بدل قتل جلسة يطحن مُرمّزها خياطة؛ | |
| بلا حشو تموت فورًا كالسابق.""" | |
| def test_blocked_write_raises_at_once_without_fill(self): | |
| stdin = _BlockedStdin(); self.addCleanup(stdin.close) | |
| started = time.monotonic() | |
| with self.assertRaises(Exception): | |
| Sender._write_deadline(stdin, b'x' * 64, started - 1.0, {'filling': False}) | |
| self.assertLess(time.monotonic() - started, 1.5) | |
| def test_blocked_write_renews_while_filling_then_raises(self): | |
| stdin = _BlockedStdin(); self.addCleanup(stdin.close) | |
| sess = {'filling': True} | |
| timer = threading.Timer(1.2, lambda: sess.update(filling=False)) | |
| started = time.monotonic(); timer.start() | |
| try: | |
| with self.assertRaises(Exception): | |
| Sender._write_deadline(stdin, b'x' * 64, started - 1.0, sess) | |
| finally: | |
| timer.cancel() | |
| elapsed = time.monotonic() - started | |
| self.assertGreater(elapsed, 1.8) # تجدّدت مرة على الأقل بدل الموت الفوري | |
| self.assertLess(elapsed, 8.0) | |
| def test_fill_output_stale_requires_running_fill_and_progress(self): | |
| sess = {'filling': True, 'fill_since': 900.0, 'ff_last': 980.0} | |
| self.assertFalse(Sender._fill_output_stale(sess, 1000.0, 25.0, 10.0)) | |
| sess['ff_last'] = 970.0 | |
| self.assertTrue(Sender._fill_output_stale(sess, 1000.0, 25.0, 10.0)) | |
| self.assertFalse(Sender._fill_output_stale(sess, 1000.0, 25.0, -1)) | |
| sess['filling'] = False | |
| self.assertFalse(Sender._fill_output_stale(sess, 1000.0, 25.0, 10.0)) | |
| def test_fill_write_renewal_requires_fresh_output(self): | |
| now = time.monotonic() | |
| fresh = {'filling': True, 'ff_last': now, 'fill_until': now + 90} | |
| stale = {'filling': True, 'ff_last': now - 60, 'fill_until': now + 90} | |
| expired = {'filling': True, 'ff_last': now, 'fill_until': now - 1} | |
| self.assertTrue(Sender._fill_write_renew_allowed(fresh)) | |
| self.assertFalse(Sender._fill_write_renew_allowed(stale)) | |
| self.assertFalse(Sender._fill_write_renew_allowed(expired)) | |
| def test_fill_output_stale_detects_blocked_emission_and_inflated_progress(self): | |
| # FFmpeg's counter looks fresh, but no fill chunk has completed writing: | |
| # the pipe/filler is blocked, so the fill must not shield the session. | |
| sess = {'filling': True, 'fill_since': 900.0, 'ff_last': 990.0, | |
| 'fill_emit_at': 900.0} | |
| self.assertTrue(Sender._fill_output_stale(sess, 1000.0, 25.0, 10.0)) | |
| # Output seconds racing far ahead of wall time during a fill is a | |
| # timestamp leap (bad splice), not live delivery YouTube will play. | |
| sess = {'filling': True, 'fill_since': 900.0, 'ff_last': 990.0, | |
| 'fill_emit_at': 990.0, 'fill_out_start': 100.0} | |
| self.assertTrue(Sender._fill_output_stale(sess, 1000.0, 25.0, 1000.0)) | |
| sess['fill_out_start'] = 905.0 | |
| self.assertFalse(Sender._fill_output_stale(sess, 1000.0, 25.0, 1000.0)) | |
| def test_fill_write_renewal_requires_fresh_emission(self): | |
| now = time.monotonic() | |
| sess = {'filling': True, 'ff_last': now, 'fill_until': now + 90, | |
| 'fill_emit_at': now - 60} | |
| self.assertFalse(Sender._fill_write_renew_allowed(sess)) | |
| class FormatLockTests(unittest.TestCase): | |
| """مصدر بفاصل إطارات مفتاحية مضطرب (قياس الجسر) لا يُنسخ: يُطبَّع حتى نهاية | |
| جلسة الناشر، لأن يوتيوب يشترط GOP ثابتًا ~2 ث والنسخ يمرّر الاضطراب كما هو.""" | |
| def test_keyframe_instability_locks_auto_to_matchgop(self): | |
| sender = Sender('/bin/true', lambda *a: None) | |
| sender.current_candidate = {'codec': 'h264', 'label': 'hd · 720x1440 · H264'} | |
| with patch.dict(os.environ, {'YOUTUBE_SEND_MODE': 'auto'}): | |
| self.assertEqual(sender.send_mode(), 'copy') | |
| sender.format_unstable = True | |
| self.assertEqual(sender.send_mode(), 'matchgop') | |
| def test_bridge_keyframe_change_sets_the_lock(self): | |
| sender = Sender('/bin/true', lambda *a: None) | |
| with patch('relay.log_event', lambda *a, **k: None): | |
| sender._bridge_event('alice', 'KEYFRAME_CHANGE', | |
| {'old_interval_ms': 1900.0, 'new_interval_ms': 400.0}) | |
| self.assertTrue(sender.format_unstable) | |
| def test_lock_blocks_copy_recovery_probe(self): | |
| sender = Sender('/bin/true', lambda *a: None) | |
| sender.sticky_mode = 'matchgop' | |
| sender.sticky_since = time.monotonic() - 600 | |
| sender._last_light_output = 120.0 | |
| sender.current_candidate = {'codec': 'h264', 'label': 'hd · 720x1280 · H264'} | |
| sender.format_unstable = True | |
| with patch.dict(os.environ, {'YOUTUBE_SEND_MODE': 'auto', | |
| 'YOUTUBE_COPY_RECOVERY_SECONDS': '60'}): | |
| self.assertEqual(sender.send_mode(), 'matchgop') | |
| def test_operator_forced_copy_is_respected(self): | |
| sender = Sender('/bin/true', lambda *a: None) | |
| sender.mode_override = 'copy' | |
| sender.format_unstable = True | |
| sender.current_candidate = {'codec': 'h264', 'label': 'hd · 720x1440 · H264'} | |
| with patch.dict(os.environ, {'YOUTUBE_SEND_MODE': 'auto'}): | |
| self.assertEqual(sender.send_mode(), 'copy') | |
| class _ExitSender: | |
| """أصغر مرسل يراه _note_early_exit عند نهاية جلسة.""" | |
| def __init__(self, unstable=False, mode='matchgop', out=0.0): | |
| self.format_unstable = unstable | |
| self.last_exit_code = 187 | |
| self.last_failure_side = 'output' | |
| self.last_source_gap = 0.0 | |
| self._snap = {'output_seconds': out, 'rtmp_phase': 'accepted', | |
| 'source_bytes': 4096, 'send_mode': mode} | |
| def snapshot(self): return dict(self._snap) | |
| def send_mode(self): return self._snap['send_mode'] | |
| def request_sticky(self, *a, **k): return False | |
| class ManagerFormatLockTests(unittest.TestCase): | |
| """قفل التطبيع محفوظ في السجلّ على مستوى (حساب، غرفة): إعادة بناء المرسل | |
| لا تمحوه، وجلسة COPY مستقرة على نفس الغرفة وحدها تفكّه.""" | |
| def make_manager(self): | |
| tmp = tempfile.TemporaryDirectory(prefix='fmt-lock-') | |
| self.addCleanup(tmp.cleanup) | |
| manager = ManagedYouTube('/bin/true', lambda u: None, lambda *a: None, | |
| lambda: ['alice'], pathlib.Path(tmp.name) / 'journal.json') | |
| manager.journal.change(lambda d: d['events'].update( | |
| {'alice:R1': {'username': 'alice', 'room_id': 'R1'}})) | |
| return manager | |
| def test_instability_is_journaled_for_the_room(self): | |
| manager = self.make_manager() | |
| manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0) | |
| self.assertEqual(manager._format_lock_room('alice'), 'R1') | |
| def test_fresh_sender_is_armed_from_the_journal(self): | |
| manager = self.make_manager() | |
| manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0) | |
| sender = Sender('/bin/true', lambda *a: None) | |
| self.assertFalse(sender.format_unstable) | |
| self.assertTrue(manager._arm_format_lock('alice', sender, 'R1')) | |
| self.assertTrue(sender.format_unstable) | |
| def test_lock_does_not_leak_into_a_new_room(self): | |
| manager = self.make_manager() | |
| manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0) | |
| sender = Sender('/bin/true', lambda *a: None) | |
| self.assertFalse(manager._arm_format_lock('alice', sender, 'R2')) | |
| self.assertFalse(sender.format_unstable) | |
| def test_verified_copy_session_releases_the_lock(self): | |
| manager = self.make_manager() | |
| manager._note_early_exit('alice', _ExitSender(unstable=True), 11.0) | |
| self.assertEqual(manager._format_lock_room('alice'), 'R1') | |
| manager._note_early_exit('alice', _ExitSender(unstable=False, mode='copy', out=60.0), 65.0) | |
| self.assertEqual(manager._format_lock_room('alice'), '') | |
| class WriteDeadlineTests(unittest.TestCase): | |
| """مهلة كتابة stdin أثناء حشو/تبديل تُقاس على تقدّم FFmpeg لا على صمت المصدر.""" | |
| def test_normal_base_is_last_source_byte(self): | |
| self.assertEqual(Sender._deadline_base({'last': 100.0, 'ff_last': 150.0}), 100.0) | |
| def test_filling_base_follows_ffmpeg_progress(self): | |
| self.assertEqual(Sender._deadline_base( | |
| {'last': 100.0, 'ff_last': 150.0, 'filling': True}), 150.0) | |
| def test_filling_base_counts_successful_fill_writes(self): | |
| self.assertEqual(Sender._deadline_base( | |
| {'last': 100.0, 'ff_last': 120.0, 'fill_emit_at': 150.0, 'filling': True}), 150.0) | |
| def test_recent_fill_seam_still_credits_last_swallow(self): | |
| now = time.monotonic() | |
| sess = {'last': now - 30, 'ff_last': now - 25, | |
| 'fill_emit_at': now - 5, 'filling': False} | |
| self.assertEqual(Sender._deadline_base(sess), now - 5) | |
| def test_stale_fill_seam_gets_no_credit(self): | |
| now = time.monotonic() | |
| sess = {'last': now - 30, 'ff_last': now - 40, | |
| 'fill_emit_at': now - 60, 'filling': False} | |
| self.assertEqual(Sender._deadline_base(sess), now - 30) | |
| def test_fill_credit_does_not_apply_while_swapping(self): | |
| self.assertEqual(Sender._deadline_base( | |
| {'last': 100.0, 'ff_last': 90.0, 'fill_emit_at': 150.0, 'swapping': True}), 100.0) | |
| def test_swapping_base_never_moves_backwards(self): | |
| self.assertEqual(Sender._deadline_base( | |
| {'last': 200.0, 'ff_last': 150.0, 'swapping': True}), 200.0) | |
| if __name__ == '__main__': | |
| unittest.main(argv=[sys.argv[0]], verbosity=2) | |