Drop reading of the pre-rename xautolock-inhibit.json state file

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

@@ -2,3 +2,4 @@ CLAUDE.md
 __pycache__/
 .mypy_cache/
 .ruff_cache/
+.pytest_cache/

PKGBUILD

@@ -3,7 +3,7 @@
 # Maintainer: Daniele Fucini <dfucini [at] gmail [dot] com>
 
 pkgname=xidle-inhibit
-pkgver=0.3
+pkgver=0.4
 pkgrel=1
 pkgdesc='Honor org.freedesktop.ScreenSaver inhibit requests by blocking xautolock or xss-lock'
 arch=('any')
@@ -20,12 +20,13 @@ conflicts=('xautolock-inhibit')
 replaces=('xautolock-inhibit')
 source=("git+$url?signed#tag=$pkgver")
 validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
-sha256sums=('13390705ad8e0abea5136dbe57f21b7d9509681ee8628109dd0a302497188fbf')
+sha256sums=('fcc12e899f80b596d854049704b1b2c0293bf73b61f837b26660622044a7b877')
 
 package()
 {
     cd "$srcdir"/"$pkgname"
     install -Dm755 "$pkgname" "$pkgdir"/usr/bin/"$pkgname"
+    install -Dm755 xidle-dim "$pkgdir"/usr/bin/xidle-dim
     install -Dm644 "$pkgname".service "$pkgdir"/usr/lib/systemd/user/"$pkgname".service
     install -Dm644 README.md "$pkgdir"/usr/share/doc/"$pkgname"/README.md
 }

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"]

tests/test_daemon.py

@@ -0,0 +1,432 @@
+"""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"
+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 <servicedir>: nothing is auto-started on this bus.
+BUS_CONFIG = """<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
+ "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
+<busconfig>
+  <type>session</type>
+  <listen>unix:dir={dir}</listen>
+  <auth>EXTERNAL</auth>
+  <policy context="default">
+    <allow send_destination="*" eavesdrop="true"/>
+    <allow eavesdrop="true"/>
+    <allow own="*"/>
+  </policy>
+</busconfig>
+"""
+
+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_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

xidle-inhibit

@@ -31,7 +31,6 @@ OBJECT_PATHS = ("/org/freedesktop/ScreenSaver", "/ScreenSaver")  # Vivaldi uses
 EXIT_NAME_TAKEN = 2
 ACTIVITY_QUIET = 300  # seconds without SimulateUserActivity that end a logged burst
 STATE_FILE = "xidle-inhibit.json"  # in $XDG_RUNTIME_DIR
-OLD_STATE_FILE = "xautolock-inhibit.json"  # before the rename (0.2); read once when upgrading
 STATE_HEARTBEAT = 10  # seconds between state file touches while inhibits are held
 CONFIG_FILE = "xidle-inhibit/config"  # in $XDG_CONFIG_HOME
 BACKENDS = ("xautolock", "xss-lock")
@@ -228,14 +227,6 @@ class Inhibitor:
         self.restore()
         self.state_ready = True
         self.save()
-        if self.state_path:
-            old = os.path.join(os.path.dirname(self.state_path), OLD_STATE_FILE)
-            try:
-                os.remove(old)
-            except FileNotFoundError:
-                pass
-            except OSError as e:
-                log(f"Cannot remove old state file {old}: {e}")
         if self.active_count():
             self.block()
         elif self.xautolock:
@@ -248,9 +239,6 @@ class Inhibitor:
             return
         assert self.conn is not None
         path = self.state_path
-        old = os.path.join(os.path.dirname(path), OLD_STATE_FILE)
-        if not os.path.exists(path) and os.path.exists(old):
-            path = old
         try:
             with open(path) as f:
                 state = json.load(f)
@@ -263,8 +251,6 @@ class Inhibitor:
         except (OSError, ValueError, KeyError, TypeError) as e:
             log(f"Ignoring unreadable state file {path}: {e}")
             return
-        if path == old:
-            log(f"Reading inhibits from {old} (xautolock-inhibit)")
         if bus_id != self.conn.get_guid():
             # Unique names like :1.42 are reused by a new bus, so they would mean other clients.
             if saved: