From 02817be3643127013cbe8ffceb3ffd9f1721698d Mon Sep 17 00:00:00 2001 From: Daniele Fucini Date: Sun, 8 Nov 2026 14:51:13 +0100 Subject: [PATCH] Add end-to-end tests for the daemon Each test runs xidle-inhibit on a private dbus-daemon with fake lockers on PATH, covering lock blocking, cookie handling, restoring after restarts and cleanup. --- .gitignore | 1 + pyproject.toml | 5 +- tests/test_daemon.py | 448 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 453 insertions(+), 1 deletion(-) create mode 100644 tests/test_daemon.py diff --git a/.gitignore b/.gitignore index 762fbdc..0d35bac 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ CLAUDE.md __pycache__/ .mypy_cache/ .ruff_cache/ +.pytest_cache/ diff --git a/pyproject.toml b/pyproject.toml index 561ad3c..6fae4b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,5 +7,8 @@ select = ["E", "F", "W", "I", "B", "UP"] [tool.mypy] # Scripts both map to __main__, so check xidle-dim separately: mypy xidle-dim -files = ["xidle-inhibit"] +files = ["xidle-inhibit", "tests"] strict = true + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/test_daemon.py b/tests/test_daemon.py new file mode 100644 index 0000000..8a002cf --- /dev/null +++ b/tests/test_daemon.py @@ -0,0 +1,448 @@ +"""End-to-end tests: run xidle-inhibit on a private D-Bus bus with fake lockers on PATH.""" + +import json +import os +import signal +import subprocess +import sys +import time +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any + +import pytest +from gi.repository import Gio, GLib + +# XIDLE_INHIBIT tests another copy, e.g. /usr/bin/xidle-inhibit. +SCRIPT = Path(os.environ.get("XIDLE_INHIBIT") or Path(__file__).resolve().parent.parent / "xidle-inhibit") +BUS_NAME = "org.freedesktop.ScreenSaver" +OBJECT_PATH = "/org/freedesktop/ScreenSaver" +STATE_FILE = "xidle-inhibit.json" +OLD_STATE_FILE = "xautolock-inhibit.json" +TIMEOUT = 5.0 +DELAY = ["--release-delay", "1"] +SETTLE = 1.5 # longer than DELAY, to check that nothing happens + +ENABLE, DISABLE = "xautolock -enable", "xautolock -disable" + +# No : nothing is auto-started on this bus. +BUS_CONFIG = """ + + session + unix:dir={dir} + EXTERNAL + + + + + + +""" + +FAKE_TOOL = '#!/bin/sh\necho "${0##*/} $*" >> "$FAKE_LOG"\n' + + +def wait_for(cond: Callable[[], bool], what: str, timeout: float = TIMEOUT) -> None: + deadline = time.monotonic() + timeout + while not cond(): + if time.monotonic() > deadline: + raise AssertionError(f"timed out waiting for {what}") + time.sleep(0.05) + + +class Bus: + def __init__(self, root: Path) -> None: + sock_dir = root / "sock" + sock_dir.mkdir(parents=True) + config = root / "bus.conf" + config.write_text(BUS_CONFIG.format(dir=sock_dir)) + self.proc = subprocess.Popen( + ["dbus-daemon", f"--config-file={config}", "--nofork", "--print-address=1"], + stdout=subprocess.PIPE, text=True, + ) + assert self.proc.stdout is not None + self.address = self.proc.stdout.readline().strip() + assert self.address, "dbus-daemon printed no address" + + def connect(self) -> Gio.DBusConnection: + flags = Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION + return Gio.DBusConnection.new_for_address_sync(self.address, flags, None, None) + + def stop(self) -> None: + self.proc.terminate() + self.proc.wait(TIMEOUT) + + +class Client: + """One bus connection, so one D-Bus sender.""" + + def __init__(self, bus: Bus) -> None: + self.conn = bus.connect() + name = self.conn.get_unique_name() + assert name is not None + self.name = name + + def call( + self, method: str, args: GLib.Variant | None = None, reply: str | None = None, + path: str = OBJECT_PATH, dest: str = BUS_NAME, iface: str = BUS_NAME, + ) -> Any: + result = self.conn.call_sync( + dest, path, iface, method, args, GLib.VariantType(reply) if reply else None, + Gio.DBusCallFlags.NO_AUTO_START, int(TIMEOUT * 1000), None, + ) + return result.unpack() + + def inhibit(self, reason: str = "Playing video", path: str = OBJECT_PATH) -> int: + return int(self.call("Inhibit", GLib.Variant("(ss)", ("test-app", reason)), "(u)", path)[0]) + + def uninhibit(self, cookie: int, path: str = OBJECT_PATH) -> None: + self.call("UnInhibit", GLib.Variant("(u)", (cookie,)), path=path) + + def has_owner(self, name: str) -> bool: + return bool(self.call( + "NameHasOwner", GLib.Variant("(s)", (name,)), "(b)", + "/org/freedesktop/DBus", "org.freedesktop.DBus", "org.freedesktop.DBus", + )[0]) + + def close(self) -> None: + if not self.conn.is_closed(): + self.conn.close_sync(None) + + +class Sandbox: + def __init__(self, root: Path) -> None: + self.root = root + self.bin = root / "bin" + self.runtime = root / "run" + self.config = root / "config" + for d in (self.bin, self.runtime, self.config / "xidle-inhibit"): + d.mkdir(parents=True) + for tool in ("xautolock", "xss-lock", "loginctl"): + path = self.bin / tool + path.write_text(FAKE_TOOL) + path.chmod(0o755) + self.fake_log = root / "fake.log" + self.daemon_log = root / "daemon.log" + self.state_file = self.runtime / STATE_FILE + self.buses: list[Bus] = [] + self.clients: list[Client] = [] + self.daemons: list[subprocess.Popen[bytes]] = [] + self.bus = self.new_bus() + self.set_backend("xautolock") + + def new_bus(self) -> Bus: + bus = Bus(self.root / f"bus{len(self.buses)}") + self.buses.append(bus) + return bus + + def client(self, bus: Bus | None = None) -> Client: + client = Client(bus or self.bus) + self.clients.append(client) + return client + + def set_backend(self, backend: str | None) -> None: + config = self.config / "xidle-inhibit" / "config" + if backend is None: + config.unlink(missing_ok=True) + else: + config.write_text(f"backend = {backend}\n") + + def environ(self, bus: Bus) -> dict[str, str]: + env = dict(os.environ) + for var in ("DISPLAY", "WAYLAND_DISPLAY"): + env.pop(var, None) + env.update( + PATH=str(self.bin), DBUS_SESSION_BUS_ADDRESS=bus.address, FAKE_LOG=str(self.fake_log), + XDG_RUNTIME_DIR=str(self.runtime), XDG_CONFIG_HOME=str(self.config), + ) + return env + + def start(self, *args: str, bus: Bus | None = None, wait: bool = True) -> subprocess.Popen[bytes]: + """Start the daemon; with wait, return once it has taken the name and set the locker.""" + bus = bus or self.bus + with open(self.daemon_log, "ab") as log: + proc = subprocess.Popen( + [sys.executable, str(SCRIPT), *args], env=self.environ(bus), stdout=log, stderr=subprocess.STDOUT, + ) + self.daemons.append(proc) + if wait: + probe = Client(bus) + try: + # Method calls are handled after start() returns, so a reply means startup is done. + def answers() -> bool: + assert proc.poll() is None, f"daemon exited:\n{self.log()}" + try: + probe.call("GetActive", reply="(b)") + return True + except GLib.Error: + return False + wait_for(answers, "the daemon to own the name") + finally: + probe.close() + return proc + + def stop(self, sig: int = signal.SIGTERM) -> int: + proc = self.daemons[-1] + proc.send_signal(sig) + return proc.wait(TIMEOUT) + + def calls(self) -> list[str]: + try: + return self.fake_log.read_text().splitlines() + except FileNotFoundError: + return [] + + def wait_calls(self, expected: list[str]) -> None: + wait_for(lambda: self.calls() == expected, f"calls {expected}, got {self.calls()}") + + def log(self) -> str: + return self.daemon_log.read_text() + + def close(self) -> None: + for proc in self.daemons: + if proc.poll() is None: + proc.kill() + proc.wait() + for client in self.clients: + client.close() + for bus in self.buses: + bus.stop() + + +@pytest.fixture +def sb(tmp_path: Path) -> Iterator[Sandbox]: + sandbox = Sandbox(tmp_path) + yield sandbox + sandbox.close() + + +# --- blocking and unblocking --------------------------------------------------- + + +def test_blocks_once_and_unblocks_after_last_release(sb: Sandbox) -> None: + sb.start(*DELAY) + assert sb.calls() == [ENABLE] # known state at startup + a = sb.client() + first, second = a.inhibit(), a.inhibit() + assert second == first + 1 + assert sb.calls() == [ENABLE, DISABLE] + a.uninhibit(first) + time.sleep(SETTLE) + assert sb.calls() == [ENABLE, DISABLE] + a.uninhibit(second) + sb.wait_calls([ENABLE, DISABLE, ENABLE]) + + +def test_burst_within_release_delay_stays_blocked(sb: Sandbox) -> None: + sb.start(*DELAY) + a = sb.client() + a.uninhibit(a.inhibit()) + a.inhibit() + time.sleep(SETTLE) + assert sb.calls() == [ENABLE, DISABLE] + + +def test_both_object_paths_share_cookies(sb: Sandbox) -> None: + sb.start(*DELAY) + a = sb.client() + cookie = a.inhibit(path="/ScreenSaver") + assert sb.calls() == [ENABLE, DISABLE] + a.uninhibit(cookie, path=OBJECT_PATH) + sb.wait_calls([ENABLE, DISABLE, ENABLE]) + + +def test_ignored_reason_does_not_block(sb: Sandbox) -> None: + sb.start(*DELAY) + sb.client().inhibit(reason="Playing audio") + assert sb.calls() == [ENABLE] + assert "reason ignored" in sb.log() + + +def test_count_all_blocks_on_any_reason(sb: Sandbox) -> None: + sb.start(*DELAY, "--count-all") + sb.client().inhibit(reason="Playing audio") + assert sb.calls() == [ENABLE, DISABLE] + + +def test_uninhibit_from_another_sender_is_ignored(sb: Sandbox) -> None: + sb.start(*DELAY) + a, b = sb.client(), sb.client() + b.uninhibit(a.inhibit()) + time.sleep(SETTLE) + assert sb.calls() == [ENABLE, DISABLE] + assert "unknown cookie, ignored" in sb.log() + + +def test_sender_leaving_the_bus_releases_its_inhibits(sb: Sandbox) -> None: + sb.start(*DELAY) + a, b = sb.client(), sb.client() + a.inhibit() + a.inhibit() + b.inhibit(reason="Playing audio") # another sender's cookie stays + a.close() + sb.wait_calls([ENABLE, DISABLE, ENABLE]) + assert sb.log().count("left the bus") == 2 + + +def test_xss_lock_backend_runs_no_locker_commands(sb: Sandbox) -> None: + sb.set_backend("xss-lock") + sb.start(*DELAY) + a = sb.client() + a.uninhibit(a.inhibit()) + time.sleep(SETTLE) + assert sb.calls() == [] + assert sb.stop() == 0 + assert sb.calls() == [] + + +@pytest.mark.parametrize(("backend", "command"), [ + ("xautolock", "xautolock -locknow"), + ("xss-lock", "loginctl lock-session"), +]) +def test_lock_and_stub_methods(sb: Sandbox, backend: str, command: str) -> None: + sb.set_backend(backend) + sb.start() + a = sb.client() + before = sb.calls() + a.call("Lock") + assert sb.calls() == [*before, command] + assert a.call("GetActive", reply="(b)") == (False,) + assert a.call("GetSessionIdleTime", reply="(u)") == (0,) + + +def test_stop_unblocks(sb: Sandbox) -> None: + sb.start(*DELAY) + sb.client().inhibit() + assert sb.stop() == 0 + assert sb.calls() == [ENABLE, DISABLE, ENABLE] + + +# --- restoring after a restart ------------------------------------------------- + + +def test_restart_restores_inhibits(sb: Sandbox) -> None: + sb.start(*DELAY) + a = sb.client() + old = a.inhibit() + assert sb.stop() == 0 + sb.start(*DELAY) + assert sb.calls() == [ENABLE, DISABLE, ENABLE, DISABLE] + new = a.inhibit() + assert new == old + 1 # numbering continues, so a late UnInhibit can't hit a new cookie + a.uninhibit(old) + a.uninhibit(new) + sb.wait_calls([ENABLE, DISABLE, ENABLE, DISABLE, ENABLE]) + + +def test_restore_skips_senders_that_left(sb: Sandbox) -> None: + sb.start() + a, probe = sb.client(), sb.client() + a.inhibit() + sb.stop() + a.close() + wait_for(lambda: not probe.has_owner(a.name), "the client to leave the bus") + sb.start() + assert sb.calls() == [ENABLE, DISABLE, ENABLE, ENABLE] + assert f"({a.name} left the bus)" in sb.log() + + +def test_no_restore_outside_window(sb: Sandbox) -> None: + sb.start() + sb.client().inhibit() + sb.stop() + sb.start("--restore-window", "0") + assert sb.calls()[-1] == ENABLE + assert "over the 0s window" in sb.log() + + +def test_no_restore_from_another_bus(sb: Sandbox) -> None: + sb.start() + sb.client().inhibit() + sb.stop() + other = sb.new_bus() + sb.start(bus=other) + assert sb.calls()[-1] == ENABLE + assert "saved on another session bus" in sb.log() + assert sb.client(other).inhibit() == 1 # cookie numbering is per bus + + +def test_crash_restores_by_mtime(sb: Sandbox) -> None: + sb.start() + sb.client().inhibit() + sb.stop(signal.SIGKILL) + assert json.loads(sb.state_file.read_text())["stopped_at"] is None + sb.start() + assert sb.calls() == [ENABLE, DISABLE, DISABLE] + + +def test_old_crash_is_not_restored(sb: Sandbox) -> None: + sb.start() + sb.client().inhibit() + sb.stop(signal.SIGKILL) + past = time.time() - 60 + os.utime(sb.state_file, (past, past)) + sb.start() + assert sb.calls()[-1] == ENABLE + assert "over the 30s window" in sb.log() + + +def test_old_state_file_is_read_once_and_deleted(sb: Sandbox) -> None: + a = sb.client() + old = sb.runtime / OLD_STATE_FILE + old.write_text(json.dumps({ + "bus_id": a.conn.get_guid(), "last_cookie": 7, "stopped_at": time.time(), + "cookies": [[7, a.name, "test-app", "Playing video"]], + })) + sb.start() + assert sb.calls() == [DISABLE] + assert "(xautolock-inhibit)" in sb.log() + assert not old.exists() + assert json.loads(sb.state_file.read_text())["cookies"] == [[7, a.name, "test-app", "Playing video"]] + assert a.inhibit() == 8 + + +def test_unreadable_state_file_is_ignored(sb: Sandbox) -> None: + sb.state_file.write_text("{not json") + sb.start() + assert sb.calls() == [ENABLE] + assert "Ignoring unreadable state file" in sb.log() + assert json.loads(sb.state_file.read_text())["cookies"] == [] + + +# --- startup and cleanup ------------------------------------------------------- + + +def test_exits_2_if_name_is_owned(sb: Sandbox) -> None: + other = sb.client() + do_not_queue = 4 + reply = other.call( + "RequestName", GLib.Variant("(su)", (BUS_NAME, do_not_queue)), "(u)", + "/org/freedesktop/DBus", "org.freedesktop.DBus", "org.freedesktop.DBus", + ) + assert reply == (1,) # primary owner + proc = sb.start(wait=False) + assert proc.wait(TIMEOUT) == 2 + assert sb.calls() == [] + assert not sb.state_file.exists() + + +@pytest.mark.parametrize(("backend", "xautolock_installed", "expected"), [ + ("xautolock", True, [ENABLE]), + ("'xss-lock'", True, []), + (None, True, [ENABLE]), # both installed: xautolock + (None, False, []), # only xss-lock + ("bogus", False, []), +]) +def test_cleanup(sb: Sandbox, backend: str | None, xautolock_installed: bool, expected: list[str]) -> None: + sb.set_backend(backend) + if not xautolock_installed: + (sb.bin / "xautolock").unlink() + result = subprocess.run( + [sys.executable, str(SCRIPT), "--cleanup"], env=sb.environ(sb.bus), + capture_output=True, text=True, timeout=TIMEOUT, + ) + assert result.returncode == 0 + assert sb.calls() == expected + if backend == "bogus": + assert "unknown backend 'bogus'" in result.stdout