diff --git a/config/dot/quickshell/scripts/panama-kdeconnect b/config/dot/quickshell/scripts/panama-kdeconnect index 5e670df..ef8d2df 100755 --- a/config/dot/quickshell/scripts/panama-kdeconnect +++ b/config/dot/quickshell/scripts/panama-kdeconnect @@ -7,7 +7,6 @@ from __future__ import annotations import json import pathlib import re -import shlex import subprocess import sys from collections.abc import Callable @@ -88,33 +87,44 @@ def normalize_device_line( } +# busctl properties are read with --json=short, not in its default text form. +# +# The text form escapes every non-ASCII byte in octal, and it escapes them into +# the OUTPUT rather than into a quoted string a shell-style parser can undo -- so +# a phone named "Gib's iPhone" with a typographic apostrophe arrived as the +# literal characters "Gib\342\200\231s iPhone" and was displayed that way. +# That is not specific to apostrophes: any name with an accent, an emoji, a +# quote, or a backslash was affected the same way. +# +# The JSON form returns real UTF-8 and needs no unescaping, which is why these +# parse a document rather than splitting words. +def parse_property(output: str, expected: str) -> object | None: + """The value of a busctl --json=short property read, or None if it is not + the type asked for.""" + try: + payload = json.loads(output) + except (json.JSONDecodeError, TypeError): + return None + if not isinstance(payload, dict) or payload.get("type") != expected: + return None + return payload.get("data") + + def parse_loaded_plugins(output: str) -> list[str]: - try: - parts = shlex.split(output) - except ValueError: + data = parse_property(output, "as") + if not isinstance(data, list): return [] - if len(parts) < 2 or parts[0] != "as": - return [] - try: - count = int(parts[1]) - except ValueError: - return [] - return parts[2 : 2 + max(0, count)] + return [str(item) for item in data] def parse_string_property(output: str) -> str: - try: - parts = shlex.split(output) - except ValueError: - return "" - return parts[1] if len(parts) == 2 and parts[0] == "s" else "" + data = parse_property(output, "s") + return data if isinstance(data, str) else "" def parse_bool_property(output: str) -> bool | None: - parts = output.split() - if len(parts) != 2 or parts[0] != "b" or parts[1] not in {"true", "false"}: - return None - return parts[1] == "true" + data = parse_property(output, "b") + return data if isinstance(data, bool) else None def run_command( @@ -141,6 +151,7 @@ def loaded_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str] [ "busctl", "--user", + "--json=short", "call", "org.kde.kdeconnect", device_object(device_id), @@ -157,6 +168,7 @@ def supported_plugins(device_id: str, runner: Runner = subprocess.run) -> list[s [ "busctl", "--user", + "--json=short", "get-property", "org.kde.kdeconnect", device_object(device_id), @@ -178,6 +190,7 @@ def reported_type(device_id: str, runner: Runner = subprocess.run) -> str: [ "busctl", "--user", + "--json=short", "get-property", "org.kde.kdeconnect", device_object(device_id), @@ -198,6 +211,7 @@ def device_property( [ "busctl", "--user", + "--json=short", "get-property", "org.kde.kdeconnect", device_object(device_id), diff --git a/tests/quickshell/kdeconnect_bridge_test.py b/tests/quickshell/kdeconnect_bridge_test.py index 41d0d73..b7517f8 100644 --- a/tests/quickshell/kdeconnect_bridge_test.py +++ b/tests/quickshell/kdeconnect_bridge_test.py @@ -89,8 +89,8 @@ class KdeConnectBridgeTest(unittest.TestCase): def test_busctl_plugin_output_is_normalized(self) -> None: output = ( - 'as 5 "kdeconnect_ping" "kdeconnect_share" ' - '"kdeconnect_clipboard" "kdeconnect_findmyphone" "unrelated"\n' + '{"type":"as","data":["kdeconnect_ping","kdeconnect_share",' + '"kdeconnect_clipboard","kdeconnect_findmyphone","unrelated"]}\n' ) self.assertEqual( @@ -104,16 +104,42 @@ class KdeConnectBridgeTest(unittest.TestCase): ], ) + 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, "as 0\n", "") + return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "") if member == "supportedPlugins": return subprocess.CompletedProcess( command, 0, - 'as 3 "kdeconnect_share" "kdeconnect_clipboard" "kdeconnect_findmyphone"\n', + '{"type":"as","data":["kdeconnect_share",' + '"kdeconnect_clipboard","kdeconnect_findmyphone"]}\n', "", ) raise AssertionError(command) @@ -139,17 +165,17 @@ class KdeConnectBridgeTest(unittest.TestCase): 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"]: + 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": 's "Fixture iPhone"\n', - "type": 's "phone"\n', - "isPaired": "b true\n", - "isReachable": "b false\n", + "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": ( - 'as 3 "kdeconnect_share" "kdeconnect_clipboard" ' - '"kdeconnect_findmyphone"\n' + '{"type":"as","data":["kdeconnect_share",' + '"kdeconnect_clipboard","kdeconnect_findmyphone"]}\n' ), } return subprocess.CompletedProcess(command, 0, values[command[-1]], "") @@ -180,14 +206,14 @@ class KdeConnectBridgeTest(unittest.TestCase): 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"]: + 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": 's "Fixture iPhone"\n', - "type": 's "phone"\n', - "isPaired": "b true\n", - "isReachable": "b false\n", + "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) @@ -217,12 +243,12 @@ class KdeConnectBridgeTest(unittest.TestCase): 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"]: + if command[:4] == ["busctl", "--user", "--json=short", "get-property"]: values = { - "name": 's "Nearby Stranger"\n', - "type": 's "phone"\n', - "isPaired": "b false\n", - "isReachable": "b true\n", + "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) @@ -240,16 +266,16 @@ class KdeConnectBridgeTest(unittest.TestCase): 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"]: + 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": '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", + "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)