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.
This commit is contained in:
Daniele Fucini
@@ -2,3 +2,4 @@ CLAUDE.md
|
||||
__pycache__/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
.pytest_cache/
|
||||
|
||||
+4
-1
@@ -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"]
|
||||
|
||||
@@ -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 <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_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
|
||||
Reference in New Issue
Block a user