diff --git a/gatelock/gatelock/__init__.py b/gatelock/gatelock/__init__.py index 22c05ed..b05bbf6 100644 --- a/gatelock/gatelock/__init__.py +++ b/gatelock/gatelock/__init__.py @@ -1,16 +1,77 @@ -"""Shared lock-window and HMAC log-integrity backend for blocking-overlay apps.""" +"""Shared lock-window and HMAC log-integrity backend for blocking-overlay apps. + +Used by ``screen-locker``, ``diet-guard`` and ``wake-alarm``, which all need +the same thing: cover every monitor the user can actually see, take the input +grab, and stay up until their own condition is met. +""" from __future__ import annotations +from gatelock._arbiter import ( + RANK_DIET_GUARD, + RANK_SCREEN_LOCKER, + RANK_WAKE_ALARM, + Arbiter, + ArbiterVerdict, + Claim, + default_runtime_dir, + grab_strength, +) +from gatelock._detect import OutputChangeDetector +from gatelock._guards import assert_not_under_pytest, wait_for_x_server +from gatelock._outputs import ( + Output, + OutputEnumerator, + OutputRect, + OutputScan, + RandrBackend, + enumerate_outputs, + parse_xrandr_query, +) +from gatelock._recovery import RecoveryLoop, RecoveryReport from gatelock._root import GateRoot +from gatelock._surfaces import ( + SurfaceBuilder, + SurfaceDelta, + SurfaceInfo, + SurfaceSet, + mirror_text_widgets, + needs_backdrop_root, +) from gatelock._vt import disable_vt_switching, restore_vt_switching from gatelock._window import LockConfig, LockWindow, LockWindowHooks __all__ = [ + "RANK_DIET_GUARD", + "RANK_SCREEN_LOCKER", + "RANK_WAKE_ALARM", + "Arbiter", + "ArbiterVerdict", + "Claim", "GateRoot", "LockConfig", "LockWindow", "LockWindowHooks", + "Output", + "OutputChangeDetector", + "OutputEnumerator", + "OutputRect", + "OutputScan", + "RandrBackend", + "RecoveryLoop", + "RecoveryReport", + "SurfaceBuilder", + "SurfaceDelta", + "SurfaceInfo", + "SurfaceSet", + "assert_not_under_pytest", + "default_runtime_dir", "disable_vt_switching", + "enumerate_outputs", + "grab_strength", + "mirror_text_widgets", + "needs_backdrop_root", + "parse_xrandr_query", "restore_vt_switching", + "wait_for_x_server", ] diff --git a/gatelock/gatelock/_arbiter.py b/gatelock/gatelock/_arbiter.py new file mode 100644 index 0000000..73cb817 --- /dev/null +++ b/gatelock/gatelock/_arbiter.py @@ -0,0 +1,397 @@ +"""Decide which of N locker apps may hold the screen. + +On 2026-07-25 two lockers armed at the same boot. One won the X grab; the +other spun forever in a retry loop, logging that "a fullscreen game" held the +grab. No game was running -- the holder was the other locker. Nothing +coordinated them because nothing could: each app only knew about itself. + +This module is that coordination. Every app publishes a *claim* before it +builds any window, and the highest-ranked live claim wins. Ranks, highest +first: + +* ``RANK_WAKE_ALARM`` -- you have to wake up before you can do anything else +* ``RANK_SCREEN_LOCKER`` -- then you work out +* ``RANK_DIET_GUARD`` -- then you eat + +Two properties do the real work. + +**Liveness is proved by the kernel, not by a heartbeat.** Each app holds an +``flock`` on its own claim file for its entire life. The kernel drops that lock +on *any* death, including ``SIGKILL``, OOM and a crashed X server. So a reader +that successfully locks somebody else's claim has thereby proved the owner is +dead. No PID scanning, no timeouts, no clock skew. + +**Rank can never weaken the lock.** Deferring to a higher-ranked app would be a +bypass if that app locked less firmly -- "start the alarm to disarm the workout +lock". So an app stands down only for an incumbent that is at least as strong +on *both* axes (grab kind and VT disabling). Otherwise it says so, loudly, and +arms anyway. See :meth:`Arbiter.evaluate`. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +import errno +import fcntl +import json +import logging +import os +from pathlib import Path +import tempfile +from typing import IO, Final, Literal +import uuid + +_logger = logging.getLogger(__name__) + +RANK_WAKE_ALARM: Final = 300 +RANK_SCREEN_LOCKER: Final = 200 +RANK_DIET_GUARD: Final = 100 + +_RUNTIME_DIR_ENV: Final = "GATELOCK_RUNTIME_DIR" +_CLAIMS_DIRNAME: Final = "claims" +_HOLDER_FILENAME: Final = "holder.lock" + +# Ordering for the strength check. A stand-down is only safe towards an +# incumbent whose grab is at least this strong. +_GRAB_STRENGTH: Final[dict[str, int]] = {"none": 0, "local": 1, "global": 2} + +VerdictReason = Literal["clear", "outranked", "weaker-incumbent-armed-anyway"] + + +def _utc_now_iso() -> str: + """Return an ISO-8601 UTC timestamp, sortable as a plain string.""" + return datetime.now(tz=timezone.utc).isoformat() + + +def grab_strength(grab: str) -> int: + """Return the comparable strength of a grab kind. + + Unknown values sort as the weakest possible, so a future grab kind can + never be silently treated as strong enough to stand down for. + """ + return _GRAB_STRENGTH.get(grab, 0) + + +@dataclass(frozen=True) +class Claim: + """One app's declared intent to hold the screen.""" + + app: str + rank: int + pid: int + started: str + grab: str + disable_vt: bool + instance_id: str + """Identifies the *arbiter*, not the process. + + A pid is not unique enough: one process may legitimately host two + arbiters, and the whole test suite does exactly that to exercise real + ``flock`` contention without spawning subprocesses. + """ + + def to_json(self) -> str: + """Serialise this claim for its on-disk file.""" + return json.dumps( + { + "app": self.app, + "rank": self.rank, + "pid": self.pid, + "started": self.started, + "grab": self.grab, + "disable_vt": self.disable_vt, + "instance_id": self.instance_id, + } + ) + + @classmethod + def from_json(cls, text: str) -> Claim | None: + """Parse a claim file's contents, or None if it is unusable. + + A malformed claim is treated as absent rather than fatal: a half-written + file must never be able to stop a locker from arming. + """ + try: + raw = json.loads(text) + except json.JSONDecodeError: + return None + if not isinstance(raw, dict): + return None + try: + return cls( + app=str(raw["app"]), + rank=int(raw["rank"]), + pid=int(raw["pid"]), + started=str(raw["started"]), + grab=str(raw["grab"]), + disable_vt=bool(raw["disable_vt"]), + instance_id=str(raw["instance_id"]), + ) + except (KeyError, TypeError, ValueError): + return None + + def at_least_as_strong_as(self, other: Claim) -> bool: + """Whether this claim locks at least as firmly as ``other``.""" + return ( + grab_strength(self.grab) >= grab_strength(other.grab) + and self.disable_vt >= other.disable_vt + ) + + def sort_key(self) -> tuple[int, str, int]: + """Rank descending, then earliest start, then lowest pid.""" + return (-self.rank, self.started, self.pid) + + +@dataclass(frozen=True) +class ArbiterVerdict: + """Whether this app may arm, and why.""" + + may_arm: bool + blocked_by: Claim | None + reason: VerdictReason + + +def default_runtime_dir() -> Path: + """Return the directory holding claims and the holder lock. + + Prefers ``$XDG_RUNTIME_DIR`` (tmpfs, so it self-clears on reboot) and falls + back to a per-uid temp directory when that is unset or unwritable. + """ + override = os.environ.get(_RUNTIME_DIR_ENV) + if override: + return Path(override) + xdg = os.environ.get("XDG_RUNTIME_DIR") + if xdg: + return Path(xdg) / "gatelock" + return Path(tempfile.gettempdir()) / f"gatelock-{os.getuid()}" + + +def _try_lock(handle: IO[str]) -> bool: + """Try to take an exclusive non-blocking flock. True if acquired.""" + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno in {errno.EACCES, errno.EAGAIN}: + return False + raise + return True + + +def _same_file(handle: IO[str], path: Path) -> bool: + """Whether ``path`` still names the file behind ``handle``. + + Guards the reap below: between locking a dead app's claim and unlinking it, + a fresh process may have recreated the same path. Comparing inodes stops us + deleting the newcomer's claim. + """ + try: + return os.fstat(handle.fileno()).st_ino == path.stat().st_ino + except OSError: + return False + + +class Arbiter: + """Publishes this app's claim and reports whether it may arm.""" + + def __init__( + self, + app: str, + rank: int, + *, + grab: str, + disable_vt: bool, + runtime_dir: Path | None = None, + ) -> None: + """Prepare an arbiter for one app. + + Args: + app: Human-readable app name, used in logs. + rank: Priority; higher wins. Use the ``RANK_*`` constants. + grab: The grab kind this app will take (``none``/``local``/``global``). + disable_vt: Whether this app will disable VT switching. + runtime_dir: Override for the state directory. Tests pass a tmp_path. + """ + self._root = runtime_dir if runtime_dir is not None else default_runtime_dir() + self._claims_dir = self._root / _CLAIMS_DIRNAME + self._claim = Claim( + app=app, + rank=rank, + pid=os.getpid(), + started=_utc_now_iso(), + grab=grab, + disable_vt=disable_vt, + instance_id=uuid.uuid4().hex, + ) + name = f"{rank:04d}-{self._claim.pid}-{self._claim.instance_id}.json" + self._claim_path = self._claims_dir / name + self._claim_handle: IO[str] | None = None + self._holder_handle: IO[str] | None = None + + @property + def claim(self) -> Claim: + """This app's own claim.""" + return self._claim + + @property + def holds_screen(self) -> bool: + """Whether this arbiter currently owns the holder lock.""" + return self._holder_handle is not None + + def publish(self) -> None: + """Write and lock this app's claim, so others can see it is alive.""" + self._claims_dir.mkdir(parents=True, exist_ok=True) + handle = self._claim_path.open("a+", encoding="utf-8") + if not _try_lock(handle): + # Only reachable if this pid's claim file is already locked, which + # would mean two Arbiters in one process. Close and carry on + # unpublished rather than dying. + handle.close() + _logger.warning( + "could not lock own claim %s; continuing unpublished", + self._claim_path, + ) + return + handle.seek(0) + handle.truncate() + handle.write(self._claim.to_json()) + handle.flush() + self._claim_handle = handle + + def live_claims(self) -> tuple[Claim, ...]: + """Return every claim whose owning process is still alive. + + Dead owners' claims are reaped as a side effect -- proving a claim is + stale and deleting it are the same operation. + """ + if not self._claims_dir.is_dir(): + return () + claims: list[Claim] = [] + for path in sorted(self._claims_dir.glob("*.json")): + claim = self._read_if_live(path) + if claim is not None: + claims.append(claim) + return tuple(sorted(claims, key=Claim.sort_key)) + + def _read_if_live(self, path: Path) -> Claim | None: + """Return the claim at ``path`` if its owner is alive; else reap it.""" + try: + handle = path.open("r+", encoding="utf-8") + except OSError: + return None + with handle: + if _try_lock(handle): + # We got the lock, so nobody holds it, so the owner is gone. + self._reap(handle, path) + return None + try: + return Claim.from_json(handle.read()) + except OSError: + return None + + def _reap(self, handle: IO[str], path: Path) -> None: + """Delete a claim whose owner has died, unless it was recreated.""" + if not _same_file(handle, path): + _logger.debug("claim %s was recreated while reaping; leaving it", path) + return + try: + path.unlink() + except OSError as exc: + _logger.debug("could not reap stale claim %s: %s", path, exc) + + def evaluate(self) -> ArbiterVerdict: + """Decide whether this app may arm. + + Standing down is only permitted for a *stronger or equal* incumbent. + Deferring to a weaker one would turn rank into a bypass, so that case + arms anyway and says so at ERROR. + """ + mine = self._claim.instance_id + others = [c for c in self.live_claims() if c.instance_id != mine] + stronger_rank = [c for c in others if c.rank > self._claim.rank] + if not stronger_rank: + return ArbiterVerdict(may_arm=True, blocked_by=None, reason="clear") + + winner = min(stronger_rank, key=Claim.sort_key) + if winner.at_least_as_strong_as(self._claim): + return ArbiterVerdict(may_arm=False, blocked_by=winner, reason="outranked") + _logger.error( + "%s (rank %d) outranks %s (rank %d) but locks more weakly " + "(grab=%s vt=%s vs grab=%s vt=%s); arming anyway rather than " + "letting rank weaken the lock", + winner.app, + winner.rank, + self._claim.app, + self._claim.rank, + winner.grab, + winner.disable_vt, + self._claim.grab, + self._claim.disable_vt, + ) + return ArbiterVerdict( + may_arm=True, + blocked_by=winner, + reason="weaker-incumbent-armed-anyway", + ) + + def acquire_holder(self) -> bool: + """Take the holder lock, marking this app as owning the screen.""" + if self._holder_handle is not None: + return True + self._root.mkdir(parents=True, exist_ok=True) + # Opened "a+", never "w": "w" truncates at open() time, which happens + # BEFORE the lock attempt. A losing contender would therefore erase the + # incumbent's claim on its way out, and the app that most needs to know + # who holds the screen would find an empty file. + handle = (self._root / _HOLDER_FILENAME).open("a+", encoding="utf-8") + if not _try_lock(handle): + handle.close() + return False + handle.seek(0) + handle.truncate() + handle.write(self._claim.to_json()) + handle.flush() + self._holder_handle = handle + return True + + def describe_holder(self) -> Claim | None: + """Return the claim of whichever app holds the screen, if any. + + Used to replace the old "a fullscreen game holds it" guess with the + truth, which on 2026-07-25 would have named diet_guard immediately. + """ + path = self._root / _HOLDER_FILENAME + if not path.is_file(): + return None + if self._holder_handle is not None: + return self._claim + try: + with path.open("r+", encoding="utf-8") as handle: + if _try_lock(handle): + # Lock was free, so no live app holds the screen. + return None + return Claim.from_json(handle.read()) + except OSError: + return None + + def release(self) -> None: + """Drop the holder lock and this app's claim. + + Must run on every exit path. ``flock`` covers a killed process for + free, but a *cleanly* exiting app has to release explicitly -- the + morning routine runs the alarm and then the workout locker as + sequential subprocesses, and a lingering claim would make the workout + locker stand down against an app that had already finished. + """ + for handle in (self._holder_handle, self._claim_handle): + if handle is not None: + try: + handle.close() + except OSError as exc: # pragma: no cover - close rarely fails + _logger.debug("ignoring error closing arbiter handle: %s", exc) + self._holder_handle = None + self._claim_handle = None + try: + self._claim_path.unlink(missing_ok=True) + except OSError as exc: + _logger.debug("could not remove own claim: %s", exc) diff --git a/gatelock/gatelock/_detect.py b/gatelock/gatelock/_detect.py new file mode 100644 index 0000000..5257779 --- /dev/null +++ b/gatelock/gatelock/_detect.py @@ -0,0 +1,183 @@ +"""Notice output changes fast, by four means at once. + +The lock must react the moment a monitor wakes up, and must not miss a monitor +going dark. No single mechanism covers both cheaply, so all of them run: + +1. **RandR events** (``python-xlib``) -- instant and authoritative, catching + every layout change including ones that leave the screen size unchanged. +2. **Tk ```` on the root** -- zero forks, fires when the X screen's + bounding box changes. Exactly the signal a monitor coming back produces. +3. **``xrandr --query`` polling** -- the backstop, driven by + :mod:`gatelock._recovery`, for when ``python-xlib`` is absent. +4. **Tk's own screen dimensions** -- the last-resort floor in + :mod:`gatelock._outputs`. + +Only the first two live here; they are the *push* signals, coalesced into one +"something changed" flag that the recovery loop drains. + +**Threading rule, enforced by a test:** Tk is not thread-safe, and a Tk call +from the RandR thread is a hard crash or silent memory corruption. The thread +may only call ``queue.put``. Every Tk touch happens on the drain tick, on the +Tk thread. +""" + +from __future__ import annotations + +import contextlib +import logging +import queue +import select +import threading +import tkinter as tk +from typing import Any + +_logger = logging.getLogger(__name__) + +_SELECT_TIMEOUT_S = 0.5 +"""How long the RandR thread blocks before re-checking the stop flag.""" + + +class _RandrEventSource: + """Watches RandR notifications on a private X connection. + + Private on purpose: Xlib connections are not safe to share across threads, + and :mod:`gatelock._outputs` is using its own on the Tk thread. + """ + + def __init__(self, sink: queue.Queue[str]) -> None: + """Prepare a source that posts change notices into ``sink``.""" + self._sink = sink + self._display: Any = None + self._thread: threading.Thread | None = None + self._stop = threading.Event() + + def start(self) -> bool: + """Begin watching. False if python-xlib is unavailable or refuses.""" + if not self._connect(): + return False + self._thread = threading.Thread( + target=self._run, name="gatelock-randr", daemon=True + ) + self._thread.start() + return True + + def _connect(self) -> bool: + """Open the connection and subscribe to layout changes.""" + try: + from Xlib import display as xdisplay + from Xlib.ext import randr + except ImportError: + _logger.info( + "python-xlib is not installed; falling back to Tk " + "plus xrandr polling for output-change detection" + ) + return False + try: + self._display = xdisplay.Display() + root = self._display.screen().root + root.xrandr_select_input( + randr.RRScreenChangeNotifyMask + | randr.RRCrtcChangeNotifyMask + | randr.RROutputChangeNotifyMask + ) + except (OSError, ValueError, AttributeError, TypeError) as exc: + _logger.warning("could not subscribe to RandR events: %s", exc) + self._close() + return False + return True + + def _run(self) -> None: + """Drain X events until stopped. Touches nothing but the queue.""" + try: + self._loop() + except (OSError, ValueError) as exc: # pragma: no cover - thread teardown + _logger.debug("RandR watch thread ended: %s", exc) + finally: + self._close() + + def _loop(self) -> None: + """Block on the X socket, posting a notice per batch of events.""" + fileno = self._display.fileno() + while not self._stop.is_set(): + readable, _, _ = select.select([fileno], [], [], _SELECT_TIMEOUT_S) + if not readable: + continue + pending = self._display.pending_events() + for _ in range(pending): + self._display.next_event() + if pending: + # One notice per batch: the drain tick re-scans regardless of + # how many events arrived, so counting them buys nothing. + self._sink.put("randr") + + def stop(self) -> None: + """Ask the thread to finish and wait briefly for it.""" + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=_SELECT_TIMEOUT_S * 4) + self._thread = None + + def _close(self) -> None: + """Drop the private X connection.""" + if self._display is None: + return + with contextlib.suppress(OSError, ValueError, AttributeError): + self._display.close() + self._display = None + + +class OutputChangeDetector: + """Coalesces every push signal into one drainable "something changed" flag.""" + + def __init__(self, root: tk.Misc) -> None: + """Prepare a detector bound to ``root``; subscribes nothing yet.""" + self._root = root + self._queue: queue.Queue[str] = queue.Queue() + self._randr = _RandrEventSource(self._queue) + self._configure_pending = False + self._binding: str | None = None + self._randr_active = False + + @property + def randr_active(self) -> bool: + """Whether RandR events are live (so polling can be less eager).""" + return self._randr_active + + def start(self) -> None: + """Subscribe to Tk ```` and, if possible, RandR events.""" + self._binding = self._root.bind("", self._on_configure, add="+") + self._randr_active = self._randr.start() + _logger.info( + "output-change detection active (randr_events=%s, tk_configure=True)", + self._randr_active, + ) + + def _on_configure(self, _event: tk.Event[tk.Misc]) -> None: + """Record that the X screen's geometry changed. + + No debounce needed: this only raises a flag, and the drain tick + coalesces. A Configure storm sets one boolean many times, and the + surface diff downstream turns an unchanged layout into a no-op. + """ + self._configure_pending = True + + def take_pending(self) -> bool: + """Consume and report whether any change signal has arrived.""" + pending = self._configure_pending + self._configure_pending = False + while True: + try: + self._queue.get_nowait() + except queue.Empty: + break + pending = True + return pending + + def stop(self) -> None: + """Unsubscribe from everything.""" + self._randr.stop() + self._randr_active = False + if self._binding is not None: + with contextlib.suppress(tk.TclError): + self._root.unbind("", self._binding) + self._binding = None diff --git a/gatelock/gatelock/_guards.py b/gatelock/gatelock/_guards.py new file mode 100644 index 0000000..7e015e2 --- /dev/null +++ b/gatelock/gatelock/_guards.py @@ -0,0 +1,112 @@ +"""Safety guards shared by every locker app. + +Both guards here were duplicated across the consuming repos (and one of them +was missing from ``wake_alarm`` entirely, which is exactly the kind of drift a +shared library exists to stop). + +The second one, :func:`wait_for_x_server`, changed meaning when it moved here, +and the change is the point. Its ancestor gated *whether to arm at all*: on a +boot where the display was not ready it logged "will retry on the next timer +tick" and exited without locking. Combined with a monitor that came up +modeless, that left the machine unlocked until the next tick. + +So the rule now is: **output count never gates arming.** Zero live outputs +means lock without showing, not decline to lock. This function may only answer +"can we talk to an X server at all", because without one there is no Tk and +genuinely nothing to do. It must never ask how many outputs are live. +""" + +from __future__ import annotations + +import logging +import sys +import time +import tkinter as tk +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + +_logger = logging.getLogger(__name__) + +_DEFAULT_TIMEOUT_S = 60.0 +_DEFAULT_INTERVAL_S = 1.0 + + +def assert_not_under_pytest(what: str) -> None: + """Refuse to build a real lock window inside a test run. + + A lock window takes a global input grab and covers every screen. One built + by accident during a test run would black out the developer's machine, so + this fails loudly instead. + + Call this from an app's entry point, never from inside gatelock: gatelock's + own tests mock the Tk *root* rather than the ``tkinter`` module, so the + check would fire throughout its own suite. + + Args: + what: Name of the thing being built, for the error message. + + Raises: + RuntimeError: If pytest is running with a real ``tkinter``. + """ + if "pytest" not in sys.modules: + return + if getattr(tk, "__name__", "") != "tkinter": + # tkinter is mocked, so no real window can appear. This is the normal + # path for a properly-isolated test. + return + message = ( + f"refusing to build {what} under pytest with real tkinter -- " + "this would grab the input of the machine running the tests" + ) + raise RuntimeError(message) + + +def wait_for_x_server( + *, + timeout_s: float = _DEFAULT_TIMEOUT_S, + interval_s: float = _DEFAULT_INTERVAL_S, + sleep: Callable[[float], None] = time.sleep, + monotonic: Callable[[], float] = time.monotonic, + probe: Callable[[], bool] | None = None, +) -> bool: + """Wait until an X server will accept a connection. + + Absorbs the cold-boot race where a unit starts before the display manager + has finished. It asks one question only -- is there an X server -- and + deliberately says nothing about whether any monitor is lit. A dark screen + is a reason to lock silently, never a reason to skip locking. + + Args: + timeout_s: Give up after this long. + interval_s: Seconds between attempts. + sleep: Injected for tests. + monotonic: Injected for tests. + probe: Injected for tests; defaults to opening a throwaway Tk root. + + Returns: + True if an X server answered, False on timeout. + """ + attempt = probe if probe is not None else _probe_x_server + deadline = monotonic() + timeout_s + while True: + if attempt(): + return True + if monotonic() >= deadline: + _logger.error( + "no X server answered within %.0fs; cannot build a lock window", + timeout_s, + ) + return False + sleep(interval_s) + + +def _probe_x_server() -> bool: + """Whether a throwaway Tk root can be created and destroyed.""" + try: + root = tk.Tk() + except tk.TclError: + return False + root.destroy() + return True diff --git a/gatelock/gatelock/_outputs.py b/gatelock/gatelock/_outputs.py new file mode 100644 index 0000000..a617217 --- /dev/null +++ b/gatelock/gatelock/_outputs.py @@ -0,0 +1,401 @@ +"""Enumerate X11 outputs and tell live ones apart from dark ones. + +The distinction this module exists to make is ``connected`` versus **live**. +On 2026-07-25 a monitor came up ``connected``, with a valid EDID, and no +mode/CRTC -- physically dark, while X still reserved its slice of the root +window. A lock window sized from ``winfo_screenwidth()`` therefore spanned a +region the user could not see, and the escape UI was rendered into the dark. + +So: an output is **live** only when it is connected *and* has a mode. Anything +else is a region no human can look at. + +Two backends compute the same predicate: + +* **RandR** via ``python-xlib`` -- authoritative. ``crtc == 0`` or ``mode == 0`` + *is* "connected but modeless". +* **``xrandr --query``** text -- a re-derivation of the same predicate for + systems without ``python-xlib``. + +``python-xlib`` is an optional dependency, so its import lives inside a +function rather than at module scope. That is deliberate and load-bearing: a +module-scope ``try/except ImportError`` is resolved at import time, which makes +the ``except`` arm permanently unreachable for branch coverage -- and this is +precisely the degradation path that must stay tested. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import logging +import re +import shutil +import subprocess +from typing import TYPE_CHECKING, Literal, Protocol + +if TYPE_CHECKING: + from collections.abc import Sequence + import tkinter as tk + + # python-xlib ships no type stubs. Rather than typing its objects as ``Any`` + # (or reaching for a suppression comment, which this repo bans), these + # protocols spell out the exact surface gatelock depends on. If a future + # python-xlib renames one of these, the type checker says so. + class _ScreenResources(Protocol): + config_timestamp: int + outputs: Sequence[int] + + class _OutputInfo(Protocol): + name: object + connection: int + crtc: int + + class _CrtcInfo(Protocol): + mode: int + x: int + y: int + width: int + height: int + + class _PrimaryReply(Protocol): + output: int + + class _XRoot(Protocol): + def xrandr_get_screen_resources(self) -> _ScreenResources: ... + + def xrandr_get_output_primary(self) -> _PrimaryReply: ... + + class _XScreen(Protocol): + root: _XRoot + + class _XDisplay(Protocol): + def screen(self) -> _XScreen: ... + + def xrandr_get_output_info( + self, output: int, timestamp: int + ) -> _OutputInfo: ... + + def xrandr_get_crtc_info(self, crtc: int, timestamp: int) -> _CrtcInfo: ... + + def close(self) -> None: ... + + +_logger = logging.getLogger(__name__) + +_XRANDR_TIMEOUT_S = 5.0 + +# RandR's `connection` field: 0 is RR_Connected. Hard-coded rather than +# imported so the constant is available without python-xlib installed. +_RR_CONNECTED = 0 + +# Output lines start at column 0; mode lines are indented four spaces. That +# anchor is the whole robustness story -- without it, the mode line +# " 2560x1440 59.95*+" parses as a geometry and every modeless output +# silently looks live. +_OUTPUT_LINE = re.compile( + r"^(?P\S+)\s+(?Pconnected|disconnected)\b(?P.*)$", + re.MULTILINE, +) +_GEOMETRY = re.compile(r"(?P\d+)x(?P\d+)\+(?P-?\d+)\+(?P-?\d+)") + +OutputSource = Literal["randr", "xrandr", "tk", "none"] +"""Which backend produced a scan. Reported so logs can explain a degradation.""" + + +@dataclass(frozen=True) +class OutputRect: + """A rectangle in root-window coordinates.""" + + x: int + y: int + width: int + height: int + + def geometry(self) -> str: + """Return the Tk geometry string for this rectangle.""" + return f"{self.width}x{self.height}+{self.x}+{self.y}" + + +@dataclass(frozen=True) +class Output: + """A single X output and whether anything can actually be seen on it.""" + + name: str + connected: bool + rect: OutputRect | None + primary: bool = False + + @property + def live(self) -> bool: + """Whether this output is connected *and* has a mode.""" + return self.connected and self.rect is not None + + +@dataclass(frozen=True) +class OutputScan: + """The result of one enumeration attempt. + + ``ok`` records whether enumeration itself succeeded, which is a different + question from whether any output is live. Callers must not conflate them: + a failed scan means "no information, change nothing", while a successful + scan with zero live outputs means "the screens really are dark". + """ + + outputs: tuple[Output, ...] + source: OutputSource + ok: bool + + @property + def live(self) -> tuple[Output, ...]: + """The subset of outputs that are connected and have a mode.""" + return tuple(output for output in self.outputs if output.live) + + +def _parse_rect(head: str) -> OutputRect | None: + """Parse a geometry token out of an xrandr output line's head segment.""" + match = _GEOMETRY.search(head) + if match is None: + return None + width = int(match.group("w")) + height = int(match.group("h")) + if width <= 0 or height <= 0: + return None + return OutputRect( + x=int(match.group("x")), + y=int(match.group("y")), + width=width, + height=height, + ) + + +def parse_xrandr_query(text: str) -> tuple[Output, ...]: + """Parse ``xrandr --query`` output into outputs. + + Args: + text: Raw stdout from ``xrandr --query``. + + Returns: + One :class:`Output` per output line, in the order xrandr listed them. + """ + outputs: list[Output] = [] + for match in _OUTPUT_LINE.finditer(text): + # Everything after the first "(" is rotation/reflection vocabulary and + # panning noise -- never geometry. Slicing it off keeps the search + # honest. + head = match.group("rest").split("(", 1)[0] + outputs.append( + Output( + name=match.group("name"), + connected=match.group("state") == "connected", + rect=_parse_rect(head), + primary="primary" in head.split(), + ) + ) + return tuple(outputs) + + +def _decode_name(raw: object) -> str: + """Normalise a RandR output name, which may arrive as bytes.""" + if isinstance(raw, bytes): + return raw.decode("utf-8", errors="replace") + return str(raw) + + +class RandrBackend: + """A persistent X connection used to enumerate outputs via RandR. + + Held open across scans -- the recovery loop scans every second, and opening + a fresh X connection each time would be pure waste. + """ + + def __init__(self, display: _XDisplay) -> None: + """Wrap an already-opened ``Xlib.display.Display``.""" + self._display = display + + @classmethod + def create(cls) -> RandrBackend | None: + """Open an X connection for RandR queries, or None if unavailable.""" + try: + from Xlib import display as xdisplay + except ImportError: + _logger.debug( + "python-xlib is not installed; RandR events and RandR " + "enumeration are unavailable, falling back to xrandr" + ) + return None + try: + return cls(xdisplay.Display()) + except (OSError, ValueError) as exc: + _logger.warning("could not open an X display for RandR: %s", exc) + return None + + def close(self) -> None: + """Close the underlying X connection, ignoring teardown errors.""" + try: + self._display.close() + except (OSError, ValueError, AttributeError) as exc: + _logger.debug("ignoring error while closing RandR display: %s", exc) + + def scan(self) -> tuple[Output, ...] | None: + """Enumerate outputs, or None if the RandR query failed.""" + try: + return self._scan() + except (OSError, ValueError, KeyError, AttributeError, TypeError) as exc: + _logger.warning("RandR enumeration failed (%s); falling back", exc) + return None + + def _scan(self) -> tuple[Output, ...]: + """Enumerate outputs via RandR. Raises on any protocol problem.""" + root = self._display.screen().root + resources = root.xrandr_get_screen_resources() + timestamp = resources.config_timestamp + primary_id = self._primary_output_id(root) + + outputs: list[Output] = [] + for output_id in resources.outputs: + info = self._display.xrandr_get_output_info(output_id, timestamp) + outputs.append( + Output( + name=_decode_name(info.name), + connected=info.connection == _RR_CONNECTED, + rect=self._crtc_rect(info.crtc, timestamp), + primary=output_id == primary_id, + ) + ) + return tuple(outputs) + + def _primary_output_id(self, root: _XRoot) -> int | None: + """Return the primary output's id, or None when RandR won't say.""" + try: + return root.xrandr_get_output_primary().output + except (OSError, ValueError, KeyError, AttributeError, TypeError): + return None + + def _crtc_rect(self, crtc: int, timestamp: int) -> OutputRect | None: + """Return the CRTC's rectangle, or None when there is no mode. + + ``crtc == 0`` means no CRTC is assigned and ``mode == 0`` means a CRTC + exists without a mode. Either one is the dark-monitor case. + """ + if not crtc: + return None + info = self._display.xrandr_get_crtc_info(crtc, timestamp) + if not info.mode or info.width <= 0 or info.height <= 0: + return None + return OutputRect(x=info.x, y=info.y, width=info.width, height=info.height) + + +def scan_xrandr() -> tuple[Output, ...] | None: + """Enumerate outputs by shelling out to ``xrandr --query``. + + Returns: + The parsed outputs, or None if xrandr is missing, failed, timed out or + produced nothing parseable. + """ + binary = shutil.which("xrandr") + if binary is None: + _logger.warning("xrandr is not on PATH; cannot enumerate outputs") + return None + try: + completed = subprocess.run( + [binary, "--query"], + capture_output=True, + text=True, + timeout=_XRANDR_TIMEOUT_S, + check=False, + ) + except (subprocess.TimeoutExpired, OSError) as exc: + _logger.warning("xrandr --query failed to run: %s", exc) + return None + if completed.returncode != 0: + _logger.warning( + "xrandr --query exited %d: %s", + completed.returncode, + completed.stderr.strip(), + ) + return None + outputs = parse_xrandr_query(completed.stdout) + if not outputs: + _logger.warning("xrandr --query produced no parseable output lines") + return None + return outputs + + +def tk_fallback_outputs(root: tk.Misc) -> tuple[Output, ...]: + """Synthesise one output covering the whole X screen. + + This is v0.1.1's behaviour, kept as a floor. It cannot tell a dark region + from a live one, so it is strictly a last resort -- but a lock covering the + full bounding box beats a lock covering nothing. + """ + width = int(root.winfo_screenwidth()) + height = int(root.winfo_screenheight()) + return ( + Output( + name="tk-screen", + connected=True, + rect=OutputRect(x=0, y=0, width=width, height=height), + ), + ) + + +class OutputEnumerator: + """Runs the backend ladder: RandR, then xrandr, then Tk, then nothing.""" + + def __init__( + self, + root: tk.Misc | None = None, + *, + backend: RandrBackend | None = None, + ) -> None: + """Build an enumerator, opening a RandR connection when possible. + + Args: + root: A Tk widget used only for the last-resort full-screen + fallback. Without it, that rung is skipped. + backend: A pre-opened RandR backend. Defaults to opening one. + """ + self._root = root + self._backend = backend if backend is not None else RandrBackend.create() + + @property + def backend(self) -> RandrBackend | None: + """The RandR backend, or None when python-xlib is unavailable.""" + return self._backend + + def close(self) -> None: + """Release the RandR connection, if any.""" + if self._backend is not None: + self._backend.close() + self._backend = None + + def scan(self) -> OutputScan: + """Enumerate outputs, degrading through the backend ladder.""" + if self._backend is not None: + outputs = self._backend.scan() + if outputs is not None: + return OutputScan(outputs=outputs, source="randr", ok=True) + outputs = scan_xrandr() + if outputs is not None: + return OutputScan(outputs=outputs, source="xrandr", ok=True) + if self._root is not None: + return OutputScan( + outputs=tk_fallback_outputs(self._root), source="tk", ok=True + ) + _logger.error( + "every output-enumeration backend failed; reporting a failed scan " + "so callers leave the lock exactly as it is" + ) + return OutputScan(outputs=(), source="none", ok=False) + + +def enumerate_outputs(root: tk.Misc | None = None) -> OutputScan: + """Enumerate outputs once, using a throwaway connection. + + Convenience for one-shot callers (CLI probes, tests). Long-lived callers + should hold an :class:`OutputEnumerator` so the X connection is reused. + """ + enumerator = OutputEnumerator(root) + try: + return enumerator.scan() + finally: + enumerator.close() diff --git a/gatelock/gatelock/_recovery.py b/gatelock/gatelock/_recovery.py new file mode 100644 index 0000000..15161d5 --- /dev/null +++ b/gatelock/gatelock/_recovery.py @@ -0,0 +1,228 @@ +"""Keep the lock covering every live output, and never weaken it. + +This loop is the watchdog the 2026-07-25 incident needed. A monitor that comes +back must show the lock immediately; a monitor that goes dark must not take the +lock's only escape UI with it. + +Monotonicity +------------ + +Let ``C`` be the set of live outputs carrying a correctly-placed mapped +surface, ``G`` whether the global grab is held, and ``V`` whether VT switching +is disabled. Every tick satisfies: + + No transition decreases ``G`` or ``V``, and no transition removes an + element of ``C`` whose output is still live. + +The proof is by exhaustion over the operations a tick can perform: create a +surface, move one, remap one, raise one, destroy the surface of an output that +is *no longer live*, re-take the grab, re-disable VT switching. None of those +is a release. ``G`` and ``V`` are only ever set. ``C`` grows except when an +output stops being live -- and such an output is outside ``C``'s domain, so +``C`` restricted to live outputs never shrinks. + +That property is *mechanical*, not aspirational. All structural change is +delegated to :mod:`gatelock._surfaces`, so this module contains none of +``grab_release``, ``restore_vt_switching``, ``close``, ``destroy``, ``quit`` or +``withdraw`` -- and a test parses this file to assert exactly that. Note the +asymmetry, which is the whole point: ``disable_vt_switching`` and +``grab_set_global`` *are* imported and called, because they strengthen. Only +their inverses are banned. + +Consequence, deliberate: with zero live outputs the loop shows nothing and +keeps the grab. Pulling a cable is therefore not an escape. +""" + +from __future__ import annotations + +import contextlib +from dataclasses import dataclass +import logging +import tkinter as tk +from typing import TYPE_CHECKING + +from gatelock._vt import disable_vt_switching + +if TYPE_CHECKING: + from gatelock._detect import OutputChangeDetector + from gatelock._outputs import OutputEnumerator + from gatelock._surfaces import SurfaceDelta, SurfaceSet + from gatelock._window import LockConfig + +_logger = logging.getLogger(__name__) + +_VT_REASSERT_EVERY = 30 +"""Re-disable VT switching every N verify ticks, in case something reset it.""" + + +@dataclass(frozen=True) +class RecoveryReport: + """What one tick observed and corrected.""" + + scan_ok: bool + source: str + live_outputs: tuple[str, ...] = () + delta: SurfaceDelta | None = None + corrected: tuple[str, ...] = () + grab_reasserted: bool = False + vt_reasserted: bool = False + blind: tuple[str, ...] = () + + +class RecoveryLoop: + """Re-asserts the lock's coverage on a timer and on change events.""" + + def __init__( + self, + root: tk.Misc, + config: LockConfig, + surfaces: SurfaceSet, + enumerator: OutputEnumerator, + detector: OutputChangeDetector, + ) -> None: + """Wire the loop to the pieces it re-asserts over.""" + self._root = root + self._config = config + self._surfaces = surfaces + self._enumerator = enumerator + self._detector = detector + self._running = False + self._ticks = 0 + self._drain_job: str | None = None + self._verify_job: str | None = None + + @property + def ticks(self) -> int: + """How many full verify passes have run.""" + return self._ticks + + def start(self) -> None: + """Begin both cadences: the cheap drain and the full verify.""" + self._running = True + self._schedule_drain() + self._schedule_verify() + + def stop(self) -> None: + """Stop rescheduling. Cancels timers; touches no window state.""" + self._running = False + for job in (self._drain_job, self._verify_job): + if job is not None: + with contextlib.suppress(tk.TclError): + self._root.after_cancel(job) + self._drain_job = None + self._verify_job = None + + def _schedule_drain(self) -> None: + """Queue the next cheap drain tick.""" + if not self._running: + return + self._drain_job = self._root.after(self._config.detect_drain_ms, self._drain) + + def _schedule_verify(self) -> None: + """Queue the next full verify tick.""" + if not self._running: + return + self._verify_job = self._root.after(self._config.recovery_tick_ms, self._verify) + + def _drain(self) -> None: + """Run a full tick only if a push signal arrived. Cheap otherwise.""" + if self._detector.take_pending(): + _logger.debug("output-change signal received; re-asserting the lock") + self.tick() + self._schedule_drain() + + def _verify(self) -> None: + """Run a full tick unconditionally.""" + self.tick() + self._schedule_verify() + + def tick(self) -> RecoveryReport: + """Re-assert coverage, placement, grab and VT. Never weakens any of them. + + Returns: + What this pass saw and corrected. + """ + self._ticks += 1 + scan = self._enumerator.scan() + + if not scan.ok: + # No information is not a reason to change anything. Keep every + # surface, keep the grab, keep VT disabled, try again next tick. + _logger.warning( + "output enumeration failed; leaving the lock exactly as it is" + ) + return RecoveryReport(scan_ok=False, source=scan.source) + + self._surfaces.update_backdrop() + delta = self._surfaces.apply(scan) + corrected = self._surfaces.enforce() + grab_reasserted = self._reassert_grab() + vt_reasserted = self._reassert_vt() + + if delta.unverified: + _logger.error( + "these outputs are live but show no lock: %s -- the escape UI " + "may be somewhere the user cannot see", + ", ".join(delta.unverified), + ) + if delta.changed: + _logger.info( + "surfaces changed (created=%s moved=%s removed=%s) from %s", + delta.created, + delta.moved, + delta.removed, + scan.source, + ) + if not scan.live: + _logger.warning( + "no live outputs; showing nothing and STAYING LOCKED " + "(grab and VT held) until one comes back" + ) + + return RecoveryReport( + scan_ok=True, + source=scan.source, + live_outputs=tuple(output.name for output in scan.live), + delta=delta, + corrected=corrected, + grab_reasserted=grab_reasserted, + vt_reasserted=vt_reasserted, + blind=delta.unverified, + ) + + def _reassert_grab(self) -> bool: + """Re-take the global grab if we lost it. Never releases one. + + Returns: + True if the grab had to be re-taken. + """ + if self._config.resolved_grab() != "global": + return False + if self._holds_grab(): + return False + with contextlib.suppress(tk.TclError): + self._root.grab_set_global() + _logger.warning("global grab had been lost; re-acquired it") + return True + return False + + def _holds_grab(self) -> bool: + """Whether our root still owns the grab.""" + try: + current = self._root.grab_current() + except tk.TclError: + return False + return current is self._root + + def _reassert_vt(self) -> bool: + """Periodically re-disable VT switching. Never re-enables it. + + Returns: + True on ticks where the re-assert ran. + """ + if not self._config.resolved_disable_vt(): + return False + if self._ticks % _VT_REASSERT_EVERY: + return False + disable_vt_switching() + return True diff --git a/gatelock/gatelock/_surfaces.py b/gatelock/gatelock/_surfaces.py new file mode 100644 index 0000000..3ac6315 --- /dev/null +++ b/gatelock/gatelock/_surfaces.py @@ -0,0 +1,380 @@ +"""Own every lock window: one per live output, plus the backdrop. + +This is the **only** module allowed to create, move or destroy windows. That +containment is what lets :mod:`gatelock._recovery` be provably monotonic -- +recovery delegates all structural change here and therefore cannot itself +contain a teardown path. A test asserts that separation by parsing both +modules, so routing destruction back into recovery fails CI. + +The window shape, and why: + +.. code-block:: text + + GateRoot (tk.Tk) <- always the grab holder, always mapped + |- overrideredirect, geometry = the whole X screen, no widgets + |- Toplevel -> "DP-0" 3840x2160+0+0 + +- Toplevel -> "HDMI-0" 2560x1440+3840+0 + +Three probe results against a live X server dictated this and are worth +recording, because each rules out an approach that looks reasonable: + +1. **A Tk grab must live on the root**, not on a Toplevel. With the grab on a + *sibling* Toplevel, clicks on another surface are swallowed. Grabbing + surface 0 would leave surface 1 visible but dead -- the 2026-07-25 incident + rebuilt on purpose. +2. **``overrideredirect`` must be set before the window is mapped.** Set it + afterwards and the placement is ignored. +3. **A window-manager-managed toplevel cannot be placed per-output** -- i3 + rewrites the geometry wholesale. Per-output placement *requires* + override-redirect, so ``attributes(fullscreen=True)`` is gone from the lock + path entirely: it is an EWMH request that an override-redirect window is + invisible to, and where it does apply it targets one monitor or the whole + bounding box. Neither is what per-output locking needs. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import logging +import tkinter as tk +from typing import TYPE_CHECKING, Protocol + +if TYPE_CHECKING: + from collections.abc import Sequence + + from gatelock._outputs import OutputRect, OutputScan + from gatelock._window import LockConfig + +_logger = logging.getLogger(__name__) + +# Tk's attributes() takes the value positionally, so a bare `True` here +# trips the boolean-positional lint. Naming it satisfies both that rule +# and the type stubs, which have no keyword overload. +_TOPMOST_ON = True + +_TEXT_START = "1.0" +_TEXT_END = "end-1c" + + +@dataclass(frozen=True) +class SurfaceInfo: + """Identifies one lock surface and the output it covers.""" + + output_name: str + rect: OutputRect + index: int + is_primary: bool + + +class SurfaceBuilder(Protocol): + """What a consuming app must implement to paint its UI on each output.""" + + def build_surface(self, parent: tk.Misc, surface: SurfaceInfo) -> None: + """Build this app's widgets inside ``parent`` for one output.""" + + def teardown_surface(self, surface: SurfaceInfo) -> None: + """Forget any per-surface state; the window is about to be destroyed.""" + + +@dataclass(frozen=True) +class SurfaceDelta: + """What one :meth:`SurfaceSet.apply` changed.""" + + created: tuple[str, ...] = () + moved: tuple[str, ...] = () + removed: tuple[str, ...] = () + unverified: tuple[str, ...] = () + + @property + def changed(self) -> bool: + """Whether anything structural happened.""" + return bool(self.created or self.moved or self.removed) + + +@dataclass +class _Surface: + """One Toplevel and the output it is pinned to.""" + + info: SurfaceInfo + window: tk.Toplevel + built: bool = field(default=False) + + +def needs_backdrop_root(config: LockConfig) -> bool: + """Whether the root must stay mapped as a backdrop and grab holder. + + Deliberately *not* simplified to "iff override-redirect". + ``LockConfig(mode="soft", grab="global")`` is a reachable combination, and + a global grab requires a viewable window -- X refuses to grab for a + withdrawn one. Collapsing this predicate would make that config fail at + runtime only. + """ + return config.resolved_overrideredirect() or config.resolved_grab() != "none" + + +class TextMirror: + """Keeps N ``tk.Text`` widgets showing the same value. + + ``tk.Text`` has no ``textvariable``, so it is the one widget that cannot + ride the shared-``StringVar`` mechanism the rest of the mirroring uses. + """ + + def __init__(self, widgets: Sequence[tk.Text], var: tk.StringVar) -> None: + """Bind ``widgets`` to each other through ``var``.""" + self._widgets = list(widgets) + self._var = var + self._syncing = False + for widget in self._widgets: + widget.bind("", self._on_key_release, add="+") + self._var.trace_add("write", self._on_var_write) + + def _on_key_release(self, event: tk.Event[tk.Text]) -> None: + """Push the edited widget's content into the shared variable.""" + if self._syncing: + return + self._syncing = True + try: + self._var.set(event.widget.get(_TEXT_START, _TEXT_END)) + finally: + self._syncing = False + self._fan_out(source=event.widget) + + def _on_var_write(self, *_args: str) -> None: + """Push the shared variable's value into every widget.""" + if self._syncing: + return + self._fan_out(source=None) + + def _fan_out(self, *, source: tk.Text | None) -> None: + """Write the variable's value into every widget but ``source``.""" + value = self._var.get() + self._syncing = True + try: + for widget in self._widgets: + if widget is source or widget.get(_TEXT_START, _TEXT_END) == value: + continue + widget.delete(_TEXT_START, "end") + widget.insert(_TEXT_START, value) + finally: + self._syncing = False + + +def mirror_text_widgets(widgets: Sequence[tk.Text], var: tk.StringVar) -> TextMirror: + """Keep several ``tk.Text`` mirrors in sync through one variable.""" + return TextMirror(widgets, var) + + +class SurfaceSet: + """The full set of lock windows, kept in step with the live outputs.""" + + def __init__( + self, + root: tk.Tk, + config: LockConfig, + builder: SurfaceBuilder, + ) -> None: + """Prepare a surface set over ``root``; creates no windows yet.""" + self._root = root + self._config = config + self._builder = builder + self._surfaces: dict[str, _Surface] = {} + + def infos(self) -> tuple[SurfaceInfo, ...]: + """Every current surface, in output order.""" + return tuple( + surface.info + for surface in sorted(self._surfaces.values(), key=lambda s: s.info.index) + ) + + def names(self) -> frozenset[str]: + """The output names currently carrying a surface.""" + return frozenset(self._surfaces) + + def update_backdrop(self) -> None: + """Size the root to the whole X screen and keep it black. + + The backdrop covers regions no live output claims -- including the dead + column a modeless monitor leaves behind -- so nothing of the desktop + shows through the gaps. + """ + if not needs_backdrop_root(self._config): + return + width = self._root.winfo_screenwidth() + height = self._root.winfo_screenheight() + self._root.geometry(f"{width}x{height}+0+0") + self._root.configure(bg=self._config.bg, cursor="arrow") + + def apply(self, scan: OutputScan) -> SurfaceDelta: + """Bring the surface set in line with ``scan``. + + A failed scan must never reach this method: callers check ``scan.ok`` + first, because "we could not enumerate" is not evidence that anything + should change. + """ + live = scan.live + wanted = {output.name: output for output in live} + + removed = tuple(name for name in sorted(self._surfaces) if name not in wanted) + for name in removed: + self._destroy(name) + + created: list[str] = [] + moved: list[str] = [] + for index, output in enumerate(live): + rect = output.rect + if rect is None: # pragma: no cover - `live` guarantees a rect + continue + existing = self._surfaces.get(output.name) + info = SurfaceInfo( + output_name=output.name, + rect=rect, + index=index, + is_primary=output.primary, + ) + if existing is None: + self._create(info) + created.append(output.name) + elif existing.info != info: + self._reposition(existing, info) + moved.append(output.name) + + return SurfaceDelta( + created=tuple(created), + moved=tuple(moved), + removed=removed, + unverified=self.verify(live_names=frozenset(wanted)), + ) + + def verify(self, *, live_names: frozenset[str] | None = None) -> tuple[str, ...]: + """Return live outputs that are NOT currently showing a lock. + + This is the alarm for the original defect: an output the user can see, + with no lock on it, means the escape UI is somewhere they cannot reach. + """ + names = live_names if live_names is not None else self.names() + blind: list[str] = [] + for name in sorted(names): + surface = self._surfaces.get(name) + if surface is None or not self._is_mapped(surface): + blind.append(name) + return tuple(blind) + + def enforce(self) -> tuple[str, ...]: + """Re-assert every surface's placement and stacking. + + Only ever *adds* visibility: remaps what became unmapped, corrects + drifted geometry, and raises. Nothing here can hide a surface. + """ + corrected = [ + surface.info.output_name + for surface in self._surfaces.values() + if self._enforce_one(surface) + ] + return tuple(sorted(corrected)) + + def _enforce_one(self, surface: _Surface) -> bool: + """Re-assert one surface. True if anything needed correcting.""" + window = surface.window + changed = False + if not self._is_mapped(surface): + window.deiconify() + changed = True + if not self._geometry_matches(surface): + self._set_geometry(window, surface.info.rect) + changed = True + # Re-assert override-redirect: it can be dropped across an unmap/map + # cycle, and a managed lock window would be moved by the WM. + if self._config.resolved_overrideredirect(): + window.overrideredirect(boolean=True) + window.lift() + return changed + + def raise_all(self) -> None: + """Lift every surface above whatever else is on screen.""" + for surface in self._surfaces.values(): + surface.window.lift() + + def focus_surface(self, index: int) -> SurfaceInfo | None: + """Move keyboard focus to one surface, returning what got it.""" + for surface in self._surfaces.values(): + if surface.info.index == index: + surface.window.focus_force() + return surface.info + return None + + def preferred_focus_index(self) -> int: + """Index of the surface that should take initial focus. + + The live primary output if there is one, else the first surface. + """ + for surface in self._surfaces.values(): + if surface.info.is_primary: + return surface.info.index + return 0 + + def _create(self, info: SurfaceInfo) -> None: + """Create one surface window and let the app build its UI inside.""" + window = tk.Toplevel(self._root) + # Order matters and is not negotiable: a window must be withdrawn while + # override-redirect is set, or the geometry below is ignored. + window.withdraw() + if self._config.resolved_overrideredirect(): + window.overrideredirect(boolean=True) + window.configure(bg=self._config.bg, cursor="arrow") + self._set_geometry(window, info.rect) + if not self._config.resolved_overrideredirect(): + window.attributes("-topmost", _TOPMOST_ON) + window.deiconify() + window.lift() + + surface = _Surface(info=info, window=window) + self._surfaces[info.output_name] = surface + self._builder.build_surface(window, info) + surface.built = True + _logger.info( + "lock surface up on %s at %s", info.output_name, info.rect.geometry() + ) + + def _reposition(self, surface: _Surface, info: SurfaceInfo) -> None: + """Move an existing surface to a new rectangle. + + Deliberately a move, never a destroy-and-recreate: a layout change must + not discard half-typed input. + """ + surface.info = info + self._set_geometry(surface.window, info.rect) + surface.window.lift() + _logger.info( + "lock surface on %s moved to %s", info.output_name, info.rect.geometry() + ) + + def _destroy(self, name: str) -> None: + """Tear down the surface for an output that is no longer live.""" + surface = self._surfaces.pop(name) + self._builder.teardown_surface(surface.info) + surface.window.destroy() + _logger.info("lock surface on %s removed; that output went dark", name) + + def destroy_all(self) -> None: + """Tear down every surface. Used only when the whole lock is closing.""" + for name in list(self._surfaces): + self._destroy(name) + + def _set_geometry(self, window: tk.Toplevel, rect: OutputRect) -> None: + """Pin a window to an output's rectangle.""" + window.geometry(rect.geometry()) + + def _is_mapped(self, surface: _Surface) -> bool: + """Whether a surface's window is currently mapped.""" + return bool(surface.window.winfo_ismapped()) + + def _geometry_matches(self, surface: _Surface) -> bool: + """Whether a surface still sits exactly on its output.""" + rect = surface.info.rect + window = surface.window + return ( + window.winfo_rootx() == rect.x + and window.winfo_rooty() == rect.y + and window.winfo_width() == rect.width + and window.winfo_height() == rect.height + ) diff --git a/gatelock/gatelock/_window.py b/gatelock/gatelock/_window.py index 2b27da9..caf5182 100644 --- a/gatelock/gatelock/_window.py +++ b/gatelock/gatelock/_window.py @@ -1,4 +1,4 @@ -"""Fullscreen lock window: setup, input grab, VT-disable, safe lifecycle. +"""Lock window orchestration: surfaces, input grab, VT-disable, safe lifecycle. Generalizes the window mechanics that wake_alarm, screen-locker, and diet_guard each implemented separately. The three differ along independent @@ -7,6 +7,19 @@ whether VT switching is disabled, how a failed global grab is retried) -- :class:`LockConfig` exposes each axis explicitly, with ``mode`` as a convenience preset, so one class can reproduce all three projects' existing behavior exactly. + +Since v0.2.0 this module only *orchestrates*. Enumeration lives in +:mod:`gatelock._outputs`, windows in :mod:`gatelock._surfaces`, change +detection in :mod:`gatelock._detect`, re-assertion in +:mod:`gatelock._recovery`, and cross-app priority in +:mod:`gatelock._arbiter`. + +Arming order matters and is deliberate: **arm first, render second.** The grab +and the VT-disable happen regardless of how many outputs are live, and the +surfaces are then built for whatever the user can actually see -- possibly +nothing. Zero live outputs means "lock without showing", never "decline to +lock". The inverse of that rule is what left the machine unlocked on +2026-07-25. """ from __future__ import annotations @@ -19,11 +32,19 @@ import signal import tkinter as tk from typing import TYPE_CHECKING, Literal, Protocol +from gatelock._arbiter import RANK_SCREEN_LOCKER +from gatelock._detect import OutputChangeDetector +from gatelock._outputs import OutputEnumerator +from gatelock._recovery import RecoveryLoop +from gatelock._surfaces import SurfaceSet, needs_backdrop_root from gatelock._vt import disable_vt_switching, restore_vt_switching if TYPE_CHECKING: from types import FrameType + from gatelock._arbiter import Arbiter + from gatelock._surfaces import SurfaceInfo + _logger = logging.getLogger(__name__) GrabKind = Literal["none", "local", "global"] @@ -31,6 +52,11 @@ LockMode = Literal["soft", "hard"] # Periodic no-op so a grabbed, event-starved loop keeps handing control back # to Python, letting SIGTERM/SIGINT be serviced promptly. +# Tk's attributes() takes the value positionally, so a bare `True` here +# trips the boolean-positional lint. Naming it satisfies both that rule +# and the type stubs, which have no keyword overload. +_TOPMOST_ON = True + _KEEPALIVE_MS = 250 # Default retry interval for a "global" grab that initially fails (e.g. a # fullscreen game holds it). Used whenever grab_retry_ms is left unset. @@ -46,20 +72,24 @@ class LockConfig: Attributes: mode: Preset bundling the common combination. "soft" = topmost only, - typeable, WM-escapable (today's wake_alarm). "hard" = - overrideredirect + global grab + VT-disable (today's diet_guard - and screen-locker production lock). + typeable, WM-escapable. "hard" = overrideredirect + global grab + + VT-disable (the production lock for all three apps). overrideredirect: Force a WM-unmanaged window. None = derive from mode. + Note that per-output placement *requires* this: a window manager + rewrites a managed window's geometry wholesale. grab: Input grab strategy. None = derive from mode. disable_vt: Disable Ctrl+Alt+Fn VT switching. None = derive from mode. grab_retry_ms: Retry interval in ms for a "global" grab that initially - fails. 0 means "try once, then fall back to a local grab" (the - original screen-locker behavior). Left unset (None), a "global" - grab retries forever every 200ms until it succeeds (the - diet_guard behavior, robust to e.g. a fullscreen game holding the - grab) -- there is no give-up/fallback in that case. + fails. 0 means "try once, then fall back to a local grab". Left + unset (None), a "global" grab retries forever every 200ms. grab_log_every: Log a warning every N failed retry-forever attempts. - bg: Background color for the root window. + app_name: This app's name, used in arbitration logs so a blocked app + can say who is actually holding the screen. + rank: Arbitration priority; higher wins. See the ``RANK_*`` constants + in :mod:`gatelock._arbiter`. + recovery_tick_ms: Interval of the full re-assertion pass. + detect_drain_ms: Interval of the cheap "did anything change?" drain. + bg: Background color for the lock surfaces. fg: Primary (near-white) text color. muted: Secondary/caption text color. field_bg: Background for "raised" surfaces (entry/spinbox fields, @@ -79,8 +109,6 @@ class LockConfig: All color/font defaults come from the ``unified-design-system`` skill (``~/.claude/skills/unified-design-system/references/tokens.md``) -- the same palette used by every one of kuhy's apps, Flutter and web included. - screen-locker, wake_alarm, and diet_guard should read these fields - instead of re-hardcoding their own hex/font literals. """ mode: LockMode = "hard" @@ -89,6 +117,10 @@ class LockConfig: disable_vt: bool | None = None grab_retry_ms: int | None = None grab_log_every: int = 25 + app_name: str = "gatelock" + rank: int = RANK_SCREEN_LOCKER + recovery_tick_ms: int = 1000 + detect_drain_ms: int = 100 bg: str = "#211D1B" fg: str = "#ECEAE9" muted: str = "#AAA09A" @@ -122,10 +154,24 @@ class LockConfig: class LockWindowHooks(Protocol): """Callbacks :class:`LockWindow` invokes; the embedding app supplies all.""" - def on_focus_ready(self) -> None: - """Called once the window is mapped and (if applicable) grabbed. + def build_surface(self, parent: tk.Misc, surface: SurfaceInfo) -> None: + """Build this app's widgets inside ``parent``, for one output. - The app should focus its first input widget here. + Called once per live output, and again whenever an output comes back. + Tk variables should stay mastered on the root so every surface shows + the same state, and every surface's submit path should call the same + handler -- solving the lock on any monitor dismisses all of them. + """ + + def teardown_surface(self, surface: SurfaceInfo) -> None: + """Forget per-surface state; that output went dark and is closing.""" + + def on_focus_ready(self, surface: SurfaceInfo | None) -> None: + """Called once the lock is mapped and (if applicable) grabbed. + + Args: + surface: The surface that took initial focus, or None when no + output is live and there is therefore nothing to focus. """ def on_callback_error(self) -> None: @@ -141,44 +187,92 @@ class LockWindowHooks(Protocol): class LockWindow: - """Fullscreen window setup, input grab, and exit-path lifecycle.""" + """Per-output lock surfaces, input grab, and exit-path lifecycle.""" def __init__( self, root: tk.Tk, config: LockConfig, hooks: LockWindowHooks, + *, + arbiter: Arbiter | None = None, ) -> None: """Initialize the lock window wrapper. Args: - root: The Tk root to configure and own the lifecycle of. + root: The Tk root to configure and own the lifecycle of. It becomes + the backdrop and the grab holder; per-output surfaces are its + children. config: Declarative lock behavior for this instance. - hooks: App-supplied callbacks for focus, errors, and teardown. + hooks: App-supplied callbacks for surfaces, focus, errors, teardown. + arbiter: The arbiter whose claim this lock is arming under. Used to + name the real holder when a grab is blocked, and released on + close so the next app can arm. """ self.root = root self._config = config self._hooks = hooks + self._arbiter = arbiter self._vt_disabled = False self._closed = False + self._focus_notified = False + self._surfaces = SurfaceSet(root, config, hooks) + self._enumerator = OutputEnumerator(root) + self._detector = OutputChangeDetector(root) + self._recovery = RecoveryLoop( + root, config, self._surfaces, self._enumerator, self._detector + ) + + @property + def surfaces(self) -> SurfaceSet: + """The live surface set, for apps that need to fan work out over it.""" + return self._surfaces # -- window mechanics ----------------------------------------------------- def setup(self) -> None: - """Configure the lock window per the resolved config.""" - screen_w = self.root.winfo_screenwidth() - screen_h = self.root.winfo_screenheight() - self.root.geometry(f"{screen_w}x{screen_h}+0+0") - self.root.attributes(topmost=True) - self.root.configure(bg=self._config.bg, cursor="arrow") - if self._config.resolved_overrideredirect(): - self.root.overrideredirect(boolean=True) - self.root.attributes(fullscreen=True) + """Configure the backdrop and build a surface on every live output. + + Disables VT switching first: strengthening the lock must not wait on + being able to draw anything. If no output is live, no surface is built + and the lock stays armed and invisible until one appears. + """ if self._config.resolved_disable_vt(): self._vt_disabled = disable_vt_switching() + if needs_backdrop_root(self._config): + # The root is a plain black backdrop spanning the whole X screen, + # including any region a modeless output left behind. It holds the + # grab (X will not grab for a window that is not viewable) and + # never carries widgets. + self.root.overrideredirect(boolean=self._config.resolved_overrideredirect()) + self._surfaces.update_backdrop() + else: + self.root.attributes("-topmost", _TOPMOST_ON) + + scan = self._enumerator.scan() + delta = self._surfaces.apply(scan) + if not scan.live: + _logger.warning( + "no live outputs at startup (source=%s); arming the lock with " + "nothing displayed -- it will appear as soon as a monitor " + "comes back", + scan.source, + ) + else: + _logger.info( + "lock armed on %d output(s): %s", + len(scan.live), + ", ".join(output.name for output in scan.live), + ) + if delta.unverified: + _logger.error( + "outputs live but not covered at startup: %s", + ", ".join(delta.unverified), + ) + def grab_input(self) -> None: - """Force focus to the window, then acquire the configured grab.""" + """Take the configured grab, then start watching for output changes.""" self.root.update_idletasks() self.root.focus_force() grab = self._config.resolved_grab() @@ -187,6 +281,8 @@ class LockWindow: elif grab == "local": with contextlib.suppress(tk.TclError): self.root.grab_set() + self._detector.start() + self._recovery.start() self.root.after(100, self._notify_focus_ready) def _acquire_global_grab(self, *, attempt: int) -> None: @@ -206,12 +302,7 @@ class LockWindow: return effective_retry_ms = retry_ms or _DEFAULT_GRAB_RETRY_MS if not attempt % self._config.grab_log_every: - _logger.warning( - "global grab still blocked after %d attempts (another " - "app -- e.g. a fullscreen game -- holds it); waiting " - "for it to free", - attempt, - ) + self._log_grab_blocked(attempt) self.root.after( effective_retry_ms, lambda: self._acquire_global_grab(attempt=attempt + 1), @@ -221,10 +312,45 @@ class LockWindow: self.root.focus_force() self._notify_focus_ready() + def _log_grab_blocked(self, attempt: int) -> None: + """Say who is actually holding the grab, rather than guessing. + + v0.1.1 always blamed "a fullscreen game". On 2026-07-25 the holder was + the other locker, and that guess sent the diagnosis in the wrong + direction for the length of the outage. + """ + holder = self._arbiter.describe_holder() if self._arbiter else None + if holder is None: + _logger.warning( + "global grab still blocked after %d attempts; no gatelock app " + "holds it -- likely another X client (e.g. a fullscreen game)", + attempt, + ) + return + _logger.warning( + "global grab still blocked after %d attempts; held by gatelock app " + "%r (rank %d, pid %d) since %s", + attempt, + holder.app, + holder.rank, + holder.pid, + holder.started, + ) + def _notify_focus_ready(self) -> None: - """Tell the app it can focus its first input widget now.""" + """Tell the app it can focus its first input widget now. + + Guarded rather than de-duplicated at the call sites: both timings are + load-bearing. The 100ms ``after`` covers a grab that succeeds + immediately, and the call from :meth:`_acquire_global_grab` covers one + that only succeeds after seconds of retrying. + """ + if self._focus_notified: + return + self._focus_notified = True with contextlib.suppress(tk.TclError): - self._hooks.on_focus_ready() + index = self._surfaces.preferred_focus_index() + self._hooks.on_focus_ready(self._surfaces.focus_surface(index)) # -- lifecycle -------------------------------------------------------- @@ -252,17 +378,29 @@ class LockWindow: self._vt_disabled = False def close(self) -> None: - """Run app teardown, restore VT switching, destroy the window. + """Run app teardown, release the screen, destroy every window. Idempotent -- safe to call directly (normal dismiss) and again from :meth:`run`'s ``finally`` (crash/signal exit) without double-running app teardown. + + Releasing the arbiter here is what makes a *clean* handoff work: the + morning routine runs the alarm and then the workout locker as + sequential subprocesses, and a claim left behind would make the second + one stand down against an app that had already exited. """ if self._closed: return self._closed = True + self._recovery.stop() + self._detector.stop() self._hooks.on_close() self._restore_vt() + if self._arbiter is not None: + self._arbiter.release() + self._enumerator.close() + with contextlib.suppress(tk.TclError): + self._surfaces.destroy_all() with contextlib.suppress(tk.TclError): self.root.destroy() diff --git a/gatelock/gatelock/tests/conftest.py b/gatelock/gatelock/tests/conftest.py index 5510598..2c9ca17 100644 --- a/gatelock/gatelock/tests/conftest.py +++ b/gatelock/gatelock/tests/conftest.py @@ -16,10 +16,68 @@ from unittest.mock import MagicMock, patch import pytest +from gatelock._outputs import Output, OutputRect from gatelock._window import LockConfig, LockWindow if TYPE_CHECKING: from collections.abc import Generator + from pathlib import Path + + +DEFAULT_OUTPUTS = ( + Output( + "VIRTUAL-1", connected=True, rect=OutputRect(0, 0, 1920, 1080), primary=True + ), +) +"""One live output matching ``mock_root``'s screen size.""" + + +def make_toplevel(_parent: object = None, **_kwargs: object) -> MagicMock: + """A Toplevel stand-in that remembers the geometry it was given. + + Real ``tk.Toplevel`` cannot be constructed over a ``MagicMock`` master -- + tkinter's widget-naming path recurses through the mock forever and the test + run hangs rather than failing. So every surface window is faked here. + """ + win = MagicMock() + state = {"x": 0, "y": 0, "w": 0, "h": 0, "mapped": False} + + def geometry(spec: str) -> None: + size, x, y = spec.split("+") + width, height = size.split("x") + state.update(x=int(x), y=int(y), w=int(width), h=int(height)) + + win.geometry.side_effect = geometry + win.deiconify.side_effect = lambda: state.update(mapped=True) + win.withdraw.side_effect = lambda: state.update(mapped=False) + win.winfo_ismapped.side_effect = lambda: state["mapped"] + win.winfo_rootx.side_effect = lambda: state["x"] + win.winfo_rooty.side_effect = lambda: state["y"] + win.winfo_width.side_effect = lambda: state["w"] + win.winfo_height.side_effect = lambda: state["h"] + return win + + +@pytest.fixture(autouse=True) +def _hermetic(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None]: + """Keep the whole suite off the real machine. + + Autouse and deliberately broad. A lock window takes a global input grab and + disables VT switching; a test that reached the real display or the real + ``setxkbmap`` would black out the developer's own machine, and one that + reached the real runtime directory could stand a *production* locker down. + Blocking all of that centrally beats remembering to patch it per test. + """ + monkeypatch.setenv("GATELOCK_RUNTIME_DIR", str(tmp_path / "runtime")) + with ( + patch("gatelock._surfaces.tk.Toplevel", side_effect=make_toplevel), + patch("gatelock._outputs.RandrBackend.create", return_value=None), + patch("gatelock._outputs.scan_xrandr", return_value=DEFAULT_OUTPUTS), + patch("gatelock._detect._RandrEventSource.start", return_value=False), + patch("gatelock._vt.shutil.which", return_value="/usr/bin/setxkbmap"), + patch("gatelock._vt.subprocess.run"), + ): + yield @pytest.fixture @@ -28,6 +86,7 @@ def mock_root() -> MagicMock: root = MagicMock() root.winfo_screenwidth.return_value = 1920 root.winfo_screenheight.return_value = 1080 + root.grab_current.return_value = root return root diff --git a/gatelock/gatelock/tests/test_arbiter.py b/gatelock/gatelock/tests/test_arbiter.py new file mode 100644 index 0000000..719a4a0 --- /dev/null +++ b/gatelock/gatelock/tests/test_arbiter.py @@ -0,0 +1,432 @@ +"""Tests for cross-app arbitration. + +The strength-check table is the most important thing here: it is the invariant +that stops priority rank from becoming a way to disarm a stronger lock. +""" + +from __future__ import annotations + +import errno +import os +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from gatelock._arbiter import ( + RANK_DIET_GUARD, + RANK_SCREEN_LOCKER, + RANK_WAKE_ALARM, + Arbiter, + Claim, + _same_file, + _try_lock, + default_runtime_dir, + grab_strength, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +DEFAULT_STARTED = "2026-07-25T12:00:00+00:00" + + +def dead_claim(app: str = "ghost", instance_id: str = "dead") -> Claim: + """A claim belonging to a process that has since died.""" + return Claim( + app=app, + rank=RANK_WAKE_ALARM, + pid=999999, + started=DEFAULT_STARTED, + grab="global", + disable_vt=True, + instance_id=instance_id, + ) + + +def make_claim( + *, + rank: int = RANK_SCREEN_LOCKER, + started: str = DEFAULT_STARTED, + grab: str = "global", + disable_vt: bool = True, +) -> Claim: + """A claim with sensible defaults.""" + return Claim( + app="test_app", + rank=rank, + pid=1234, + started=started, + grab=grab, + disable_vt=disable_vt, + instance_id="tok", + ) + + +@pytest.fixture +def arb(tmp_path: Path) -> Iterator[Arbiter]: + """A published hard-locking arbiter in an isolated runtime dir.""" + arbiter = Arbiter( + "screen_locker", + RANK_SCREEN_LOCKER, + grab="global", + disable_vt=True, + runtime_dir=tmp_path / "rt", + ) + arbiter.publish() + yield arbiter + arbiter.release() + + +def hard(name: str, rank: int, root: Path) -> Arbiter: + """A hard-locking arbiter.""" + return Arbiter(name, rank, grab="global", disable_vt=True, runtime_dir=root) + + +class TestGrabStrength: + """Unknown grab kinds must never rank as strong.""" + + @pytest.mark.parametrize( + ("kind", "expected"), + [("none", 0), ("local", 1), ("global", 2), ("something-new", 0)], + ) + def test_ordering(self, kind: str, expected: int) -> None: + """Known kinds order none < local < global; unknown sorts weakest.""" + assert grab_strength(kind) == expected + + +class TestStrengthCheck: + """THE anti-bypass invariant, as a table.""" + + @pytest.mark.parametrize( + ("their_grab", "their_vt", "our_grab", "our_vt", "strong_enough"), + [ + ("global", True, "global", True, True), + ("global", True, "local", False, True), + ("global", True, "none", False, True), + ("none", False, "global", True, False), + ("local", True, "global", True, False), + ("global", False, "global", True, False), + ("local", False, "local", False, True), + ("none", True, "none", True, True), + ], + ) + def test_table( + self, + their_grab: str, + our_grab: str, + *, + their_vt: bool, + our_vt: bool, + strong_enough: bool, + ) -> None: + """Standing down requires being no weaker on BOTH axes.""" + theirs = make_claim(grab=their_grab, disable_vt=their_vt) + ours = make_claim(grab=our_grab, disable_vt=our_vt) + assert theirs.at_least_as_strong_as(ours) is strong_enough + + +class TestClaimSerialisation: + """A malformed claim must never stop a locker arming.""" + + def test_round_trip(self) -> None: + """A claim survives serialisation.""" + claim = make_claim() + assert Claim.from_json(claim.to_json()) == claim + + @pytest.mark.parametrize( + "text", ["not json", "[]", '"a string"', "123", '{"app": "x"}'] + ) + def test_unusable_input_returns_none(self, text: str) -> None: + """Bad JSON, wrong type, or missing fields all yield None.""" + assert Claim.from_json(text) is None + + def test_wrong_field_type_returns_none(self) -> None: + """A non-numeric rank is rejected rather than raising.""" + assert ( + Claim.from_json( + '{"app":"a","rank":"x","pid":1,"started":"s",' + '"grab":"none","disable_vt":false,"instance_id":"t"}' + ) + is None + ) + + def test_sort_key_orders_by_rank_then_start_then_pid(self) -> None: + """Highest rank first, then earliest, then lowest pid.""" + high = make_claim(rank=300) + low_early = make_claim(rank=100, started="2026-01-01T00:00:00+00:00") + low_late = make_claim(rank=100, started="2026-12-01T00:00:00+00:00") + ordered = sorted([low_late, high, low_early], key=Claim.sort_key) + assert [c.rank for c in ordered] == [300, 100, 100] + assert ordered[1] is low_early + + +class TestRuntimeDir: + """Where claims live.""" + + def test_env_override_wins( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """GATELOCK_RUNTIME_DIR takes precedence.""" + monkeypatch.setenv("GATELOCK_RUNTIME_DIR", str(tmp_path / "override")) + assert default_runtime_dir() == tmp_path / "override" + + def test_xdg_runtime_dir( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """Falls back to XDG_RUNTIME_DIR/gatelock.""" + monkeypatch.delenv("GATELOCK_RUNTIME_DIR", raising=False) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(tmp_path / "xdg")) + assert default_runtime_dir() == tmp_path / "xdg" / "gatelock" + + def test_tempdir_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None: + """With neither env var, a per-uid temp directory is used.""" + monkeypatch.delenv("GATELOCK_RUNTIME_DIR", raising=False) + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + assert str(os.getuid()) in str(default_runtime_dir()) + + +class TestTryLock: + """flock wrapper behaviour.""" + + def test_contended_returns_false(self, tmp_path: Path) -> None: + """EAGAIN/EACCES mean somebody else holds it.""" + path = tmp_path / "f" + path.write_text("x", encoding="utf-8") + with path.open("r+", encoding="utf-8") as first: + assert _try_lock(first) is True + with path.open("r+", encoding="utf-8") as second: + assert _try_lock(second) is False + + def test_other_oserror_propagates(self, tmp_path: Path) -> None: + """An unexpected errno is a real problem and must not be swallowed.""" + path = tmp_path / "f" + path.write_text("x", encoding="utf-8") + with path.open("r+", encoding="utf-8") as handle: + exc = OSError() + exc.errno = errno.ENOSPC + with ( + patch("gatelock._arbiter.fcntl.flock", side_effect=exc), + pytest.raises(OSError, match=r"^$"), + ): + _try_lock(handle) + + def test_same_file_false_on_missing_path(self, tmp_path: Path) -> None: + """A vanished path is not the same file.""" + path = tmp_path / "gone" + path.write_text("x", encoding="utf-8") + with path.open("r+", encoding="utf-8") as handle: + path.unlink() + assert _same_file(handle, path) is False + + +class TestPublishAndLiveClaims: + """Publishing, liveness and reaping.""" + + def test_own_claim_is_live(self, arb: Arbiter) -> None: + """A published claim shows up as live.""" + assert [c.app for c in arb.live_claims()] == ["screen_locker"] + + def test_no_claims_dir(self, tmp_path: Path) -> None: + """Before anyone publishes, there are no claims.""" + assert hard("a", 1, tmp_path / "empty").live_claims() == () + + def test_dead_claim_is_reaped(self, arb: Arbiter, tmp_path: Path) -> None: + """An unlocked claim file proves its owner died; it is deleted.""" + dead = tmp_path / "rt" / "claims" / "0300-999-dead.json" + dead.write_text(dead_claim().to_json(), "utf-8") + assert "ghost" not in [c.app for c in arb.live_claims()] + assert not dead.exists() + + def test_unreadable_claim_ignored(self, arb: Arbiter, tmp_path: Path) -> None: + """A claim that cannot be opened is skipped, not fatal.""" + bad = tmp_path / "rt" / "claims" / "0100-1-bad.json" + bad.write_text("{}", encoding="utf-8") + with patch("gatelock._arbiter.Path.open", side_effect=OSError("denied")): + assert arb.live_claims() == () + + def test_reap_skips_recreated_file(self, arb: Arbiter, tmp_path: Path) -> None: + """A claim recreated between lock and unlink is not deleted.""" + dead = tmp_path / "rt" / "claims" / "0300-999-dead.json" + dead.write_text(dead_claim().to_json(), "utf-8") + with patch("gatelock._arbiter._same_file", return_value=False): + arb.live_claims() + assert dead.exists() + + def test_reap_tolerates_unlink_failure(self, arb: Arbiter, tmp_path: Path) -> None: + """A failed unlink is logged, not raised.""" + dead = tmp_path / "rt" / "claims" / "0300-999-dead.json" + dead.write_text(dead_claim().to_json(), "utf-8") + with patch("gatelock._arbiter.Path.unlink", side_effect=OSError("busy")): + live = arb.live_claims() + # The reap is best-effort: the claim is excluded from the live set even + # when the file itself cannot be removed. + assert "ghost" not in [c.app for c in live] + + def test_publish_twice_warns_and_continues(self, tmp_path: Path) -> None: + """A second publish on a locked path continues unpublished.""" + arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt") + arbiter.publish() + with patch("gatelock._arbiter._try_lock", return_value=False): + arbiter.publish() # must not raise + arbiter.release() + + +class TestEvaluate: + """Who may arm.""" + + def test_clear_when_alone(self, arb: Arbiter) -> None: + """Nothing pending means arm.""" + verdict = arb.evaluate() + assert verdict.may_arm is True + assert verdict.reason == "clear" + assert verdict.blocked_by is None + + def test_lower_rank_does_not_block(self, arb: Arbiter, tmp_path: Path) -> None: + """A weaker-ranked peer never blocks.""" + diet = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt") + diet.publish() + assert arb.evaluate().may_arm is True + diet.release() + + def test_stronger_higher_rank_blocks(self, arb: Arbiter, tmp_path: Path) -> None: + """A higher-ranked, equally strong app wins.""" + alarm = hard("wake_alarm", RANK_WAKE_ALARM, tmp_path / "rt") + alarm.publish() + verdict = arb.evaluate() + assert verdict.may_arm is False + assert verdict.reason == "outranked" + assert verdict.blocked_by is not None + assert verdict.blocked_by.app == "wake_alarm" + alarm.release() + + def test_weaker_higher_rank_arms_anyway(self, arb: Arbiter, tmp_path: Path) -> None: + """Rank must never be able to reduce total lock strength.""" + soft_alarm = Arbiter( + "wake_alarm", + RANK_WAKE_ALARM, + grab="none", + disable_vt=False, + runtime_dir=tmp_path / "rt", + ) + soft_alarm.publish() + verdict = arb.evaluate() + assert verdict.may_arm is True + assert verdict.reason == "weaker-incumbent-armed-anyway" + soft_alarm.release() + + +class TestHolder: + """Owning and naming the screen.""" + + def test_acquire_and_reacquire(self, arb: Arbiter) -> None: + """Acquiring twice is idempotent.""" + assert arb.acquire_holder() is True + assert arb.holds_screen is True + assert arb.acquire_holder() is True + + def test_second_app_is_blocked(self, arb: Arbiter, tmp_path: Path) -> None: + """Only one app holds the screen.""" + arb.acquire_holder() + other = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt") + other.publish() + assert other.acquire_holder() is False + other.release() + + def test_blocked_app_learns_the_holder(self, arb: Arbiter, tmp_path: Path) -> None: + """A failed acquire must not erase the incumbent's claim.""" + arb.acquire_holder() + other = hard("diet_guard", RANK_DIET_GUARD, tmp_path / "rt") + other.publish() + other.acquire_holder() + holder = other.describe_holder() + assert holder is not None + assert holder.app == "screen_locker" + other.release() + + def test_describe_holder_when_we_hold_it(self, arb: Arbiter) -> None: + """The holder describes itself without touching the file.""" + arb.acquire_holder() + holder = arb.describe_holder() + assert holder is not None + assert holder.app == "screen_locker" + + def test_describe_holder_no_file(self, arb: Arbiter) -> None: + """No holder file means nobody holds the screen.""" + assert arb.describe_holder() is None + + def test_describe_holder_stale_lock(self, arb: Arbiter, tmp_path: Path) -> None: + """A holder file nobody locks means the owner died.""" + (tmp_path / "rt").mkdir(parents=True, exist_ok=True) + (tmp_path / "rt" / "holder.lock").write_text( + dead_claim().to_json(), encoding="utf-8" + ) + assert arb.describe_holder() is None + + def test_describe_holder_oserror(self, arb: Arbiter, tmp_path: Path) -> None: + """An unreadable holder file is reported as no holder.""" + (tmp_path / "rt").mkdir(parents=True, exist_ok=True) + (tmp_path / "rt" / "holder.lock").write_text("{}", encoding="utf-8") + with patch("gatelock._arbiter.Path.open", side_effect=OSError("denied")): + assert arb.describe_holder() is None + + +class TestRelease: + """Clean exit hands the screen over.""" + + def test_release_lets_the_next_app_in(self, tmp_path: Path) -> None: + """The alarm -> workout handoff: a clean exit must free the screen.""" + alarm = hard("wake_alarm", RANK_WAKE_ALARM, tmp_path / "rt") + alarm.publish() + assert alarm.acquire_holder() is True + + workout = hard("screen_locker", RANK_SCREEN_LOCKER, tmp_path / "rt") + workout.publish() + assert workout.acquire_holder() is False + assert workout.evaluate().may_arm is False + + alarm.release() + + assert workout.acquire_holder() is True + assert workout.evaluate().may_arm is True + workout.release() + + def test_release_is_idempotent(self, tmp_path: Path) -> None: + """Releasing twice is safe.""" + arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt") + arbiter.publish() + arbiter.release() + arbiter.release() + assert arbiter.holds_screen is False + + def test_release_tolerates_unlink_failure(self, tmp_path: Path) -> None: + """A claim that cannot be removed does not break teardown.""" + arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt") + arbiter.publish() + with patch("gatelock._arbiter.Path.unlink", side_effect=OSError("busy")): + arbiter.release() + + def test_claim_property(self, arb: Arbiter) -> None: + """The arbiter exposes its own claim.""" + assert arb.claim.app == "screen_locker" + assert arb.claim.rank == RANK_SCREEN_LOCKER + + +class TestReleaseHandleClose: + """Handle-close failures are non-fatal.""" + + def test_close_oserror_is_swallowed(self, tmp_path: Path) -> None: + """A failing close does not stop the release.""" + arbiter = hard("a", RANK_DIET_GUARD, tmp_path / "rt") + arbiter.publish() + arbiter.acquire_holder() + bad = MagicMock() + bad.close.side_effect = OSError("nope") + real_handle = arbiter._holder_handle + assert real_handle is not None + real_handle.close() + arbiter._holder_handle = bad + arbiter.release() + assert arbiter.holds_screen is False diff --git a/gatelock/gatelock/tests/test_detect.py b/gatelock/gatelock/tests/test_detect.py new file mode 100644 index 0000000..9453672 --- /dev/null +++ b/gatelock/gatelock/tests/test_detect.py @@ -0,0 +1,188 @@ +"""Tests for layered output-change detection.""" + +from __future__ import annotations + +import queue +import tkinter as tk +from unittest.mock import MagicMock, patch + +from gatelock._detect import OutputChangeDetector, _RandrEventSource + +REAL_START = _RandrEventSource.start + + +class TestTakePending: + """Coalescing the push signals.""" + + def test_nothing_pending(self, mock_root: MagicMock) -> None: + """A quiet period reports no change.""" + assert OutputChangeDetector(mock_root).take_pending() is False + + def test_configure_event_sets_the_flag(self, mock_root: MagicMock) -> None: + """A Tk is a change signal.""" + detector = OutputChangeDetector(mock_root) + detector._on_configure(MagicMock()) + assert detector.take_pending() is True + assert detector.take_pending() is False + + def test_configure_storm_coalesces(self, mock_root: MagicMock) -> None: + """Many events drain as one -- no rebuild thrash.""" + detector = OutputChangeDetector(mock_root) + for _ in range(50): + detector._on_configure(MagicMock()) + assert detector.take_pending() is True + assert detector.take_pending() is False + + def test_randr_events_drain(self, mock_root: MagicMock) -> None: + """Queued RandR notices are consumed.""" + detector = OutputChangeDetector(mock_root) + detector._queue.put("randr") + detector._queue.put("randr") + assert detector.take_pending() is True + assert detector._queue.empty() + + +class TestStartStop: + """Subscription lifecycle.""" + + def test_start_binds_configure(self, mock_root: MagicMock) -> None: + """Starting binds the Tk event.""" + detector = OutputChangeDetector(mock_root) + detector.start() + mock_root.bind.assert_called_once() + assert mock_root.bind.call_args.args[0] == "" + + def test_start_without_xlib_still_works(self, mock_root: MagicMock) -> None: + """The autouse patch makes RandR unavailable; detection still runs.""" + detector = OutputChangeDetector(mock_root) + detector.start() + assert detector.randr_active is False + + def test_start_with_randr(self, mock_root: MagicMock) -> None: + """When RandR subscribes, it is reported active.""" + detector = OutputChangeDetector(mock_root) + with patch.object(_RandrEventSource, "start", return_value=True): + detector.start() + assert detector.randr_active is True + + def test_stop_unbinds(self, mock_root: MagicMock) -> None: + """Stopping releases the Tk binding.""" + detector = OutputChangeDetector(mock_root) + mock_root.bind.return_value = "binding-id" + detector.start() + detector.stop() + mock_root.unbind.assert_called_once_with("", "binding-id") + assert detector.randr_active is False + + def test_stop_swallows_tclerror(self, mock_root: MagicMock) -> None: + """Unbinding a dead widget is not an error.""" + detector = OutputChangeDetector(mock_root) + mock_root.bind.return_value = "binding-id" + detector.start() + mock_root.unbind.side_effect = tk.TclError("gone") + detector.stop() + + def test_stop_before_start(self, mock_root: MagicMock) -> None: + """Stopping an unstarted detector is safe.""" + OutputChangeDetector(mock_root).stop() + + +def _xlib(display: MagicMock) -> dict[str, MagicMock]: + """A fake python-xlib package exposing ``display``.""" + randr = MagicMock( + RRScreenChangeNotifyMask=1, RRCrtcChangeNotifyMask=2, RROutputChangeNotifyMask=4 + ) + ext = MagicMock(randr=randr) + return { + "Xlib": MagicMock(display=display, ext=ext), + "Xlib.display": display, + "Xlib.ext": ext, + "Xlib.ext.randr": randr, + } + + +class TestRandrEventSource: + """The private-connection watcher.""" + + def test_connect_without_xlib(self) -> None: + """A missing python-xlib degrades to no events.""" + source = _RandrEventSource(queue.Queue()) + with patch.dict("sys.modules", {"Xlib": None}): + assert source._connect() is False + + def test_connect_subscribes_to_all_three_masks(self) -> None: + """Screen, CRTC and output changes are all watched.""" + display = MagicMock() + source = _RandrEventSource(queue.Queue()) + with patch.dict("sys.modules", _xlib(display)): + assert source._connect() is True + root = display.Display.return_value.screen.return_value.root + root.xrandr_select_input.assert_called_once_with(1 | 2 | 4) + + def test_connect_failure_closes_and_reports(self) -> None: + """A subscription error degrades cleanly.""" + display = MagicMock() + display.Display.return_value.screen.side_effect = OSError("refused") + source = _RandrEventSource(queue.Queue()) + with patch.dict("sys.modules", _xlib(display)): + assert source._connect() is False + assert source._display is None + + def test_start_returns_false_without_connection(self) -> None: + """No connection means no thread.""" + source = _RandrEventSource(queue.Queue()) + with patch.object(source, "_connect", return_value=False): + assert REAL_START(source) is False + + def test_loop_posts_one_notice_per_batch(self) -> None: + """A burst of X events becomes a single change notice.""" + sink: queue.Queue[str] = queue.Queue() + source = _RandrEventSource(sink) + source._display = MagicMock() + source._display.pending_events.side_effect = [3, 0] + stop_after = {"n": 0} + + def is_set() -> bool: + stop_after["n"] += 1 + return stop_after["n"] > 2 + + source._stop = MagicMock() + source._stop.is_set.side_effect = is_set + with patch( + "gatelock._detect.select.select", side_effect=[([1], [], []), ([], [], [])] + ): + source._loop() + assert sink.qsize() == 1 + assert source._display.next_event.call_count == 3 + + def test_run_closes_on_exit(self) -> None: + """The thread always drops its connection.""" + source = _RandrEventSource(queue.Queue()) + source._display = MagicMock() + with patch.object(source, "_loop"): + source._run() + assert source._display is None + + def test_stop_joins_thread(self) -> None: + """Stopping sets the flag and joins.""" + source = _RandrEventSource(queue.Queue()) + thread = MagicMock() + source._thread = thread + source.stop() + thread.join.assert_called_once() + assert source._thread is None + + def test_close_is_idempotent(self) -> None: + """Closing twice is safe.""" + source = _RandrEventSource(queue.Queue()) + source._display = MagicMock() + source._close() + source._close() + + def test_close_swallows_errors(self) -> None: + """A failing close does not escape the thread.""" + source = _RandrEventSource(queue.Queue()) + source._display = MagicMock() + source._display.close.side_effect = OSError("already gone") + source._close() + assert source._display is None diff --git a/gatelock/gatelock/tests/test_guards.py b/gatelock/gatelock/tests/test_guards.py new file mode 100644 index 0000000..3c94ddd --- /dev/null +++ b/gatelock/gatelock/tests/test_guards.py @@ -0,0 +1,97 @@ +"""Tests for the shared safety guards.""" + +from __future__ import annotations + +import sys +import tkinter as tk +from unittest.mock import MagicMock, patch + +import pytest + +from gatelock._guards import ( + _probe_x_server, + assert_not_under_pytest, + wait_for_x_server, +) + + +class TestAssertNotUnderPytest: + """Refuse to build a real grabbing window inside a test run.""" + + def test_raises_under_pytest_with_real_tkinter(self) -> None: + """The dangerous combination is rejected.""" + with pytest.raises(RuntimeError, match="refusing to build"): + assert_not_under_pytest("the workout lock") + + def test_allows_when_tkinter_is_mocked(self) -> None: + """A properly isolated test cannot open a window, so it is allowed.""" + fake_tk = MagicMock() + fake_tk.__name__ = "MagicMock" + with patch("gatelock._guards.tk", fake_tk): + assert_not_under_pytest("the workout lock") + + def test_allows_outside_pytest(self) -> None: + """In production there is no pytest, so nothing is blocked.""" + modules = {k: v for k, v in sys.modules.items() if k != "pytest"} + with patch.dict("sys.modules", modules, clear=True): + assert_not_under_pytest("the workout lock") + + +class TestWaitForXServer: + """Gate on an X server existing -- never on how many outputs are live.""" + + def test_returns_true_immediately(self) -> None: + """A ready display returns at once.""" + assert wait_for_x_server(probe=lambda: True) is True + + def test_retries_then_succeeds(self) -> None: + """A slow display is waited for.""" + results = iter([False, False, True]) + sleeps: list[float] = [] + assert ( + wait_for_x_server( + probe=lambda: next(results), + sleep=sleeps.append, + monotonic=lambda: 0.0, + ) + is True + ) + assert sleeps == [1.0, 1.0] + + def test_times_out(self) -> None: + """A display that never arrives gives up and says so.""" + clock = iter([0.0, 100.0]) + assert ( + wait_for_x_server( + probe=lambda: False, + sleep=lambda _s: None, + monotonic=lambda: next(clock), + timeout_s=1.0, + ) + is False + ) + + def test_zero_live_outputs_does_not_block_arming(self) -> None: + """THE 2b rule: a connected X server with dark monitors still arms. + + This function must answer "is there an X server", nothing more. If it + ever consulted output liveness, a dark monitor would once again mean + "do not lock" instead of "lock without showing". + """ + assert wait_for_x_server(probe=lambda: True) is True + + +class TestProbeXServer: + """The default probe.""" + + def test_success_destroys_the_probe_window(self) -> None: + """A working display yields True and leaves nothing behind.""" + fake_root = MagicMock() + with patch("gatelock._guards.tk.Tk", return_value=fake_root): + assert _probe_x_server() is True + fake_root.destroy.assert_called_once_with() + + def test_failure_returns_false(self) -> None: + """No display yields False rather than raising.""" + with patch("gatelock._guards.tk.Tk", side_effect=tk.TclError("no display")): + assert _probe_x_server() is False diff --git a/gatelock/gatelock/tests/test_outputs.py b/gatelock/gatelock/tests/test_outputs.py new file mode 100644 index 0000000..1f0456a --- /dev/null +++ b/gatelock/gatelock/tests/test_outputs.py @@ -0,0 +1,418 @@ +"""Tests for output enumeration and the live-vs-dark predicate. + +The regression that matters most is :data:`XRANDR_INCIDENT`: on 2026-07-25 an +output reported ``connected`` with a valid EDID and no geometry, and every +consumer treated it as usable screen real estate. +""" + +from __future__ import annotations + +import subprocess +from unittest.mock import MagicMock, patch + +import pytest + +from gatelock._outputs import ( + Output, + OutputEnumerator, + OutputRect, + OutputScan, + RandrBackend, + _decode_name, + enumerate_outputs, + parse_xrandr_query, +) +from gatelock._outputs import scan_xrandr as real_scan_xrandr + +# Bound at import, so the autouse hermetic patch (which replaces these +# attributes) does not hide the real implementations from their own tests. +REAL_CREATE = RandrBackend.create.__func__ + +XRANDR_REAL = """\ +Screen 0: minimum 8 x 8, current 6400 x 2160, maximum 32767 x 32767 +HDMI-0 connected 2560x1440+3840+0 (normal left inverted right x axis) 597mm x 336mm + 2560x1440 59.95*+ 143.98 120.00 + 1920x1080 60.00 59.94 +DP-0 connected primary 3840x2160+0+0 (normal left inverted right x axis) 621mm x 341mm + 3840x2160 60.00*+ 144.00 +DP-1 disconnected (normal left inverted right x axis y axis) +""" + +XRANDR_INCIDENT = """\ +Screen 0: minimum 8 x 8, current 6400 x 1440, maximum 32767 x 32767 +HDMI-0 connected 2560x1440+3840+0 (normal left inverted right x axis) 600mm x 330mm + 2560x1440 59.95*+ +DP-0 connected primary (normal left inverted right x axis y axis) 621mm x 341mm + 3840x2160 60.00 + +""" + +XRANDR_ALL_DARK = """\ +HDMI-0 connected (normal left inverted right x axis) +DP-0 connected primary (normal left inverted right x axis y axis) +""" + +XRANDR_ROTATED = "DP-0 connected primary 1080x1920+0+0 left (normal left inverted)\n" +XRANDR_NEGATIVE = "DP-0 connected 1920x1080+-1920+0 (normal left inverted)\n" +XRANDR_ZERO_SIZE = "DP-0 connected 0x0+0+0 (normal left inverted)\n" +XRANDR_MODE_LINES_ONLY = " 2560x1440 59.95*+ 143.98\n 1920x1080 60.00\n" +XRANDR_GARBAGE = "segfault\n" + + +class TestOutputRect: + """Tests for OutputRect.""" + + def test_geometry_string(self) -> None: + """A rect renders as Tk's WxH+X+Y form.""" + assert OutputRect(3840, 0, 2560, 1440).geometry() == "2560x1440+3840+0" + + def test_geometry_with_negative_offset(self) -> None: + """Negative offsets survive the round trip.""" + assert OutputRect(-1920, 0, 1920, 1080).geometry() == "1920x1080+-1920+0" + + +class TestLivePredicate: + """Connected is not the same as live.""" + + def test_connected_with_mode_is_live(self) -> None: + """A connected output with a rect is live.""" + assert Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)).live + + def test_connected_without_mode_is_not_live(self) -> None: + """THE incident: connected, EDID present, no mode -> not live.""" + assert not Output("DP-0", connected=True, rect=None).live + + def test_disconnected_is_not_live(self) -> None: + """A disconnected output is not live even if a rect lingers.""" + assert not Output("DP-1", connected=False, rect=OutputRect(0, 0, 1, 1)).live + + +class TestParseXrandrQuery: + """Tests for the xrandr text parser.""" + + def test_real_layout(self) -> None: + """Both live outputs are found with exact geometry, plus primary.""" + outputs = parse_xrandr_query(XRANDR_REAL) + assert [o.name for o in outputs] == ["HDMI-0", "DP-0", "DP-1"] + by_name = {o.name: o for o in outputs} + assert by_name["DP-0"].rect == OutputRect(0, 0, 3840, 2160) + assert by_name["DP-0"].primary is True + assert by_name["HDMI-0"].rect == OutputRect(3840, 0, 2560, 1440) + assert by_name["HDMI-0"].primary is False + assert by_name["DP-1"].connected is False + + def test_the_incident(self) -> None: + """DP-0 connected-but-modeless is parsed as not live.""" + outputs = parse_xrandr_query(XRANDR_INCIDENT) + by_name = {o.name: o for o in outputs} + assert by_name["DP-0"].connected is True + assert by_name["DP-0"].rect is None + assert by_name["DP-0"].live is False + assert [o.name for o in outputs if o.live] == ["HDMI-0"] + + def test_all_dark(self) -> None: + """Every output connected and modeless yields zero live.""" + assert not [o for o in parse_xrandr_query(XRANDR_ALL_DARK) if o.live] + + def test_mode_lines_are_never_geometry(self) -> None: + """Indented mode lines must not be read as output lines.""" + assert parse_xrandr_query(XRANDR_MODE_LINES_ONLY) == () + + def test_rotated_reports_post_rotation_size(self) -> None: + """xrandr already reports rotated geometry; no special handling.""" + (output,) = parse_xrandr_query(XRANDR_ROTATED) + assert output.rect == OutputRect(0, 0, 1080, 1920) + + def test_negative_offset(self) -> None: + """A monitor placed left of origin parses correctly.""" + (output,) = parse_xrandr_query(XRANDR_NEGATIVE) + assert output.rect == OutputRect(-1920, 0, 1920, 1080) + + def test_zero_size_rejected(self) -> None: + """A 0x0 geometry is not a usable screen.""" + (output,) = parse_xrandr_query(XRANDR_ZERO_SIZE) + assert output.rect is None + + def test_garbage_and_empty(self) -> None: + """Unparseable input yields no outputs rather than raising.""" + assert parse_xrandr_query(XRANDR_GARBAGE) == () + assert parse_xrandr_query("") == () + + +class TestScanXrandr: + """Tests for the xrandr subprocess wrapper, including every failure mode.""" + + def _run(self, **kwargs: object) -> tuple[Output, ...] | None: + with ( + patch("gatelock._outputs.shutil.which", return_value="/usr/bin/xrandr"), + patch("gatelock._outputs.subprocess.run", **kwargs), + ): + return real_scan_xrandr() + + def test_success(self) -> None: + """A clean run parses stdout.""" + completed = MagicMock(returncode=0, stdout=XRANDR_REAL, stderr="") + assert self._run(return_value=completed) is not None + + def test_missing_binary(self) -> None: + """xrandr not on PATH degrades rather than raising.""" + with patch("gatelock._outputs.shutil.which", return_value=None): + assert real_scan_xrandr() is None + + def test_nonzero_exit(self) -> None: + """A non-zero exit is a failed scan.""" + completed = MagicMock(returncode=1, stdout="", stderr="no display") + assert self._run(return_value=completed) is None + + def test_timeout(self) -> None: + """A hung xrandr is a failed scan, not a hung locker.""" + exc = subprocess.TimeoutExpired(cmd="xrandr", timeout=5) + assert self._run(side_effect=exc) is None + + def test_oserror(self) -> None: + """An OSError launching xrandr is a failed scan.""" + assert self._run(side_effect=OSError("boom")) is None + + def test_unparseable_output(self) -> None: + """Exit 0 with nothing parseable is still a failed scan.""" + completed = MagicMock(returncode=0, stdout=XRANDR_GARBAGE, stderr="") + assert self._run(return_value=completed) is None + + +class TestDecodeName: + """RandR output names may arrive as bytes.""" + + @pytest.mark.parametrize( + ("raw", "expected"), + [(b"DP-0", "DP-0"), ("HDMI-0", "HDMI-0"), (b"\xff", "�")], + ) + def test_decode(self, raw: object, expected: str) -> None: + """Bytes are decoded, str passes through, bad bytes are replaced.""" + assert _decode_name(raw) == expected + + +def _reply(**fields: object) -> MagicMock: + """An Xlib reply exposing its fields as attributes, as python-xlib does.""" + reply = MagicMock() + for key, value in fields.items(): + setattr(reply, key, value) + return reply + + +def _randr_display(*, outputs: dict[int, dict[str, object]]) -> MagicMock: + """Build a fake Xlib display exposing ``outputs``.""" + display = MagicMock() + root = MagicMock() + root.xrandr_get_screen_resources.return_value = _reply( + config_timestamp=1, outputs=list(outputs) + ) + root.xrandr_get_output_primary.return_value = _reply(output=1) + display.screen.return_value.root = root + + def output_info(output_id: int, _ts: int) -> MagicMock: + return _reply(**outputs[output_id]["info"]) + + def crtc_info(crtc: int, _ts: int) -> MagicMock: + return _reply(**outputs[crtc]["crtc"]) + + display.xrandr_get_output_info.side_effect = output_info + display.xrandr_get_crtc_info.side_effect = crtc_info + return display + + +class TestRandrBackend: + """Tests for the RandR backend.""" + + def test_scan_maps_crtc_to_rect(self) -> None: + """A connected output with a CRTC and mode becomes a live rect.""" + display = _randr_display( + outputs={ + 1: { + "info": {"name": "DP-0", "connection": 0, "crtc": 1}, + "crtc": {"mode": 55, "x": 0, "y": 0, "width": 3840, "height": 2160}, + } + } + ) + (output,) = RandrBackend(display).scan() or () + assert output.name == "DP-0" + assert output.live is True + assert output.primary is True + assert output.rect == OutputRect(0, 0, 3840, 2160) + + def test_no_crtc_is_not_live(self) -> None: + """crtc == 0 is the dark-monitor case.""" + display = _randr_display( + outputs={1: {"info": {"name": "DP-0", "connection": 0, "crtc": 0}}} + ) + (output,) = RandrBackend(display).scan() or () + assert output.connected is True + assert output.rect is None + + def test_crtc_without_mode_is_not_live(self) -> None: + """A CRTC exists but carries no mode: still dark.""" + display = _randr_display( + outputs={ + 1: { + "info": {"name": "DP-0", "connection": 0, "crtc": 1}, + "crtc": {"mode": 0, "x": 0, "y": 0, "width": 0, "height": 0}, + } + } + ) + (output,) = RandrBackend(display).scan() or () + assert output.rect is None + + def test_zero_size_crtc_is_not_live(self) -> None: + """A mode with no pixels is not a screen.""" + display = _randr_display( + outputs={ + 1: { + "info": {"name": "DP-0", "connection": 0, "crtc": 1}, + "crtc": {"mode": 55, "x": 0, "y": 0, "width": 0, "height": 1080}, + } + } + ) + (output,) = RandrBackend(display).scan() or () + assert output.rect is None + + def test_disconnected(self) -> None: + """connection != 0 means disconnected.""" + display = _randr_display( + outputs={1: {"info": {"name": "DP-1", "connection": 1, "crtc": 0}}} + ) + (output,) = RandrBackend(display).scan() or () + assert output.connected is False + + def test_scan_failure_returns_none(self) -> None: + """A protocol error degrades to None instead of raising.""" + display = MagicMock() + display.screen.side_effect = OSError("connection reset") + assert RandrBackend(display).scan() is None + + def test_primary_query_failure_is_tolerated(self) -> None: + """RandR refusing to name a primary is not fatal.""" + display = _randr_display( + outputs={1: {"info": {"name": "DP-0", "connection": 0, "crtc": 0}}} + ) + root = display.screen.return_value.root + root.xrandr_get_output_primary.side_effect = OSError("nope") + (output,) = RandrBackend(display).scan() or () + assert output.primary is False + + def test_close_swallows_errors(self) -> None: + """Teardown never raises.""" + display = MagicMock() + display.close.side_effect = OSError("already gone") + RandrBackend(display).close() + + def test_close_normal(self) -> None: + """A clean close closes the display.""" + display = MagicMock() + RandrBackend(display).close() + display.close.assert_called_once_with() + + +class TestCreateWithoutXlib: + """RandrBackend.create's degradation path, exercised without the patch.""" + + def test_import_error_returns_none(self) -> None: + """A missing python-xlib degrades to None, never an exception. + + This is the branch the function-scope import exists to keep reachable. + """ + with patch.dict("sys.modules", {"Xlib": None}): + assert REAL_CREATE(RandrBackend) is None + + def test_display_error_returns_none(self) -> None: + """A display that refuses to open returns None.""" + fake_display = MagicMock() + fake_display.Display.side_effect = OSError("cannot connect") + # `from Xlib import display` reads the attribute off the parent + # module, so wiring sys.modules alone is not enough. + fake_xlib = MagicMock(display=fake_display) + with patch.dict( + "sys.modules", + {"Xlib": fake_xlib, "Xlib.display": fake_display}, + ): + assert REAL_CREATE(RandrBackend) is None + + def test_success_returns_backend(self) -> None: + """A working python-xlib yields a usable backend.""" + fake_display = MagicMock() + fake_xlib = MagicMock(display=fake_display) + with patch.dict( + "sys.modules", + {"Xlib": fake_xlib, "Xlib.display": fake_display}, + ): + backend = REAL_CREATE(RandrBackend) + assert isinstance(backend, RandrBackend) + + +class TestOutputEnumerator: + """Tests for the backend ladder.""" + + def test_prefers_randr(self, mock_root: MagicMock) -> None: + """When RandR answers, xrandr is never consulted.""" + backend = MagicMock() + backend.scan.return_value = ( + Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)), + ) + scan = OutputEnumerator(mock_root, backend=backend).scan() + assert scan.source == "randr" + assert scan.ok is True + + def test_falls_back_to_xrandr(self, mock_root: MagicMock) -> None: + """A failed RandR scan drops to xrandr.""" + backend = MagicMock() + backend.scan.return_value = None + with patch( + "gatelock._outputs.scan_xrandr", + return_value=(Output("DP-0", connected=True, rect=OutputRect(0, 0, 1, 1)),), + ): + scan = OutputEnumerator(mock_root, backend=backend).scan() + assert scan.source == "xrandr" + + def test_falls_back_to_tk(self, mock_root: MagicMock) -> None: + """With neither backend, Tk's screen size is the floor.""" + with patch("gatelock._outputs.scan_xrandr", return_value=None): + scan = OutputEnumerator(mock_root, backend=None).scan() + assert scan.source == "tk" + assert scan.live[0].rect == OutputRect(0, 0, 1920, 1080) + + def test_total_failure_is_not_ok(self) -> None: + """With no root and no backend, the scan reports failure.""" + with patch("gatelock._outputs.scan_xrandr", return_value=None): + scan = OutputEnumerator(None, backend=None).scan() + assert scan.ok is False + assert scan.source == "none" + assert scan.live == () + + def test_backend_property_and_close(self) -> None: + """close() releases the backend once and is safe to repeat.""" + backend = MagicMock() + enumerator = OutputEnumerator(None, backend=backend) + assert enumerator.backend is backend + enumerator.close() + enumerator.close() + backend.close.assert_called_once_with() + + def test_enumerate_outputs_helper(self, mock_root: MagicMock) -> None: + """The one-shot helper works and cleans up after itself.""" + with patch("gatelock._outputs.scan_xrandr", return_value=None): + scan = enumerate_outputs(mock_root) + assert scan.source == "tk" + + +class TestOutputScan: + """Tests for OutputScan's live filter.""" + + def test_live_filters_dark_outputs(self) -> None: + """Only connected-with-mode outputs are live.""" + scan = OutputScan( + outputs=( + Output("DP-0", connected=True, rect=None), + Output("HDMI-0", connected=True, rect=OutputRect(0, 0, 1, 1)), + Output("DP-1", connected=False, rect=None), + ), + source="randr", + ok=True, + ) + assert [o.name for o in scan.live] == ["HDMI-0"] diff --git a/gatelock/gatelock/tests/test_recovery.py b/gatelock/gatelock/tests/test_recovery.py new file mode 100644 index 0000000..a42939b --- /dev/null +++ b/gatelock/gatelock/tests/test_recovery.py @@ -0,0 +1,317 @@ +"""Tests for the monotonic recovery loop. + +Three layers, per the release plan: a static AST check that the weakening calls +are absent, a behavioural walk through the 2026-07-25 sequence asserting the +lock never weakens, and per-branch coverage. +""" + +from __future__ import annotations + +import ast +from pathlib import Path +import tkinter as tk +from unittest.mock import MagicMock, patch + +import pytest + +from gatelock._outputs import Output, OutputRect, OutputScan +from gatelock._recovery import RecoveryLoop +from gatelock._surfaces import SurfaceSet +from gatelock._window import LockConfig + +DP0 = Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True) +HDMI = Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440)) +DP0_DARK = Output("DP-0", connected=True, rect=None, primary=True) +HDMI_DARK = Output("HDMI-0", connected=True, rect=None) + +BOTH = OutputScan((DP0, HDMI), "randr", ok=True) +DEAD_PRIMARY = OutputScan((DP0_DARK, HDMI), "randr", ok=True) +ALL_DARK = OutputScan((DP0_DARK, HDMI_DARK), "randr", ok=True) +FAILED = OutputScan((), "none", ok=False) + +WEAKENING_CALLS = frozenset( + {"grab_release", "restore_vt_switching", "close", "destroy", "quit", "withdraw"} +) +SOURCE_ROOT = Path(__file__).resolve().parent.parent + + +def symbols(module: str) -> set[str]: + """Every attribute and name referenced in a module's source.""" + tree = ast.parse((SOURCE_ROOT / module).read_text(encoding="utf-8")) + return {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)} | { + n.id for n in ast.walk(tree) if isinstance(n, ast.Name) + } + + +@pytest.fixture +def loop(mock_root: MagicMock) -> tuple[RecoveryLoop, MagicMock, SurfaceSet]: + """A recovery loop over mocked surroundings.""" + config = LockConfig(mode="hard") + surfaces = SurfaceSet(mock_root, config, MagicMock()) + enumerator = MagicMock() + detector = MagicMock() + return ( + RecoveryLoop(mock_root, config, surfaces, enumerator, detector), + enumerator, + surfaces, + ) + + +class TestStaticInvariant: + """The separation that makes monotonicity mechanical rather than hoped-for.""" + + def test_recovery_contains_no_weakening_call(self) -> None: + """_recovery.py must not be able to release, restore or destroy.""" + assert symbols("_recovery.py") & WEAKENING_CALLS == set() + + def test_recovery_may_still_strengthen(self) -> None: + """The asymmetry is the point: strengthening calls ARE present.""" + found = symbols("_recovery.py") + assert "disable_vt_switching" in found + assert "grab_set_global" in found + + def test_surfaces_never_touches_grab_or_vt(self) -> None: + """Window ownership and lock strength stay separate concerns.""" + banned = {"grab_release", "grab_set", "grab_set_global", "restore_vt_switching"} + assert symbols("_surfaces.py") & banned == set() + + def test_detect_thread_never_touches_tk(self) -> None: + """Tk is not thread-safe; the RandR thread may only queue.""" + source = (SOURCE_ROOT / "_detect.py").read_text(encoding="utf-8") + loop_body = source.split("def _loop(")[1].split("def ")[0] + assert "self._sink.put" in loop_body + for forbidden in ("geometry(", "deiconify(", "lift(", "grab_"): + assert forbidden not in loop_body + + +class TestMonotonicSequence: + """The whole incident, walked end to end.""" + + def test_lock_never_weakens( + self, + loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], + mock_root: MagicMock, + ) -> None: + """Across every transition the grab and VT are only ever asserted.""" + recovery, enumerator, surfaces = loop + hooks = surfaces._builder + with patch("gatelock._recovery.disable_vt_switching") as vt: + for step in (BOTH, DEAD_PRIMARY, ALL_DARK, FAILED, BOTH): + enumerator.scan.return_value = step + recovery.tick() + mock_root.grab_release.assert_not_called() + mock_root.destroy.assert_not_called() + hooks.on_close.assert_not_called() + assert "restore" not in str(vt.mock_calls) + + def test_dark_primary_keeps_the_visible_monitor( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """The exact 2026-07-25 state: the lock lands where it can be seen.""" + recovery, enumerator, surfaces = loop + enumerator.scan.return_value = BOTH + recovery.tick() + enumerator.scan.return_value = DEAD_PRIMARY + report = recovery.tick() + assert report.live_outputs == ("HDMI-0",) + assert sorted(surfaces.names()) == ["HDMI-0"] + assert report.blind == () + + +class TestTick: + """Per-branch behaviour of one pass.""" + + def test_failed_scan_changes_nothing( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """No information is never a reason to touch anything.""" + recovery, enumerator, surfaces = loop + enumerator.scan.return_value = BOTH + recovery.tick() + before = sorted(surfaces.names()) + enumerator.scan.return_value = FAILED + report = recovery.tick() + assert report.scan_ok is False + assert report.delta is None + assert sorted(surfaces.names()) == before + + def test_zero_live_outputs_stays_armed( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """All monitors dark: nothing shown, nothing released.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = ALL_DARK + report = recovery.tick() + assert report.live_outputs == () + mock_root.grab_release.assert_not_called() + + def test_blind_output_is_reported( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """A live output that failed to map is surfaced as an error.""" + recovery, enumerator, surfaces = loop + enumerator.scan.return_value = BOTH + recovery.tick() + surfaces._surfaces["DP-0"].window.withdraw() + report = recovery.tick() + assert "DP-0" in report.blind + + def test_tick_counter( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """Ticks are counted, for the periodic VT re-assert.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + recovery.tick() + recovery.tick() + assert recovery.ticks == 2 + + +class TestGrabReassertion: + """Re-taking a lost grab, never releasing one.""" + + def test_reasserts_when_lost( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Something stole the grab; we take it back.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + report = recovery.tick() + assert report.grab_reasserted is True + mock_root.grab_set_global.assert_called() + + def test_noop_when_still_held( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Holding the grab already means no work.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = mock_root + assert recovery.tick().grab_reasserted is False + + def test_tclerror_while_regrabbing_is_swallowed( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """A failed re-grab is retried next tick, not raised.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + mock_root.grab_set_global.side_effect = tk.TclError("held") + assert recovery.tick().grab_reasserted is False + + def test_grab_current_tclerror( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """A TclError querying the grab counts as not holding it.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + mock_root.grab_current.side_effect = tk.TclError("gone") + assert recovery.tick().grab_reasserted is True + + def test_no_reassert_when_grab_not_global(self, mock_root: MagicMock) -> None: + """Soft mode takes no grab, so there is nothing to re-assert.""" + config = LockConfig(mode="soft") + surfaces = SurfaceSet(mock_root, config, MagicMock()) + enumerator = MagicMock() + enumerator.scan.return_value = BOTH + recovery = RecoveryLoop(mock_root, config, surfaces, enumerator, MagicMock()) + assert recovery.tick().grab_reasserted is False + + +class TestVtReassertion: + """Periodically re-disabling VT switching, never re-enabling it.""" + + def test_reasserts_on_the_thirtieth_tick( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """VT is re-disabled every 30 ticks.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + with patch("gatelock._recovery.disable_vt_switching") as vt: + reports = [recovery.tick() for _ in range(30)] + assert [i for i, r in enumerate(reports, 1) if r.vt_reasserted] == [30] + vt.assert_called_once_with() + + def test_never_when_vt_not_managed(self, mock_root: MagicMock) -> None: + """An app that does not disable VT never re-asserts it.""" + config = LockConfig(mode="soft") + surfaces = SurfaceSet(mock_root, config, MagicMock()) + enumerator = MagicMock() + enumerator.scan.return_value = BOTH + recovery = RecoveryLoop(mock_root, config, surfaces, enumerator, MagicMock()) + assert all(not recovery.tick().vt_reasserted for _ in range(31)) + + +class TestScheduling: + """The two cadences.""" + + def test_start_schedules_both( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Starting queues a drain and a verify.""" + recovery, _enumerator, _surfaces = loop + recovery.start() + assert mock_root.after.call_count == 2 + + def test_drain_ticks_only_when_signalled( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """No change signal means no full pass.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + recovery._detector.take_pending.return_value = False + recovery._running = True + recovery._drain() + enumerator.scan.assert_not_called() + + def test_drain_ticks_when_signalled( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """A pending change triggers a full pass.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + recovery._detector.take_pending.return_value = True + recovery._running = True + recovery._drain() + enumerator.scan.assert_called_once_with() + + def test_verify_reschedules( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet] + ) -> None: + """The verify cadence re-arms itself.""" + recovery, enumerator, _surfaces = loop + enumerator.scan.return_value = BOTH + recovery._running = True + recovery._verify() + assert recovery.ticks == 1 + + def test_stop_cancels_and_halts( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Stopping cancels timers and prevents rescheduling.""" + recovery, _enumerator, _surfaces = loop + recovery.start() + recovery.stop() + assert mock_root.after_cancel.call_count == 2 + mock_root.after.reset_mock() + recovery._schedule_drain() + recovery._schedule_verify() + mock_root.after.assert_not_called() + + def test_stop_swallows_cancel_error( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """A timer already gone is not an error.""" + recovery, _enumerator, _surfaces = loop + recovery.start() + mock_root.after_cancel.side_effect = tk.TclError("no such id") + recovery.stop() + + def test_stop_before_start( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Stopping a loop that never started is safe.""" + recovery, _enumerator, _surfaces = loop + recovery.stop() + mock_root.after_cancel.assert_not_called() diff --git a/gatelock/gatelock/tests/test_surfaces.py b/gatelock/gatelock/tests/test_surfaces.py new file mode 100644 index 0000000..3d7f08f --- /dev/null +++ b/gatelock/gatelock/tests/test_surfaces.py @@ -0,0 +1,282 @@ +"""Tests for the surface set: one lock window per live output.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from gatelock._outputs import Output, OutputRect, OutputScan +from gatelock._surfaces import ( + SurfaceInfo, + SurfaceSet, + TextMirror, + mirror_text_widgets, + needs_backdrop_root, +) +from gatelock._window import LockConfig + +DP0 = Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True) +HDMI = Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440)) +DP0_DARK = Output("DP-0", connected=True, rect=None, primary=True) +DP0_MOVED = Output( + "DP-0", connected=True, rect=OutputRect(0, 0, 1920, 1080), primary=True +) + + +def scan(*outputs: Output) -> OutputScan: + """Build a successful scan over ``outputs``.""" + return OutputScan(outputs=outputs, source="randr", ok=True) + + +@pytest.fixture +def surfaces(mock_root: MagicMock) -> SurfaceSet: + """A hard-mode surface set over a mock root.""" + return SurfaceSet(mock_root, LockConfig(mode="hard"), MagicMock()) + + +class TestNeedsBackdropRoot: + """The predicate that must NOT be simplified to `iff overrideredirect`.""" + + @pytest.mark.parametrize( + ("config", "expected"), + [ + (LockConfig(mode="hard"), True), + (LockConfig(mode="soft"), False), + (LockConfig(mode="soft", grab="global"), True), + (LockConfig(mode="soft", grab="local"), True), + (LockConfig(mode="hard", overrideredirect=False, grab="none"), False), + ], + ) + def test_predicate(self, config: LockConfig, *, expected: bool) -> None: + """A global grab needs a viewable root even without overrideredirect.""" + assert needs_backdrop_root(config) is expected + + +class TestApply: + """Bringing the surface set in line with a scan.""" + + def test_creates_one_per_live_output(self, surfaces: SurfaceSet) -> None: + """Each live output gets its own window at its own rectangle.""" + delta = surfaces.apply(scan(DP0, HDMI)) + assert sorted(delta.created) == ["DP-0", "HDMI-0"] + assert delta.changed is True + infos = surfaces.infos() + assert [i.output_name for i in infos] == ["DP-0", "HDMI-0"] + assert infos[0].rect == OutputRect(0, 0, 3840, 2160) + assert infos[0].is_primary is True + + def test_skips_dark_outputs(self, surfaces: SurfaceSet) -> None: + """A connected-but-modeless output gets no surface.""" + surfaces.apply(scan(DP0_DARK, HDMI)) + assert sorted(surfaces.names()) == ["HDMI-0"] + + def test_removes_surface_when_output_goes_dark(self, surfaces: SurfaceSet) -> None: + """Losing a mode tears that surface down and tells the app.""" + surfaces.apply(scan(DP0, HDMI)) + builder = surfaces._builder + delta = surfaces.apply(scan(DP0_DARK, HDMI)) + assert delta.removed == ("DP-0",) + assert sorted(surfaces.names()) == ["HDMI-0"] + builder.teardown_surface.assert_called_once() + + def test_moves_in_place_rather_than_recreating(self, surfaces: SurfaceSet) -> None: + """A resolution change must not discard half-typed input.""" + surfaces.apply(scan(DP0)) + builder = surfaces._builder + builder.build_surface.reset_mock() + delta = surfaces.apply(scan(DP0_MOVED)) + assert delta.moved == ("DP-0",) + assert delta.created == () + builder.build_surface.assert_not_called() + assert surfaces.infos()[0].rect == OutputRect(0, 0, 1920, 1080) + + def test_unchanged_layout_is_a_noop(self, surfaces: SurfaceSet) -> None: + """Re-applying the same scan changes nothing (Configure-storm guard).""" + surfaces.apply(scan(DP0, HDMI)) + delta = surfaces.apply(scan(DP0, HDMI)) + assert delta.changed is False + + def test_zero_live_outputs_removes_everything(self, surfaces: SurfaceSet) -> None: + """All dark: no surfaces, and nothing blind (there is nothing to see).""" + surfaces.apply(scan(DP0, HDMI)) + delta = surfaces.apply(scan(DP0_DARK)) + assert sorted(delta.removed) == ["DP-0", "HDMI-0"] + assert surfaces.names() == frozenset() + assert delta.unverified == () + + +class TestVerify: + """The blind-output alarm.""" + + def test_nothing_blind_when_all_mapped(self, surfaces: SurfaceSet) -> None: + """A mapped surface on every live output means nothing is blind.""" + delta = surfaces.apply(scan(DP0, HDMI)) + assert delta.unverified == () + + def test_unmapped_surface_is_blind(self, surfaces: SurfaceSet) -> None: + """A live output whose window is unmapped is reported.""" + surfaces.apply(scan(DP0)) + surfaces._surfaces["DP-0"].window.withdraw() + assert surfaces.verify() == ("DP-0",) + + def test_missing_surface_is_blind(self, surfaces: SurfaceSet) -> None: + """A live output with no surface at all is reported.""" + assert surfaces.verify(live_names=frozenset({"DP-0"})) == ("DP-0",) + + +class TestEnforce: + """Re-assertion only ever adds visibility.""" + + def test_remaps_an_unmapped_surface(self, surfaces: SurfaceSet) -> None: + """An unmapped surface is deiconified again.""" + surfaces.apply(scan(DP0)) + window = surfaces._surfaces["DP-0"].window + window.withdraw() + assert surfaces.enforce() == ("DP-0",) + assert window.winfo_ismapped() is True + + def test_corrects_drifted_geometry(self, surfaces: SurfaceSet) -> None: + """A surface moved by something else is put back.""" + surfaces.apply(scan(DP0)) + window = surfaces._surfaces["DP-0"].window + window.geometry("100x100+50+50") + assert surfaces.enforce() == ("DP-0",) + assert window.winfo_width() == 3840 + + def test_stable_surface_needs_no_correction(self, surfaces: SurfaceSet) -> None: + """A correct surface reports nothing, but is still raised.""" + surfaces.apply(scan(DP0)) + assert surfaces.enforce() == () + surfaces._surfaces["DP-0"].window.lift.assert_called() + + def test_soft_mode_does_not_reassert_overrideredirect( + self, mock_root: MagicMock + ) -> None: + """A managed window must not be forced override-redirect.""" + soft = SurfaceSet(mock_root, LockConfig(mode="soft"), MagicMock()) + soft.apply(scan(DP0)) + soft.enforce() + soft._surfaces["DP-0"].window.overrideredirect.assert_not_called() + + def test_raise_all(self, surfaces: SurfaceSet) -> None: + """Every surface can be lifted at once.""" + surfaces.apply(scan(DP0, HDMI)) + surfaces.raise_all() + for surface in surfaces._surfaces.values(): + surface.window.lift.assert_called() + + +class TestFocus: + """Which surface takes the keyboard.""" + + def test_focus_surface_by_index(self, surfaces: SurfaceSet) -> None: + """Focusing an index returns that surface's info.""" + surfaces.apply(scan(DP0, HDMI)) + info = surfaces.focus_surface(1) + assert info is not None + assert info.output_name == "HDMI-0" + + def test_focus_unknown_index(self, surfaces: SurfaceSet) -> None: + """An index with no surface returns None.""" + surfaces.apply(scan(DP0)) + assert surfaces.focus_surface(9) is None + + def test_preferred_index_is_the_primary(self, surfaces: SurfaceSet) -> None: + """The live primary output takes initial focus.""" + surfaces.apply(scan(HDMI, DP0)) + assert surfaces.preferred_focus_index() == 1 + + def test_preferred_index_defaults_to_first(self, surfaces: SurfaceSet) -> None: + """With no primary marked, the first surface wins.""" + surfaces.apply(scan(HDMI)) + assert surfaces.preferred_focus_index() == 0 + + +class TestBackdropAndTeardown: + """The root backdrop and full teardown.""" + + def test_backdrop_covers_the_whole_screen( + self, surfaces: SurfaceSet, mock_root: MagicMock + ) -> None: + """The backdrop spans the X screen, covering dead columns.""" + surfaces.update_backdrop() + mock_root.geometry.assert_called_once_with("1920x1080+0+0") + + def test_soft_mode_has_no_backdrop(self, mock_root: MagicMock) -> None: + """Without a grab or overrideredirect there is nothing to back.""" + SurfaceSet(mock_root, LockConfig(mode="soft"), MagicMock()).update_backdrop() + mock_root.geometry.assert_not_called() + + def test_destroy_all(self, surfaces: SurfaceSet) -> None: + """Closing tears down every surface.""" + surfaces.apply(scan(DP0, HDMI)) + surfaces.destroy_all() + assert surfaces.names() == frozenset() + + +class TestTextMirror: + """tk.Text has no textvariable, so it needs explicit mirroring.""" + + def _widget(self, text: str = "") -> MagicMock: + widget = MagicMock() + state = {"text": text} + widget.get.side_effect = lambda *_a: state["text"] + widget.delete.side_effect = lambda *_a: state.update(text="") + widget.insert.side_effect = lambda _i, value: state.update(text=value) + widget._state = state + return widget + + def test_typing_in_one_updates_the_other(self) -> None: + """A key release in surface A reaches surface B.""" + a, b = self._widget("hello"), self._widget() + var = MagicMock() + var.get.return_value = "hello" + mirror = TextMirror([a, b], var) + event = MagicMock() + event.widget = a + mirror._on_key_release(event) + var.set.assert_called_once_with("hello") + assert b._state["text"] == "hello" + + def test_var_write_fans_out(self) -> None: + """Setting the shared variable updates every widget.""" + a, b = self._widget(), self._widget() + var = MagicMock() + var.get.return_value = "typed" + mirror = TextMirror([a, b], var) + mirror._on_var_write("x", "", "w") + assert a._state["text"] == "typed" + assert b._state["text"] == "typed" + + def test_reentrancy_is_blocked(self) -> None: + """A fan-out must not trigger another fan-out.""" + a = self._widget() + var = MagicMock() + var.get.return_value = "v" + mirror = TextMirror([a], var) + mirror._syncing = True + mirror._on_var_write("x", "", "w") + assert a._state["text"] == "" + event = MagicMock() + event.widget = a + mirror._on_key_release(event) + var.set.assert_not_called() + + def test_helper_returns_mirror(self) -> None: + """The public helper wires the widgets up.""" + widget = self._widget() + var = MagicMock() + assert isinstance(mirror_text_widgets([widget], var), TextMirror) + widget.bind.assert_called_once() + + +class TestSurfaceInfo: + """SurfaceInfo equality drives the move-vs-noop decision.""" + + def test_identical_infos_compare_equal(self) -> None: + """Equality is by value, so an unchanged layout is a no-op.""" + rect = OutputRect(0, 0, 100, 100) + first = SurfaceInfo("DP-0", rect, 0, is_primary=True) + second = SurfaceInfo("DP-0", rect, 0, is_primary=True) + assert first == second diff --git a/gatelock/gatelock/tests/test_window.py b/gatelock/gatelock/tests/test_window.py index 42edc2b..66454f3 100644 --- a/gatelock/gatelock/tests/test_window.py +++ b/gatelock/gatelock/tests/test_window.py @@ -58,8 +58,14 @@ class TestSetup: window.setup() mock_root.overrideredirect.assert_called_once_with(boolean=True) - mock_root.attributes.assert_any_call(fullscreen=True) - mock_root.attributes.assert_any_call(topmost=True) + # No attributes(fullscreen=True): that is an EWMH request an + # override-redirect window is invisible to, and where it does apply it + # snaps to one monitor or the whole bounding box. Per-output surfaces + # use explicit geometry instead. + assert not any( + call.kwargs.get("fullscreen") or "-fullscreen" in call.args + for call in mock_root.attributes.call_args_list + ) mock_disable.assert_called_once() assert window._vt_disabled is True @@ -84,7 +90,11 @@ class TestSetup: window.setup() - mock_root.configure.assert_called_once_with(bg="#000000", cursor="arrow") + # Soft mode with no grab needs no backdrop, so the root is left to the + # window manager and only the surfaces carry the colour. + assert window.surfaces.infos() + topmost_on = True + mock_root.attributes.assert_any_call("-topmost", topmost_on) class TestGrabInput: @@ -100,7 +110,7 @@ class TestGrabInput: window.grab_input() mock_acquire.assert_called_once_with(attempt=1) - mock_root.after.assert_called_once_with(100, window._notify_focus_ready) + mock_root.after.assert_any_call(100, window._notify_focus_ready) def test_local_grab_calls_grab_set(self, mock_root: MagicMock) -> None: """grab="local" calls grab_set directly, no retry logic.""" @@ -144,7 +154,7 @@ class TestAcquireGlobalGrab: mock_root.grab_set_global.assert_called_once_with() mock_root.focus_force.assert_called_once_with() - hooks.on_focus_ready.assert_called_once_with() + hooks.on_focus_ready.assert_called_once_with(None) def test_success_swallows_tclerror_from_focus(self, mock_root: MagicMock) -> None: """A TclError while focusing after a successful grab is swallowed.""" @@ -248,7 +258,7 @@ class TestNotifyFocusReady: window._notify_focus_ready() - hooks.on_focus_ready.assert_called_once_with() + hooks.on_focus_ready.assert_called_once_with(None) def test_swallows_tclerror_from_hook(self, mock_root: MagicMock) -> None: """A TclError raised by the hook (widget already destroyed) is swallowed.""" diff --git a/gatelock/gatelock/tests/test_window_surfaces.py b/gatelock/gatelock/tests/test_window_surfaces.py new file mode 100644 index 0000000..f6e97b0 --- /dev/null +++ b/gatelock/gatelock/tests/test_window_surfaces.py @@ -0,0 +1,173 @@ +"""Tests for LockWindow's v0.2.0 behaviour: arming, holder naming, teardown.""" + +from __future__ import annotations + +import queue +from unittest.mock import MagicMock, patch + +from gatelock._arbiter import RANK_DIET_GUARD, Claim +from gatelock._detect import _RandrEventSource +from gatelock._outputs import Output, OutputRect +from gatelock._window import LockConfig +from gatelock.tests.conftest import make_window + +# Captured before the autouse hermetic fixture replaces it. +REAL_RANDR_START = _RandrEventSource.start + +DARK_ONLY = (Output("DP-0", connected=True, rect=None, primary=True),) +TWO_LIVE = ( + Output("DP-0", connected=True, rect=OutputRect(0, 0, 3840, 2160), primary=True), + Output("HDMI-0", connected=True, rect=OutputRect(3840, 0, 2560, 1440)), +) + + +class TestArmWithoutDisplay: + """Output count must never gate arming.""" + + def test_zero_live_outputs_still_arms(self, mock_root: MagicMock) -> None: + """The 2b rule: dark monitors mean lock silently, not decline to lock.""" + window, _hooks = make_window(mock_root, config=LockConfig(mode="hard")) + with patch("gatelock._outputs.scan_xrandr", return_value=DARK_ONLY): + window.setup() + assert window.surfaces.names() == frozenset() + # Armed regardless: VT was disabled and the backdrop is up. + assert window._vt_disabled is True + mock_root.overrideredirect.assert_called_once_with(boolean=True) + + def test_reports_outputs_live_but_uncovered(self, mock_root: MagicMock) -> None: + """A live output with no surface is an error at startup, not silence.""" + window, _hooks = make_window(mock_root, config=LockConfig(mode="hard")) + with ( + patch("gatelock._outputs.scan_xrandr", return_value=TWO_LIVE), + patch( + "gatelock._surfaces.SurfaceSet.verify", + return_value=("HDMI-0",), + ), + ): + window.setup() # must log, not raise + + def test_builds_one_surface_per_live_output(self, mock_root: MagicMock) -> None: + """Two live monitors means two surfaces, each at its own rectangle.""" + window, hooks = make_window(mock_root, config=LockConfig(mode="hard")) + with patch("gatelock._outputs.scan_xrandr", return_value=TWO_LIVE): + window.setup() + assert sorted(window.surfaces.names()) == ["DP-0", "HDMI-0"] + assert hooks.build_surface.call_count == 2 + + +class TestGrabBlockedLogging: + """Naming the real holder instead of blaming a fullscreen game.""" + + def test_names_the_holding_gatelock_app(self, mock_root: MagicMock) -> None: + """The 2026-07-25 diagnosis, available immediately.""" + arbiter = MagicMock() + arbiter.describe_holder.return_value = Claim( + app="diet_guard", + rank=RANK_DIET_GUARD, + pid=3353, + started="2026-07-25T12:49:40+00:00", + grab="global", + disable_vt=True, + instance_id="t", + ) + window, _hooks = make_window(mock_root, config=LockConfig(mode="hard")) + window._arbiter = arbiter + with patch("gatelock._window._logger") as logger: + window._log_grab_blocked(25) + assert "diet_guard" in str(logger.warning.call_args) + + def test_falls_back_when_no_gatelock_app_holds_it( + self, mock_root: MagicMock + ) -> None: + """With no gatelock holder, a foreign X client is the honest guess.""" + window, _hooks = make_window(mock_root, config=LockConfig(mode="hard")) + with patch("gatelock._window._logger") as logger: + window._log_grab_blocked(25) + assert "fullscreen game" in str(logger.warning.call_args) + + def test_arbiter_without_holder(self, mock_root: MagicMock) -> None: + """An arbiter that reports no holder takes the same path.""" + arbiter = MagicMock() + arbiter.describe_holder.return_value = None + window, _hooks = make_window(mock_root, config=LockConfig(mode="hard")) + window._arbiter = arbiter + with patch("gatelock._window._logger") as logger: + window._log_grab_blocked(50) + assert "fullscreen game" in str(logger.warning.call_args) + + +class TestFocusNotifiedOnce: + """Both call sites are load-bearing, so the guard is on the callee.""" + + def test_second_notification_is_suppressed(self, mock_root: MagicMock) -> None: + """The hook fires once even though two timings can reach it.""" + window, hooks = make_window(mock_root, config=LockConfig(mode="hard")) + window._notify_focus_ready() + window._notify_focus_ready() + hooks.on_focus_ready.assert_called_once() + + +class TestCloseReleasesTheScreen: + """A clean exit must hand the screen to the next app.""" + + def test_arbiter_is_released(self, mock_root: MagicMock) -> None: + """Without this the alarm -> workout handoff strands the workout lock.""" + arbiter = MagicMock() + window, _hooks = make_window(mock_root, config=LockConfig(mode="hard")) + window._arbiter = arbiter + window.close() + arbiter.release.assert_called_once_with() + + def test_close_without_arbiter(self, mock_root: MagicMock) -> None: + """An unarbitrated lock still closes cleanly.""" + window, hooks = make_window(mock_root, config=LockConfig(mode="hard")) + window.close() + hooks.on_close.assert_called_once_with() + + +class TestRemainingBranches: + """Small paths the other suites do not reach.""" + + def test_claim_read_oserror(self, tmp_path: object) -> None: + """A claim whose contents cannot be read is treated as absent.""" + from gatelock._arbiter import Arbiter + + arbiter = Arbiter("a", RANK_DIET_GUARD, grab="none", disable_vt=False) + arbiter.publish() + handle = MagicMock() + handle.__enter__.return_value = handle + handle.read.side_effect = OSError("io error") + with ( + patch("gatelock._arbiter.Path.open", return_value=handle), + patch("gatelock._arbiter._try_lock", return_value=False), + ): + assert arbiter.live_claims() == () + arbiter.release() + + def test_randr_start_spawns_thread(self) -> None: + """A successful connect starts the watcher thread.""" + source = _RandrEventSource(queue.Queue()) + with ( + patch.object(source, "_connect", return_value=True), + patch("gatelock._detect.threading.Thread") as thread_cls, + ): + assert REAL_RANDR_START(source) is True + thread_cls.return_value.start.assert_called_once_with() + + def test_randr_loop_ignores_empty_batch(self) -> None: + """A readable socket with no events posts no notice.""" + sink: queue.Queue[str] = queue.Queue() + source = _RandrEventSource(sink) + source._display = MagicMock() + source._display.pending_events.return_value = 0 + calls = {"n": 0} + + def is_set() -> bool: + calls["n"] += 1 + return calls["n"] > 1 + + source._stop = MagicMock() + source._stop.is_set.side_effect = is_set + with patch("gatelock._detect.select.select", return_value=([1], [], [])): + source._loop() + assert sink.empty() diff --git a/gatelock/pyproject.toml b/gatelock/pyproject.toml index 11d097a..333a430 100644 --- a/gatelock/pyproject.toml +++ b/gatelock/pyproject.toml @@ -1,10 +1,25 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + [project] name = "gatelock" -version = "0.1.1" -description = "Shared fullscreen lock-window + HMAC log-integrity backend for blocking-overlay apps" +version = "0.2.0" +description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps" requires-python = ">=3.10" dependencies = [] # pure stdlib — tkinter is bundled with Python +[project.optional-dependencies] +# python-xlib buys instant RandR change events. Without it the lock still +# detects output changes via Tk and xrandr polling, just less +# promptly — so this must stay OPTIONAL: a locker that refuses to start +# because a dependency is missing is itself a way to end up unlocked. +# Consumers' install scripts install it unconditionally all the same. +randr = ["python-xlib>=0.33"] + +[tool.setuptools.packages.find] +include = ["gatelock*"] + [tool.ruff] target-version = "py310" include = ["*.py", "**/*.py"] @@ -18,6 +33,12 @@ ignore = [ "COM812", "ISC001", "S603", + # Optional dependencies (python-xlib) are imported inside functions on + # purpose. A module-scope try/except ImportError resolves at import time, + # which makes the except arm permanently unreachable for branch coverage -- + # and that arm is the degradation path we must keep tested. Per-line + # suppressions are banned here, so the rule is disabled repo-wide. + "PLC0415", ] fixable = ["ALL"] unfixable = []