"""Keyboard and pointer: event checks, and the agent's KDE Connect protocol against a fake kdeconnectd. The fake follows what KDE Connect 24.02 does (read from its source, core/backends/lan/lanlinkprovider.cpp): it accepts the TCP connection, reads the identity line, then starts TLS as the *client*. Run: python3 -m unittest discover -s tests """ import sandbox # noqa: F401 (first: keeps tests off real data and services) import json import os import shutil import socket import ssl import subprocess import sys import tempfile import threading import time import unittest from unittest import mock from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "ui")) class InputEvents(unittest.TestCase): @classmethod def setUpClass(cls): import server cls.server = server def check(self, event): return self.server.input_event(event) def test_keeps_known_fields(self): self.assertEqual(self.check({"dx": 3, "dy": -1.234}), {"dx": 3.0, "dy": -1.23}) self.assertEqual(self.check({"singleclick": True, "other": 1}), {"singleclick": True}) self.assertEqual(self.check({"key": "héllo", "shift": True}), {"key": "héllo", "shift": True}) self.assertEqual(self.check({"specialKey": 12, "ctrl": True}), {"specialKey": 12, "ctrl": True}) self.assertEqual(self.check({"scroll": True, "dy": 1}), {"scroll": True, "dy": 1.0}) def test_clamps_movement(self): self.assertEqual(self.check({"dx": 1e9})["dx"], self.server.INPUT_MOVE_LIMIT) self.assertEqual(self.check({"dy": -1e9})["dy"], -self.server.INPUT_MOVE_LIMIT) def test_rejects_bad_events(self): bad = [None, [], "a", {}, {"shift": True}, {"dx": "1"}, {"dx": True}, {"dx": float("nan")}, {"key": ""}, {"key": 5}, {"key": "x" * 501}, {"specialKey": 0}, {"specialKey": 33}, {"specialKey": True}, {"specialKey": 1.5}, {"singleclick": "yes"}] for event in bad: with self.assertRaises(self.server.Failure, msg=repr(event)): self.check(event) def test_send_while_link_is_down_says_not_sent(self): # The page keeps unsent keys and clicks and sends them once the agent is ready again. from types import SimpleNamespace class Probe(self.server.InputAgent): def start(self): self.proc, self.status = SimpleNamespace(poll=lambda: None, stdin=None), {"state": "starting"} agent = Probe() agent.status, agent.proc = {"state": "ready"}, SimpleNamespace(poll=lambda: 255) # ssh has exited self.assertEqual(agent.send([{"key": "x"}]), {"state": "starting", "sent": False}) def test_command_passes_client_quoted(self): os.environ["FRAME_CLIENT"] = "test-client-1" try: cmd = self.server.InputAgent().command() finally: del os.environ["FRAME_CLIENT"] self.assertTrue(cmd.startswith("python3 -u -c '")) self.assertIn(" test-client-1 ", cmd) cmd = self.server.InputAgent(packages=[("a.pkg.tar.zst", "ab")]).command("~/in/x") self.assertTrue(cmd.endswith(""" '~/in/x' '[["a.pkg.tar.zst","ab"]]'"""), cmd) def test_batch_limits(self): with self.assertRaises(self.server.Failure): self.server.remote_input({"events": "dx"}) with self.assertRaises(self.server.Failure): self.server.remote_input({"events": [{"dx": 1}] * (self.server.INPUT_BATCH_LIMIT + 1)}) class Bundled(unittest.TestCase): """KDE Connect ships with Frame Control: the manifest, the notice, the copy to the Frame.""" @classmethod def setUpClass(cls): import server cls.server = server cls.manifest = json.loads((ROOT / "frame/kdeconnect/packages.json").read_text()) def test_manifest_pins_every_package(self): packages = self.manifest["packages"] self.assertEqual({p["name"] for p in packages}, {"kdeconnect", "kcontacts", "kpeople", "libfakekey", "modemmanager-qt", "pulseaudio-qt"}) for p in packages: self.assertRegex(p["sha256"], r"^[0-9a-f]{64}$") self.assertTrue(p["file"].startswith(f"{p['name']}-{p['version']}-") and p["file"].endswith("-aarch64.pkg.tar.zst"), p) self.assertTrue(p["source"].startswith(self.manifest["release"]), p) self.assertRegex(p["source_sha256"], r"^[0-9a-f]{64}$") self.assertTrue(self.manifest["release"].startswith("https://github.com/") and self.manifest["release"].endswith("/")) self.assertNotIn("DO_NOT_SHARE", json.dumps(self.manifest)) def test_notice_names_each_version_and_its_licence_texts_ship(self): notice = (ROOT / "frame/kdeconnect/NOTICE.md").read_text() for p in self.manifest["packages"]: self.assertIn(f"{p['name']} {p['version']}", notice) self.assertIn(p["source"], notice) self.assertIn(p["upstream"], notice) for spdx in p["licenses"]: self.assertTrue((ROOT / "frame/kdeconnect/LICENSES" / p["name"] / f"{spdx}.txt").is_file(), spdx) self.assertIn("frame/kdeconnect/NOTICE.md", (ROOT / "THIRD_PARTY_NOTICES.md").read_text()) def test_about_dialog_has_the_licences(self): titles = [n["title"] for n in self.server.licenses()] self.assertIn("Frame Control (MIT)", titles) self.assertIn("KDE Connect for the Frame", titles) self.assertIn("kdeconnect: GPL-2.0-only", titles) def test_both_apps_bundle_the_packages(self): pkg = json.loads((ROOT / "app/package.json").read_text())["build"]["extraResources"] kde = next(r for r in pkg if r["from"] == "../frame/kdeconnect") self.assertIn("packages/*.pkg.tar.zst", kde["filter"]) self.assertIn("LICENSES/**/*", kde["filter"]) bundle = (ROOT / "ios/scripts/make_frame_bundle.py").read_text() for pattern in ("frame/kdeconnect/packages/*.pkg.tar.zst", "frame/kdeconnect/LICENSES/*/*", "THIRD_PARTY_NOTICES.md"): self.assertIn(pattern, bundle) def fake_bundle(self, damaged=False): folder = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, folder) (folder / "packages").mkdir() data = {"a-1-1-aarch64.pkg.tar.zst": b"first", "b-2-1-aarch64.pkg.tar.zst": b"second"} packages = [] for name, body in data.items(): (folder / "packages" / name).write_bytes(b"x" + body if damaged else body) packages.append((name, __import__("hashlib").sha256(body).hexdigest())) return folder, packages def deliver(self, frame_has, damaged=False): folder, packages = self.fake_bundle(damaged) calls = [] def ssh(remote, stdin=None, timeout=30, text=True): calls.append((remote, stdin)) answer = "yes\n" if remote.startswith("{ test -x") and frame_has else "" return answer if text else answer.encode() old = self.server.ssh, self.server.KDECONNECT, self.server.LOCAL self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = ssh, folder, False try: agent = self.server.InputAgent(packages=packages) return agent.deliver(lambda m: calls.append(("report", m))), calls, packages finally: self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = old def test_copies_nothing_when_the_frame_has_them(self): folder, calls, packages = self.deliver(frame_has=True) self.assertEqual(folder, "") self.assertEqual(len(calls), 1) # Bytes: on Windows a text pipe would send CRLF and never match. self.assertEqual(calls[0][1], "".join(f"{sha} {name}\n" for name, sha in packages).encode()) def test_copies_each_package_over_ssh(self): folder, calls, packages = self.deliver(frame_has=False) self.assertTrue(folder.startswith("~/.local/share/frame-control/kdeconnect/incoming/")) copies = [c for c in calls if c[0] != "report" and "cat >" in c[0]] self.assertEqual([c[1] for c in copies], [b"first", b"second"]) self.assertIn("a-1-1-aarch64.pkg.tar.zst.part", copies[0][0]) def test_refuses_a_damaged_bundle(self): with self.assertRaises(self.server.Failure): self.deliver(frame_has=False, damaged=True) def test_a_failed_copy_is_removed(self): calls = [] def ssh(remote, stdin=None, timeout=30, text=True): calls.append(remote) if "cat >" in remote and len([c for c in calls if "cat >" in c]) == 2: raise self.server.Failure("Broken pipe") return b"" if not text else "" folder, packages = self.fake_bundle() old = self.server.ssh, self.server.KDECONNECT, self.server.LOCAL self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = ssh, folder, False try: with self.assertRaises(self.server.Failure): self.server.InputAgent(packages=packages).deliver(lambda m: None) finally: self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = old self.assertTrue(calls[-1].startswith("rm -rf .local/share/frame-control/kdeconnect/incoming/"), calls[-1]) def lifecycle(self, agent_lines, deliver=None): """An InputAgent whose ssh and agent are fakes. Returns it, the folders each launch used, and the fake agents. A fake agent reports its lines, then keeps running (unless its lines end with need-packages) until the test ends.""" launches, procs, done = [], [], threading.Event() self.addCleanup(done.set) class Agent(self.server.InputAgent): def deliver(self, report, force=False): if deliver: deliver() return "~/copied" if force else "" def command(self, folder=""): launches.append(folder) return "agent" class Proc: stdin = None def __init__(self, lines): self.lines, self.ended = lines, threading.Event() procs.append(self) @property def stdout(self): yield from self.lines if self.lines and b"need-packages" not in self.lines[-1]: while not (done.is_set() or self.ended.is_set()): self.ended.wait(0.02) self.ended.set() def wait(self): self.ended.wait(5) return 0 def poll(self): return 0 if self.ended.is_set() else None def terminate(self): self.ended.set() def popen(*a, **k): return Proc(agent_lines.pop(0) if agent_lines else [b'{"state": "ready"}\n']) old = self.server.ensure_master, self.server.subprocess.Popen self.server.ensure_master, self.server.subprocess.Popen = lambda: None, popen self.addCleanup(lambda: (setattr(self.server, "ensure_master", old[0]), setattr(self.server.subprocess, "Popen", old[1]))) agent = Agent(packages=[("a.pkg.tar.zst", "0" * 64)]) def settle(): # nothing of this test may still be launching when the next one starts agent.stop() for _ in range(250): if agent.launching is None and all(p.ended.is_set() for p in procs): return time.sleep(0.02) self.addCleanup(settle) return agent, launches, procs def wait_for(self, check): for _ in range(250): if check(): return time.sleep(0.02) self.fail("timed out") def test_agent_asking_for_packages_gets_them_and_starts_again(self): agent, launches, procs = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']]) agent.start() self.wait_for(lambda: agent.status == {"state": "ready"}) self.assertEqual(launches, ["", "~/copied"]) self.assertIs(agent.proc, procs[1]) def test_asking_twice_is_an_error_not_a_loop(self): need = [b'{"state": "need-packages"}\n'] agent, launches, _ = self.lifecycle([need, list(need)]) agent.start() self.wait_for(lambda: agent.status.get("state") == "error") self.assertEqual(launches, ["", "~/copied"]) self.assertIn("didn't reach", agent.status["message"]) def test_start_after_stop_during_the_copy_still_starts(self): gate, entered = threading.Event(), threading.Event() agent, launches, procs = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5))) agent.start() self.assertTrue(entered.wait(5)) agent.stop() entered.clear() agent.start() # while the first launch is still copying self.assertTrue(entered.wait(5), "the second start didn't launch") gate.set() self.wait_for(lambda: agent.status == {"state": "ready"}) time.sleep(0.1) # The stopped launch started no agent; the new one is the one in use. self.assertEqual(len(procs), 1) self.assertIs(agent.proc, procs[0]) agent.stop() self.wait_for(lambda: all(p.ended.is_set() for p in procs)) def test_retry_and_a_new_start_never_run_two_agents(self): # A start() landing just as an agent that asked for the packages exits: exactly one # of it and the retry launches, and stop() ends everything. The new start is held # in its copy until the retry has decided, so neither can see the other's agent. first, decided = [], threading.Event() agent, launches, procs = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']]) watch, launch, deliver = agent._watch, agent._launch, agent.deliver def racing_watch(proc, errors, retry=False): wanted, heard = watch(proc, errors, retry) if wanted and not first: first.append(threading.current_thread()) agent.start() # lands between the agent exiting and the retry return wanted, heard def held_deliver(report, force=False): if force: decided.set() # the retry went ahead elif first and threading.current_thread() is not first[0]: decided.wait(5) # the new start waits until the retry has decided return deliver(report, force) def first_launch(generation, force=False): try: launch(generation, force) finally: if first and threading.current_thread() is first[0]: decided.set() # the retry declined agent._watch, agent._launch, agent.deliver = racing_watch, first_launch, held_deliver agent.start() # Both have decided once the new start has launched its agent (it always does). self.wait_for(lambda: decided.is_set() and launches.count("") == 2 and len(procs) == len(launches)) self.wait_for(lambda: agent.status == {"state": "ready"}) self.assertEqual(sum(not p.ended.is_set() for p in procs), 1, launches) agent.stop() self.wait_for(lambda: all(p.ended.is_set() for p in procs)) def test_stopped_during_the_copy_removes_it(self): gate, entered = threading.Event(), threading.Event() agent, launches, procs = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5))) removed = [] agent.discard = removed.append agent.deliver = lambda report, force=False: (entered.set(), gate.wait(5), "~/incoming/x")[2] self.addCleanup(lambda: self.assertEqual(removed, ["~/incoming/x"])) agent.start() self.assertTrue(entered.wait(5)) agent.stop() gate.set() self.wait_for(lambda: removed == ["~/incoming/x"]) self.assertEqual(procs, []) # no agent started for it def test_a_copy_whose_agent_never_starts_is_removed(self): agent, launches, procs = self.lifecycle([[]]) # the agent dies before saying anything removed = [] agent.discard = removed.append agent.deliver = lambda report, force=False: "~/incoming/y" agent.start() self.wait_for(lambda: removed == ["~/incoming/y"]) def test_discard_only_touches_copies(self): calls = [] old = self.server.ssh, self.server.LOCAL self.server.ssh, self.server.LOCAL = (lambda remote, **k: calls.append(remote)), False try: agent = self.server.InputAgent(packages=[]) agent.discard("") agent.discard("~/.local/share/frame-control/kdeconnect") agent.discard("~/.local/share/frame-control/kdeconnect/incoming/ab12") self.server.LOCAL = True agent.discard("~/.local/share/frame-control/kdeconnect/incoming/ab12") finally: self.server.ssh, self.server.LOCAL = old self.assertEqual(calls, ["rm -rf .local/share/frame-control/kdeconnect/incoming/ab12"]) def test_a_launch_error_leaves_it_ready_to_try_again(self): # Popen fails (say, out of file handles) and so does removing the copy: the # page gets the error and the next start launches. agent, launches, procs = self.lifecycle([]) agent.deliver = lambda report, force=False: "~/.local/share/frame-control/kdeconnect/incoming/z" old_popen, old_ssh = self.server.subprocess.Popen, self.server.ssh self.server.subprocess.Popen = self.server.ssh = lambda *a, **k: (_ for _ in ()).throw(OSError(24, "Too many open files")) try: agent.start() self.wait_for(lambda: agent.status.get("state") == "error") finally: self.server.subprocess.Popen, self.server.ssh = old_popen, old_ssh self.assertIsNone(agent.launching) agent.start() self.wait_for(lambda: agent.status == {"state": "ready"}) def test_any_unexpected_error_still_reports_and_allows_a_retry(self): agent, launches, procs = self.lifecycle([]) old = self.server.tempfile.TemporaryFile self.server.tempfile.TemporaryFile = lambda: (_ for _ in ()).throw(OSError(24, "Too many open files")) try: agent.start() self.wait_for(lambda: agent.status.get("state") == "error") finally: self.server.tempfile.TemporaryFile = old self.assertIsNone(agent.launching) agent.deliver = lambda report, force=False: (_ for _ in ()).throw(ValueError("something odd")) agent.start() self.wait_for(lambda: "something odd" in agent.status.get("message", "")) self.assertIsNone(agent.launching) def test_copies_go_to_a_folder_of_their_own(self): first, _, _ = self.deliver(frame_has=False) second, _, _ = self.deliver(frame_has=False) self.assertNotEqual(first, second) @unittest.skipIf(sys.platform == "win32", "the agent runs on the Frame (Linux)") class AgentInstall(unittest.TestCase): """The agent unpacks what the server copied, after checking each SHA-256.""" @classmethod def setUpClass(cls): import frame_input_agent cls.agent = frame_input_agent def setUp(self): self.dir = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.dir) saved = self.agent.BASE, self.agent.ROOT, self.agent.stop_daemon, self.agent.say self.addCleanup(lambda: setattr_all(self.agent, saved)) self.agent.BASE, self.agent.ROOT = self.dir / "base", self.dir / "base/root" self.agent.stop_daemon, self.agent.say = lambda: None, lambda *a, **k: None self.agent.BASE.mkdir() def package(self): src = self.dir / "src" (src / "usr/lib").mkdir(parents=True) (src / "usr/lib/kdeconnectd").write_text("#!/bin/sh\n") out = self.dir / "incoming/kdeconnect-24.02.2-1-aarch64.pkg.tar.zst" out.parent.mkdir() if subprocess.run(["tar", "--zstd", "-cf", str(out), "-C", str(src), "usr"], capture_output=True).returncode: self.skipTest("this tar can't write zstd") return out, [(out.name, self.agent.sha256(out))] def test_unpacks_and_stamps(self): path, packages = self.package() self.assertFalse(self.agent.installed(packages)) self.agent.install(path.parent, packages) self.assertTrue((self.agent.ROOT / "usr/lib/kdeconnectd").is_file()) self.assertTrue(self.agent.installed(packages)) self.assertFalse(self.agent.installed([(path.name, "0" * 64)])) def test_refuses_a_damaged_package(self): path, packages = self.package() with open(path, "ab") as f: f.write(b"!") with self.assertRaisesRegex(RuntimeError, "damaged"): self.agent.install(path.parent, packages) self.assertFalse(self.agent.ROOT.exists()) def test_missing_package_and_empty_manifest(self): with self.assertRaisesRegex(RuntimeError, "didn't reach"): self.agent.install(self.dir, [("nope.pkg.tar.zst", "0" * 64)]) with self.assertRaisesRegex(RuntimeError, "doesn't include"): self.agent.install(self.dir, []) def test_leaves_another_devices_running_copy_alone(self): # A different build is running for another device: use it, don't stop it to reinstall. self.agent.listening, old = (lambda: True), self.agent.listening self.agent.our_daemons, old_ours = (lambda: [123]), self.agent.our_daemons try: self.agent.ensure_daemon("", [("new.pkg.tar.zst", "1" * 64)]) finally: self.agent.listening, self.agent.our_daemons = old, old_ours self.assertFalse(self.agent.ROOT.exists()) def test_asks_for_packages_it_was_not_sent(self): self.agent.listening, old = (lambda: False), self.agent.listening try: for folder in ("", str(self.dir / "gone")): # none sent, or already tidied away with self.assertRaises(self.agent.NeedPackages): self.agent.ensure_daemon(folder, [("new.pkg.tar.zst", "1" * 64)]) finally: self.agent.listening = old def test_restart_can_still_unpack_its_copy_then_tidies_it(self): folder = self.agent.BASE / "incoming/abc" folder.mkdir(parents=True) seen = [] stubs = {"ensure_daemon": lambda f, p: seen.append(Path(f).is_dir()), "identity": lambda c: ("id", "cert", "key"), "our_daemons": lambda: [123], "connect": lambda *a: (_ for _ in ()).throw(OSError("no answer"))} saved = {k: getattr(self.agent, k) for k in stubs} self.addCleanup(lambda: [setattr(self.agent, k, v) for k, v in saved.items()]) for k, v in stubs.items(): setattr(self.agent, k, v) self.assertEqual(self.agent.run("c", "n", str(folder), []), 1) self.assertEqual(seen, [True, True]) # there for the first start and the restart self.assertFalse(folder.exists()) def test_tidy_keeps_copies_in_use(self): incoming = self.agent.BASE / "incoming" held_dir = incoming / "held" held_dir.mkdir(parents=True) held = self.agent.hold_incoming(str(held_dir)) self.addCleanup(held.close) os.utime(held_dir, (time.time() - 7200,) * 2) self.agent.tidy_incoming("") self.assertTrue(held_dir.is_dir()) def test_tidy_keeps_other_starts_copies(self): incoming = self.agent.BASE / "incoming" mine, theirs, stale = incoming / "mine", incoming / "theirs", incoming / "stale" for d in (mine, theirs, stale): d.mkdir(parents=True) os.utime(stale, (time.time() - 7200,) * 2) self.agent.tidy_incoming(str(mine)) self.assertEqual(sorted(p.name for p in incoming.iterdir()), ["theirs"]) def test_server_and_agent_agree_on_the_stamp(self): import server packages = [("a.pkg.tar.zst", "1" * 64), ("b.pkg.tar.zst", "2" * 64)] self.assertEqual(server.kdeconnect_stamp(packages), self.agent.stamp(packages)) def setattr_all(module, saved): module.BASE, module.ROOT, module.stop_daemon, module.say = saved class FakeKdeConnect: """Just enough of kdeconnectd: pairs when asked and records remote-input packets.""" def __init__(self, paired=False): self.listener = socket.socket() self.listener.bind(("127.0.0.1", 0)) self.listener.listen(1) self.port = self.listener.getsockname()[1] self.paired, self.identity, self.received, self.pair_requests = paired, None, [], 0 threading.Thread(target=self.serve, daemon=True).start() def serve(self): conn, _ = self.listener.accept() line = b"" while not line.endswith(b"\n"): line += conn.recv(1) self.identity = json.loads(line) ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ctx.check_hostname, ctx.verify_mode = False, ssl.CERT_NONE tls = ctx.wrap_socket(conn) # "Starting client ssl (but I'm the server TCP socket)" tls.sendall(b'{"id": 1, "type": "kdeconnect.mousepad.keyboardstate", "body": {"state": true}}\n') buf = b"" while True: try: chunk = tls.recv(65536) except OSError: return if not chunk: return buf += chunk while b"\n" in buf: raw, buf = buf.split(b"\n", 1) p = json.loads(raw) if p["type"] == "kdeconnect.pair": self.pair_requests += 1 elif p["type"] == "kdeconnect.mousepad.request" and self.paired: self.received.append(p["body"]) @unittest.skipIf(sys.platform == "win32", "the agent runs on the Frame (Linux)") @unittest.skipUnless(shutil.which("openssl"), "needs openssl to make a certificate") class AgentProtocol(unittest.TestCase): @classmethod def setUpClass(cls): import frame_input_agent cls.agent = frame_input_agent cls.dir = tempfile.TemporaryDirectory() cls.cert, cls.key = Path(cls.dir.name, "cert.pem"), Path(cls.dir.name, "key.pem") subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1", "-subj", "/CN=framecontrol_test", "-keyout", str(cls.key), "-out", str(cls.cert)], check=True, capture_output=True) @classmethod def tearDownClass(cls): cls.dir.cleanup() def wait_for(self, check): for _ in range(100): if check(): return time.sleep(0.02) self.fail("timed out") def test_pairs_then_forwards_events(self): fake = FakeKdeConnect() link = self.agent.Link("framecontrol_0123456789abcdef01234567", self.cert, self.key, port=fake.port) link.pair(lambda: fake.paired, lambda: setattr(fake, "paired", fake.pair_requests > 0), timeout=5) self.assertEqual(fake.pair_requests, 1) ident = fake.identity["body"] self.assertEqual(fake.identity["type"], "kdeconnect.identity") self.assertEqual(ident["protocolVersion"], 7) self.assertIn("kdeconnect.mousepad.request", ident["outgoingCapabilities"]) link.read(0.5) self.assertTrue(link.keyboard) for body in self.agent.events(b'[{"dx": 4, "dy": -2}, {"key": "hi"}]'): link.send(body) self.wait_for(lambda: len(fake.received) == 2) self.assertEqual(fake.received, [{"dx": 4, "dy": -2}, {"key": "hi"}]) def test_already_paired_sends_no_pair_request(self): # A pair request to a device that's already paired makes KDE Connect unpair it. fake = FakeKdeConnect(paired=True) link = self.agent.Link("framecontrol_0123456789abcdef01234567", self.cert, self.key, port=fake.port) link.pair(lambda: True, lambda: self.fail("accepted a pairing that wasn't needed")) link.send({"singleclick": True}) self.wait_for(lambda: fake.received == [{"singleclick": True}]) self.assertEqual(fake.pair_requests, 0) def test_client_args_are_folder_safe(self): old = sys.argv try: sys.argv = ["-c", "../../etc x", "Alex's Mac"] self.assertEqual(self.agent.client_args(), ("etcx", "Frame Control (Alex's Mac)", "", [])) sys.argv = ["-c"] self.assertEqual(self.agent.client_args(), ("default", "Frame Control", "", [])) sys.argv = ["-c", "mac", "Mac", "~/x", '[["a.pkg.tar.zst", "ab"]]'] self.assertEqual(self.agent.client_args()[2:], (os.path.expanduser("~/x"), [("a.pkg.tar.zst", "ab")])) finally: sys.argv = old def test_events_parsing(self): self.assertEqual(self.agent.events(b'{"dx": 1}'), [{"dx": 1}]) self.assertEqual(self.agent.events(b'[{"dx": 1}, 5, {}]'), [{"dx": 1}]) self.assertEqual(self.agent.events(b"not json"), []) @unittest.skipUnless(shutil.which("node"), "needs node") class PageQueue(unittest.TestCase): """The page's input queue (tests/page/pad_queue.mjs runs the real functions from index.html).""" def test_queue_rules(self): r = subprocess.run(["node", str(ROOT / "tests" / "page" / "pad_queue.mjs")], capture_output=True, text=True) self.assertEqual(r.returncode, 0, r.stdout + r.stderr) if __name__ == "__main__": unittest.main() class PadEvents(unittest.TestCase): """The trackpad and key row go through gamescope; accents and emoji go the KDE Connect way.""" @classmethod def setUpClass(cls): import server cls.pad = staticmethod(server.pad_events) def test_moves_clicks_and_scroll(self): out, extra = self.pad([{"dx": 3, "dy": -2}, {"singleclick": True}, {"singlehold": True}, {"singlerelease": True}, {"scroll": True, "dy": 1}, {"rightclick": True}]) self.assertEqual(out[0], {"dx": 3, "dy": -2}) self.assertEqual(out[1:3], [{"button": "left", "down": True}, {"button": "left", "down": False}]) self.assertEqual(out[3:5], [{"button": "left", "down": True}, {"button": "left", "down": False}]) self.assertEqual(out[5], {"scroll": [0, -15]}) # up = content follows the finger self.assertEqual(out[6:], [{"button": "right", "down": True}, {"button": "right", "down": False}]) self.assertEqual(extra, []) def test_keys_and_modifiers(self): out, _ = self.pad([{"specialKey": 12}, {"specialKey": 1, "ctrl": True}, {"key": "c", "ctrl": True}]) self.assertEqual(out[:2], [{"key": 28, "down": True}, {"key": 28, "down": False}]) self.assertEqual([e["key"] for e in out[2:6]], [29, 14, 14, 29]) # Ctrl+Backspace self.assertEqual([(e["key"], e["down"]) for e in out[6:]], [(29, True), (46, True), (46, False), (29, False)]) # Ctrl+C def test_text_stays_ascii_through_gamescope(self): out, slow = self.pad([{"key": "Hi "}, {"key": "there"}]) self.assertEqual(out, [{"text": "Hi there"}]) self.assertEqual(slow, []) def test_keyboard_goes_to_kde_whole_and_in_order(self): events = [{"key": "a"}, {"key": "é"}, {"specialKey": 12}, {"key": "c", "ctrl": True}, {"dx": 2, "dy": 1}] out, slow = self.pad(events, kde=True) self.assertEqual(out, [{"dx": 2, "dy": 1}]) # the pointer still goes through gamescope self.assertEqual(slow, events[:4]) # the keyboard as sent, in order def test_shifted_key_with_modifier(self): out, _ = self.pad([{"key": "A", "ctrl": True}]) self.assertEqual([e["key"] for e in out], [29, 42, 30, 30, 42, 29]) class PadDelivery(unittest.TestCase): """remote_input's promises: "sent" means the whole batch is taken; accents go once, in order.""" class Agent: def __init__(self, sent=True, state="ready"): self.sent, self.state, self.got = sent, state, [] def send(self, events): if events and self.sent: self.got.append(events) return {"state": self.state, "sent": self.sent and bool(events)} def setUp(self): import server self.s = server self.saved = (server._touch, server._input, server.KDE_WAIT) server._touch, server._input = self.Agent(), self.Agent() self.reset() def tearDown(self): self.s._touch, self.s._input, self.s.KDE_WAIT = self.saved self.settle() self.reset() def reset(self): self.s._kde_queue.clear() self.s._kde_error = None self.s._kde_until = self.s._typed_until = 0.0 self.s._kde_worker[0] = None def settle(self): worker = self.s._kde_worker[0] if worker: worker.join(5) def typed(self): return [e for batch in self.s._input.got for e in batch] def test_accent_only_batch_is_acknowledged_and_typed_once(self): status = self.s.remote_input({"events": [{"key": "é"}]}) self.assertTrue(status["sent"]) # so the page doesn't send it again self.settle() self.assertEqual(self.typed(), [{"key": "é"}]) def test_mixed_text_stays_in_one_ordered_stream(self): self.s.remote_input({"events": [{"key": "aéb"}, {"specialKey": 12}]}) self.settle() self.assertEqual(self.typed(), [{"key": "aéb"}, {"specialKey": 12}]) self.assertEqual(self.s._touch.got, []) # none of it split off through gamescope def test_batch_gamescope_didnt_take_is_not_queued(self): self.s._touch = self.Agent(sent=False, state="starting") status = self.s.remote_input({"events": [{"dx": 1, "dy": 1}, {"key": "é"}]}) self.assertFalse(status["sent"]) # the page tries the whole batch again... self.assertEqual(self.s._kde_queue, []) # ...so the accent isn't queued twice def test_waits_for_kde_connect_then_sends_without_more_input(self): self.s._input = self.Agent(sent=False, state="starting") self.s.remote_input({"events": [{"key": "é"}]}) time.sleep(0.2) self.assertEqual(self.s._kde_queue, [{"key": "é"}]) self.s._input.sent = True self.settle() self.assertEqual(self.typed(), [{"key": "é"}]) self.assertEqual(self.s._kde_queue, []) def test_later_ascii_waits_behind_a_pending_accent(self): self.s._input = self.Agent(sent=False, state="starting") self.s.remote_input({"events": [{"key": "é"}]}) self.s.remote_input({"events": [{"key": "x"}]}) self.assertEqual(self.s._touch.got, []) # "x" can't overtake the "é" self.assertEqual(self.s._kde_queue, [{"key": "é"}, {"key": "x"}]) self.s._input.sent = True self.settle() self.assertEqual(self.typed(), [{"key": "é"}, {"key": "x"}]) def test_errors_and_timeout_retain_accepted_text_for_explicit_retry(self): events = [{"key": "é"}] * 150 for state in ("error", "starting"): with self.subTest(state=state): self.reset() self.s._input = self.Agent(sent=False, state=state) self.s.KDE_WAIT = 0 self.assertTrue(self.s.remote_input({"events": events})["sent"]) self.settle() self.assertEqual(self.s._kde_queue, events) self.assertIsNone(self.s._kde_worker[0]) self.assertEqual(self.s._kde_until, 0) status = self.s.pad_status({"state": "ready"}) self.assertEqual(status["state"], "error") self.assertIn("retained", status["message"]) self.assertFalse(self.s.remote_input({"events": [{"key": "x"}]})["sent"]) self.s._input = self.Agent() self.s.start_pad() self.settle() self.assertEqual(self.typed(), events) self.assertEqual(self.s._kde_queue, []) self.assertIsNone(self.s._kde_error) def test_worker_exception_and_start_failure_are_visible_and_retryable(self): for start_failure in (False, True): with self.subTest(start_failure=start_failure): self.reset() events = [{"key": "é"}] if start_failure: with mock.patch.object(threading.Thread, "start", side_effect=RuntimeError("no thread")): self.assertTrue(self.s.remote_input({"events": events})["sent"]) else: with mock.patch.object(self.s._input, "send", side_effect=OSError("send failed")): self.assertTrue(self.s.remote_input({"events": events})["sent"]) self.settle() self.assertEqual(self.s._kde_queue, events) self.assertIsNone(self.s._kde_worker[0]) self.assertEqual(self.s._kde_until, 0) self.assertEqual(self.s.pad_status({"state": "ready"})["state"], "error") self.s._input = self.Agent() self.s.start_pad() self.settle() self.assertEqual(self.typed(), events) def test_ascii_just_after_an_accent_stays_behind_it(self): self.s.remote_input({"events": [{"key": "é"}]}) self.settle() self.s.remote_input({"events": [{"key": "x"}]}) # KDE Connect may still be typing the é self.settle() self.assertEqual(self.s._touch.got, []) self.assertEqual(self.typed(), [{"key": "é"}, {"key": "x"}]) def test_accent_just_after_gamescope_text_waits_for_it(self): self.s.remote_input({"events": [{"key": "x" * 50}]}) self.assertEqual(self.s._touch.got, [[{"text": "x" * 50}]]) start = time.time() self.s.remote_input({"events": [{"key": "é"}]}) self.settle() self.assertGreater(time.time() - start, 0.6) # 50 keys at 16 ms, plus the margin self.assertEqual(self.typed(), [{"key": "é"}]) def test_typing_time_counts_every_key_transition(self): t = self.s.typing_seconds self.assertAlmostEqual(t([{"text": "ab"}]), 4 * 0.008) self.assertAlmostEqual(t([{"text": "A"}]), 4 * 0.008) # shift down, key down, key up, shift up self.assertAlmostEqual(t([{"text": "x" * 500}]), 8.0) # a long paste self.assertAlmostEqual(t([{"key": 29, "down": True}, {"dx": 3, "dy": 0}]), 0.008) def test_the_margin_is_added_once_not_per_request(self): for _ in range(10): self.s.remote_input({"events": [{"key": "ab"}]}) # ten requests of 4 transitions each: the work queues up, with no margin added per request self.assertLess(self.s._typed_until - time.time(), 10 * 4 * 0.008 + 0.05) def hold_worker(self): entered, release = threading.Event(), threading.Event() class Gate: def wait(self, timeout): entered.set() if not release.wait(3): raise RuntimeError("test did not release worker") return False patch = mock.patch.object(self.s, "_kde_cancel", Gate()) patch.start() self.addCleanup(patch.stop) self.addCleanup(release.set) return entered, release def test_saturation_rejects_whole_batch_and_retry_delivers_every_accepted_event_once(self): entered, release = self.hold_worker() self.s._typed_until = time.time() + 60 first = [{"key": "é"}] * 150 remainder = [{"key": "ü"}] * 50 overflow = [{"singleclick": True}, {"key": "ß"}] * 75 self.assertTrue(self.s.remote_input({"events": first})["sent"]) self.assertTrue(entered.wait(1)) # batch is in flight, waiting for prior ASCII try: self.assertTrue(self.s.remote_input({"events": remainder})["sent"]) status = self.s.remote_input({"events": overflow}) self.assertFalse(status["sent"]) self.assertEqual(status["state"], "waiting") self.assertEqual(self.s._kde_queue, first + remainder) self.assertEqual(len(self.s._kde_queue), self.s.KDE_QUEUE_LIMIT) self.assertEqual(self.s._touch.got, []) # rejected clicks cannot be duplicated finally: self.s._typed_until = 0 release.set() self.settle() self.assertEqual(self.typed(), first + remainder) self.assertTrue(self.s.remote_input({"events": overflow})["sent"]) self.settle() self.assertEqual(self.typed(), first + remainder + [{"key": "ß"}] * 75) self.assertEqual(self.s._touch.got, [[e for _ in range(75) for e in ( {"button": "left", "down": True}, {"button": "left", "down": False})]]) def test_retrying_batch_keeps_its_capacity_and_order(self): entered, release = self.hold_worker() self.s._input = self.Agent(sent=False, state="starting") first, second = [{"key": "é"}] * 150, [{"key": "ü"}] * 150 self.assertTrue(self.s.remote_input({"events": first})["sent"]) self.assertTrue(entered.wait(1)) # first send failed; worker is in retry delay try: self.assertFalse(self.s.remote_input({"events": second})["sent"]) self.assertEqual(self.s._kde_queue, first) finally: self.s._input.sent = True release.set() self.settle() self.assertTrue(self.s.remote_input({"events": second})["sent"]) self.settle() self.assertEqual(self.typed(), first + second) def test_concurrent_requests_cannot_both_claim_the_last_capacity(self): entered, release = self.hold_worker() self.s._typed_until = time.time() + 60 first = [{"key": "é"}] self.s.remote_input({"events": first}) self.assertTrue(entered.wait(1)) barrier, results = threading.Barrier(3), [] def request(key): barrier.wait(2) batch = [{"key": key}] * 150 results.append((batch, self.s.remote_input({"events": batch})["sent"])) threads = [threading.Thread(target=request, args=(key,)) for key in ("ü", "ß")] try: for thread in threads: thread.start() barrier.wait(2) for thread in threads: thread.join(2) self.assertFalse(thread.is_alive()) self.assertEqual(sorted(sent for _, sent in results), [False, True]) finally: self.s._typed_until = 0 release.set() self.settle() accepted = next(batch for batch, sent in results if sent) rejected = next(batch for batch, sent in results if not sent) self.assertEqual(self.typed(), first + accepted) self.assertTrue(self.s.remote_input({"events": rejected})["sent"]) self.settle() self.assertEqual(self.typed(), first + accepted + rejected) def test_input_queued_as_the_worker_finishes_is_still_sent(self): for _ in range(40): self.reset() self.s._input.got.clear() self.s.remote_input({"events": [{"key": "é"}]}) time.sleep(0.002) self.s.remote_input({"events": [{"key": "ü"}]}) self.settle() self.assertEqual(self.s._kde_queue, []) self.assertEqual(self.typed(), [{"key": "é"}, {"key": "ü"}]) class PadRoute(unittest.TestCase): def test_unicode_start_and_headset_switch_do_not_deadlock(self): # Real InputAgent.send/start takes the route lock. Race it with route(), # which cancels the pad worker; mocks of send() hide this lock inversion. # A child keeps a regression's blocked daemon threads out of this suite. code = r''' import sandbox import server as s import threading from unittest import mock entered, release, switching = threading.Event(), threading.Event(), threading.Event() s._input = s.InputAgent(packages=[]) s._input._launch = lambda *args: None # exercise real start, without any SSH s._touch = mock.Mock() send = s._input.send def held(events): entered.set() assert release.wait(2) return send(events) s._input.send = held s._power.stop = lambda *args: switching.set() s._queue_for_kde([{'key': 'é'}]) worker = s._kde_worker[0] assert entered.wait(2), 'Unicode worker never reached send' switch = threading.Thread(target=s.route, args=('frame-pr66-new', []), daemon=True) switch.start() switching.wait(0.2) release.set() worker.join(2) switch.join(2) assert not worker.is_alive(), 'Unicode worker deadlocked starting input during headset switch' assert not switch.is_alive(), 'headset switch deadlocked cancelling Unicode input' assert not s._kde_queue ''' env = {**os.environ, "PYTHONPATH": os.pathsep.join([str(ROOT / 'tests'), str(ROOT / 'ui')])} result = subprocess.run([sys.executable, '-c', code], env=env, capture_output=True, text=True, timeout=8) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) def setUp(self): import server self.s = server self.agents = mock.patch.multiple(server, _touch=mock.Mock(), _input=mock.Mock()) self.agents.start() self.addCleanup(self.agents.stop) self.s._touch.send.return_value = {"state": "ready", "sent": True} self.s._input.send.return_value = {"state": "ready", "sent": True} self.s.reset_pad_route() self.s._touch.reset_mock(); self.s._input.reset_mock() self.addCleanup(self.s.reset_pad_route) def switch(self): with mock.patch.multiple(self.s, FRAME="frame-a", HOST_OPTS=[], MUX=[], SSH=[], macview=None), \ mock.patch.multiple(self.s.frame_android, FRAME="frame-a", SSH_OPTS=[]): self.s.route("frame-b", []) def test_stale_worker_exception_cannot_damage_new_route(self): entered, release = threading.Event(), threading.Event() class BrokenWait: def wait(self, timeout): entered.set() if not release.wait(2): raise RuntimeError("test did not release old worker") raise OSError("old route wait failed") def set(self): pass # hold its exception until the new route has been published self.s._kde_cancel = BrokenWait() self.s._typed_until = time.time() + 60 self.s.remote_input({"events": [{"key": "é"}]}) worker = self.s._kde_worker[0] try: self.assertTrue(entered.wait(1)) self.switch() marker = object() with self.s._kde_lock: self.s._kde_queue.append({"key": "new"}) self.s._kde_worker[0] = marker self.s._kde_until = 42 finally: release.set() worker.join(2) self.assertFalse(worker.is_alive()) self.assertEqual(self.s._kde_queue, [{"key": "new"}]) self.assertIs(self.s._kde_worker[0], marker) self.assertEqual(self.s._kde_until, 42) self.assertIsNone(self.s._kde_error) def test_switch_stops_both_input_agents_and_discards_old_queue(self): self.s._kde_queue.append({"key": "old"}) self.s._kde_until = self.s._typed_until = time.time() + 60 cancel = self.s._kde_cancel self.switch() self.s._input.stop.assert_called_once() self.s._touch.stop.assert_called_once() self.assertEqual(self.s._kde_queue, []) self.assertEqual((self.s._kde_until, self.s._typed_until), (0, 0)) self.assertTrue(cancel.is_set()) def test_switch_cancels_delayed_and_retrying_text_before_new_input(self): for delaying in (True, False): with self.subTest(delaying=delaying): waiting, cancel = threading.Event(), threading.Event() class ObservedCancel: def wait(self, timeout): waiting.set() return cancel.wait(timeout) def set(self): cancel.set() self.s.reset_pad_route() self.s._kde_cancel = ObservedCancel() self.s._typed_until = time.time() + 60 if delaying else 0 self.s._input.reset_mock() self.s._input.send.return_value = {"state": "starting", "sent": False} self.s.remote_input({"events": [{"key": "é"}]}) worker = self.s._kde_worker[0] self.assertTrue(waiting.wait(1), "worker never reached its cancellable delay") self.switch() worker.join(1) self.assertFalse(worker.is_alive()) attempts = self.s._input.send.call_count self.assertEqual(attempts, 0 if delaying else 1) self.s._input.send.return_value = {"state": "ready", "sent": True} self.s.remote_input({"events": [{"key": "ü"}]}) next_worker = self.s._kde_worker[0] if next_worker: next_worker.join(1) self.assertFalse(next_worker.is_alive()) self.assertEqual(self.s._input.send.call_count, attempts + 1) self.assertEqual(self.s._input.send.call_args.args[0], [{"key": "ü"}]) self.assertEqual(self.s._kde_queue, [])