feat(x4): add per-camera manual wake and hide power action when connected

plugins/insta360-x4/tests/check_recovery.py

@@ -1,5 +1,6 @@
 """Synthetic recovery/privilege-boundary tests. No USB, Bluetooth or service changes."""
 
+import io
 import json
 import sys
 import tempfile
@@ -149,6 +150,72 @@ class RecoveryTests(unittest.TestCase):
             with self.assertRaises(ValueError):
                 parameters("recovery.configure", value)
 
+    def test_registry_retains_offline_wake_identity_without_automatic_intent(self):
+        with patch.object(recovery, "request", return_value={"items": [A, B]}):
+            self.value.sync_registry()
+        self.assertTrue(self.value.view(A)["wake_available"])
+        self.assertFalse(self.value.view(A)["enabled"])
+        self.assertEqual(self.value.observe([], 900), [])
+        self.assertEqual(len(self.value.augment([])), 2)
+
+    def test_manual_wake_is_durable_once_and_does_not_enable_policy_or_preview(self):
+        self.value.states[A] = self.value.default()
+        session = self.value.states[A]["session_id"]
+        value = command(A, session, "power.wake")
+        with patch.object(recovery, "request", return_value={"state": "connected"}) as call:
+            self.assertEqual(self.value.operation(A, value)["state"], "complete")
+            self.assertEqual(self.value.operation(A, value)["state"], "complete")
+        call.assert_called_once_with(
+            recovery.HELPER,
+            "/wake-once",
+            {
+                "device_id": A,
+                "attempt": value["operation_id"][3:],
+            },
+            timeout=52,
+        )
+        self.assertFalse(self.value.view(A)["enabled"])
+        self.assertFalse(self.value.view(A)["preview_wanted"])
+        with self.assertRaises(ValueError):
+            parameters("power.wake", {"serial": "ARBITRARY"})
+
+    def test_manual_wake_failure_does_not_schedule_automatic_retry(self):
+        self.value.states[A] = self.value.default()
+        value = command(A, self.value.states[A]["session_id"], "power.wake")
+        with (
+            patch.object(recovery, "request", return_value={"state": "unavailable"}) as call,
+            patch.object(recovery.time, "monotonic", side_effect=[0, 53]),
+        ):
+            result = self.value.operation(A, value)
+            self.assertEqual(self.value.operation(A, value), result)
+        self.assertEqual(result["state"], "error")
+        call.assert_called_once()
+        self.assertEqual(self.value.observe([], 1000), [])
+
+    def test_manual_wake_waits_for_late_sdk_return_without_another_radio_command(self):
+        with (
+            patch.object(recovery, "request", return_value={"state": "unavailable"}) as call,
+            patch.object(recovery.time, "monotonic", side_effect=[0, 22, 37]),
+            patch.object(self.value.stopping, "wait") as wait,
+            patch.object(self.broker, "snapshots", side_effect=[[snapshot(B)], [snapshot(A)]]),
+        ):
+            result = self.value.wake_once(A, command(A, "x4_one", "power.wake"))
+        self.assertEqual(result["state"], "complete")
+        call.assert_called_once()
+        wait.assert_called_once_with(1)
+        self.assertFalse(self.value.view(A)["enabled"])
+
+    def test_manual_wake_does_not_wait_for_other_camera_or_retry_after_deadline(self):
+        with (
+            patch.object(recovery, "request", return_value={"state": "unavailable"}) as call,
+            patch.object(recovery.time, "monotonic", side_effect=[0, 22, 53]),
+            patch.object(self.value.stopping, "wait"),
+            patch.object(self.broker, "snapshots", return_value=[snapshot(B)]),
+        ):
+            result = self.value.wake_once(A, command(A, "x4_one", "power.wake"))
+        self.assertEqual(result["state"], "error")
+        call.assert_called_once()
+
 
 class WakeServiceTests(unittest.TestCase):
     def setUp(self):
@@ -231,6 +298,49 @@ class WakeServiceTests(unittest.TestCase):
             self.service.advertiser.release()
         self.assertEqual(list(self.root.iterdir()), [])
 
+    def test_registry_enrolls_usb_without_enabling_recovery_and_keeps_it_when_unplugged(self):
+        with patch.object(wake_service, "connected", return_value=[self.binding]):
+            self.assertEqual(self.service.registry(), {"items": [A]})
+        self.assertFalse(self.service.load(A)["enabled"])
+        self.assertEqual(self.service.load(A)["attempts"], [])
+        with patch.object(wake_service, "connected", return_value=[]):
+            self.assertEqual(self.service.registry(), {"items": [A]})
+
+    def test_registry_never_enrolls_ambiguous_usb_identity(self):
+        with patch.object(wake_service, "connected", return_value=[self.binding, self.binding]):
+            self.assertEqual(self.service.registry(), {"items": []})
+
+    def test_manual_wake_works_while_disabled_and_shares_the_persisted_attempt_budget(self):
+        self.enable()
+        self.service.control(A, False)
+
+        def start(*args, **kwargs):
+            kwargs["stdout"].write(b'{"sdk_usb_returned": true}')
+            kwargs["stdout"].flush()
+            return Mock(stdin=io.BytesIO(), returncode=0)
+
+        with (
+            patch.object(wake_service, "connected", return_value=[]),
+            patch.object(
+                wake_service.subprocess,
+                "Popen",
+                side_effect=start,
+            ) as spawn,
+        ):
+            self.assertEqual(self.service.wake_locked(A, "1" * 32), {"state": "disabled"})
+            self.assertEqual(
+                self.service.wake_locked(A, "1" * 32, manual=True), {"state": "connected"}
+            )
+            self.assertEqual(
+                self.service.wake_locked(A, "1" * 32, manual=True), {"state": "already_attempted"}
+            )
+            self.assertEqual(
+                self.service.wake_locked(B, "2" * 32, manual=True), {"state": "disabled"}
+            )
+        spawn.assert_called_once()
+        self.assertFalse(self.service.load(A)["enabled"])
+        self.assertEqual(len(self.service.load(A)["attempts"]), 1)
+
 
 if __name__ == "__main__":
     unittest.main()