#!/usr/bin/env python3 from __future__ import annotations import importlib.machinery import importlib.util import json import pathlib import subprocess import tempfile import unittest ROOT = pathlib.Path(__file__).resolve().parents[2] HELPER = ROOT / "config/dot/quickshell/scripts/panama-kdeconnect" loader = importlib.machinery.SourceFileLoader("panama_kdeconnect", str(HELPER)) spec = importlib.util.spec_from_loader(loader.name, loader) bridge = importlib.util.module_from_spec(spec) loader.exec_module(bridge) class KdeConnectBridgeTest(unittest.TestCase): def test_status_omits_network_and_exposes_supported_ios_actions(self) -> None: line = ( "- Fixture iPhone: BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB " "on 192.0.2.7 via LAN (paired and reachable)" ) plugins = [ "kdeconnect_clipboard", "kdeconnect_findmyphone", "kdeconnect_ping", "kdeconnect_share", "kdeconnect_shareinputdevices", ] device = bridge.normalize_device_line(line, plugins) self.assertEqual(device["name"], "Fixture iPhone") self.assertEqual(device["type"], "phone") self.assertTrue(device["paired"]) self.assertTrue(device["reachable"]) self.assertEqual(device["actions"], ["clipboard", "ping", "ring", "share"]) self.assertNotIn("192.0.2.7", json.dumps(device)) self.assertNotIn("LAN", json.dumps(device)) def test_status_parses_an_offline_paired_device(self) -> None: line = "- Pocket: AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA (paired)" device = bridge.normalize_device_line(line, ["kdeconnect_share"]) self.assertTrue(device["paired"]) self.assertFalse(device["reachable"]) self.assertEqual(device["actions"], ["share"]) def test_invalid_device_id_is_rejected(self) -> None: with self.assertRaisesRegex(ValueError, "invalid-device"): bridge.validate_device_id("phone; shutdown") def test_send_file_requires_a_regular_file(self) -> None: with tempfile.TemporaryDirectory() as directory: with self.assertRaisesRegex(ValueError, "invalid-file"): bridge.validate_file(pathlib.Path(directory)) def test_send_file_resolves_a_regular_file(self) -> None: with tempfile.TemporaryDirectory() as directory: file_path = pathlib.Path(directory) / "hello world.txt" file_path.write_text("fixture", encoding="utf-8") self.assertEqual(bridge.validate_file(file_path), file_path.resolve()) def test_action_commands_use_separate_arguments(self) -> None: device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" self.assertEqual( bridge.action_command("ring", device_id), ["kdeconnect-cli", "-d", device_id, "--ring"], ) self.assertEqual( bridge.action_command("clipboard", device_id), ["kdeconnect-cli", "-d", device_id, "--send-clipboard"], ) def test_unknown_action_is_rejected(self) -> None: with self.assertRaisesRegex(ValueError, "unsupported-action"): bridge.action_command( "unlock", "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB", ) def test_busctl_plugin_output_is_normalized(self) -> None: output = ( '{"type":"as","data":["kdeconnect_ping","kdeconnect_share",' '"kdeconnect_clipboard","kdeconnect_findmyphone","unrelated"]}\n' ) self.assertEqual( bridge.parse_loaded_plugins(output), [ "kdeconnect_ping", "kdeconnect_share", "kdeconnect_clipboard", "kdeconnect_findmyphone", "unrelated", ], ) def test_device_name_keeps_its_typographic_characters(self) -> None: """A phone named "Gib's iPhone" is displayed that way. busctl's default TEXT output escapes every non-ASCII byte in octal, and escapes it into the output rather than into a quoted string a shell-style parser can undo -- so the name arrived as the literal characters "Gib\\342\\200\\231s iPhone" and was shown on the Home & Phone page exactly like that. Apostrophes were only the visible case; accents, emoji, quotes and backslashes were all affected. """ output = '{"type":"s","data":"Gib\u2019s iPhone"}\n' self.assertEqual(bridge.parse_string_property(output), "Gib\u2019s iPhone") def test_octal_escaped_name_is_not_accepted_as_a_value(self) -> None: """The old text form must not parse at all, rather than parse wrongly. Reading it as a value is what produced the mangled name; refusing it means a future change back to text output fails loudly instead of displaying escape sequences to someone. """ self.assertEqual(bridge.parse_string_property('s "Gib\\342\\200\\231s iPhone"\n'), "") self.assertEqual(bridge.parse_bool_property("b true\n"), None) self.assertEqual(bridge.parse_loaded_plugins('as 1 "kdeconnect_ping"\n'), []) def test_offline_device_falls_back_to_supported_plugins(self) -> None: def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: member = command[-1] if member == "loadedPlugins": return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "") if member == "supportedPlugins": return subprocess.CompletedProcess( command, 0, '{"type":"as","data":["kdeconnect_share",' '"kdeconnect_clipboard","kdeconnect_findmyphone"]}\n', "", ) raise AssertionError(command) self.assertEqual( bridge.device_plugins( "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB", runner, ), [ "kdeconnect_share", "kdeconnect_clipboard", "kdeconnect_findmyphone", ], ) 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[:4] == ["busctl", "--user", "--json=short", "call"]: return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "") if command[:4] == ["busctl", "--user", "--json=short", "get-property"]: values = { "name": '{"type":"s","data":"Fixture iPhone"}\n', "type": '{"type":"s","data":"phone"}\n', "isPaired": '{"type":"b","data":true}\n', "isReachable": '{"type":"b","data":false}\n', "supportedPlugins": ( '{"type":"as","data":["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[:4] == ["busctl", "--user", "--json=short", "call"]: return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "") if command[:4] == ["busctl", "--user", "--json=short", "get-property"]: values = { "name": '{"type":"s","data":"Fixture iPhone"}\n', "type": '{"type":"s","data":"phone"}\n', "isPaired": '{"type":"b","data":true}\n', "isReachable": '{"type":"b","data":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[:4] == ["busctl", "--user", "--json=short", "get-property"]: values = { "name": '{"type":"s","data":"Nearby Stranger"}\n', "type": '{"type":"s","data":"phone"}\n', "isPaired": '{"type":"b","data":false}\n', "isReachable": '{"type":"b","data":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[:4] == ["busctl", "--user", "--json=short", "call"]: return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "") if command[:4] == ["busctl", "--user", "--json=short", "get-property"]: is_dbus_device = dbus_id in command[4] values = { "name": '{"type":"s","data":"Fixture iPhone"}\n', "type": '{"type":"s","data":"phone"}\n' if is_dbus_device else '{"type":"s","data":"desktop"}\n', "isPaired": '{"type":"b","data":true}\n', "isReachable": '{"type":"b","data":false}\n', "supportedPlugins": '{"type":"as","data":[]}\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()