Blame for test_smart_mode.py

#!/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)