"""Synthetic recovery/privilege-boundary tests. No USB, Bluetooth or service changes.""" import json import sys import tempfile import unittest from datetime import UTC, datetime, timedelta from pathlib import Path from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[1] sys.path[:0] = [str(ROOT), str(ROOT / "packaging")] import layout # noqa: E402 import wake_service # noqa: E402 from check_runtime import command # noqa: E402 from runtime import broker, identity, recovery # noqa: E402 from runtime.native import parameters # noqa: E402 A = identity.device_id("SYNTHETICAAAAAA") B = identity.device_id("SYNTHETICBBBBBB") def snapshot(ident=A, session="x4_one", preview=0, recording=0): now = datetime.now(UTC).isoformat() return { "id": ident, "session_id": session, "opened_at": now, "observed_at": now, "revision": 2, "prepared": True, "verified": True, "preparation_safe": True, "message": None, "online": True, "status": {"connected": True, "preview": preview, "recording": recording}, } class RecoveryTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(dir=ROOT / "build") self.root = Path(self.temporary.name) self.broker = broker.Broker() self.value = recovery.Recovery(self.broker, self.root / "recovery") self.broker.recovery = self.value self.addCleanup(self.value.pool.shutdown, wait=True) self.addCleanup(self.temporary.cleanup) def enable(self, ident=A): self.value.observe([snapshot(ident)], 0) with patch.object(recovery, "request", return_value={"enabled": True}): self.value.configure(ident, True) def test_default_has_no_wake_and_independent_instances_keep_distinct_intents(self): self.assertEqual(self.value.observe([], 0), []) self.enable() self.value.states[B] = self.value.default() self.assertEqual(self.value.observe([], 0), []) self.assertEqual(self.value.observe([], 8), [(A, "wake")]) self.assertFalse(self.value.view(B)["enabled"]) def test_grace_three_attempts_and_restart_do_not_reset_budget(self): self.enable() self.value.observe([], 0) with patch.object(recovery, "request", return_value={"state": "unavailable"}): for number, threshold in enumerate((8, 40, 100)): self.assertEqual(self.value.observe([], threshold - 1), []) self.assertEqual(self.value.observe([], threshold), [(A, "wake")]) self.value.work(A, "wake") self.assertEqual(self.value.states[A]["attempts"], number + 1) self.assertEqual(self.value.observe([], 900), []) restored = recovery.Recovery(self.broker, self.root / "recovery") self.addCleanup(restored.pool.shutdown, wait=True) self.assertEqual(restored.observe([], 1000), []) def test_stable_return_resets_budget_only_after_one_minute(self): self.enable() self.value.states[A]["attempts"] = 3 self.value.observe([], 0) self.value.observe([snapshot()], 10) self.assertEqual(self.value.states[A]["attempts"], 3) self.value.observe([snapshot()], 70) self.assertEqual(self.value.states[A]["attempts"], 0) def test_disable_cancels_helper_even_while_offline_and_clears_preview(self): self.enable() self.value.states[A]["preview_wanted"] = True self.value.observe([], 1) with patch.object(recovery, "request", return_value={"enabled": False}) as call: self.value.configure(A, False) call.assert_called_once() self.assertFalse(self.value.view(A)["preview_wanted"]) self.assertEqual(self.value.observe([], 500), []) def test_no_preview_restore_without_confirmed_intent_or_when_recording_unknown(self): self.enable() self.assertEqual(self.value.observe([snapshot(session="x4_two")], 1), []) self.value.states[A]["preview_wanted"] = True for recording in (1, -1, None): self.assertEqual(self.value.observe([snapshot(recording=recording)], 2), []) def test_restore_once_for_new_session_and_never_replay_sd_commands(self): self.enable() self.value.states[A]["preview_wanted"] = True self.value.observe([snapshot(session="x4_two")], 2) with patch.object( self.broker, "capture_operation", return_value={"state": "unknown"} ) as call: self.value.work(A, "preview") self.value.work(A, "preview") self.assertEqual(call.call_count, 1) self.assertEqual(call.call_args.args[1]["action_id"], "preview.start") self.assertEqual(self.value.observe([snapshot(session="x4_two")], 3), []) def test_cancelled_intent_blocks_previously_scheduled_preview_job(self): self.enable() self.value.states[A]["preview_wanted"] = True self.assertEqual(self.value.observe([snapshot(session="x4_two")], 1), [(A, "preview")]) self.value.states[A]["preview_wanted"] = False with patch.object(self.broker, "capture_operation") as call: self.value.work(A, "preview") call.assert_not_called() def test_old_start_receipt_cannot_reenable_preview_after_stop(self): self.enable() start = command(A, "x4_one", "preview.start") stop = command(A, "x4_one", "preview.stop") stop["requested_at"] = (datetime.now(UTC) + timedelta(seconds=1)).isoformat() for value in (start, stop, start): self.value.preview_intent(A, value, {"state": "complete"}) self.assertFalse(self.value.view(A)["preview_wanted"]) def test_offline_inventory_preserves_control_but_never_claims_online(self): self.enable() offline = self.value.augment([])[0] self.assertFalse(offline["online"]) self.assertFalse(offline["preparation_safe"]) self.assertEqual(offline["status"], {"recovery": self.value.view(A)}) self.assertNotIn("serial", json.dumps(offline)) def test_recovery_parameters_require_explicit_camera_setting_confirmation(self): parameters("recovery.configure", {"enabled": False, "wakeup_confirmed": False}) for value in ( {"enabled": True, "wakeup_confirmed": False}, {"enabled": 1, "wakeup_confirmed": True}, {"enabled": True}, ): with self.assertRaises(ValueError): parameters("recovery.configure", value) class WakeServiceTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(dir=ROOT / "build") self.root = Path(self.temporary.name) self.addCleanup(self.temporary.cleanup) self.service = wake_service.WakeService(self.root) self.trust = patch.object(wake_service, "trusted", side_effect=lambda path, *args: path) self.trust.start() self.addCleanup(self.trust.stop) layout_trust = patch.object(layout, "trusted", side_effect=lambda path, *args: path) layout_trust.start() self.addCleanup(layout_trust.stop) self.binding = identity.Binding(A, "1-2", 1, 2, "SYNTHETICAAAAAA") def enable(self): with patch.object(wake_service, "connected", return_value=[self.binding]): self.service.control(A, True) def test_enrollment_uses_os_serial_and_refuses_arbitrary_transport_input(self): self.enable() self.assertEqual(self.service.load(A)["serial"], self.binding.serial) with self.assertRaises(ValueError): self.service.dispatch( "POST", "/control", {"device_id": A, "enabled": True, "serial": "ATTACKER"}, {} ) with ( patch.object(wake_service, "connected", return_value=[]), self.assertRaises(ValueError), ): self.service.control(B, True) def test_present_sdk_and_duplicate_serials_never_launch_advertiser(self): self.enable() with ( patch.object(wake_service, "connected", return_value=[self.binding]), patch.object(wake_service.subprocess, "Popen") as spawn, ): self.assertEqual(self.service.wake_locked(A, "1" * 32)["state"], "connected") spawn.assert_not_called() with ( patch.object(wake_service, "connected", return_value=[self.binding, self.binding]), self.assertRaises(ValueError), ): self.service.control(A, True) def test_persisted_rate_limit_and_duplicate_attempt_prevent_commands(self): self.enable() value = self.service.load(A) value["attempts"] = [{"id": str(i) * 32, "at": wake_service.time.time()} for i in range(3)] self.service.save(A, value) with ( patch.object(wake_service, "connected", return_value=[]), patch.object(wake_service.subprocess, "Popen") as spawn, ): self.assertEqual(self.service.wake_locked(A, "0" * 32)["state"], "already_attempted") self.assertEqual(self.service.wake_locked(A, "4" * 32)["state"], "exhausted") spawn.assert_not_called() def test_known_radio_suffix_collision_blocks_wake(self): self.enable() self.service.save(B, {"serial": "DIFFERENTAAAAAA", "enabled": True, "attempts": []}) with patch.object(wake_service, "connected", return_value=[]): self.assertEqual(self.service.wake_locked(A, "1" * 32)["state"], "identity_conflict") def test_disable_persists_before_cancelling_only_selected_child(self): self.enable() left, right = Mock(), Mock() self.service.running = {A: left, B: right} self.service.control(A, False) left.terminate.assert_called_once() right.terminate.assert_not_called() self.assertFalse(self.service.load(A)["enabled"]) def test_one_adapter_busy_does_not_consume_another_camera_attempt(self): self.service.advertiser.acquire() try: self.assertEqual(self.service.wake(A, "1" * 32), {"state": "busy"}) finally: self.service.advertiser.release() self.assertEqual(list(self.root.iterdir()), []) if __name__ == "__main__": unittest.main()