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