From b3e629e3f0af06dff0fc5f3f8cbe9264c90e1f86 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 28 Aug 2026 19:24:14 +0200 Subject: [PATCH] Bill gaming time only while the user is actually playing A running game process was treated as proof of play. On 2026-08-28 the budget billed ~2h23m while a gatelock screen locker covered the display and the machine could not be touched at all -- the game was simply still resident. A tick now bills only when all three hold, evaluated together rather than short-circuited so the audit log can answer "was the screen ALSO locked?": * no gatelock guard owns the screen (_screen_hold, via /proc/locks) * the focused window belongs to a qualifying process or its ancestry * some input source has been used within idle_grace_seconds (default 300) Any probe that cannot answer BILLS the tick and is recorded in `degraded`. A broken detector must never become an unlimited-gaming exploit. Three findings from measuring this machine rather than assuming: * /dev/input/js0 here is an *ASRock LED Controller* -- the kernel exposes the motherboard RGB device as a joystick. A presence check would have disabled the idle criterion permanently, so _controller_input watches for real events instead (skipping joydev's synthetic open burst). * Aseprite's window carries no _NET_WM_PID at all, so focus attribution falls back to XRes, which asks the server who owns the window rather than trusting the client to have announced itself. * /proc/locks reports gatelock's hold without taking it; gatelock's own read_claim_if_held detects held-ness by acquiring, which would race a guard every 3 seconds. Idle pauses also refund the grace period once per walk-away, on the engaged->paused edge only, never across a day rollover, and never on the first tick after a restart -- the only place the counter moves backwards. Everything is recorded to /var/log/steam-backlog-enforcer/budget.jsonl, on change plus a heartbeat rather than one line per 3s tick. Records name the qualifying PIDs: reconstructing this incident from journald alone was only possible by inferring accrual from the spacing of two warning notifications. Also: refuse to bounce Steam while a game is running. ensure_steam_debug_port relaunches Steam when the CDP port is closed, which kills the running game -- that is how restarting the daemon to deploy this change ended a live session. It now defers exactly as it already did for in-flight game updates. Talks to X through ctypes over one cached connection: at a 3s tick, forking xprintidle and xprop would cost ~40 forks a minute. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_019CEsG6BzHtYR9cHcdxPZqG --- steam_backlog_enforcer/_controller_input.py | 153 +++++++++++++ steam_backlog_enforcer/_enforce_loop.py | 10 +- steam_backlog_enforcer/_engagement_probes.py | 168 ++++++++++++++ steam_backlog_enforcer/_engagement_types.py | 91 ++++++++ steam_backlog_enforcer/_playtime.py | 26 ++- .../_playtime_engagement.py | 164 ++++++++++++++ steam_backlog_enforcer/_playtime_log.py | 209 ++++++++++++++++++ steam_backlog_enforcer/_playtime_session.py | 83 +++++++ steam_backlog_enforcer/_playtime_state.py | 6 + steam_backlog_enforcer/_screen_hold.py | 124 +++++++++++ steam_backlog_enforcer/_steam_errors.py | 12 + steam_backlog_enforcer/_steam_launch.py | 30 +-- .../_steam_restart_guard.py | 66 ++++++ steam_backlog_enforcer/_x_focus.py | 176 +++++++++++++++ steam_backlog_enforcer/_x_probe.py | 185 ++++++++++++++++ steam_backlog_enforcer/config.py | 8 + steam_backlog_enforcer/tests/_fake_session.py | 123 +++++++++++ steam_backlog_enforcer/tests/_fake_xlib.py | 99 +++++++++ steam_backlog_enforcer/tests/conftest.py | 21 ++ .../tests/test_controller_input.py | 137 ++++++++++++ .../tests/test_enforce_loop_part2.py | 15 +- .../tests/test_enforce_loop_part3.py | 15 +- .../tests/test_engagement_probes.py | 155 +++++++++++++ .../tests/test_engagement_types.py | 50 +++++ .../tests/test_playtime_engagement.py | 201 +++++++++++++++++ .../tests/test_playtime_log.py | 192 ++++++++++++++++ .../tests/test_playtime_session.py | 33 +++ .../tests/test_playtime_tick.py | 17 +- .../tests/test_screen_hold.py | 111 ++++++++++ .../tests/test_steam_restart_guard.py | 71 ++++++ .../tests/test_update_restart_guard.py | 17 +- steam_backlog_enforcer/tests/test_x_focus.py | 187 ++++++++++++++++ steam_backlog_enforcer/tests/test_x_probe.py | 120 ++++++++++ 33 files changed, 3031 insertions(+), 44 deletions(-) create mode 100644 steam_backlog_enforcer/_controller_input.py create mode 100644 steam_backlog_enforcer/_engagement_probes.py create mode 100644 steam_backlog_enforcer/_engagement_types.py create mode 100644 steam_backlog_enforcer/_playtime_engagement.py create mode 100644 steam_backlog_enforcer/_playtime_log.py create mode 100644 steam_backlog_enforcer/_playtime_session.py create mode 100644 steam_backlog_enforcer/_screen_hold.py create mode 100644 steam_backlog_enforcer/_steam_restart_guard.py create mode 100644 steam_backlog_enforcer/_x_focus.py create mode 100644 steam_backlog_enforcer/_x_probe.py create mode 100644 steam_backlog_enforcer/tests/_fake_session.py create mode 100644 steam_backlog_enforcer/tests/_fake_xlib.py create mode 100644 steam_backlog_enforcer/tests/test_controller_input.py create mode 100644 steam_backlog_enforcer/tests/test_engagement_probes.py create mode 100644 steam_backlog_enforcer/tests/test_engagement_types.py create mode 100644 steam_backlog_enforcer/tests/test_playtime_engagement.py create mode 100644 steam_backlog_enforcer/tests/test_playtime_log.py create mode 100644 steam_backlog_enforcer/tests/test_playtime_session.py create mode 100644 steam_backlog_enforcer/tests/test_screen_hold.py create mode 100644 steam_backlog_enforcer/tests/test_steam_restart_guard.py create mode 100644 steam_backlog_enforcer/tests/test_x_focus.py create mode 100644 steam_backlog_enforcer/tests/test_x_probe.py diff --git a/steam_backlog_enforcer/_controller_input.py b/steam_backlog_enforcer/_controller_input.py new file mode 100644 index 0000000..4f119c8 --- /dev/null +++ b/steam_backlog_enforcer/_controller_input.py @@ -0,0 +1,153 @@ +"""Noticing that a game controller is being used. + +The X screensaver idle counter only tracks keyboards and pointers. A gamepad +session that goes five minutes without touching either would look idle, and +pausing the budget there would hand out free gaming — the single worst failure +mode of the engagement gate, because it rewards the behaviour it fails to see. + +Presence is not a usable proxy for use. Measured on this machine, ``/dev/input/js0`` +is an *ASRock LED Controller*: the kernel exposes the motherboard's RGB device as +a joystick. Treating "a js device exists" as "a controller is in play" would +disable the idle criterion permanently, on a machine with no gamepad attached. + +So this watches for actual events instead. Each joystick device is opened +non-blocking and drained every tick; joydev gives every reader its own event +queue, so draining ours cannot steal input from the game. The synthetic burst +joydev replays on open is flagged ``JS_EVENT_INIT`` and is skipped — without that +filter, merely opening the device would look like a button press. +""" + +from __future__ import annotations + +import contextlib +import errno +import logging +import os +from pathlib import Path +from typing import Final + +logger = logging.getLogger(__name__) + +_DEVICE_DIR: Final = Path("/dev/input") +_DEVICE_PREFIX: Final = "js" + +# struct js_event { __u32 time; __s16 value; __u8 type; __u8 number; } +_EVENT_SIZE: Final = 8 +_TYPE_OFFSET: Final = 6 +_JS_EVENT_INIT: Final = 0x80 + +# Cap per device per tick so a flood cannot stall the enforce loop. +_MAX_EVENTS_PER_POLL: Final = 64 + + +class ControllerActivity: + """Tracks when a joystick device last produced a real event.""" + + def __init__(self, *, device_dir: Path = _DEVICE_DIR) -> None: + """Start with no devices open and no activity seen. + + Args: + device_dir: Directory holding ``js*`` nodes. + """ + self._device_dir = device_dir + self._fds: dict[Path, int] = {} + self._last_event_at: float | None = None + + def poll(self, *, now: float) -> None: + """Drain pending controller events, recording any real activity. + + Args: + now: Monotonic timestamp for this tick. + """ + self._sync_devices() + for path in list(self._fds): + if self._drain(path): + self._last_event_at = now + + def seconds_since_activity(self, *, now: float) -> float | None: + """Return seconds since the last real controller event. + + Args: + now: Monotonic timestamp for this tick. + + Returns: + The elapsed seconds, or ``None`` if no event has been seen at all — + which the caller must not read as "idle", only as "no evidence". + """ + if self._last_event_at is None: + return None + return now - self._last_event_at + + def close(self) -> None: + """Close every open device.""" + for path in list(self._fds): + self._forget(path) + + def _sync_devices(self) -> None: + """Open newly appeared ``js*`` devices and forget vanished ones.""" + try: + present = { + entry + for entry in self._device_dir.iterdir() + if entry.name.startswith(_DEVICE_PREFIX) + } + except OSError: + present = set() + + for path in list(self._fds): + if path not in present: + self._forget(path) + + for path in present - set(self._fds): + self._open(path) + + def _open(self, path: Path) -> None: + """Open *path* for reading, ignoring devices we may not read. + + Args: + path: Device to open. + """ + try: + self._fds[path] = os.open(path, os.O_RDONLY | os.O_NONBLOCK) + except OSError: + # A js node we cannot open tells us nothing either way; the idle + # criterion simply falls back to keyboard and pointer input. + logger.debug("Cannot open controller device %s", path) + + def _drain(self, path: Path) -> bool: + """Read pending events from *path*. + + Args: + path: Device to drain. + + Returns: + Whether at least one non-synthetic event was read. + """ + fd = self._fds[path] + seen = False + for _ in range(_MAX_EVENTS_PER_POLL): + try: + chunk = os.read(fd, _EVENT_SIZE) + except OSError as exc: + if exc.errno != errno.EAGAIN: + self._forget(path) + break + if len(chunk) < _EVENT_SIZE: + break + if not chunk[_TYPE_OFFSET] & _JS_EVENT_INIT: + seen = True + return seen + + def _forget(self, path: Path) -> None: + """Close and drop *path*. + + Args: + path: Device to close. + """ + fd = self._fds.pop(path, None) + if fd is not None: + # The usual reason to forget a device is that reading it just + # failed, which often means the fd is already unusable; closing it + # again must not take the enforce loop down with it. + with contextlib.suppress(OSError): + os.close(fd) diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index b7b7d24..4d587d9 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -17,6 +17,7 @@ from steam_backlog_enforcer._enforce_steps import ( # from here, so splitting it into _owned_apps_cache stays invisible to them. from steam_backlog_enforcer._owned_apps_cache import get_all_owned_app_ids from steam_backlog_enforcer._playtime import playtime_tick +from steam_backlog_enforcer._playtime_session import PlaytimeSession, new_session from steam_backlog_enforcer._steam_launch import steam_is_installed from steam_backlog_enforcer._total_block import ( end_total_block_cleanup, @@ -103,6 +104,7 @@ def _enforce_loop_iteration( config: Config, state: State, *, + session: PlaytimeSession, demo: bool = False, ) -> None: """Perform one iteration of the enforcement loop. @@ -110,13 +112,14 @@ def _enforce_loop_iteration( Args: config: Enforcer configuration. state: Current enforcer state. + session: Cross-tick engagement and logging state. demo: Run the gaming budget on a 60-second demo budget. """ # Daily gaming budget runs FIRST and unconditionally. Every guard below # returns early in situations where gaming time is still being spent (total # block active, Steam absent, nothing assigned) and where a 06:00 release # still has to happen — gating this on any of them would strand the block. - playtime_tick(config, interval=ENFORCE_INTERVAL, demo=demo) + playtime_tick(config, interval=ENFORCE_INTERVAL, session=session, demo=demo) # Total block takes priority over the assigned-game enforcement below - # while active, don't fight ourselves (e.g. installing the assigned @@ -204,6 +207,9 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None: _echo(f" Enforce loop: ACTIVE (every {ENFORCE_INTERVAL}s)") _echo(" Guarding: processes + installs + store") _echo(" Press Ctrl+C to stop.\n") + # One session for the whole daemon: the engagement backdate needs to see + # the previous tick's verdict, so it cannot be rebuilt per iteration. + session = new_session() try: while True: # Reload state from disk so CLI changes (e.g. new game @@ -219,7 +225,7 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None: state.current_game_name = fresh.current_game_name state.finished_app_ids = fresh.finished_app_ids - _enforce_loop_iteration(config, state, demo=demo) + _enforce_loop_iteration(config, state, session=session, demo=demo) time.sleep(ENFORCE_INTERVAL) except KeyboardInterrupt: _echo("\nEnforcer stopped.") diff --git a/steam_backlog_enforcer/_engagement_probes.py b/steam_backlog_enforcer/_engagement_probes.py new file mode 100644 index 0000000..59bf5c1 --- /dev/null +++ b/steam_backlog_enforcer/_engagement_probes.py @@ -0,0 +1,168 @@ +"""Turning each raw engagement signal into a pause cause. + +Split from :mod:`_playtime_engagement`, which owns the per-tick decision and the +backdate; this owns the three probes and the rule each one applies. Every probe +follows the same contract: append to *causes* when the criterion says "not +playing", append to *degraded* when the probe could not answer, and never raise. + +**A probe that fails does not pause the tick.** Failure means the tick is billed. +A detector that stopped working must never become an unlimited-gaming exploit, +so the burden of proof sits on "not playing", not on "playing". +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._engagement_types import ( + CAUSE_FOCUS, + CAUSE_IDLE, + CAUSE_SCREEN_HELD, + PROBE_FOCUS, + PROBE_IDLE, + PROBE_SCREEN_HELD, + CauseTally, +) +from steam_backlog_enforcer._playtime_kill import _MAX_PROCESS_TREE_DEPTH, _read_ppid +from steam_backlog_enforcer._screen_hold import ScreenHoldError, screen_hold +from steam_backlog_enforcer._x_focus import focused_pid +from steam_backlog_enforcer._x_probe import XProbeError + +if TYPE_CHECKING: + from pathlib import Path + + from steam_backlog_enforcer._controller_input import ControllerActivity + from steam_backlog_enforcer._playtime_state import PlaytimeRules + from steam_backlog_enforcer._x_probe import XProbe + +logger = logging.getLogger(__name__) + +_INIT_PID = 1 + + +def assess_screen( + holder_path: Path, tally: CauseTally +) -> tuple[bool | None, int | None]: + """Add the screen-held cause if a gatelock guard owns the display. + + Args: + holder_path: The gatelock holder lock. + tally: Accumulating causes and probe failures. + + Returns: + The hold state and holding PID, both ``None`` when the probe failed. + """ + try: + hold = screen_hold(holder_path) + except ScreenHoldError as exc: + logger.warning("Screen-hold probe failed; billing this tick: %s", exc) + tally.degraded.append(PROBE_SCREEN_HELD) + return None, None + + if hold.held: + tally.causes.append(CAUSE_SCREEN_HELD) + return hold.held, hold.holder_pid + + +def assess_focus( + probe: XProbe, + rules: PlaytimeRules, + qualifying: set[int], + tally: CauseTally, +) -> int | None: + """Add the focus cause unless a qualifying process owns the focus. + + Args: + probe: X connection. + rules: Policy for this tick. + qualifying: PIDs that would otherwise count. + tally: Accumulating causes and probe failures. + + Returns: + The focused PID, or ``None``. + """ + if not rules.require_game_focus: + return None + try: + pid = focused_pid(probe.connect()) + except XProbeError as exc: + logger.warning("Focus probe failed; billing this tick: %s", exc) + tally.degraded.append(PROBE_FOCUS) + probe.close() + return None + + if not focus_qualifies(pid, qualifying): + tally.causes.append(CAUSE_FOCUS) + return pid + + +def assess_idle( + probe: XProbe, + controller: ControllerActivity, + rules: PlaytimeRules, + tally: CauseTally, + now_monotonic: float, +) -> tuple[float | None, float | None]: + """Add the idle cause when every input source has gone quiet. + + The controller is polled first and its age folded in with ``min``: the X + idle counter does not see a gamepad, and pausing during controller play + would be the gate's worst failure. + + Args: + probe: X connection. + controller: Controller-activity watcher. + rules: Policy for this tick. + tally: Accumulating causes and probe failures. + now_monotonic: Monotonic timestamp for controller ages. + + Returns: + The effective idle seconds and the controller idle seconds. + """ + controller.poll(now=now_monotonic) + controller_idle = controller.seconds_since_activity(now=now_monotonic) + + try: + pointer_idle = probe.idle_seconds() + except XProbeError as exc: + logger.warning("Idle probe failed; billing this tick: %s", exc) + tally.degraded.append(PROBE_IDLE) + probe.close() + return None, controller_idle + + idle = pointer_idle + if controller_idle is not None: + idle = min(idle, controller_idle) + if idle >= rules.idle_grace_seconds: + tally.causes.append(CAUSE_IDLE) + return idle, controller_idle + + +def focus_qualifies(focus_pid: int | None, qualifying: set[int]) -> bool: + """Whether *focus_pid* is, or descends from, a qualifying process. + + A game's visible window is often owned by a child of the process carrying + ``SteamAppId`` — Proton in particular — so the ancestry is walked rather + than compared directly. + + Args: + focus_pid: PID owning the focused window, or ``None``. + qualifying: PIDs that count as gaming. + + Returns: + Whether the focused window belongs to the game. + """ + if focus_pid is None: + return False + pid = focus_pid + for _ in range(_MAX_PROCESS_TREE_DEPTH): + if pid in qualifying: + return True + if pid <= _INIT_PID: + return False + parent = _read_ppid(pid) + if parent is None: + return False + pid = parent + return False diff --git a/steam_backlog_enforcer/_engagement_types.py b/steam_backlog_enforcer/_engagement_types.py new file mode 100644 index 0000000..f229419 --- /dev/null +++ b/steam_backlog_enforcer/_engagement_types.py @@ -0,0 +1,91 @@ +"""The verdict describing whether this tick should bill the gaming budget. + +Split from :mod:`_playtime_engagement` so the decision *record* stays a plain +frozen value with no probes attached: everything that writes it is testable +without an X server, and everything that reads it — the accumulator and the +audit log — sees the same shape. + +The verdict deliberately records *every* cause rather than short-circuiting on +the first one. The idle backdate only fires when idle is the sole cause, and a +log that recorded just the first-matching reason could not answer "was the +screen also locked?" after the fact. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Final + +STATE_ENGAGED: Final = "engaged" +STATE_PAUSED: Final = "paused" +STATE_NOT_APPLICABLE: Final = "not_applicable" + +CAUSE_IDLE: Final = "idle" +CAUSE_SCREEN_HELD: Final = "screen_held" +CAUSE_FOCUS: Final = "focus" + +PROBE_IDLE: Final = "idle" +PROBE_SCREEN_HELD: Final = "screen_held" +PROBE_FOCUS: Final = "focus" + + +@dataclass +class CauseTally: + """Pause causes and probe failures accumulated across one tick's probes. + + Passed to each probe instead of two parallel lists: it keeps the probe + signatures short, and it makes "this probe appends to both" a single idea + rather than a convention two arguments have to remember. + """ + + causes: list[str] = field(default_factory=list) + degraded: list[str] = field(default_factory=list) + + +@dataclass(frozen=True) +class EngagementVerdict: + """Why this tick did or did not count against the budget.""" + + state: str + causes: tuple[str, ...] = () + degraded: tuple[str, ...] = () + idle_seconds: float | None = None + controller_idle_seconds: float | None = None + screen_held: bool | None = None + holder_pid: int | None = None + focus_pid: int | None = None + qualifying: tuple[int, ...] = field(default_factory=tuple) + + @property + def engaged(self) -> bool: + """Whether this tick should be billed. + + ``not_applicable`` counts as engaged: nothing qualifies, so the + accumulator will credit nothing regardless, and calling it "paused" + would fill the audit log with a pause reason during ordinary desktop + use. + + Returns: + Whether the tick bills. + """ + return self.state != STATE_PAUSED + + @property + def reason(self) -> str: + """A short human-readable summary of *causes*. + + Returns: + The joined causes, or the state when there are none. + """ + return "+".join(self.causes) if self.causes else self.state + + def idle_only(self) -> bool: + """Whether idleness is the sole reason this tick was paused. + + The backdate refunds the idle grace period, which is only correct when + nothing else would have paused the tick anyway. + + Returns: + Whether *causes* is exactly ``(CAUSE_IDLE,)``. + """ + return self.causes == (CAUSE_IDLE,) diff --git a/steam_backlog_enforcer/_playtime.py b/steam_backlog_enforcer/_playtime.py index 3de5064..ce63871 100644 --- a/steam_backlog_enforcer/_playtime.py +++ b/steam_backlog_enforcer/_playtime.py @@ -30,6 +30,7 @@ from __future__ import annotations from dataclasses import replace from datetime import datetime, timezone import logging +import time from typing import TYPE_CHECKING from steam_backlog_enforcer._playtime_block import ( @@ -57,6 +58,7 @@ from steam_backlog_enforcer._playtime_state import ( from steam_backlog_enforcer._total_block import is_total_block_active if TYPE_CHECKING: + from steam_backlog_enforcer._playtime_session import PlaytimeSession from steam_backlog_enforcer.config import Config logger = logging.getLogger(__name__) @@ -66,16 +68,27 @@ _SECONDS_PER_MINUTE = 60 _MINUTES_PER_HOUR = 60 -def playtime_tick(config: Config, *, interval: float, demo: bool = False) -> None: +def playtime_tick( + config: Config, + *, + interval: float, + session: PlaytimeSession, + demo: bool = False, +) -> None: """Account for this tick's gaming time and enforce the daily budget. Called first in every enforce-loop iteration, before any other guard, so that time is still counted (and a 06:00 release still happens) in the situations where the rest of the loop early-returns. + Time only accrues while the user is actually engaged: a resident game + process behind a locked screen, an unfocused window or five minutes of + silence is not play. See :mod:`_playtime_engagement`. + Args: config: Loaded user configuration. interval: Nominal seconds between enforce-loop ticks. + session: Cross-tick engagement and logging state. demo: Whether this is a short-budget demo run. """ now = datetime.now(timezone.utc).astimezone() @@ -90,13 +103,20 @@ def playtime_tick(config: Config, *, interval: float, demo: bool = False) -> Non save_state(state, demo=demo) return + pids = qualifying_pids(rules) + monotonic = time.monotonic() + verdict = session.tracker.assess(rules, qualifying=pids, now_monotonic=monotonic) + state = accumulate( state, now=now, - qualifying=qualifying_pids(rules), + qualifying=pids if verdict.engaged else set(), interval=interval, ) - save_state(_policy(state, rules, now=now), demo=demo) + state = session.tracker.backdate(state, verdict, rules=rules) + state = _policy(state, rules, now=now) + session.journal.observe(verdict, state, rules=rules, now_monotonic=monotonic) + save_state(state, demo=demo) def _state_or_recover(rules: PlaytimeRules, *, now: datetime) -> PlaytimeState: diff --git a/steam_backlog_enforcer/_playtime_engagement.py b/steam_backlog_enforcer/_playtime_engagement.py new file mode 100644 index 0000000..f58e56a --- /dev/null +++ b/steam_backlog_enforcer/_playtime_engagement.py @@ -0,0 +1,164 @@ +"""Deciding whether the user is actually playing, not just leaving a game open. + +A running game process is not evidence of play. The budget was previously billed +for two hours during which a screen locker covered the display and the machine +could not be touched at all — the game was simply still resident. + +Three signals gate a tick, evaluated together rather than short-circuited, so +that the log can answer "was the screen *also* locked?" after the fact and the +backdate can tell an idle-only pause from a compound one. The probes themselves +live in :mod:`_engagement_probes`. + +This module owns the per-tick verdict and the one piece of state a pure function +cannot hold: the previous verdict, which the idle backdate needs to recognise an +engaged-to-paused edge. +""" + +from __future__ import annotations + +from dataclasses import replace +import logging +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._controller_input import ControllerActivity +from steam_backlog_enforcer._engagement_probes import ( + assess_focus, + assess_idle, + assess_screen, +) +from steam_backlog_enforcer._engagement_types import ( + STATE_ENGAGED, + STATE_NOT_APPLICABLE, + STATE_PAUSED, + CauseTally, + EngagementVerdict, +) +from steam_backlog_enforcer._screen_hold import holder_lock_path +from steam_backlog_enforcer._x_probe import XProbe + +if TYPE_CHECKING: + from pathlib import Path + + from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + +logger = logging.getLogger(__name__) + + +class EngagementTracker: + """Assesses engagement per tick and owns the state the backdate needs.""" + + def __init__( + self, + *, + uid: int, + probe: XProbe | None = None, + controller: ControllerActivity | None = None, + holder_path: Path | None = None, + ) -> None: + """Build a tracker over the desktop session owned by *uid*. + + Args: + uid: Owner of the desktop session. + probe: X probe, injected by tests. + controller: Controller-activity watcher, injected by tests. + holder_path: gatelock holder lock, injected by tests. + """ + self._probe = probe if probe is not None else XProbe() + self._controller = ( + controller if controller is not None else ControllerActivity() + ) + self._holder_path = ( + holder_path if holder_path is not None else holder_lock_path(uid) + ) + self._previous: EngagementVerdict | None = None + self._previous_day: str | None = None + + def close(self) -> None: + """Release the X connection and controller devices.""" + self._probe.close() + self._controller.close() + + def assess( + self, + rules: PlaytimeRules, + *, + qualifying: set[int], + now_monotonic: float, + ) -> EngagementVerdict: + """Judge whether this tick counts against the budget. + + Args: + rules: Policy for this tick. + qualifying: PIDs that would otherwise count. + now_monotonic: Monotonic timestamp, for controller-activity ages. + + Returns: + The verdict for this tick. + """ + if not qualifying: + return EngagementVerdict(state=STATE_NOT_APPLICABLE) + + pids = tuple(sorted(qualifying)) + if not rules.engagement_gate: + return EngagementVerdict(state=STATE_ENGAGED, qualifying=pids) + + tally = CauseTally() + + held, holder_pid = assess_screen(self._holder_path, tally) + focus_pid = assess_focus(self._probe, rules, qualifying, tally) + idle, controller_idle = assess_idle( + self._probe, self._controller, rules, tally, now_monotonic + ) + + return EngagementVerdict( + state=STATE_PAUSED if tally.causes else STATE_ENGAGED, + causes=tuple(tally.causes), + degraded=tuple(tally.degraded), + idle_seconds=idle, + controller_idle_seconds=controller_idle, + screen_held=held, + holder_pid=holder_pid, + focus_pid=focus_pid, + qualifying=pids, + ) + + def backdate( + self, + state: PlaytimeState, + verdict: EngagementVerdict, + *, + rules: PlaytimeRules, + ) -> PlaytimeState: + """Refund the idle grace period once per walk-away. + + Time billed *before* the idle threshold tripped was not play either, so + the first paused tick of an idle-only spell gives it back. Deliberately + conservative: it fires only on the engaged-to-paused edge, only when + idleness is the sole cause, and never across a day rollover — and after + a daemon restart there is no previous verdict, so nothing is refunded. + + Args: + state: Accounting state after this tick's accumulation. + verdict: This tick's verdict. + rules: Policy for this tick. + + Returns: + The state, with the grace period deducted when all of that holds. + """ + previous, previous_day = self._previous, self._previous_day + self._previous, self._previous_day = verdict, state.day_key + + if previous is None or previous.state != STATE_ENGAGED: + return state + if previous_day != state.day_key: + return state + if not verdict.idle_only(): + return state + + refunded = max(0.0, state.seconds - rules.idle_grace_seconds) + logger.info( + "Idle for %.0fs; refunding the %.0fs grace period.", + verdict.idle_seconds or 0.0, + state.seconds - refunded, + ) + return replace(state, seconds=refunded) diff --git a/steam_backlog_enforcer/_playtime_log.py b/steam_backlog_enforcer/_playtime_log.py new file mode 100644 index 0000000..1a529cd --- /dev/null +++ b/steam_backlog_enforcer/_playtime_log.py @@ -0,0 +1,209 @@ +"""A structured audit trail for every decision the gaming budget makes. + +Reconstructing the incident that prompted this module — two hours billed while +a screen locker covered the display — took cross-referencing journald against +the screen locker's own JSONL, and even then the accrual could only be inferred +from the spacing of two warning notifications. That is not an audit trail. + +So every budget decision is recorded here as one JSON object per line, readable +without ``sudo`` (unlike the state file, whose ``PermissionError`` is swallowed +on load and silently reported as "no state recorded yet"). + +Volume is controlled by only writing when something *changes*. At a three-second +tick, one line per tick would be 28 800 lines a day of almost entirely identical +records; instead a verdict is written when it differs from the last one, with a +periodic heartbeat so a long unchanging session still leaves a trace. + +Nothing here may raise. A logging failure must not take down the enforcer, so +the sink degrades to a warning on the daemon's own logger and carries on. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json +import logging +import logging.handlers +from pathlib import Path +from typing import TYPE_CHECKING, Final, cast + +from steam_backlog_enforcer._engagement_types import STATE_NOT_APPLICABLE + +if TYPE_CHECKING: + from steam_backlog_enforcer._engagement_types import EngagementVerdict + from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + +logger = logging.getLogger(__name__) + +BUDGET_LOG_FILE: Final = Path("/var/log/steam-backlog-enforcer/budget.jsonl") + +_MAX_BYTES: Final = 5 * 1024 * 1024 +_BACKUP_COUNT: Final = 5 +_HEARTBEAT_SECONDS: Final = 300.0 + +EVENT_VERDICT_CHANGE: Final = "verdict_change" +EVENT_HEARTBEAT: Final = "heartbeat" +EVENT_DETECTOR_FAILURE: Final = "detector_failure" +EVENT_MANUAL_ADJUSTMENT: Final = "manual_adjustment" + + +class BudgetLog: + """Append-only JSONL sink, rotated by size.""" + + def __init__(self, *, path: Path = BUDGET_LOG_FILE) -> None: + """Prepare a sink; the file is opened on first write. + + Args: + path: Destination JSONL file. + """ + self._path = path + self._sink: logging.Logger | None = None + + def _open(self) -> logging.Logger | None: + """Return the rotating sink, creating it on first use. + + Returns: + The sink, or ``None`` if it could not be created. + """ + if self._sink is not None: + return self._sink + try: + self._path.parent.mkdir(parents=True, exist_ok=True) + handler = logging.handlers.RotatingFileHandler( + self._path, maxBytes=_MAX_BYTES, backupCount=_BACKUP_COUNT + ) + except OSError as exc: + logger.warning("Budget log unavailable at %s: %s", self._path, exc) + return None + + handler.setFormatter(logging.Formatter("%(message)s")) + # Named per destination: a single shared logger would let two sinks + # with different paths silently evict each other's handler. + sink = logging.getLogger(f"{__name__}.sink.{self._path}") + sink.setLevel(logging.INFO) + sink.propagate = False + for stale in sink.handlers: + stale.close() + sink.handlers = [handler] + self._sink = sink + return self._sink + + def close(self) -> None: + """Close the underlying file handler.""" + if self._sink is None: + return + for handler in self._sink.handlers: + handler.close() + self._sink.handlers = [] + self._sink = None + + def record(self, event: str, **fields: object) -> None: + """Append one record. + + Args: + event: Event name. + fields: Arbitrary JSON-serialisable payload. + """ + sink = self._open() + if sink is None: + return + payload = { + "timestamp": datetime.now(timezone.utc).astimezone().isoformat(), + "event": event, + **fields, + } + try: + sink.info(json.dumps(payload, default=str)) + except (OSError, ValueError) as exc: + logger.warning("Could not write budget log record %r: %s", event, exc) + + +class TickJournal: + """Decides which ticks are worth recording, and records them.""" + + def __init__( + self, log: BudgetLog, *, heartbeat: float = _HEARTBEAT_SECONDS + ) -> None: + """Start with nothing observed. + + Args: + log: Sink to write through. + heartbeat: Seconds between heartbeats during an unchanged verdict. + """ + self._log = log + self._heartbeat = heartbeat + self._last_reason: str | None = None + self._last_written_at: float | None = None + + def observe( + self, + verdict: EngagementVerdict, + state: PlaytimeState, + *, + rules: PlaytimeRules, + now_monotonic: float, + ) -> None: + """Record *verdict* if it changed, failed, or is due a heartbeat. + + Args: + verdict: This tick's verdict. + state: Accounting state after this tick. + rules: Policy for this tick. + now_monotonic: Monotonic timestamp for heartbeat spacing. + """ + if verdict.degraded: + self._log.record( + EVENT_DETECTOR_FAILURE, + degraded=list(verdict.degraded), + billed=True, + **_snapshot(verdict, state, rules), + ) + + if verdict.reason != self._last_reason: + self._last_reason = verdict.reason + self._last_written_at = now_monotonic + self._log.record(EVENT_VERDICT_CHANGE, **_snapshot(verdict, state, rules)) + return + + if verdict.state == STATE_NOT_APPLICABLE: + return + # The change branch above always fires on the first observation and + # sets both fields together, so by here this is never None. + last = cast("float", self._last_written_at) + if now_monotonic - last < self._heartbeat: + return + + self._last_written_at = now_monotonic + self._log.record(EVENT_HEARTBEAT, **_snapshot(verdict, state, rules)) + + +def _snapshot( + verdict: EngagementVerdict, state: PlaytimeState, rules: PlaytimeRules +) -> dict[str, object]: + """Flatten a tick into the fields every record carries. + + The qualifying PIDs are included deliberately: the accrual that continued + after the last game exited could not be attributed without them. + + Args: + verdict: This tick's verdict. + state: Accounting state after this tick. + rules: Policy for this tick. + + Returns: + The record body. + """ + return { + "state": verdict.state, + "reason": verdict.reason, + "causes": list(verdict.causes), + "idle_seconds": verdict.idle_seconds, + "controller_idle_seconds": verdict.controller_idle_seconds, + "screen_held": verdict.screen_held, + "holder_pid": verdict.holder_pid, + "focus_pid": verdict.focus_pid, + "qualifying": list(verdict.qualifying), + "day_key": state.day_key, + "billed_seconds": round(state.seconds, 3), + "remaining_seconds": round(rules.budget_seconds - state.seconds, 3), + } diff --git a/steam_backlog_enforcer/_playtime_session.py b/steam_backlog_enforcer/_playtime_session.py new file mode 100644 index 0000000..9ddccc7 --- /dev/null +++ b/steam_backlog_enforcer/_playtime_session.py @@ -0,0 +1,83 @@ +"""The cross-tick state the daily budget carries, and how it is built. + +Two things cannot be recomputed from scratch each tick. The engagement backdate +must recognise an engaged-to-paused *edge*, so it needs the previous verdict; +and the audit journal only writes on change, so it needs the previous record. +Both live on a session that the enforce loop creates once and threads through, +rather than in module globals. + +The session's fields are typed by :class:`Protocol` so a test can supply an +honest stub. A ``MagicMock`` would satisfy every attribute lookup — including +the ones a refactor forgot to update — which is exactly the failure a type +would otherwise have caught. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Protocol + +from steam_backlog_enforcer._desktop_env import desktop_uid, resolve_desktop_user +from steam_backlog_enforcer._playtime_engagement import EngagementTracker +from steam_backlog_enforcer._playtime_log import BudgetLog, TickJournal + +if TYPE_CHECKING: + from steam_backlog_enforcer._engagement_types import EngagementVerdict + from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + + +class EngagementSource(Protocol): + """What the budget tick needs from an engagement tracker.""" + + def assess( + self, + rules: PlaytimeRules, + *, + qualifying: set[int], + now_monotonic: float, + ) -> EngagementVerdict: + """Judge whether this tick counts against the budget.""" + + def backdate( + self, + state: PlaytimeState, + verdict: EngagementVerdict, + *, + rules: PlaytimeRules, + ) -> PlaytimeState: + """Refund the idle grace period on an engaged-to-idle edge.""" + + +class TickRecorder(Protocol): + """What the budget tick needs from the audit journal.""" + + def observe( + self, + verdict: EngagementVerdict, + state: PlaytimeState, + *, + rules: PlaytimeRules, + now_monotonic: float, + ) -> None: + """Record this tick if it is worth recording.""" + + +@dataclass(frozen=True) +class PlaytimeSession: + """Per-daemon engagement and logging state.""" + + tracker: EngagementSource + journal: TickRecorder + + +def new_session() -> PlaytimeSession: + """Build the session the enforce loop threads through every tick. + + Returns: + A session bound to the desktop user's X display and runtime dir. + """ + uid = desktop_uid(resolve_desktop_user()) + return PlaytimeSession( + tracker=EngagementTracker(uid=uid), + journal=TickJournal(BudgetLog()), + ) diff --git a/steam_backlog_enforcer/_playtime_state.py b/steam_backlog_enforcer/_playtime_state.py index fcbb76f..10f41e0 100644 --- a/steam_backlog_enforcer/_playtime_state.py +++ b/steam_backlog_enforcer/_playtime_state.py @@ -90,6 +90,9 @@ class PlaytimeRules: count_launchers: bool enforcement: bool demo: bool + engagement_gate: bool = True + idle_grace_seconds: float = 300.0 + require_game_focus: bool = True def rules_for(config: Config, *, demo: bool) -> PlaytimeRules: @@ -115,6 +118,9 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules: count_launchers=config.count_launcher_processes, enforcement=config.playtime_enforcement, demo=demo, + engagement_gate=config.engagement_gate, + idle_grace_seconds=float(config.idle_grace_seconds), + require_game_focus=config.require_game_focus, ) diff --git a/steam_backlog_enforcer/_screen_hold.py b/steam_backlog_enforcer/_screen_hold.py new file mode 100644 index 0000000..6533e04 --- /dev/null +++ b/steam_backlog_enforcer/_screen_hold.py @@ -0,0 +1,124 @@ +"""Detecting that a gatelock guard currently owns the screen. + +The screen lockers (``screen_locker``, ``leetcode_guard``) coordinate through a +single advisory ``flock`` on ``$XDG_RUNTIME_DIR/gatelock/holder.lock``. The file's +*contents* cannot answer "is it held right now" — a stale record left by a dead +process is byte-identical to a live one — and gatelock's own +``read_claim_if_held`` answers the question by *taking* the lock. + +The enforcer must not do that. Taking the lock every three seconds, even for +microseconds, would race a guard trying to claim the screen. ``/proc/locks`` +reports the same fact by observation, so this reads that instead and never +touches the file. + +Line format is ``idx: FLOCK ADVISORY WRITE ``, +with major and minor in hex. Lines containing ``->`` describe processes *waiting* +for a lock, not holding one, and are skipped. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path +import re +from typing import Final + +_PROC_LOCKS: Final = Path("/proc/locks") +_HOLDER_RELATIVE: Final = Path("gatelock/holder.lock") + +# `maj:min:inode`, major and minor hex, inode decimal. +_DEVICE_FIELD: Final = re.compile(r"^[0-9a-f]+:[0-9a-f]+:\d+$") + +_WAITER_MARKER: Final = "->" +_PID_FIELD_INDEX: Final = 4 + + +class ScreenHoldError(RuntimeError): + """Whether the screen is held could not be determined.""" + + +@dataclass(frozen=True) +class ScreenHold: + """Whether a guard owns the screen, and which process does.""" + + held: bool + holder_pid: int | None = None + + +def holder_lock_path(uid: int) -> Path: + """Return the gatelock holder-lock path for *uid*. + + Args: + uid: Owner of the desktop session. + + Returns: + Path to ``holder.lock``. + """ + return Path(f"/run/user/{uid}") / _HOLDER_RELATIVE + + +def screen_hold(path: Path) -> ScreenHold: + """Report whether *path*'s advisory lock is currently held. + + Args: + path: The gatelock holder-lock file. + + Returns: + The hold state, with the holding PID when there is one. + + Raises: + ScreenHoldError: If ``/proc/locks`` cannot be read. A missing *path* is + not an error — it means no guard has ever run this boot. + """ + try: + stat = path.stat() + except OSError: + return ScreenHold(held=False) + + target = f"{os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x}:{stat.st_ino}" + + try: + raw = _PROC_LOCKS.read_text(encoding="utf-8") + except OSError as exc: + msg = f"cannot read {_PROC_LOCKS}: {exc}" + raise ScreenHoldError(msg) from exc + + return _find_holder(raw, target) + + +def _find_holder(raw: str, target: str) -> ScreenHold: + """Scan ``/proc/locks`` text for a live lock on *target*. + + Args: + raw: Contents of ``/proc/locks``. + target: ``maj:min:inode`` string to match. + + Returns: + The hold state. + """ + for line in raw.splitlines(): + fields = line.split() + if _WAITER_MARKER in fields: + continue + if not any(field == target for field in fields if _DEVICE_FIELD.match(field)): + continue + return ScreenHold(held=True, holder_pid=_pid_of(fields)) + return ScreenHold(held=False) + + +def _pid_of(fields: list[str]) -> int | None: + """Return the PID column of a ``/proc/locks`` row. + + Args: + fields: Whitespace-split row. + + Returns: + The PID, or ``None`` if the row is shaped unexpectedly. + """ + if len(fields) <= _PID_FIELD_INDEX: + return None + try: + return int(fields[_PID_FIELD_INDEX]) + except ValueError: + return None diff --git a/steam_backlog_enforcer/_steam_errors.py b/steam_backlog_enforcer/_steam_errors.py index b174ab4..a6b314f 100644 --- a/steam_backlog_enforcer/_steam_errors.py +++ b/steam_backlog_enforcer/_steam_errors.py @@ -38,3 +38,15 @@ class DesktopSessionNotReadyError(SteamUnavailableError): that way for the whole session — the Kingdom Come: Deliverance II startup crash. Waiting a pass costs 3s; getting it wrong costs a reboot. """ + + +class GameInProgressError(SteamUnavailableError): + """Raised to defer a Steam restart while a game is actually being played. + + Subclasses :class:`SteamUnavailableError` so existing callers already + degrade gracefully (skip this pass, retry next loop). Bouncing Steam to + open the CDP port takes every running game down with it, losing unsaved + progress — observed on 2026-08-28, when restarting the daemon to deploy an + unrelated change killed a live session. Library hiding can wait; a player's + afternoon cannot be given back. + """ diff --git a/steam_backlog_enforcer/_steam_launch.py b/steam_backlog_enforcer/_steam_launch.py index f7101c9..10eb1ad 100644 --- a/steam_backlog_enforcer/_steam_launch.py +++ b/steam_backlog_enforcer/_steam_launch.py @@ -28,9 +28,12 @@ from steam_backlog_enforcer._desktop_env import ( from steam_backlog_enforcer._steam_errors import ( DesktopSessionNotReadyError, SteamUnavailableError, - SteamUpdateInProgressError, ) from steam_backlog_enforcer._steam_process import _run_as_user +from steam_backlog_enforcer._steam_restart_guard import ( + assert_safe_to_restart, + game_is_running, +) from steam_backlog_enforcer._steam_state import steam_update_in_progress logger = logging.getLogger(__name__) @@ -193,18 +196,7 @@ def ensure_steam_debug_port() -> None: logger.info("Steam CDP port not available — (re)starting Steam...") if _is_steam_running(): - # Never bounce a running Steam while a game update is downloading or - # committing: the shutdown suspends it and can leave a partially - # written install (the root cause of the AoE2 launch crash). Defer and - # retry on the next enforce pass. - if steam_update_in_progress(): - msg = ( - "Deferring Steam restart: a game update is in progress. " - "Restarting now would interrupt and can corrupt it; will " - "retry once the update settles." - ) - logger.info(msg) - raise SteamUpdateInProgressError(msg) + assert_safe_to_restart() _shutdown_steam() _launch_steam_with_debug() @@ -228,9 +220,17 @@ def ensure_steam_debug_port() -> None: def restart_steam() -> None: """Gracefully restart the Steam client with CEF debugging enabled. - Skips the restart if a game update is downloading or committing, so the - update is not interrupted (interrupting it can corrupt the install). + Skips the restart while a game is running (the restart would kill it) or + while a game update is downloading or committing (interrupting it can + corrupt the install). """ + if game_is_running(): + logger.warning( + "Skipping Steam restart — a game is running; restarting now " + "would kill it and lose unsaved progress.", + ) + return + if steam_update_in_progress(): logger.warning( "Skipping Steam restart — a game update is in progress; " diff --git a/steam_backlog_enforcer/_steam_restart_guard.py b/steam_backlog_enforcer/_steam_restart_guard.py new file mode 100644 index 0000000..d192f14 --- /dev/null +++ b/steam_backlog_enforcer/_steam_restart_guard.py @@ -0,0 +1,66 @@ +"""When it is *not* safe to bounce a running Steam. + +Opening the CDP debug port requires restarting Steam, and a restart is never +free: it takes down whatever the client is currently doing. Two situations make +that unacceptable, and both defer rather than abort — the enforce loop retries +every few seconds, so waiting costs a pass and nothing else. + +Split out of :mod:`_steam_launch` to keep it under the 250-line cap, and because +"is a restart safe right now" is a question worth being able to ask on its own. +""" + +from __future__ import annotations + +import logging + +from steam_backlog_enforcer._steam_errors import ( + GameInProgressError, + SteamUpdateInProgressError, +) +from steam_backlog_enforcer._steam_state import steam_update_in_progress +from steam_backlog_enforcer.enforcer import get_running_steam_game_pids + +logger = logging.getLogger(__name__) + + +def game_is_running() -> bool: + """Whether a Steam game is currently running. + + Uses the same ``SteamAppId != 0`` predicate as the playtime budget, which + excludes the Steam client's own process tree — the client being up is not a + reason to defer, only an actual game is. + + Returns: + Whether any game process is live. + """ + return any(app_id != 0 for app_id in get_running_steam_game_pids().values()) + + +def assert_safe_to_restart() -> None: + """Refuse to restart Steam while doing so would destroy something. + + Raises: + GameInProgressError: If a game is running. Bouncing Steam kills it and + any unsaved progress with it — observed on 2026-08-28, when + restarting the daemon to deploy an unrelated change ended a live + session. Library hiding can wait; an afternoon cannot be returned. + SteamUpdateInProgressError: If a game update is downloading or + committing. The shutdown suspends it and can leave a partially + written install (the root cause of the AoE2 launch crash). + """ + if game_is_running(): + msg = ( + "Deferring Steam restart: a game is running. Restarting now would " + "kill it and lose unsaved progress; will retry once it exits." + ) + logger.info(msg) + raise GameInProgressError(msg) + + if steam_update_in_progress(): + msg = ( + "Deferring Steam restart: a game update is in progress. Restarting " + "now would interrupt and can corrupt it; will retry once the " + "update settles." + ) + logger.info(msg) + raise SteamUpdateInProgressError(msg) diff --git a/steam_backlog_enforcer/_x_focus.py b/steam_backlog_enforcer/_x_focus.py new file mode 100644 index 0000000..f02db6e --- /dev/null +++ b/steam_backlog_enforcer/_x_focus.py @@ -0,0 +1,176 @@ +"""Resolving which process owns the focused X window. + +``_NET_WM_PID`` is the obvious answer and the one every window-manager guide +names, but it is a *convention* toolkits opt into, not a guarantee. Measured on +this machine: Aseprite's window carries no ``_NET_WM_PID`` at all. Treating that +as "cannot tell" and billing the tick would make the focus criterion useless for +every such application — including, plausibly, games. + +So the absent property falls through to the X-Resource extension, which asks the +*server* which client owns the window rather than trusting the client to have +announced itself. Verified against the same Aseprite window: ``XResQueryClientIds`` +returned the correct PID where ``_NET_WM_PID`` returned nothing. +""" + +from __future__ import annotations + +import ctypes +from typing import Final + +from steam_backlog_enforcer._x_probe import Conn, XProbeError, root_window + +# XGetWindowProperty's `req_type`: accept whatever type the property carries. +_ANY_PROPERTY_TYPE: Final = 0 +_SUCCESS: Final = 0 + +# XInternAtom's `only_if_exists`: never create an atom just by asking for it. +# An int, not a bool: it is a C ``Bool`` argument, not a Python flag. +_ONLY_IF_EXISTS: Final = 1 + +# XRes client-id spec masks, from . +_XRES_PID_MASK: Final = 1 << 1 +_XRES_PID_BYTES: Final = 4 + + +class _XResSpec(ctypes.Structure): + """Mirror of libXRes's ``XResClientIdSpec``.""" + + _fields_ = (("client", ctypes.c_ulong), ("mask", ctypes.c_uint)) + + +class _XResValue(ctypes.Structure): + """Mirror of libXRes's ``XResClientIdValue``.""" + + _fields_ = ( + ("spec", _XResSpec), + ("length", ctypes.c_long), + ("value", ctypes.c_void_p), + ) + + +def focused_pid(conn: Conn) -> int | None: + """Return the PID owning the focused window. + + Args: + conn: Live connection. + + Returns: + The PID, or ``None`` when nothing is focused — a real answer, not a + failure. + + Raises: + XProbeError: If a window is focused but cannot be attributed to a + process, which the caller must treat as "cannot tell" and bill. + """ + active = read_cardinal(conn, root_window(conn), b"_NET_ACTIVE_WINDOW") + if not active: + return None + + declared = read_cardinal(conn, active, b"_NET_WM_PID") + if declared is not None: + return declared + + resolved = _xres_pid(conn, active) + if resolved is None: + msg = f"window {active:#x} has no _NET_WM_PID and XRes could not name it" + raise XProbeError(msg) + return resolved + + +def _xres_pid(conn: Conn, window: int) -> int | None: + """Ask the X server which process owns *window*. + + Args: + conn: Live connection. + window: Window to attribute. + + Returns: + The owning PID, or ``None`` if XRes is unavailable or has no answer. + """ + if conn.xres is None: + return None + + spec = _XResSpec(client=window, mask=_XRES_PID_MASK) + count = ctypes.c_long() + values = ctypes.POINTER(_XResValue)() + + status = conn.xres.XResQueryClientIds( + conn.display, + ctypes.c_long(1), + ctypes.byref(spec), + ctypes.byref(count), + ctypes.byref(values), + ) + if status != _SUCCESS or not values: + return None + + try: + return _first_pid(values, count.value) + finally: + conn.xres.XResClientIdsDestroy(count, values) + + +def _first_pid(values: ctypes._Pointer[_XResValue], count: int) -> int | None: + """Return the first PID among *count* XRes client-id values. + + Args: + values: Array returned by ``XResQueryClientIds``. + count: Number of entries in *values*. + + Returns: + The PID, or ``None`` if no entry carries one. + """ + for index in range(count): + entry = values[index] + if entry.spec.mask != _XRES_PID_MASK or entry.length < _XRES_PID_BYTES: + continue + if not entry.value: + continue + return int(ctypes.cast(entry.value, ctypes.POINTER(ctypes.c_uint32))[0]) + return None + + +def read_cardinal(conn: Conn, window: int, name: bytes) -> int | None: + """Read the first 32-bit value of window property *name*. + + Args: + conn: Live connection. + window: Window to query. + name: Property name, e.g. ``b"_NET_WM_PID"``. + + Returns: + The value, or ``None`` if the property is absent or empty. + """ + conn.x11.XInternAtom.restype = ctypes.c_ulong + atom = conn.x11.XInternAtom(conn.display, name, _ONLY_IF_EXISTS) + if not atom: + return None + + actual_type = ctypes.c_ulong() + actual_format = ctypes.c_int() + nitems = ctypes.c_ulong() + bytes_after = ctypes.c_ulong() + prop = ctypes.POINTER(ctypes.c_ulong)() + + status = conn.x11.XGetWindowProperty( + conn.display, + ctypes.c_ulong(window), + ctypes.c_ulong(atom), + ctypes.c_long(0), + ctypes.c_long(1), + ctypes.c_int(0), + ctypes.c_ulong(_ANY_PROPERTY_TYPE), + ctypes.byref(actual_type), + ctypes.byref(actual_format), + ctypes.byref(nitems), + ctypes.byref(bytes_after), + ctypes.byref(prop), + ) + if status != _SUCCESS or not prop: + return None + try: + if nitems.value == 0: + return None + return int(prop[0]) + finally: + conn.x11.XFree(prop) diff --git a/steam_backlog_enforcer/_x_probe.py b/steam_backlog_enforcer/_x_probe.py new file mode 100644 index 0000000..05b89c8 --- /dev/null +++ b/steam_backlog_enforcer/_x_probe.py @@ -0,0 +1,185 @@ +"""An X11 connection for reading input-idle time, without forking. + +The enforce loop ticks every three seconds, so shelling out to ``xprintidle`` +would cost roughly twenty forks a minute for a number the X server hands over +directly. This talks to libX11 and libXss through ``ctypes`` over one cached +connection instead. Focused-window resolution lives in :mod:`_x_focus`. + +The daemon runs as root with ``DISPLAY=:0`` and ``HOME=/home/kuhy`` set by its +unit file, which is what lets it authenticate against the user's session. + +Two details are load-bearing: + +**A no-op X error handler is installed.** Xlib's default handler calls +``exit(1)``. Asking about a window that vanished mid-query is an ordinary race, +not a fatal condition, and the enforcer must not die of it. + +**Every failure raises :class:`XProbeError`.** The caller bills the tick when a +probe cannot answer, so returning a bland ``None`` would be indistinguishable +from a real "not idle" and would quietly become free gaming. +""" + +from __future__ import annotations + +import ctypes +from dataclasses import dataclass +from typing import Final + +_MILLISECONDS_PER_SECOND: Final = 1000.0 + + +class XProbeError(RuntimeError): + """The X server could not be reached or did not answer.""" + + +class _ScreenSaverInfo(ctypes.Structure): + """Mirror of libXss's ``XScreenSaverInfo``.""" + + _fields_ = ( + ("window", ctypes.c_ulong), + ("state", ctypes.c_int), + ("kind", ctypes.c_int), + ("til_or_since", ctypes.c_ulong), + ("idle", ctypes.c_ulong), + ("event_mask", ctypes.c_ulong), + ) + + +@dataclass(frozen=True) +class Conn: + """An open display and the libraries it was opened with. + + ``xres`` is optional: the X-Resource extension is the fallback for windows + that advertise no ``_NET_WM_PID``, and a server without it simply means + those windows cannot be attributed. + """ + + display: ctypes.c_void_p + x11: ctypes.CDLL + xss: ctypes.CDLL + xres: ctypes.CDLL | None + + +_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p) + + +def _ignore_x_error(_display: ctypes.c_void_p, _event: ctypes.c_void_p) -> int: + """Swallow an X protocol error instead of letting Xlib call ``exit``.""" + return 0 + + +def load_libraries() -> tuple[ctypes.CDLL, ctypes.CDLL, ctypes.CDLL | None]: + """Load libX11, libXss and — if present — libXRes. + + Returns: + The ``(x11, xss, xres)`` handles; ``xres`` is ``None`` when absent. + + Raises: + XProbeError: If libX11 or libXss is missing. + """ + try: + x11 = ctypes.CDLL("libX11.so.6") + xss = ctypes.CDLL("libXss.so.1") + except OSError as exc: + msg = f"X client libraries unavailable: {exc}" + raise XProbeError(msg) from exc + try: + xres: ctypes.CDLL | None = ctypes.CDLL("libXRes.so.1") + except OSError: + xres = None + return x11, xss, xres + + +Libraries = tuple[ctypes.CDLL, ctypes.CDLL, ctypes.CDLL | None] + + +class XProbe: + """A cached X connection answering idle-time queries.""" + + def __init__(self, *, libraries: Libraries | None = None) -> None: + """Store the library handles; the connection is opened lazily. + + Args: + libraries: Pre-loaded ``(x11, xss, xres)`` handles, for tests. + """ + self._libraries = libraries + self._conn: Conn | None = None + # Xlib keeps only a borrowed pointer, so the trampoline must outlive it. + self._handler = _ERROR_HANDLER(_ignore_x_error) + + def connect(self) -> Conn: + """Return the open connection, opening it on first use. + + Returns: + The live connection. + + Raises: + XProbeError: If the display cannot be opened. + """ + if self._conn is not None: + return self._conn + + if self._libraries is None: + self._libraries = load_libraries() + x11, xss, xres = self._libraries + + x11.XSetErrorHandler(self._handler) + x11.XOpenDisplay.restype = ctypes.c_void_p + display = x11.XOpenDisplay(None) + if not display: + msg = "XOpenDisplay returned NULL; no reachable X session" + raise XProbeError(msg) + + self._conn = Conn(display=ctypes.c_void_p(display), x11=x11, xss=xss, xres=xres) + return self._conn + + def close(self) -> None: + """Drop the connection so the next query reconnects.""" + if self._conn is not None: + self._conn.x11.XCloseDisplay(self._conn.display) + self._conn = None + + def idle_seconds(self) -> float: + """Return seconds since the last keyboard or pointer input. + + Note that a game controller does **not** reset this counter unless it + is presented to X as a keyboard or pointer, which is why the caller + drops the idle criterion while a controller is connected. + + Returns: + The idle time in seconds. + + Raises: + XProbeError: If the query fails. + """ + conn = self.connect() + + conn.xss.XScreenSaverAllocInfo.restype = ctypes.POINTER(_ScreenSaverInfo) + info = conn.xss.XScreenSaverAllocInfo() + if not info: + msg = "XScreenSaverAllocInfo returned NULL" + raise XProbeError(msg) + + try: + queried = conn.xss.XScreenSaverQueryInfo( + conn.display, ctypes.c_ulong(root_window(conn)), info + ) + if not queried: + msg = "XScreenSaverQueryInfo failed; MIT-SCREEN-SAVER missing?" + raise XProbeError(msg) + return float(info.contents.idle) / _MILLISECONDS_PER_SECOND + finally: + conn.x11.XFree(info) + + +def root_window(conn: Conn) -> int: + """Return the default root window id. + + Args: + conn: Live connection. + + Returns: + The root window id. + """ + conn.x11.XDefaultRootWindow.restype = ctypes.c_ulong + return int(conn.x11.XDefaultRootWindow(conn.display)) diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index 4a9d119..7c374c1 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -85,6 +85,14 @@ class Config: Disabling stops the cutoff but never strands a live block — the release path runs regardless, so "disabled" cannot come to mean "blocked forever". """ + # Engagement gating: without it a resident game process bills the budget + # while the screen is locked. See :mod:`_playtime_engagement`. + engagement_gate: bool = True + """Whether a tick must show active engagement before it bills.""" + idle_grace_seconds: int = 300 + """Input silence before billing stops; controller input counts as input.""" + require_game_focus: bool = True + """Whether the focused window must belong to the game for a tick to bill.""" def save(self) -> None: """Persist config to disk.""" diff --git a/steam_backlog_enforcer/tests/_fake_session.py b/steam_backlog_enforcer/tests/_fake_session.py new file mode 100644 index 0000000..0d82c75 --- /dev/null +++ b/steam_backlog_enforcer/tests/_fake_session.py @@ -0,0 +1,123 @@ +"""A :class:`PlaytimeSession` stub for tests that are not about engagement. + +Most budget tests predate the engagement gate and care only about accounting, +warnings and the cutoff. They get a session that always says "engaged", which +reproduces the pre-gate behaviour exactly, so those tests keep asserting what +they were written to assert. + +Deliberately hand-written rather than ``MagicMock``: a mock satisfies every +attribute lookup, including ones a later refactor renamed, so a stub that must +actually match the protocol is the point. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from steam_backlog_enforcer._engagement_types import ( + STATE_ENGAGED, + STATE_PAUSED, + EngagementVerdict, +) +from steam_backlog_enforcer._playtime_session import PlaytimeSession + +if TYPE_CHECKING: + from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + + +class StubTracker: + """An engagement tracker with a fixed answer and no probes.""" + + def __init__(self, *, engaged: bool = True) -> None: + """Record the answer this tracker always gives. + + Args: + engaged: Whether every tick should count. + """ + self.engaged = engaged + self.verdicts: list[EngagementVerdict] = [] + + def assess( + self, + rules: PlaytimeRules, + *, + qualifying: set[int], + now_monotonic: float, + ) -> EngagementVerdict: + """Return the fixed verdict. + + Args: + rules: Ignored. + qualifying: PIDs recorded on the verdict. + now_monotonic: Ignored. + + Returns: + The verdict. + """ + del rules, now_monotonic + verdict = EngagementVerdict( + state=STATE_ENGAGED if self.engaged else STATE_PAUSED, + causes=() if self.engaged else ("focus",), + qualifying=tuple(sorted(qualifying)), + ) + self.verdicts.append(verdict) + return verdict + + def backdate( + self, + state: PlaytimeState, + verdict: EngagementVerdict, + *, + rules: PlaytimeRules, + ) -> PlaytimeState: + """Return *state* untouched. + + Args: + state: Accounting state. + verdict: Ignored. + rules: Ignored. + + Returns: + The unchanged state. + """ + del verdict, rules + return state + + +class StubJournal: + """A journal that keeps records in memory instead of on disk.""" + + def __init__(self) -> None: + """Start with nothing recorded.""" + self.observed: list[EngagementVerdict] = [] + + def observe( + self, + verdict: EngagementVerdict, + state: PlaytimeState, + *, + rules: PlaytimeRules, + now_monotonic: float, + ) -> None: + """Remember *verdict*. + + Args: + verdict: This tick's verdict. + state: Ignored. + rules: Ignored. + now_monotonic: Ignored. + """ + del state, rules, now_monotonic + self.observed.append(verdict) + + +def fake_session(*, engaged: bool = True) -> PlaytimeSession: + """Build a session that never touches X, /proc/locks or the log file. + + Args: + engaged: Whether every tick should count against the budget. + + Returns: + The session. + """ + return PlaytimeSession(tracker=StubTracker(engaged=engaged), journal=StubJournal()) diff --git a/steam_backlog_enforcer/tests/_fake_xlib.py b/steam_backlog_enforcer/tests/_fake_xlib.py new file mode 100644 index 0000000..db1ca6e --- /dev/null +++ b/steam_backlog_enforcer/tests/_fake_xlib.py @@ -0,0 +1,99 @@ +"""Fake libX11 / libXss / libXRes handles so no test needs an X server. + +The real modules talk to X through ``ctypes``, which means the seam worth +faking is the shared-library handle itself. These stand-ins return the same +kinds of values Xlib does — an integer display pointer, a struct pointer, a +status code — and, where the C API writes through a pointer argument, actually +write through it, so the calling code is exercised rather than bypassed. +""" + +from __future__ import annotations + +import ctypes +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +if TYPE_CHECKING: + from collections.abc import Callable + + +def fake_x11(*, display: int = 0x1234, root: int = 0x50, atom: int = 7) -> MagicMock: + """Build a libX11 stand-in. + + Args: + display: Value ``XOpenDisplay`` returns; ``0`` means "cannot connect". + root: Value ``XDefaultRootWindow`` returns. + atom: Value ``XInternAtom`` returns; ``0`` means "no such atom". + + Returns: + The fake handle. + """ + lib = MagicMock(name="libX11") + lib.XOpenDisplay.return_value = display + lib.XDefaultRootWindow.return_value = root + lib.XInternAtom.return_value = atom + return lib + + +def fake_xss(*, idle_ms: int = 0, allocated: bool = True, ok: bool = True) -> MagicMock: + """Build a libXss stand-in. + + Args: + idle_ms: Idle milliseconds reported by ``XScreenSaverQueryInfo``. + allocated: Whether ``XScreenSaverAllocInfo`` returns a struct. + ok: Whether ``XScreenSaverQueryInfo`` succeeds. + + Returns: + The fake handle. + """ + lib = MagicMock(name="libXss") + if allocated: + info = MagicMock(name="XScreenSaverInfo") + info.contents.idle = idle_ms + lib.XScreenSaverAllocInfo.return_value = info + else: + lib.XScreenSaverAllocInfo.return_value = 0 + lib.XScreenSaverQueryInfo.return_value = 1 if ok else 0 + return lib + + +def property_writer( + *, status: int = 0, nitems: int = 1, value: int = 4321, null: bool = False +) -> Callable[..., int]: + """Build an ``XGetWindowProperty`` stand-in that writes its out-params. + + Args: + status: Return code; non-zero means failure. + nitems: Number of items to report. + value: The first 32-bit value to hand back. + null: Whether to leave the property pointer NULL. + + Returns: + A callable with ``XGetWindowProperty``'s signature. + """ + # Kept alive for the lifetime of the fake: the pointer written into the + # caller's out-param must not dangle. + keepalive: list[ctypes.c_ulong] = [] + + def impl( # signature mirrors the C function + _display: object, + _window: object, + _atom: object, + _offset: object, + _length: object, + _delete: object, + _req_type: object, + _actual_type: object, + _actual_format: object, + nitems_ref: object, + _bytes_after: object, + prop_ref: object, + ) -> int: + nitems_ref._obj.value = nitems # type-erased ctypes byref holder + if not null: + holder = ctypes.c_ulong(value) + keepalive.append(holder) + prop_ref._obj.contents = holder + return status + + return impl diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index a769513..3344305 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -201,3 +201,24 @@ def total_block_paths(tmp_path: Path) -> Iterator[Paths]: patch(f"{PURGE}._STEAM_REMNANT_PATHS", built.remnant_paths), ): yield built + + +# ────────────────────────────────────────────────────────────── +# Live process-table isolation +# +# The Steam-restart guard asks whether a game is running by scanning +# /proc for SteamAppId. Unpatched, that reads the *developer's* real +# process table: running the suite while playing a game made three +# restart tests fail on 2026-08-28. Tests that care about the guard +# patch this themselves. +# ────────────────────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def _no_live_games() -> Iterator[None]: + """Report no running games unless a test says otherwise.""" + with patch( + "steam_backlog_enforcer._steam_restart_guard.get_running_steam_game_pids", + return_value={}, + ): + yield diff --git a/steam_backlog_enforcer/tests/test_controller_input.py b/steam_backlog_enforcer/tests/test_controller_input.py new file mode 100644 index 0000000..86bc347 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_controller_input.py @@ -0,0 +1,137 @@ +"""Tests for noticing real gamepad input. + +The distinction that matters: this machine exposes an *ASRock LED Controller* +as ``/dev/input/js0``, so mere presence of a js device must not register as +play — only actual, non-synthetic events do. +""" + +from __future__ import annotations + +import errno +import os +from typing import TYPE_CHECKING +from unittest.mock import patch + +from steam_backlog_enforcer._controller_input import ControllerActivity + +if TYPE_CHECKING: + from pathlib import Path + +CI = "steam_backlog_enforcer._controller_input" + + +def _event(*, init: bool) -> bytes: + """Build one 8-byte ``struct js_event``; byte 6 is the type field.""" + return bytes([0, 0, 0, 0, 0, 0, 0x81 if init else 0x01, 0]) + + +class TestSyncDevices: + def test_missing_directory_is_not_an_error(self, tmp_path: Path) -> None: + activity = ControllerActivity(device_dir=tmp_path / "absent") + activity.poll(now=1.0) + assert activity.seconds_since_activity(now=1.0) is None + + def test_non_js_entries_are_ignored(self, tmp_path: Path) -> None: + (tmp_path / "event0").write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=1.0) + assert activity.seconds_since_activity(now=1.0) is None + + def test_a_vanished_device_is_forgotten(self, tmp_path: Path) -> None: + node = tmp_path / "js0" + node.write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=1.0) + node.unlink() + activity.poll(now=2.0) + activity.close() + + def test_an_unopenable_device_is_skipped(self, tmp_path: Path) -> None: + node = tmp_path / "js0" + node.write_bytes(_event(init=False)) + with patch(f"{CI}.os.open", side_effect=OSError("denied")): + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=1.0) + assert activity.seconds_since_activity(now=1.0) is None + + +class TestDrain: + def test_a_real_event_is_activity(self, tmp_path: Path) -> None: + (tmp_path / "js0").write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + assert activity.seconds_since_activity(now=12.5) == 2.5 + activity.close() + + def test_the_synthetic_open_burst_is_not_activity(self, tmp_path: Path) -> None: + # joydev replays current state on open; treating that as a button press + # would make simply having an LED device look like a gaming session. + (tmp_path / "js0").write_bytes(_event(init=True) * 3) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + assert activity.seconds_since_activity(now=10.0) is None + activity.close() + + def test_a_truncated_record_stops_the_drain(self, tmp_path: Path) -> None: + (tmp_path / "js0").write_bytes(b"\x00\x00\x00") + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + assert activity.seconds_since_activity(now=10.0) is None + activity.close() + + def test_the_per_tick_cap_bounds_the_drain(self, tmp_path: Path) -> None: + (tmp_path / "js0").write_bytes(_event(init=False) * 200) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + assert activity.seconds_since_activity(now=10.0) == 0.0 + activity.close() + + def test_eagain_leaves_the_device_open(self, tmp_path: Path) -> None: + # "No events pending" is the normal case on a quiet controller; it must + # not be mistaken for the device going away. + (tmp_path / "js0").write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + with patch(f"{CI}.os.read", side_effect=OSError(errno.EAGAIN, "again")): + activity.poll(now=10.0) + assert list(activity._fds) == [tmp_path / "js0"] + assert activity.seconds_since_activity(now=10.0) is None + activity.close() + + def test_an_empty_device_yields_nothing(self, tmp_path: Path) -> None: + fifo = tmp_path / "js0" + os.mkfifo(fifo) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + assert activity.seconds_since_activity(now=10.0) is None + activity.close() + + def test_a_device_that_stays_is_kept_open_across_ticks( + self, tmp_path: Path + ) -> None: + node = tmp_path / "js0" + node.write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + first = dict(activity._fds) + activity.poll(now=11.0) + assert activity._fds == first + activity.close() + + def test_forgetting_an_unknown_device_is_a_no_op(self, tmp_path: Path) -> None: + activity = ControllerActivity(device_dir=tmp_path) + activity._forget(tmp_path / "js9") + assert activity._fds == {} + + def test_a_hard_read_error_drops_the_device(self, tmp_path: Path) -> None: + (tmp_path / "js0").write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + with patch(f"{CI}.os.read", side_effect=OSError(errno.EIO, "gone")): + activity.poll(now=10.0) + assert activity.seconds_since_activity(now=10.0) is None + + def test_forgetting_survives_an_already_closed_fd(self, tmp_path: Path) -> None: + (tmp_path / "js0").write_bytes(_event(init=False)) + activity = ControllerActivity(device_dir=tmp_path) + activity.poll(now=10.0) + with patch(f"{CI}.os.close", side_effect=OSError("bad fd")): + activity.close() diff --git a/steam_backlog_enforcer/tests/test_enforce_loop_part2.py b/steam_backlog_enforcer/tests/test_enforce_loop_part2.py index 0379380..de59fc5 100644 --- a/steam_backlog_enforcer/tests/test_enforce_loop_part2.py +++ b/steam_backlog_enforcer/tests/test_enforce_loop_part2.py @@ -11,6 +11,7 @@ from steam_backlog_enforcer._enforce_loop import ( _enforce_loop_iteration, ) from steam_backlog_enforcer.config import Config, State +from steam_backlog_enforcer.tests._fake_session import fake_session if TYPE_CHECKING: from collections.abc import Iterator @@ -48,7 +49,7 @@ class TestEnforceLoopIteration: patch(f"{PKG}._echo"), patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) def test_no_kill(self) -> None: config = Config( @@ -60,7 +61,7 @@ class TestEnforceLoopIteration: patch(f"{PKG}.enforce_allowed_game") as mock_enforce, patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_enforce.assert_not_called() def test_guards_installed(self) -> None: @@ -74,7 +75,7 @@ class TestEnforceLoopIteration: patch(f"{PKG}._echo"), patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) def test_guard_removes_zero(self) -> None: config = Config( @@ -86,7 +87,7 @@ class TestEnforceLoopIteration: patch(f"{PKG}._guard_installed_games", return_value=0), patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) def test_skips_when_steam_absent(self) -> None: """With Steam uninstalled the iteration must do nothing. @@ -106,7 +107,7 @@ class TestEnforceLoopIteration: patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed, patch(f"{ENFORCE_STEPS_PKG}.install_game") as mock_install, ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_enforce.assert_not_called() mock_guard.assert_not_called() @@ -123,7 +124,7 @@ class TestEnforceLoopIteration: patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=False), patch(f"{ENFORCE_STEPS_PKG}.install_game") as mock_install, ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_install.assert_called_once() def test_no_app_id_skip_reinstall(self) -> None: @@ -137,7 +138,7 @@ class TestEnforceLoopIteration: patch(f"{PKG}._guard_installed_games") as mock_guard, patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed, ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_enforce.assert_not_called() mock_guard.assert_not_called() mock_installed.assert_not_called() diff --git a/steam_backlog_enforcer/tests/test_enforce_loop_part3.py b/steam_backlog_enforcer/tests/test_enforce_loop_part3.py index 2671307..4b7d4d3 100644 --- a/steam_backlog_enforcer/tests/test_enforce_loop_part3.py +++ b/steam_backlog_enforcer/tests/test_enforce_loop_part3.py @@ -9,6 +9,7 @@ from steam_backlog_enforcer._enforce_loop import ( do_enforce, ) from steam_backlog_enforcer.config import Config, State +from steam_backlog_enforcer.tests._fake_session import fake_session PKG = "steam_backlog_enforcer._enforce_loop" ENFORCE_STEPS_PKG = "steam_backlog_enforcer._enforce_steps" @@ -27,7 +28,7 @@ class TestEnforceLoopIterationTotalBlock: patch(f"{PKG}._guard_installed_games") as mock_guard, patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed, ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_tick.assert_called_once() mock_enforce.assert_not_called() mock_guard.assert_not_called() @@ -42,7 +43,7 @@ class TestEnforceLoopIterationTotalBlock: patch(f"{PKG}.enforce_total_block_tick") as mock_tick, patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_tick.assert_not_called() def test_expired_lock_triggers_cleanup_once(self) -> None: @@ -54,7 +55,7 @@ class TestEnforceLoopIterationTotalBlock: patch(f"{PKG}.end_total_block_cleanup") as mock_cleanup, patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_cleanup.assert_called_once() def test_no_lock_no_cleanup_call(self) -> None: @@ -66,7 +67,7 @@ class TestEnforceLoopIterationTotalBlock: patch(f"{PKG}.end_total_block_cleanup") as mock_cleanup, patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_cleanup.assert_not_called() @@ -145,7 +146,7 @@ class TestPlaytimeIsEnforcedFirst: patch(f"{PKG}.is_total_block_active", return_value=True), patch(f"{PKG}.enforce_total_block_tick"), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) mock_playtime.assert_called_once() def test_receives_the_loop_interval(self) -> None: @@ -156,7 +157,7 @@ class TestPlaytimeIsEnforcedFirst: patch(f"{PKG}.is_total_block_active", return_value=True), patch(f"{PKG}.enforce_total_block_tick"), ): - _enforce_loop_iteration(config, state) + _enforce_loop_iteration(config, state, session=fake_session()) assert mock_playtime.call_args.kwargs["interval"] == 3 assert mock_playtime.call_args.kwargs["demo"] is False @@ -168,7 +169,7 @@ class TestPlaytimeIsEnforcedFirst: patch(f"{PKG}.is_total_block_active", return_value=True), patch(f"{PKG}.enforce_total_block_tick"), ): - _enforce_loop_iteration(config, state, demo=True) + _enforce_loop_iteration(config, state, session=fake_session(), demo=True) assert mock_playtime.call_args.kwargs["demo"] is True def test_do_enforce_plumbs_demo_into_the_iteration(self) -> None: diff --git a/steam_backlog_enforcer/tests/test_engagement_probes.py b/steam_backlog_enforcer/tests/test_engagement_probes.py new file mode 100644 index 0000000..7645854 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_engagement_probes.py @@ -0,0 +1,155 @@ +"""Tests for turning each engagement signal into a pause cause. + +The rule under test everywhere: a probe that *fails* must not pause the tick. +Failing closed keeps a broken detector from becoming free gaming. +""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path +from unittest.mock import MagicMock, patch + +from steam_backlog_enforcer._engagement_probes import ( + assess_focus, + assess_idle, + assess_screen, + focus_qualifies, +) +from steam_backlog_enforcer._engagement_types import CauseTally +from steam_backlog_enforcer._playtime_state import PlaytimeRules +from steam_backlog_enforcer._screen_hold import ScreenHold, ScreenHoldError +from steam_backlog_enforcer._x_probe import XProbeError + +EP = "steam_backlog_enforcer._engagement_probes" + +RULES = PlaytimeRules( + budget_seconds=28800.0, + warn_at=(3600,), + sigkill_after=30.0, + count_launchers=True, + enforcement=True, + demo=False, + engagement_gate=True, + idle_grace_seconds=300.0, + require_game_focus=True, +) + + +def _probe(idle: float = 0.0) -> MagicMock: + probe = MagicMock(name="XProbe") + probe.idle_seconds.return_value = idle + return probe + + +class TestAssessScreen: + def test_a_held_screen_pauses(self) -> None: + tally = CauseTally() + held = ScreenHold(held=True, holder_pid=99) + with patch(f"{EP}.screen_hold", return_value=held): + assert assess_screen(Path("x"), tally) == (True, 99) + assert tally.causes == ["screen_held"] + assert tally.degraded == [] + + def test_a_free_screen_does_not_pause(self) -> None: + tally = CauseTally() + with patch(f"{EP}.screen_hold", return_value=ScreenHold(held=False)): + assert assess_screen(Path("x"), tally) == (False, None) + assert tally.causes == [] + + def test_a_failed_probe_bills_and_is_recorded(self) -> None: + tally = CauseTally() + with patch(f"{EP}.screen_hold", side_effect=ScreenHoldError("boom")): + assert assess_screen(Path("x"), tally) == (None, None) + assert tally.causes == [] + assert tally.degraded == ["screen_held"] + + +class TestAssessFocus: + def test_disabled_criterion_reports_nothing(self) -> None: + tally = CauseTally() + relaxed = replace(RULES, require_game_focus=False) + assert assess_focus(_probe(), relaxed, {1}, tally) is None + assert tally.causes == [] + + def test_the_game_owning_focus_does_not_pause(self) -> None: + tally = CauseTally() + with patch(f"{EP}.focused_pid", return_value=42): + assert assess_focus(_probe(), RULES, {42}, tally) == 42 + assert tally.causes == [] + + def test_another_window_owning_focus_pauses(self) -> None: + tally = CauseTally() + with patch(f"{EP}.focused_pid", return_value=7): + assert assess_focus(_probe(), RULES, {42}, tally) == 7 + assert tally.causes == ["focus"] + + def test_a_failed_probe_bills_and_drops_the_connection(self) -> None: + tally = CauseTally() + probe = _probe() + with patch(f"{EP}.focused_pid", side_effect=XProbeError("no display")): + assert assess_focus(probe, RULES, {42}, tally) is None + assert tally.causes == [] + assert tally.degraded == ["focus"] + probe.close.assert_called_once() + + +class TestAssessIdle: + def test_recent_input_does_not_pause(self) -> None: + tally = CauseTally() + controller = MagicMock() + controller.seconds_since_activity.return_value = None + assert assess_idle(_probe(1.0), controller, RULES, tally, 5.0) == (1.0, None) + assert tally.causes == [] + + def test_silence_past_the_grace_pauses(self) -> None: + tally = CauseTally() + controller = MagicMock() + controller.seconds_since_activity.return_value = None + assess_idle(_probe(400.0), controller, RULES, tally, 5.0) + assert tally.causes == ["idle"] + + def test_recent_controller_input_keeps_the_tick_alive(self) -> None: + # The X idle counter cannot see a gamepad; without this, a controller + # session would pause the budget and hand out free gaming. + tally = CauseTally() + controller = MagicMock() + controller.seconds_since_activity.return_value = 2.0 + assert assess_idle(_probe(9999.0), controller, RULES, tally, 5.0) == (2.0, 2.0) + assert tally.causes == [] + + def test_a_failed_probe_bills_and_drops_the_connection(self) -> None: + tally = CauseTally() + probe = _probe() + probe.idle_seconds.side_effect = XProbeError("no display") + controller = MagicMock() + controller.seconds_since_activity.return_value = 7.0 + assert assess_idle(probe, controller, RULES, tally, 5.0) == (None, 7.0) + assert tally.causes == [] + assert tally.degraded == ["idle"] + probe.close.assert_called_once() + + +class TestFocusQualifies: + def test_nothing_focused_does_not_qualify(self) -> None: + assert focus_qualifies(None, {1234}) is False + + def test_the_process_itself_qualifies(self) -> None: + assert focus_qualifies(1234, {1234}) is True + + def test_a_descendant_window_qualifies(self) -> None: + # Proton games commonly own their window from a child process. + with patch(f"{EP}._read_ppid", side_effect=[500, 42]): + assert focus_qualifies(9000, {42}) is True + + def test_reaching_init_does_not_qualify(self) -> None: + with patch(f"{EP}._read_ppid", return_value=1): + assert focus_qualifies(9000, {42}) is False + + def test_an_unreadable_parent_does_not_qualify(self) -> None: + with patch(f"{EP}._read_ppid", return_value=None): + assert focus_qualifies(9000, {42}) is False + + def test_a_pathological_chain_terminates(self) -> None: + with patch(f"{EP}._read_ppid", side_effect=lambda pid: pid + 1): + assert focus_qualifies(9000, {42}) is False diff --git a/steam_backlog_enforcer/tests/test_engagement_types.py b/steam_backlog_enforcer/tests/test_engagement_types.py new file mode 100644 index 0000000..7b56f42 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_engagement_types.py @@ -0,0 +1,50 @@ +"""Tests for the engagement verdict value object.""" + +from __future__ import annotations + +from steam_backlog_enforcer._engagement_types import ( + CAUSE_FOCUS, + CAUSE_IDLE, + STATE_ENGAGED, + STATE_NOT_APPLICABLE, + STATE_PAUSED, + EngagementVerdict, +) + + +class TestEngaged: + def test_engaged_state_bills(self) -> None: + assert EngagementVerdict(state=STATE_ENGAGED).engaged is True + + def test_nothing_qualifying_still_bills(self) -> None: + # Nothing accrues anyway; calling it "paused" would fill the audit log + # with a pause reason during ordinary desktop use. + assert EngagementVerdict(state=STATE_NOT_APPLICABLE).engaged is True + + def test_paused_state_does_not_bill(self) -> None: + assert EngagementVerdict(state=STATE_PAUSED).engaged is False + + +class TestReason: + def test_falls_back_to_the_state_when_there_are_no_causes(self) -> None: + assert EngagementVerdict(state=STATE_ENGAGED).reason == STATE_ENGAGED + + def test_joins_every_cause(self) -> None: + verdict = EngagementVerdict( + state=STATE_PAUSED, causes=(CAUSE_IDLE, CAUSE_FOCUS) + ) + assert verdict.reason == "idle+focus" + + +class TestIdleOnly: + def test_true_for_a_sole_idle_cause(self) -> None: + assert EngagementVerdict(state=STATE_PAUSED, causes=(CAUSE_IDLE,)).idle_only() + + def test_false_when_something_else_also_paused_the_tick(self) -> None: + verdict = EngagementVerdict( + state=STATE_PAUSED, causes=(CAUSE_IDLE, CAUSE_FOCUS) + ) + assert verdict.idle_only() is False + + def test_false_when_engaged(self) -> None: + assert EngagementVerdict(state=STATE_ENGAGED).idle_only() is False diff --git a/steam_backlog_enforcer/tests/test_playtime_engagement.py b/steam_backlog_enforcer/tests/test_playtime_engagement.py new file mode 100644 index 0000000..9ca606c --- /dev/null +++ b/steam_backlog_enforcer/tests/test_playtime_engagement.py @@ -0,0 +1,201 @@ +"""Tests for the per-tick engagement verdict and the idle backdate. + +The backdate is deliberately hard to trigger: it is the only place the budget +counter ever moves backwards, so every guard around it is tested here. +""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path +from unittest.mock import MagicMock, patch + +from steam_backlog_enforcer._engagement_types import ( + CAUSE_FOCUS, + CAUSE_IDLE, + STATE_ENGAGED, + STATE_NOT_APPLICABLE, + STATE_PAUSED, + CauseTally, + EngagementVerdict, +) +from steam_backlog_enforcer._playtime_engagement import EngagementTracker +from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + +PE = "steam_backlog_enforcer._playtime_engagement" + +RULES = PlaytimeRules( + budget_seconds=28800.0, + warn_at=(3600,), + sigkill_after=30.0, + count_launchers=True, + enforcement=True, + demo=False, + engagement_gate=True, + idle_grace_seconds=300.0, + require_game_focus=True, +) + + +def _tracker(**kwargs: object) -> EngagementTracker: + return EngagementTracker( + uid=1000, + probe=MagicMock(name="XProbe"), + controller=MagicMock(name="ControllerActivity"), + holder_path=Path("/nonexistent/holder.lock"), + **kwargs, + ) + + +def _engaged() -> EngagementVerdict: + return EngagementVerdict(state=STATE_ENGAGED) + + +def _idle_paused() -> EngagementVerdict: + return EngagementVerdict(state=STATE_PAUSED, causes=(CAUSE_IDLE,)) + + +class TestAssess: + def test_nothing_qualifying_is_not_applicable(self) -> None: + verdict = _tracker().assess(RULES, qualifying=set(), now_monotonic=1.0) + assert verdict.state == STATE_NOT_APPLICABLE + assert verdict.engaged is True + + def test_a_disabled_gate_bills_without_probing(self) -> None: + tracker = _tracker() + with patch(f"{PE}.assess_screen") as screen: + verdict = tracker.assess( + replace(RULES, engagement_gate=False), + qualifying={7}, + now_monotonic=1.0, + ) + assert verdict.state == STATE_ENGAGED + assert verdict.qualifying == (7,) + screen.assert_not_called() + + def test_every_cause_is_recorded_not_just_the_first(self) -> None: + # The log has to be able to answer "was the screen ALSO locked?", and + # the backdate must be able to tell an idle-only pause from a compound. + tracker = _tracker() + with ( + patch(f"{PE}.assess_screen", side_effect=_add("screen_held", (True, 5))), + patch(f"{PE}.assess_focus", side_effect=_add_focus("focus", 9)), + patch(f"{PE}.assess_idle", side_effect=_add_idle("idle", (900.0, None))), + ): + verdict = tracker.assess(RULES, qualifying={7}, now_monotonic=1.0) + assert verdict.state == STATE_PAUSED + assert set(verdict.causes) == {"screen_held", "focus", "idle"} + assert verdict.holder_pid == 5 + assert verdict.focus_pid == 9 + + def test_no_causes_means_engaged(self) -> None: + tracker = _tracker() + with ( + patch(f"{PE}.assess_screen", return_value=(False, None)), + patch(f"{PE}.assess_focus", return_value=7), + patch(f"{PE}.assess_idle", return_value=(1.0, None)), + ): + verdict = tracker.assess(RULES, qualifying={7}, now_monotonic=1.0) + assert verdict.state == STATE_ENGAGED + + +class TestClose: + def test_releases_both_resources(self) -> None: + probe = MagicMock() + controller = MagicMock() + tracker = EngagementTracker( + uid=1000, probe=probe, controller=controller, holder_path=Path("x") + ) + tracker.close() + probe.close.assert_called_once() + controller.close.assert_called_once() + + +class TestBackdate: + def _state(self, seconds: float = 1000.0) -> PlaytimeState: + return PlaytimeState(day_key="2026-08-28", seconds=seconds) + + def test_refunds_the_grace_on_the_engaged_to_idle_edge(self) -> None: + tracker = _tracker() + state = tracker.backdate(self._state(), _engaged(), rules=RULES) + state = tracker.backdate(state, _idle_paused(), rules=RULES) + assert state.seconds == 700.0 + + def test_refunds_only_once_per_walk_away(self) -> None: + tracker = _tracker() + state = tracker.backdate(self._state(), _engaged(), rules=RULES) + state = tracker.backdate(state, _idle_paused(), rules=RULES) + state = tracker.backdate(state, _idle_paused(), rules=RULES) + assert state.seconds == 700.0 + + def test_never_refunds_below_zero(self) -> None: + tracker = _tracker() + state = tracker.backdate(self._state(seconds=10.0), _engaged(), rules=RULES) + state = tracker.backdate(state, _idle_paused(), rules=RULES) + assert state.seconds == 0.0 + + def test_no_refund_on_the_first_tick_after_a_restart(self) -> None: + # With no previous verdict there is no edge to recognise, and inventing + # one would hand out a free refund on every daemon restart. + tracker = _tracker() + state = tracker.backdate(self._state(), _idle_paused(), rules=RULES) + assert state.seconds == 1000.0 + + def test_no_refund_across_a_day_rollover(self) -> None: + tracker = _tracker() + tracker.backdate(self._state(), _engaged(), rules=RULES) + tomorrow = PlaytimeState(day_key="2026-08-29", seconds=1000.0) + assert tracker.backdate(tomorrow, _idle_paused(), rules=RULES).seconds == 1000.0 + + def test_no_refund_when_something_else_also_paused_the_tick(self) -> None: + tracker = _tracker() + state = tracker.backdate(self._state(), _engaged(), rules=RULES) + compound = EngagementVerdict( + state=STATE_PAUSED, causes=(CAUSE_IDLE, CAUSE_FOCUS) + ) + assert tracker.backdate(state, compound, rules=RULES).seconds == 1000.0 + + def test_no_refund_while_still_engaged(self) -> None: + tracker = _tracker() + state = tracker.backdate(self._state(), _engaged(), rules=RULES) + assert tracker.backdate(state, _engaged(), rules=RULES).seconds == 1000.0 + + def test_reports_the_measured_idle_time(self) -> None: + tracker = _tracker() + tracker.backdate(self._state(), _engaged(), rules=RULES) + verdict = EngagementVerdict( + state=STATE_PAUSED, causes=(CAUSE_IDLE,), idle_seconds=301.0 + ) + assert tracker.backdate(self._state(), verdict, rules=RULES).seconds == 700.0 + + +def _add(cause: str, result: tuple[bool | None, int | None]) -> object: + def impl(_path: object, tally: CauseTally) -> object: + tally.causes.append(cause) + return result + + return impl + + +def _add_focus(cause: str, pid: int) -> object: + def impl( + _probe: object, _rules: object, _qualifying: object, tally: CauseTally + ) -> int: + tally.causes.append(cause) + return pid + + return impl + + +def _add_idle(cause: str, result: tuple[float | None, float | None]) -> object: + def impl( + _probe: object, + _controller: object, + _rules: object, + tally: CauseTally, + _now: float, + ) -> object: + tally.causes.append(cause) + return result + + return impl diff --git a/steam_backlog_enforcer/tests/test_playtime_log.py b/steam_backlog_enforcer/tests/test_playtime_log.py new file mode 100644 index 0000000..500c37b --- /dev/null +++ b/steam_backlog_enforcer/tests/test_playtime_log.py @@ -0,0 +1,192 @@ +"""Tests for the budget audit trail. + +Two properties matter. It must never raise — a logging failure taking down the +enforcer would be worse than the missing record — and it must not write a line +per tick, which at a three-second interval would be 28 800 lines a day. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer._engagement_types import ( + CAUSE_IDLE, + STATE_ENGAGED, + STATE_NOT_APPLICABLE, + STATE_PAUSED, + EngagementVerdict, +) +from steam_backlog_enforcer._playtime_log import ( + EVENT_DETECTOR_FAILURE, + EVENT_HEARTBEAT, + EVENT_VERDICT_CHANGE, + BudgetLog, + TickJournal, +) +from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +PL = "steam_backlog_enforcer._playtime_log" + +RULES = PlaytimeRules( + budget_seconds=28800.0, + warn_at=(3600,), + sigkill_after=30.0, + count_launchers=True, + enforcement=True, + demo=False, +) +STATE = PlaytimeState(day_key="2026-08-28", seconds=100.0) + + +def _records(path: Path) -> list[dict]: + if not path.exists(): + return [] + return [json.loads(line) for line in path.read_text().splitlines() if line] + + +class TestBudgetLog: + def test_creates_its_directory_on_first_write(self, tmp_path: Path) -> None: + target = tmp_path / "deep" / "nested" / "budget.jsonl" + log = BudgetLog(path=target) + log.record("hello", answer=42) + log.close() + assert _records(target)[0]["answer"] == 42 + + def test_closing_an_unopened_sink_is_a_no_op(self, tmp_path: Path) -> None: + BudgetLog(path=tmp_path / "budget.jsonl").close() + + def test_a_second_sink_on_the_same_path_replaces_the_first( + self, tmp_path: Path + ) -> None: + # Two sinks share a logger name; the stale handler must be closed + # rather than silently dropped, or the file descriptor leaks. + target = tmp_path / "budget.jsonl" + first = BudgetLog(path=target) + first.record("one") + second = BudgetLog(path=target) + second.record("two") + second.close() + assert [r["event"] for r in _records(target)] == ["one", "two"] + + def test_every_record_carries_a_timestamp_and_event(self, tmp_path: Path) -> None: + target = tmp_path / "budget.jsonl" + log = BudgetLog(path=target) + log.record("thing") + log.close() + record = _records(target)[0] + assert record["event"] == "thing" + assert record["timestamp"] + + def test_reuses_the_sink_across_records(self, tmp_path: Path) -> None: + target = tmp_path / "budget.jsonl" + log = BudgetLog(path=target) + log.record("one") + log.record("two") + log.close() + assert [r["event"] for r in _records(target)] == ["one", "two"] + + def test_an_uncreatable_path_does_not_raise(self, tmp_path: Path) -> None: + blocker = tmp_path / "file" + blocker.write_text("not a directory") + BudgetLog(path=blocker / "budget.jsonl").record("dropped") + + def test_an_unserialisable_payload_does_not_raise(self, tmp_path: Path) -> None: + target = tmp_path / "budget.jsonl" + log = BudgetLog(path=target) + with patch(f"{PL}.json.dumps", side_effect=ValueError("nope")): + log.record("bad") + log.close() + + def test_a_write_failure_does_not_raise(self, tmp_path: Path) -> None: + target = tmp_path / "budget.jsonl" + log = BudgetLog(path=target) + log.record("first") + with patch(f"{PL}.json.dumps", side_effect=OSError("disk gone")): + log.record("second") + log.close() + + +class TestTickJournal: + def _journal(self, tmp_path: Path, **kwargs: float) -> tuple[TickJournal, Path]: + target = tmp_path / "budget.jsonl" + log = BudgetLog(path=target) + self._logs.append(log) + return TickJournal(log, **kwargs), target + + @pytest.fixture(autouse=True) + def _close_logs(self) -> Iterator[None]: + self._logs: list[BudgetLog] = [] + yield + for log in self._logs: + log.close() + + def _observe( + self, journal: TickJournal, verdict: EngagementVerdict, at: float + ) -> None: + journal.observe(verdict, STATE, rules=RULES, now_monotonic=at) + + def test_a_changed_verdict_is_recorded(self, tmp_path: Path) -> None: + journal, target = self._journal(tmp_path) + self._observe(journal, EngagementVerdict(state=STATE_ENGAGED), 0.0) + events = [r["event"] for r in _records(target)] + assert events == [EVENT_VERDICT_CHANGE] + + def test_an_unchanged_verdict_is_not_repeated(self, tmp_path: Path) -> None: + journal, target = self._journal(tmp_path, heartbeat=300.0) + verdict = EngagementVerdict(state=STATE_ENGAGED) + for tick in range(5): + self._observe(journal, verdict, float(tick) * 3.0) + assert len(_records(target)) == 1 + + def test_a_heartbeat_is_written_once_the_interval_passes( + self, tmp_path: Path + ) -> None: + journal, target = self._journal(tmp_path, heartbeat=10.0) + verdict = EngagementVerdict(state=STATE_ENGAGED) + self._observe(journal, verdict, 0.0) + self._observe(journal, verdict, 5.0) + self._observe(journal, verdict, 50.0) + assert [r["event"] for r in _records(target)] == [ + EVENT_VERDICT_CHANGE, + EVENT_HEARTBEAT, + ] + + def test_ordinary_desktop_use_produces_no_heartbeat(self, tmp_path: Path) -> None: + journal, target = self._journal(tmp_path, heartbeat=0.0) + verdict = EngagementVerdict(state=STATE_NOT_APPLICABLE) + self._observe(journal, verdict, 0.0) + self._observe(journal, verdict, 1000.0) + assert len(_records(target)) == 1 + + def test_a_degraded_probe_is_always_recorded(self, tmp_path: Path) -> None: + journal, target = self._journal(tmp_path) + verdict = EngagementVerdict(state=STATE_ENGAGED, degraded=("idle",)) + self._observe(journal, verdict, 0.0) + self._observe(journal, verdict, 3.0) + events = [r["event"] for r in _records(target)] + assert events.count(EVENT_DETECTOR_FAILURE) == 2 + assert all(r.get("billed", True) for r in _records(target)) + + def test_the_snapshot_names_the_qualifying_processes(self, tmp_path: Path) -> None: + # Without these, accrual from an orphaned helper process cannot be + # attributed after the fact. + journal, target = self._journal(tmp_path) + verdict = EngagementVerdict( + state=STATE_PAUSED, + causes=(CAUSE_IDLE,), + idle_seconds=310.0, + qualifying=(11, 22), + ) + self._observe(journal, verdict, 0.0) + record = _records(target)[0] + assert record["qualifying"] == [11, 22] + assert record["reason"] == "idle" + assert record["remaining_seconds"] == 28700.0 diff --git a/steam_backlog_enforcer/tests/test_playtime_session.py b/steam_backlog_enforcer/tests/test_playtime_session.py new file mode 100644 index 0000000..f43d6b9 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_playtime_session.py @@ -0,0 +1,33 @@ +"""Tests for building the per-daemon budget session.""" + +from __future__ import annotations + +from unittest.mock import patch + +from steam_backlog_enforcer._playtime_engagement import EngagementTracker +from steam_backlog_enforcer._playtime_log import TickJournal +from steam_backlog_enforcer._playtime_session import new_session + +PS = "steam_backlog_enforcer._playtime_session" + + +class TestNewSession: + def test_builds_a_tracker_and_a_journal(self) -> None: + with ( + patch(f"{PS}.resolve_desktop_user", return_value="kuhy"), + patch(f"{PS}.desktop_uid", return_value=1000) as uid, + ): + session = new_session() + assert isinstance(session.tracker, EngagementTracker) + assert isinstance(session.journal, TickJournal) + uid.assert_called_once_with("kuhy") + + def test_binds_the_tracker_to_the_desktop_users_runtime_dir(self) -> None: + # Root's own /run/user/0 would be the wrong place to look for the + # gatelock holder file. + with ( + patch(f"{PS}.resolve_desktop_user", return_value="kuhy"), + patch(f"{PS}.desktop_uid", return_value=1000), + ): + session = new_session() + assert "/run/user/1000/" in str(session.tracker._holder_path) diff --git a/steam_backlog_enforcer/tests/test_playtime_tick.py b/steam_backlog_enforcer/tests/test_playtime_tick.py index 105952c..82b0439 100644 --- a/steam_backlog_enforcer/tests/test_playtime_tick.py +++ b/steam_backlog_enforcer/tests/test_playtime_tick.py @@ -6,9 +6,7 @@ from datetime import ( timedelta, timezone, ) -from unittest.mock import ( - patch, -) +from unittest.mock import patch import pytest @@ -24,6 +22,7 @@ from steam_backlog_enforcer._playtime_state import ( save_state, ) from steam_backlog_enforcer.config import Config +from steam_backlog_enforcer.tests._fake_session import fake_session PKG = "steam_backlog_enforcer._playtime" LOCAL = timezone(timedelta(hours=2)) @@ -164,7 +163,7 @@ class TestPlaytimeTick: patch(f"{PKG}.datetime") as mock_dt, ): mock_dt.now.return_value = NOW - playtime_tick(Config(), interval=3.0) + playtime_tick(Config(), interval=3.0, session=fake_session()) assert load_state(demo=False).seconds == 3.0 def test_total_block_releases_and_stops_accruing(self, quiet_tick: dict) -> None: @@ -181,7 +180,7 @@ class TestPlaytimeTick: patch(f"{PKG}.datetime") as mock_dt, ): mock_dt.now.return_value = NOW - playtime_tick(Config(), interval=3.0) + playtime_tick(Config(), interval=3.0, session=fake_session()) quiet_tick["reconcile"].assert_called_once_with(should_block=False) assert load_state(demo=False).seconds == 5.0 @@ -200,7 +199,7 @@ class TestPlaytimeTick: patch(f"{PKG}.datetime") as mock_dt, ): mock_dt.now.return_value = NOW - playtime_tick(Config(), interval=3.0) + playtime_tick(Config(), interval=3.0, session=fake_session()) stored = load_state(demo=False) assert stored.day_key == TODAY assert stored.seconds == 0.0 @@ -222,7 +221,7 @@ class TestPlaytimeTick: patch(f"{PKG}.datetime") as mock_dt, ): mock_dt.now.return_value = NOW - playtime_tick(Config(), interval=3.0, demo=True) + playtime_tick(Config(), interval=3.0, session=fake_session(), demo=True) stored = load_state(demo=True) assert stored.is_blocked() is True quiet_tick["request_steam_shutdown"].assert_called_once() @@ -234,7 +233,7 @@ class TestPlaytimeTick: patch(f"{PKG}.datetime") as mock_dt, ): mock_dt.now.return_value = NOW - playtime_tick(Config(), interval=3.0, demo=True) + playtime_tick(Config(), interval=3.0, session=fake_session(), demo=True) assert load_state(demo=True) is not None assert load_state(demo=False) is None @@ -244,7 +243,7 @@ class TestPlaytimeTick: patch(f"{PKG}.datetime") as mock_dt, ): mock_dt.now.return_value = NOW - playtime_tick(Config(), interval=3.0) + playtime_tick(Config(), interval=3.0, session=fake_session()) stored = load_state(demo=False) assert stored is not None assert stored.day_key == TODAY diff --git a/steam_backlog_enforcer/tests/test_screen_hold.py b/steam_backlog_enforcer/tests/test_screen_hold.py new file mode 100644 index 0000000..446b59b --- /dev/null +++ b/steam_backlog_enforcer/tests/test_screen_hold.py @@ -0,0 +1,111 @@ +"""Tests for observing a gatelock screen hold through ``/proc/locks``. + +The probe must never take the lock itself, so every case here is driven by +synthetic ``/proc/locks`` text rather than by acquiring anything. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer._screen_hold import ( + ScreenHoldError, + holder_lock_path, + screen_hold, +) + +SH = "steam_backlog_enforcer._screen_hold" + + +def _device_field(path: Path) -> str: + stat = path.stat() + return f"{os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x}:{stat.st_ino}" + + +class TestHolderLockPath: + def test_uses_the_runtime_dir_of_the_session_owner(self) -> None: + assert holder_lock_path(1000) == Path("/run/user/1000/gatelock/holder.lock") + + def test_differs_per_uid(self) -> None: + assert holder_lock_path(1000) != holder_lock_path(1001) + + +class TestScreenHold: + def test_missing_file_is_not_held_and_not_an_error(self, tmp_path: Path) -> None: + result = screen_hold(tmp_path / "never-created.lock") + assert result.held is False + assert result.holder_pid is None + + def test_reports_held_with_the_holding_pid(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + locks = tmp_path / "locks" + locks.write_text( + f"1: FLOCK ADVISORY WRITE 4242 {_device_field(lock)} 0 EOF\n" + ) + with patch(f"{SH}._PROC_LOCKS", locks): + result = screen_hold(lock) + assert result.held is True + assert result.holder_pid == 4242 + + def test_reports_free_when_another_file_is_locked(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + locks = tmp_path / "locks" + locks.write_text("1: FLOCK ADVISORY WRITE 4242 00:19:999999 0 EOF\n") + with patch(f"{SH}._PROC_LOCKS", locks): + assert screen_hold(lock).held is False + + def test_a_waiter_row_is_not_a_holder(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + locks = tmp_path / "locks" + locks.write_text( + f"1: -> FLOCK ADVISORY WRITE 4242 {_device_field(lock)} 0 EOF\n" + ) + with patch(f"{SH}._PROC_LOCKS", locks): + assert screen_hold(lock).held is False + + def test_rows_without_a_device_field_are_skipped(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + locks = tmp_path / "locks" + locks.write_text( + "garbage row with no device\n" + f"2: FLOCK ADVISORY READ 7 {_device_field(lock)} 0 EOF\n" + ) + with patch(f"{SH}._PROC_LOCKS", locks): + assert screen_hold(lock).held is True + + def test_short_row_yields_no_pid(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + locks = tmp_path / "locks" + locks.write_text(f"1: FLOCK {_device_field(lock)}\n") + with patch(f"{SH}._PROC_LOCKS", locks): + result = screen_hold(lock) + assert result.held is True + assert result.holder_pid is None + + def test_non_numeric_pid_yields_no_pid(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + locks = tmp_path / "locks" + locks.write_text( + f"1: FLOCK ADVISORY WRITE nope {_device_field(lock)} 0 EOF\n" + ) + with patch(f"{SH}._PROC_LOCKS", locks): + assert screen_hold(lock).holder_pid is None + + def test_unreadable_proc_locks_raises(self, tmp_path: Path) -> None: + lock = tmp_path / "holder.lock" + lock.touch() + with ( + patch(f"{SH}._PROC_LOCKS", tmp_path / "absent"), + pytest.raises(ScreenHoldError), + ): + screen_hold(lock) diff --git a/steam_backlog_enforcer/tests/test_steam_restart_guard.py b/steam_backlog_enforcer/tests/test_steam_restart_guard.py new file mode 100644 index 0000000..80130d4 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_steam_restart_guard.py @@ -0,0 +1,71 @@ +"""Tests for refusing to bounce Steam when the bounce would destroy something. + +The game case is the one that cost a real session: restarting the daemon on +2026-08-28 relaunched Steam for its CDP port and killed a live game with it. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer._steam_errors import ( + GameInProgressError, + SteamUpdateInProgressError, +) +from steam_backlog_enforcer._steam_restart_guard import ( + assert_safe_to_restart, + game_is_running, +) + +GUARD = "steam_backlog_enforcer._steam_restart_guard" + + +class TestGameIsRunning: + def test_no_processes_means_no_game(self) -> None: + with patch(f"{GUARD}.get_running_steam_game_pids", return_value={}): + assert game_is_running() is False + + def test_the_steam_client_itself_does_not_count(self) -> None: + with patch(f"{GUARD}.get_running_steam_game_pids", return_value={42: 0}): + assert game_is_running() is False + + def test_a_real_app_id_counts(self) -> None: + with patch( + f"{GUARD}.get_running_steam_game_pids", return_value={42: 0, 43: 3164500} + ): + assert game_is_running() is True + + +class TestAssertSafeToRestart: + def test_refuses_while_a_game_runs(self) -> None: + with ( + patch(f"{GUARD}.get_running_steam_game_pids", return_value={7: 412830}), + patch(f"{GUARD}.steam_update_in_progress", return_value=False), + pytest.raises(GameInProgressError), + ): + assert_safe_to_restart() + + def test_a_running_game_outranks_an_update(self) -> None: + with ( + patch(f"{GUARD}.get_running_steam_game_pids", return_value={7: 412830}), + patch(f"{GUARD}.steam_update_in_progress", return_value=True), + pytest.raises(GameInProgressError), + ): + assert_safe_to_restart() + + def test_refuses_during_an_update(self) -> None: + with ( + patch(f"{GUARD}.get_running_steam_game_pids", return_value={}), + patch(f"{GUARD}.steam_update_in_progress", return_value=True), + pytest.raises(SteamUpdateInProgressError), + ): + assert_safe_to_restart() + + def test_allows_when_idle(self) -> None: + with ( + patch(f"{GUARD}.get_running_steam_game_pids", return_value={}), + patch(f"{GUARD}.steam_update_in_progress", return_value=False), + ): + assert_safe_to_restart() diff --git a/steam_backlog_enforcer/tests/test_update_restart_guard.py b/steam_backlog_enforcer/tests/test_update_restart_guard.py index c7f6f5f..a6021d8 100644 --- a/steam_backlog_enforcer/tests/test_update_restart_guard.py +++ b/steam_backlog_enforcer/tests/test_update_restart_guard.py @@ -24,6 +24,9 @@ if TYPE_CHECKING: SS = "steam_backlog_enforcer._steam_state" LH = "steam_backlog_enforcer._steam_launch" +# ensure_steam_debug_port defers through the shared restart guard, which +# imports steam_update_in_progress into its own namespace. +GUARD = "steam_backlog_enforcer._steam_restart_guard" def _manifest(*, to_download: int, downloaded: int, to_stage: int, staged: int) -> str: @@ -106,7 +109,7 @@ class TestEnsureSteamDebugPortDefersUpdate: patch(f"{LH}.steam_is_installed", return_value=True), patch(f"{LH}._steam_has_debug_port", return_value=False), patch(f"{LH}._is_steam_running", return_value=True), - patch(f"{LH}.steam_update_in_progress", return_value=True), + patch(f"{GUARD}.steam_update_in_progress", return_value=True), patch(f"{LH}._shutdown_steam") as shutdown, patch(f"{LH}._launch_steam_with_debug") as launch, ): @@ -117,6 +120,18 @@ class TestEnsureSteamDebugPortDefersUpdate: class TestRestartSteamGuard: + def test_skips_while_a_game_is_running(self) -> None: + # The interactive restart path has the same teeth as the daemon's: + # bouncing Steam kills the game and any unsaved progress with it. + with ( + patch(f"{LH}.game_is_running", return_value=True), + patch(f"{LH}._shutdown_steam") as shutdown, + patch(f"{LH}._launch_steam_with_debug") as launch, + ): + restart_steam() + shutdown.assert_not_called() + launch.assert_not_called() + def test_skips_when_update_in_progress(self) -> None: with ( patch(f"{LH}.steam_update_in_progress", return_value=True), diff --git a/steam_backlog_enforcer/tests/test_x_focus.py b/steam_backlog_enforcer/tests/test_x_focus.py new file mode 100644 index 0000000..19c5189 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_x_focus.py @@ -0,0 +1,187 @@ +"""Tests for attributing the focused window to a process. + +The XRes fallback is not a nicety: measured on this machine, Aseprite's window +carries no ``_NET_WM_PID`` at all, and treating that as "cannot tell" would +make the focus criterion useless for every application that behaves the same. +""" + +from __future__ import annotations + +import ctypes +from typing import TYPE_CHECKING, cast +from unittest.mock import MagicMock + +import pytest + +from steam_backlog_enforcer._x_focus import _XResValue, focused_pid, read_cardinal +from steam_backlog_enforcer._x_probe import Conn, XProbeError +from steam_backlog_enforcer.tests._fake_xlib import fake_x11, property_writer + +if TYPE_CHECKING: + from collections.abc import Callable + + +def _conn(x11: MagicMock, xres: MagicMock | None = None) -> Conn: + """Wrap fake handles in a Conn, casting only at this one seam.""" + return Conn( + display=ctypes.c_void_p(1), + x11=cast("ctypes.CDLL", x11), + xss=cast("ctypes.CDLL", MagicMock()), + xres=None if xres is None else cast("ctypes.CDLL", xres), + ) + + +def _xres( + *, + pids: list[int] | None = None, + entries: list[tuple[int, int, int | None]] | None = None, + status: int = 0, + null: bool = False, +) -> MagicMock: + """Build a libXRes stand-in. + + Args: + pids: Shorthand for well-formed PID entries. + entries: Raw ``(mask, length, pid)`` rows; ``pid=None`` means a NULL + value pointer, which the server can legitimately return. + status: Return code; non-zero means the query failed. + null: Whether to leave the results pointer NULL. + """ + rows = ( + entries + if entries is not None + else [(1 << 1, 4, pid) for pid in (pids if pids is not None else [])] + ) + lib = MagicMock(name="libXRes") + array = (_XResValue * max(len(rows), 1))() + keepalive: list[ctypes.c_uint32] = [] + for index, (mask, length, pid) in enumerate(rows): + array[index].spec.mask = mask + array[index].length = length + if pid is None: + array[index].value = None + else: + holder = ctypes.c_uint32(pid) + keepalive.append(holder) + array[index].value = ctypes.cast( + ctypes.byref(holder), ctypes.c_void_p + ).value + + def impl( + _display: object, + _n: object, + _spec: object, + count_ref: object, + values_ref: object, + ) -> int: + count_ref._obj.value = len(rows) + if not null: + values_ref._obj.contents = array[0] + return status + + lib.XResQueryClientIds.side_effect = impl + lib._keepalive = (array, keepalive) + return lib + + +class TestReadCardinal: + def test_returns_the_first_value(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = property_writer(value=99) + assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") == 99 + + def test_a_missing_atom_is_none(self) -> None: + assert read_cardinal(_conn(fake_x11(atom=0)), 0x1, b"_NOPE") is None + + def test_a_failed_request_is_none(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = property_writer(status=1) + assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") is None + + def test_a_null_property_is_none(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = property_writer(null=True) + assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") is None + + def test_an_empty_property_is_none_and_still_freed(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = property_writer(nitems=0) + assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") is None + x11.XFree.assert_called_once() + + +class TestFocusedPid: + def test_no_focused_window_is_none(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = property_writer(value=0) + assert focused_pid(_conn(x11)) is None + + def test_uses_net_wm_pid_when_present(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = property_writer(value=555) + assert focused_pid(_conn(x11)) == 555 + + def test_falls_back_to_xres(self) -> None: + x11 = fake_x11() + # The active window resolves, then _NET_WM_PID comes back absent. + x11.XGetWindowProperty.side_effect = _sequence( + property_writer(value=0x400001), property_writer(nitems=0) + ) + assert focused_pid(_conn(x11, _xres(pids=[8080]))) == 8080 + + def test_raises_when_neither_source_can_attribute_the_window(self) -> None: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = _sequence( + property_writer(value=0x400001), property_writer(nitems=0) + ) + with pytest.raises(XProbeError, match="XRes"): + focused_pid(_conn(x11, None)) + + +class TestXResFallback: + def _focus_without_wm_pid(self) -> MagicMock: + x11 = fake_x11() + x11.XGetWindowProperty.side_effect = _sequence( + property_writer(value=0x400001), property_writer(nitems=0) + ) + return x11 + + def test_a_failed_query_gives_no_pid(self) -> None: + with pytest.raises(XProbeError): + focused_pid(_conn(self._focus_without_wm_pid(), _xres(status=1))) + + def test_a_null_result_gives_no_pid(self) -> None: + with pytest.raises(XProbeError): + focused_pid(_conn(self._focus_without_wm_pid(), _xres(null=True))) + + def test_no_entries_gives_no_pid(self) -> None: + with pytest.raises(XProbeError): + focused_pid(_conn(self._focus_without_wm_pid(), _xres(pids=[]))) + + def test_a_non_pid_entry_is_skipped(self) -> None: + # XRes can return XID rows alongside PID rows; only the PID row counts. + xres = _xres(entries=[(1 << 0, 4, 111), (1 << 1, 4, 222)]) + assert focused_pid(_conn(self._focus_without_wm_pid(), xres)) == 222 + + def test_a_short_entry_is_skipped(self) -> None: + xres = _xres(entries=[(1 << 1, 0, 111), (1 << 1, 4, 333)]) + assert focused_pid(_conn(self._focus_without_wm_pid(), xres)) == 333 + + def test_a_null_value_is_skipped(self) -> None: + xres = _xres(entries=[(1 << 1, 4, None), (1 << 1, 4, 444)]) + assert focused_pid(_conn(self._focus_without_wm_pid(), xres)) == 444 + + def test_the_result_is_destroyed(self) -> None: + xres = _xres(pids=[4242]) + focused_pid(_conn(self._focus_without_wm_pid(), xres)) + xres.XResClientIdsDestroy.assert_called_once() + + +def _sequence(*writers: Callable[..., int]) -> Callable[..., int]: + """Chain property writers so successive calls use successive fakes.""" + calls = iter(writers) + + def impl(*args: object) -> int: + return next(calls)(*args) + + return impl diff --git a/steam_backlog_enforcer/tests/test_x_probe.py b/steam_backlog_enforcer/tests/test_x_probe.py new file mode 100644 index 0000000..a0a797c --- /dev/null +++ b/steam_backlog_enforcer/tests/test_x_probe.py @@ -0,0 +1,120 @@ +"""Tests for the cached X connection and the input-idle query. + +Every case runs against fake library handles, so the suite never needs a +display and never depends on whether the developer happens to be at the +keyboard. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from steam_backlog_enforcer._x_probe import ( + XProbe, + XProbeError, + _ignore_x_error, + load_libraries, + root_window, +) +from steam_backlog_enforcer.tests._fake_xlib import fake_x11, fake_xss + +XP = "steam_backlog_enforcer._x_probe" + + +class TestLoadLibraries: + def test_returns_all_three_when_present(self) -> None: + x11, xss, xres = MagicMock(), MagicMock(), MagicMock() + with patch(f"{XP}.ctypes.CDLL", side_effect=[x11, xss, xres]): + loaded = load_libraries() + assert loaded[0] is x11 + assert loaded[1] is xss + assert loaded[2] is xres + + def test_missing_xres_is_tolerated(self) -> None: + x11, xss = MagicMock(), MagicMock() + with patch(f"{XP}.ctypes.CDLL", side_effect=[x11, xss, OSError("no")]): + loaded = load_libraries() + assert loaded[0] is x11 + assert loaded[2] is None + + def test_missing_core_libraries_raise(self) -> None: + with ( + patch(f"{XP}.ctypes.CDLL", side_effect=OSError("no libX11")), + pytest.raises(XProbeError), + ): + load_libraries() + + +class TestConnect: + def test_opens_once_and_caches(self) -> None: + x11 = fake_x11() + probe = XProbe(libraries=(x11, fake_xss(), None)) + first = probe.connect() + assert probe.connect() is first + assert x11.XOpenDisplay.call_count == 1 + + def test_loads_libraries_when_none_were_injected(self) -> None: + libs = (fake_x11(), fake_xss(), None) + with patch(f"{XP}.load_libraries", return_value=libs) as loader: + XProbe().connect() + loader.assert_called_once_with() + + def test_a_null_display_raises(self) -> None: + probe = XProbe(libraries=(fake_x11(display=0), fake_xss(), None)) + with pytest.raises(XProbeError, match="XOpenDisplay"): + probe.connect() + + def test_an_error_handler_is_installed(self) -> None: + # Xlib's default handler calls exit(1); a window vanishing mid-query + # must not be able to take the daemon down. + x11 = fake_x11() + XProbe(libraries=(x11, fake_xss(), None)).connect() + x11.XSetErrorHandler.assert_called_once() + + +class TestClose: + def test_closing_an_open_connection_releases_it(self) -> None: + x11 = fake_x11() + probe = XProbe(libraries=(x11, fake_xss(), None)) + probe.connect() + probe.close() + x11.XCloseDisplay.assert_called_once() + probe.connect() + assert x11.XOpenDisplay.call_count == 2 + + def test_closing_an_unopened_connection_is_a_no_op(self) -> None: + x11 = fake_x11() + XProbe(libraries=(x11, fake_xss(), None)).close() + x11.XCloseDisplay.assert_not_called() + + +class TestIdleSeconds: + def test_converts_milliseconds_to_seconds(self) -> None: + probe = XProbe(libraries=(fake_x11(), fake_xss(idle_ms=4500), None)) + assert probe.idle_seconds() == 4.5 + + def test_a_failed_allocation_raises(self) -> None: + probe = XProbe(libraries=(fake_x11(), fake_xss(allocated=False), None)) + with pytest.raises(XProbeError, match="AllocInfo"): + probe.idle_seconds() + + def test_a_failed_query_raises_and_still_frees(self) -> None: + x11 = fake_x11() + probe = XProbe(libraries=(x11, fake_xss(ok=False), None)) + with pytest.raises(XProbeError, match="QueryInfo"): + probe.idle_seconds() + x11.XFree.assert_called_once() + + +class TestErrorHandler: + def test_swallows_the_error_and_reports_success(self) -> None: + assert _ignore_x_error(None, None) == 0 + + +class TestRootWindow: + def test_returns_the_default_root(self) -> None: + conn = MagicMock() + conn.x11.XDefaultRootWindow.return_value = 0x99 + assert root_window(conn) == 0x99