import tempfile import unittest from pathlib import Path from controller.config import ControllerConfig, StorageConfig from controller.playback import OperationInProgressError, PlaybackReducer from controller.store import ControllerStore class Clock: def __init__(self, value): self.value = value def __call__(self): return self.value class PlaybackReducerTests(unittest.TestCase): def setUp(self): self.tempdir = tempfile.TemporaryDirectory() base = Path(self.tempdir.name) self.wall = Clock(1_700_000_000.0) self.monotonic = Clock(100.0) self.config = ControllerConfig( storage=StorageConfig( media_base=base / "media", data_base=base / "data", retention_seconds=100, uncertain_review_after_seconds=30, ) ) self.store = ControllerStore(self.config.storage, now=self.wall).open() self.reducer = PlaybackReducer( self.config, self.store, monotonic=self.monotonic, wall_clock=self.wall, operation_timeout_seconds=5, ) self._ready(1, "one") self._ready(2, "two") def tearDown(self): self.store.close() self.tempdir.cleanup() def _ready(self, slot, suffix): reservation = self.store.reserve_download( f"https://example.test/{suffix}", None, None, job_id=f"job-{suffix}" ) self.assertEqual(reservation.slot, slot) self.store.mark_download_running( reservation.job_id, reservation.slot, reservation.generation ) self.store.complete_download( reservation.job_id, reservation.slot, reservation.generation, title=suffix.title(), items=((suffix.title(), f"{suffix}.mp4"),), ) def test_play_intent_is_not_optimistically_reported_playing(self): operation = self.reducer.begin_operation( "play", slot=1, expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u", ) self.assertEqual(operation.state, "awaiting_player") self.assertEqual(self.store.player_observation().phase, "unknown") self.assertEqual(self.reducer.projected_slot_state(1), "ready") with self.assertRaises(OperationInProgressError): self.reducer.begin_operation("stop") def test_matching_player_playing_confirms_play_and_protects_slot(self): self.reducer.begin_operation( "play", slot=1, expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u", ) effects = self.reducer.handle_status( { "state": "playing", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", } ) self.assertIn("operation_confirmed", [effect.kind for effect in effects]) self.assertIsNone(self.reducer.pending) self.assertEqual(self.store.player_observation().phase, "playing") self.assertEqual(self.reducer.projected_slot_state(1), "playing") guards = self.store.list_live_guards() self.assertEqual([(guard.slot, guard.kind) for guard in guards], [(1, "player_active")]) def test_reconnect_online_idle_with_path_is_not_terminal(self): self.reducer.handle_status( { "state": "playing", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", } ) effects = self.reducer.handle_status( { "state": "idle", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", "message": "player online", } ) self.assertEqual(effects[0].kind, "player_reconnected_active") self.assertEqual(self.store.player_observation().phase, "playing") self.assertEqual(self.store.list_live_guards()[0].slot, 1) def test_reconnect_online_idle_without_path_is_fresh_idle_evidence(self): self.reducer.handle_status( { "state": "playing", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", } ) self.reducer.handle_status({"state": "idle", "message": "player online"}) self.assertEqual(self.store.player_observation().phase, "idle") self.assertEqual(self.store.list_live_guards(), []) def test_offline_without_slot_preserves_active_protection_as_unknown(self): self.reducer.handle_status( { "state": "playing", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", } ) effects = self.reducer.handle_status({"state": "offline"}) observation = self.store.player_observation() self.assertEqual(effects[0].kind, "player_offline") self.assertEqual(observation.active_slot, 1) self.assertEqual(observation.phase, "unknown") self.assertEqual(self.reducer.projected_slot_state(1), "unknown") self.assertEqual(self.store.list_live_guards()[0].kind, "restart_uncertain") def test_busy_identifies_existing_playback_not_requested_slot(self): self.reducer.handle_status( { "state": "playing", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", } ) self.reducer.begin_operation( "play", slot=2, expected_path="/mnt/Media/YTPlayer/slot2/playlist.m3u", ) effects = self.reducer.handle_status( { "state": "busy", "slot": 1, "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u", "message": "mpv already playing; stop first", } ) self.assertIn("operation_failed", [effect.kind for effect in effects]) self.assertEqual(self.store.player_observation().active_slot, 1) self.assertEqual(self.reducer.projected_slot_state(1), "playing") self.assertEqual(self.reducer.projected_slot_state(2), "ready") def test_pause_status_does_not_override_actual_telemetry(self): path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" self.reducer.handle_telemetry( {"state": "telemetry", "slot": 1, "path": path, "paused": False} ) self.reducer.begin_operation("pause", slot=1, expected_path=path) effects = self.reducer.handle_status( {"state": "paused", "slot": 1, "path": path} ) self.assertIn("control_issued", [effect.kind for effect in effects]) self.assertIsNotNone(self.reducer.pending) self.assertEqual(self.reducer.projected_slot_state(1), "playing") effects = self.reducer.handle_telemetry( {"state": "telemetry", "slot": 1, "path": path, "paused": True} ) self.assertIn("operation_confirmed", [effect.kind for effect in effects]) self.assertEqual(self.reducer.projected_slot_state(1), "paused") def test_terminal_idle_and_cleared_telemetry_release_protection(self): path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" self.reducer.handle_telemetry( {"state": "telemetry", "slot": 1, "path": path, "paused": False} ) self.reducer.begin_operation("stop", slot=1, expected_path=path) effects = self.reducer.handle_telemetry( {"state": "telemetry", "slot": 1, "path": None, "paused": False} ) self.assertIn("operation_confirmed", [effect.kind for effect in effects]) self.assertEqual(self.store.player_observation().phase, "idle") self.assertEqual(self.store.list_live_guards(), []) self.reducer.handle_status({"state": "idle", "slot": 1, "reason": "stop"}) self.assertEqual(self.reducer.projected_slot_state(1), "ready") def test_error_does_not_release_owned_playback(self): path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" self.reducer.handle_status({"state": "playing", "slot": 1, "path": path}) self.reducer.handle_status( {"state": "error", "slot": 1, "path": path, "message": "cannot stop"} ) self.assertEqual(self.store.player_observation().phase, "error") self.assertEqual(self.store.list_live_guards()[0].slot, 1) self.assertEqual(self.reducer.projected_slot_state(1), "unknown") def test_play_timeout_becomes_unknown_then_operator_review_without_unprotecting(self): self.reducer.begin_operation( "play", slot=1, expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u", ) self.monotonic.value += 6 effect = self.reducer.check_timeout() self.assertEqual(effect.kind, "operation_timed_out") self.assertEqual(self.reducer.projected_slot_state(1), "unknown") self.wall.value += 31 review = self.reducer.mark_due_reconciliation() self.assertEqual(review[0].kind, "operator_review") self.assertEqual(self.reducer.projected_slot_state(1), "operator_review") self.assertEqual(self.store.list_live_guards()[0].state, "review_required") def test_malformed_and_unknown_statuses_are_isolated(self): for payload in (None, [], {}, {"state": "mystery"}): effects = self.reducer.handle_status(payload) self.assertEqual(effects[0].kind, "warning") def test_contradictory_slot_and_path_cannot_confirm_play(self): expected = str(self.config.storage.media_base / "slot1" / "playlist.m3u") self.reducer.begin_operation("play", slot=1, expected_path=expected) effects = self.reducer.handle_status( { "state": "playing", "slot": 1, "path": str( self.config.storage.media_base / "slot2" / "playlist.m3u" ), } ) self.assertEqual(effects[0].kind, "warning") self.assertIsNotNone(self.reducer.pending) self.assertEqual(self.store.player_observation().phase, "unknown") def test_mismatched_active_evidence_fails_intent_and_is_not_overwritten(self): self.reducer.begin_operation( "play", slot=1, expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u", ) effects = self.reducer.handle_status( { "state": "playing", "slot": 2, "path": "/mnt/Media/YTPlayer/slot2/playlist.m3u", } ) self.assertIn("operation_failed", [effect.kind for effect in effects]) self.assertIsNone(self.reducer.pending) self.monotonic.value += 100 self.assertIsNone(self.reducer.check_timeout()) observation = self.store.player_observation() self.assertEqual(observation.active_slot, 2) self.assertEqual(observation.phase, "playing") self.assertEqual(self.store.list_live_guards()[0].slot, 2) def test_bare_idle_does_not_clear_live_playback(self): path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" self.reducer.handle_status({"state": "playing", "slot": 1, "path": path}) effects = self.reducer.handle_status({"state": "idle", "slot": 1}) self.assertEqual(effects[0].kind, "warning") self.assertEqual(self.store.player_observation().phase, "playing") self.assertEqual(self.store.list_live_guards()[0].slot, 1) def test_path_only_evidence_derives_and_protects_managed_slot(self): path = str(self.config.storage.media_base / "slot2" / "playlist.m3u") self.reducer.handle_telemetry( {"state": "telemetry", "slot": None, "path": path, "paused": False} ) observation = self.store.player_observation() self.assertEqual(observation.active_slot, 2) self.assertEqual(self.store.list_live_guards()[0].slot, 2) def test_duplicate_offline_evidence_does_not_postpone_review(self): path = "/mnt/Media/YTPlayer/slot1/playlist.m3u" self.reducer.handle_status({"state": "playing", "slot": 1, "path": path}) self.reducer.handle_status({"state": "offline"}) first_deadline = self.store.list_live_guards()[0].review_after self.wall.value += 20 self.reducer.handle_status({"state": "offline"}) self.assertEqual( self.store.list_live_guards()[0].review_after, first_deadline, ) def test_repeated_active_telemetry_reuses_guard_without_history_churn(self): path = str(self.config.storage.media_base / "slot1" / "playlist.m3u") payload = { "state": "telemetry", "slot": 1, "path": path, "paused": False, } self.reducer.handle_telemetry(payload) first_guard = self.store.list_live_guards()[0] for _ in range(20): self.reducer.handle_telemetry(payload) self.assertEqual(self.store.list_live_guards()[0].guard_id, first_guard.guard_id) count = self.store.connection.execute( "SELECT COUNT(*) FROM slot_guards" ).fetchone()[0] self.assertEqual(count, 1) def test_external_play_timeout_transitions_to_review_without_assuming_stop(self): path = "/srv/operator-playlists/special-event.m3u" self.reducer.begin_operation("play", expected_path=path) self.monotonic.value += 6 timeout = self.reducer.check_timeout() self.assertEqual(timeout.kind, "operation_timed_out") uncertain = self.store.player_observation() self.assertEqual(uncertain.active_path, path) self.assertEqual(uncertain.reconciliation_state, "awaiting_evidence") self.assertIsNotNone(uncertain.reconciliation_id) self.wall.value += 31 effects = self.reducer.mark_due_reconciliation() self.assertEqual(len(effects), 1) self.assertEqual(effects[0].kind, "operator_review") self.assertEqual(effects[0].operation_id, uncertain.reconciliation_id) review = self.store.player_observation() self.assertEqual(review.reconciliation_state, "operator_review") self.assertEqual(review.active_path, path) def test_external_pause_does_not_bleed_to_a_different_path(self): first_path = "/srv/operator-playlists/first.m3u" second_path = "/srv/operator-playlists/second.m3u" self.reducer.handle_telemetry( { "state": "telemetry", "path": first_path, "paused": True, } ) self.assertIs(self.store.player_observation().paused, True) self.reducer.handle_status( {"state": "playing", "path": second_path} ) observation = self.store.player_observation() self.assertEqual(observation.active_path, second_path) self.assertIsNone(observation.paused) self.assertEqual(observation.phase, "playing") def test_stop_timeout_preserves_existing_operator_review_and_token(self): path = str(self.config.storage.media_base / "slot1" / "playlist.m3u") self.reducer.handle_status( {"state": "playing", "slot": 1, "path": path} ) self.reducer.handle_status({"state": "offline"}) self.wall.value += 31 self.reducer.mark_due_reconciliation() before = self.store.player_observation() guard_before = self.store.list_live_guards()[0] self.assertEqual(before.reconciliation_state, "operator_review") self.reducer.begin_operation("stop", slot=1, expected_path=path) self.monotonic.value += 6 effect = self.reducer.check_timeout() self.assertEqual(effect.kind, "operation_timed_out") after = self.store.player_observation() guard_after = self.store.list_live_guards()[0] self.assertEqual(after.reconciliation_state, "operator_review") self.assertEqual(after.reconciliation_id, before.reconciliation_id) self.assertEqual(guard_after.guard_id, guard_before.guard_id) self.assertEqual(guard_after.state, "review_required") if __name__ == "__main__": unittest.main()