Gabriel Brown
dbd1472e6b
A phone named "Gib's iPhone" with a typographic apostrophe was shown as "Gib\342\200\231s iPhone". 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 shlex handed back the escape sequences as literal characters and they went straight to the page. Apostrophes were only the visible case: accents, emoji, quotes and backslashes were all affected, and a name containing a quote could have split a field. Property and method reads now use --json=short, which returns real UTF-8, and the parsers read a document rather than splitting words. That removes the class rather than unescaping octal by hand. The fixtures were the reason this stayed invisible: every test fed the text form and passed against output the helper is no longer asking for. They now carry what busctl actually emits in the mode used, plus a case for a non-ASCII name and one asserting the old text form is refused rather than parsed wrongly. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
290 lines
13 KiB
Python
290 lines
13 KiB
Python
#!/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()
|