Fix live network and phone discovery

2026-08-18 12:34:02 -04:00
parent 60a321e6f4
commit 08b16fa03f
3 changed files with 235 additions and 7 deletions
@@ -130,6 +130,134 @@ class KdeConnectBridgeTest(unittest.TestCase):
            ],
        )

+    def test_status_falls_back_to_paired_dbus_device_when_cli_is_empty(self) -> None:
+        device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
+        device_path = f"/modules/kdeconnect/devices/{device_id}"
+
+        def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
+            if command == ["kdeconnect-cli", "--list-devices"]:
+                return subprocess.CompletedProcess(command, 0, "0 devices found\n", "")
+            if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
+                return subprocess.CompletedProcess(command, 0, f"└─ {device_path}\n", "")
+            if command[:3] == ["busctl", "--user", "call"]:
+                return subprocess.CompletedProcess(command, 0, "as 0\n", "")
+            if command[:3] == ["busctl", "--user", "get-property"]:
+                values = {
+                    "name": 's "Fixture iPhone"\n',
+                    "type": 's "phone"\n',
+                    "isPaired": "b true\n",
+                    "isReachable": "b false\n",
+                    "supportedPlugins": (
+                        'as 3 "kdeconnect_share" "kdeconnect_clipboard" '
+                        '"kdeconnect_findmyphone"\n'
+                    ),
+                }
+                return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
+            raise AssertionError(command)
+
+        self.assertEqual(
+            bridge.collect_status(runner),
+            {
+                "available": True,
+                "devices": [
+                    {
+                        "id": device_id,
+                        "name": "Fixture iPhone",
+                        "type": "phone",
+                        "paired": True,
+                        "reachable": False,
+                        "actions": ["clipboard", "ring", "share"],
+                    }
+                ],
+                "error": "",
+            },
+        )
+
+    def test_dbus_plugin_timeout_keeps_device_with_no_actions(self) -> None:
+        device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
+
+        def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
+            if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
+                path = f"/modules/kdeconnect/devices/{device_id}"
+                return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "")
+            if command[:3] == ["busctl", "--user", "call"]:
+                return subprocess.CompletedProcess(command, 0, "as 0\n", "")
+            if command[:3] == ["busctl", "--user", "get-property"]:
+                values = {
+                    "name": 's "Fixture iPhone"\n',
+                    "type": 's "phone"\n',
+                    "isPaired": "b true\n",
+                    "isReachable": "b false\n",
+                }
+                if command[-1] == "supportedPlugins":
+                    raise subprocess.TimeoutExpired(command, 8)
+                return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
+            raise AssertionError(command)
+
+        self.assertEqual(bridge.dbus_devices(runner)[0]["actions"], [])
+
+    def test_cli_device_plugin_timeout_fails_closed(self) -> None:
+        device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
+        listing = f"- Fixture iPhone: {device_id} (paired and reachable)\n"
+
+        def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
+            if command == ["kdeconnect-cli", "--list-devices"]:
+                return subprocess.CompletedProcess(command, 0, listing, "")
+            raise subprocess.TimeoutExpired(command, 8)
+
+        status = bridge.collect_status(runner)
+
+        self.assertEqual(status["devices"][0]["type"], "phone")
+        self.assertEqual(status["devices"][0]["actions"], [])
+
+    def test_dbus_inventory_excludes_unpaired_peers(self) -> None:
+        device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
+
+        def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
+            if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
+                path = f"/modules/kdeconnect/devices/{device_id}"
+                return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "")
+            if command[:3] == ["busctl", "--user", "get-property"]:
+                values = {
+                    "name": 's "Nearby Stranger"\n',
+                    "type": 's "phone"\n',
+                    "isPaired": "b false\n",
+                    "isReachable": "b true\n",
+                }
+                return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
+            raise AssertionError(command)
+
+        self.assertEqual(bridge.dbus_devices(runner), [])
+
+    def test_status_merges_paired_dbus_device_missing_from_cli(self) -> None:
+        cli_id = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+        dbus_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"
+        listing = f"- Fixture Laptop: {cli_id} (paired and reachable)\n"
+
+        def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
+            if command == ["kdeconnect-cli", "--list-devices"]:
+                return subprocess.CompletedProcess(command, 0, listing, "")
+            if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
+                path = f"/modules/kdeconnect/devices/{dbus_id}"
+                return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "")
+            if command[:3] == ["busctl", "--user", "call"]:
+                return subprocess.CompletedProcess(command, 0, "as 0\n", "")
+            if command[:3] == ["busctl", "--user", "get-property"]:
+                is_dbus_device = dbus_id in command[4]
+                values = {
+                    "name": 's "Fixture iPhone"\n',
+                    "type": 's "phone"\n' if is_dbus_device else 's "desktop"\n',
+                    "isPaired": "b true\n",
+                    "isReachable": "b false\n",
+                    "supportedPlugins": "as 0\n",
+                }
+                return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
+            raise AssertionError(command)
+
+        status = bridge.collect_status(runner)
+
+        self.assertEqual({device["id"] for device in status["devices"]}, {cli_id, dbus_id})
+

 if __name__ == "__main__":
    unittest.main()