gatelock v0.2.0: per-output locking, monotonic recovery, N-app arbitration

On 2026-07-25 the PC appeared frozen and needed an SSH rescue. Three defects
stacked:

1. DP-0 came up `connected` with a valid EDID and no mode/CRTC -- physically
   dark, while X still reserved its 3840px slice of the root window.
2. diet-guard-gate and workout-locker each mapped an override-redirect window
   spanning the WHOLE root (6400x1440+0+0), centring their controls on the
   primary output -- the dark one. The escape UI was invisible.
3. They deadlocked on the X grab, and the loser logged "another app -- e.g. a
   fullscreen game -- holds it". No game was running; the holder was the other
   locker. That guess sent the diagnosis the wrong way for the whole outage.

The library had no concept of an individual output: setup() sized the window
from winfo_screenwidth(), the union bounding box of every output.

What changes:

* _outputs.py  An output is LIVE only if connected AND has a mode/CRTC.
               RandR (crtc==0 or mode==0) and an xrandr --query parser derive
               that predicate independently; verified to agree exactly on the
               real DP-0/HDMI-0 layout. Fails closed three distinct ways.
* _surfaces.py One window per live output at its own geometry, and the only
               module allowed to create/move/destroy windows. Explicit
               geometry replaces attributes(fullscreen=True), which an
               override-redirect window is invisible to anyway.
* _detect.py   RandR events + Tk <Configure> + xrandr polling, coalesced. The
               RandR thread runs on a private X connection and may only call
               queue.put -- Tk is not thread-safe.
* _recovery.py Re-asserts coverage, placement, grab and VT on a timer. Proven
               monotonic: it may only ADD coverage or RE-ACQUIRE the grab. An
               ast test asserts the file contains none of grab_release /
               restore_vt_switching / close / destroy / quit / withdraw, while
               disable_vt_switching and grab_set_global ARE present. Yanking
               every cable therefore does not unlock anything.
* _arbiter.py  N-app priority: wake_alarm > screen_locker > diet_guard.
               Liveness is proved by flock-on-death -- no heartbeat, no PID
               scan, no clock. A two-axis strength check stops rank ever
               weakening the lock: an app stands down only for an incumbent
               at least as strong on BOTH grab kind and VT disabling.
* _guards.py   wait_for_x_server replaces wait_for_display and gates ONLY on
               an X server existing. Output count must never gate arming:
               zero live outputs means lock without showing, not decline to
               lock. That inversion is what left the machine unlocked.
* _window.py   Arms first, renders second. Names the real grab holder instead
               of blaming a fullscreen game. Releases the arbiter on close so
               the alarm -> workout handoff cannot strand the next app.

Two bugs found by running it, not reading it:

* acquire_holder() opened the holder file "w", which truncates at open() --
  before the lock attempt. A losing contender erased the incumbent's claim, so
  the app that most needed to know who held the screen found an empty file.
* tk.Toplevel over a MagicMock sends real tkinter into infinite mock recursion;
  the suite hung instead of failing, and an unpatched setup() would have run
  setxkbmap against the developer's own machine. conftest now has an autouse
  hermetic fixture; the suite went from >120s to 0.7s.

235 tests, 100% branch coverage, ruff (select=ALL) and mypy --strict clean.
No suppressions added: Protocols replace Any for python-xlib, public attribute
access replaces ._data, and type-only Protocols live under TYPE_CHECKING.

Consumers still pin gatelock-v0.1.1 and are unmigrated; nothing ships until a
tag and a reinstall into system python.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-07-25 14:30:03 +02:00
parent a50d27688f
commit a9b65812d7
18 changed files with 3943 additions and 46 deletions

View File

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

View File

@ -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)

View File

@ -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 ``<Configure>`` 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 <Configure> "
"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 ``<Configure>`` and, if possible, RandR events."""
self._binding = self._root.bind("<Configure>", 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("<Configure>", self._binding)
self._binding = None

View File

@ -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

View File

@ -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<name>\S+)\s+(?P<state>connected|disconnected)\b(?P<rest>.*)$",
re.MULTILINE,
)
_GEOMETRY = re.compile(r"(?P<w>\d+)x(?P<h>\d+)\+(?P<x>-?\d+)\+(?P<y>-?\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()

View File

@ -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

View File

@ -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("<KeyRelease>", 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
)

View File

@ -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()

View File

@ -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

View File

@ -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

View File

@ -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 <Configure> 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] == "<Configure>"
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("<Configure>", "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

View File

@ -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

View File

@ -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", "<EFBFBD>")],
)
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"]

View File

@ -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()

View File

@ -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

View File

@ -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."""

View File

@ -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()

View File

@ -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 <Configure> 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 = []