Add a daily gaming-time budget with a binary-level cutoff

Accumulates time spent in Steam games and other launchers, and once the
daily budget (default 8h) is spent, shuts Steam down, kills the games, and
makes them unlaunchable until the next 06:00.

The gaming day starts at 06:00 local rather than midnight so a late session
counts against the day it began in; rollover is checked lazily against a
stored day key, so it survives suspend, reboot and daemon restarts with no
timer of its own. Elapsed time comes from wall-clock deltas clamped to two
tick intervals, which keeps suspends, missed ticks and NTP steps honest.

The block is a read-only bind mount of a refusal stub over all four launcher
entry points -- /usr/bin/steam is only a two-line exec wrapper, so masking it
alone leaves bin_steam.sh, Steam's own steam.sh and lutris live. A bind mount
mutates no bytes, umount restores exactly, and a reboot clears it
unconditionally; chmod/rename would leave Steam broken if the daemon died
mid-block. /proc/self/mountinfo is the authority on what is masked, never the
saved state, which is what makes a stale block release even if the state file
was deleted. The mount is deferred one tick past the cutoff so
`steam -shutdown` still works and games close cleanly.

Launchers are matched by cmdline, not comm: /usr/bin/lutris is a python3
script, so the kernel records its comm as "python3" and the existing scan
never sees it. The kill set is wider than the budget predicate, walking down
from each launcher, because a Lutris-launched Wine game carries no SteamAppId.

State is chattr +i'd -- ~/.config is user-writable, so root ownership alone
would not stop `rm` from resetting the counter -- and a missing state file
while mounts are live fails closed rather than lifting the block.

Also fixes a pre-existing bug: do_enforce returned when no game was assigned,
exiting the process. Since the budget is accounted for from inside that loop,
it silently stopped enforcing whenever a game was finished but not rescanned.

Verified on the desktop with a real game: AoE2 counted 18 processes, tripped
at the 60s demo budget, all processes killed, all four paths masked and
refusing, then gaming-unblock restored them byte-identical with
`pacman -Qkk steam lutris` reporting 0 altered files.

1322 tests pass at 100.00% branch coverage; pacman validated the ALPM hook.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbEaXapjMSkxyzgvbdkUFb
This commit is contained in:
Krzysztof Rudnicki 2026-08-04 20:00:08 +02:00
parent bfdd69453e
commit 0c05d71008
21 changed files with 3477 additions and 32 deletions

View File

@ -53,6 +53,16 @@ if [[ "${ans,,}" == "y" ]]; then
s|__DESKTOP_USER__|$DESKTOP_USER|" \
"$SERVICE_SRC" > "$SERVICE_DST"
# Daily-gaming-budget pacman hook. While the budget is exhausted a refusal
# stub is bind-mounted over /usr/bin/steam and friends; a bind mount makes
# package extraction fail EBUSY, so the mounts must be dropped before any
# transaction touching those packages.
HOOK_SRC="$SCRIPT_DIR/pacman-hooks/50-steam-backlog-playtime-unblock.hook"
HOOK_DST="/etc/pacman.d/hooks/50-steam-backlog-playtime-unblock.hook"
mkdir -p /etc/pacman.d/hooks
install -m 644 "$HOOK_SRC" "$HOOK_DST"
echo "Installed pacman hook: $HOOK_DST"
systemctl daemon-reload
systemctl enable steam-backlog-enforcer
echo "Service installed and enabled."

View File

@ -0,0 +1,29 @@
# steam-backlog-enforcer: release daily-gaming-budget bind mounts before pacman
# touches the packages they mask.
#
# While the daily gaming budget is exhausted, a refusal stub is bind-mounted
# over the launcher entry points. A bind-mounted file is a mountpoint, so
# unlink(2)/rename(2) over it returns EBUSY and package extraction fails. This
# hook clears the mounts for the duration of the transaction; the enforcer's
# 3-second tick re-applies them afterwards, so there is no PostTransaction half.
#
# The daemon side also refuses to (re-)mount while /var/lib/pacman/db.lck
# exists, which is what stops a tick from remounting mid-transaction.
#
# KEEP THE TARGET LIST IN SYNC with BLOCK_TARGETS in
# steam_backlog_enforcer/_playtime_block.py. The user-owned
# ~/.local/share/Steam/steam.sh is deliberately absent here: it is not owned by
# any package, so no pacman transaction can collide with it.
[Trigger]
Operation = Install
Operation = Upgrade
Operation = Remove
Type = Package
Target = steam
Target = lutris
[Action]
Description = steam-backlog-enforcer: releasing playtime bind mounts
When = PreTransaction
Exec = /bin/sh -c 'for t in /usr/bin/steam /usr/lib/steam/bin_steam.sh /usr/bin/lutris; do i=0; while mountpoint -q "$t" && [ "$i" -lt 20 ]; do umount -l "$t" || break; i=$((i+1)); done; done'

View File

@ -16,6 +16,15 @@ Environment=STEAM_ENFORCER_DESKTOP_USER=__DESKTOP_USER__
Environment=DISPLAY=:0
# Hardening: enforcer must not be easily killed.
OOMScoreAdjust=-900
#
# DO NOT ADD PrivateMounts=, PrivateTmp=, ProtectSystem= OR ProtectHome= HERE.
# The daily gaming budget enforces its cutoff by bind-mounting a refusal stub
# over /usr/bin/steam and the other launcher entry points. Any of those
# directives gives this service a private mount namespace, so those mounts stay
# invisible to the rest of the system: the launchers keep working while the
# counter keeps running, and the block silently becomes a no-op. That failure
# mode is invisible from the outside, which is why it is spelled out here.
# _playtime_block.mounts_are_visible() logs an error if it ever happens.
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,150 @@
"""CLI handlers for the daily gaming budget.
Every ``_echo`` / ``input`` / ``sys.exit`` for this feature lives here rather
than in :mod:`steam_backlog_enforcer._playtime` or
:mod:`steam_backlog_enforcer._playtime_block`, so the MCP server can import
those modules' leaf helpers without any risk of writing to stdout — which is
the JSON-RPC channel.
"""
from __future__ import annotations
from datetime import datetime, timezone
import os
from typing import TYPE_CHECKING
from steam_backlog_enforcer._enforce_loop import do_enforce
from steam_backlog_enforcer._playtime import (
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,
state_path,
)
from steam_backlog_enforcer._playtime_block import (
block_targets,
mounted_targets,
release_block,
)
from steam_backlog_enforcer.game_install import _echo
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config, State
_DEMO_FLAG = "--demo"
def cmd_enforce(config: Config, state: State, args: list[str]) -> int:
"""Run the enforcement loop.
Args:
config: Enforcer configuration.
state: Current enforcer state.
args: Arguments after the command name.
Returns:
Process exit code.
"""
unknown = [arg for arg in args if arg != _DEMO_FLAG]
if unknown:
# Never ignore an unrecognised flag here: a mistyped "--Demo" would
# silently run the real 8-hour budget during what was meant to be a
# 60-second demo, and the mistake only shows up hours later.
_echo(f"Unknown argument(s): {' '.join(unknown)}")
_echo(f"Usage: enforce [{_DEMO_FLAG}]")
return 2
demo = _DEMO_FLAG in args
if demo:
_echo("DEMO MODE: gaming budget is 60 seconds, using a separate state file.")
do_enforce(config, state, demo=demo)
return 0
def cmd_gaming_status(config: Config, _state: State) -> None:
"""Print today's gaming-budget usage and block state.
Args:
config: Enforcer configuration.
_state: Unused; required by the command dispatch signature.
"""
for demo in (False, True):
rules = rules_for(config, demo=demo)
stored = load_state(demo=demo)
if stored is None and demo:
continue
_echo(f"{'Demo' if demo else 'Gaming'} budget:")
_echo(f" state file: {state_path(demo=demo)}")
_echo(f" budget_seconds: {rules.budget_seconds:.0f}")
if stored is None:
_echo(" (no state recorded yet)")
else:
remaining = max(0.0, rules.budget_seconds - stored.seconds)
_echo(f" gaming day: {stored.day_key} (starts 06:00 local)")
_echo(f" used: {stored.seconds:.0f}s")
_echo(f" remaining: {remaining:.0f}s")
_echo(f" blocked: {stored.is_blocked()}")
masked = mounted_targets()
_echo(f"Masked launchers: {len(masked)}/{len(block_targets())}")
for target in sorted(masked):
_echo(f" {target}")
def cmd_gaming_reset(config: Config, _state: State) -> int:
"""Zero today's gaming counter and lift the block.
Args:
config: Enforcer configuration.
_state: Unused; required by the command dispatch signature.
Returns:
Process exit code.
"""
if os.geteuid() != 0:
_echo("gaming-reset requires root (it releases bind mounts).")
return 1
_echo("This resets today's gaming counter and lifts the block.")
if input("Type YES to confirm: ").strip() != "YES":
_echo("Aborted.")
return 1
released = release_block()
now = datetime.now(timezone.utc).astimezone()
save_state(
PlaytimeState(day_key=gaming_day_key(now), last_tick_at=now.timestamp()),
demo=rules_for(config, demo=False).demo,
)
_echo(f"Gaming counter reset. Released {len(released)} mount(s).")
return 0
def cmd_gaming_unblock(args: list[str]) -> int:
"""Release every playtime mount unconditionally.
The recovery hatch: it consults no state at all, so it works when the state
file is unreadable, the daemon is down, or a block was left behind by a
previous run.
Args:
args: Arguments after the command name.
Returns:
Process exit code.
"""
if args:
_echo(f"Unknown argument(s): {' '.join(args)}")
return 2
if os.geteuid() != 0:
_echo("gaming-unblock requires root (it releases bind mounts).")
return 1
released = release_block()
if not released:
_echo("No playtime mounts were active.")
return 0
for target in released:
_echo(f"Released {target}")
return 0

View File

@ -8,6 +8,7 @@ import time
from typing import Any
from steam_backlog_enforcer._actions import allowed_app_ids, allowed_games
from steam_backlog_enforcer._playtime import playtime_tick
from steam_backlog_enforcer._total_block import (
end_total_block_cleanup,
enforce_total_block_tick,
@ -283,13 +284,25 @@ def _enforce_hide_games(config: Config, state: State) -> None:
_echo(" Library: games already hidden")
def _enforce_loop_iteration(config: Config, state: State) -> None:
def _enforce_loop_iteration(
config: Config,
state: State,
*,
demo: bool = False,
) -> None:
"""Perform one iteration of the enforcement loop.
Args:
config: Enforcer configuration.
state: Current enforcer 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)
# Total block takes priority over the assigned-game enforcement below -
# while active, don't fight ourselves (e.g. installing the assigned
# game while total-block tries to keep Steam uninstalled).
@ -336,7 +349,7 @@ def _enforce_loop_iteration(config: Config, state: State) -> None:
lock_enforcement_files(CONFIG_FILE)
def do_enforce(config: Config, state: State) -> None:
def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
"""Run the enforcer: block store, uninstall other games, kill processes.
This is a persistent loop that continuously:
@ -344,14 +357,24 @@ def do_enforce(config: Config, state: State) -> None:
2. Removes any newly-installed unauthorized games.
3. Auto-installs the assigned game if missing.
4. Kills any running unauthorized game processes.
5. Accounts for the daily gaming budget and enforces its cutoff.
Args:
config: Enforcer configuration.
state: Current enforcer state.
demo: Run the gaming budget on a 60-second demo budget.
"""
if is_total_block_active():
_echo(
"Total gaming block ACTIVE - enforcing that instead of any assigned game."
)
elif state.current_app_id is None:
# Fall through to the idle loop for the same reason as the Steam
# branch below, plus one of its own: the daily gaming budget is
# enforced from this loop, so returning here would silently stop
# counting playtime whenever a game is finished but not yet rescanned.
_echo("No game assigned. Run 'scan' first.")
return
_echo(" (Daily gaming budget is still enforced.)")
elif not steam_is_installed():
# Fall through to the idle loop rather than returning: returning exits
# the process, and under Restart=always that is just the crash loop
@ -381,7 +404,7 @@ def do_enforce(config: Config, state: State) -> None:
state.current_game_name = fresh.current_game_name
state.finished_app_ids = fresh.finished_app_ids
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, demo=demo)
time.sleep(ENFORCE_INTERVAL)
except KeyboardInterrupt:
_echo("\nEnforcer stopped.")

View File

@ -25,7 +25,9 @@ Safety invariants (do not break when adding tools):
from __future__ import annotations
from datetime import datetime, timezone
import logging
import os
import sys
from typing import Any
@ -43,6 +45,14 @@ from steam_backlog_enforcer._actions import (
status_payload,
)
from steam_backlog_enforcer._allowed_games import MANUAL_LOCK_DAYS
from steam_backlog_enforcer._playtime import (
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,
)
from steam_backlog_enforcer._playtime_block import mounted_targets, release_block
from steam_backlog_enforcer._total_block import start_total_block
from steam_backlog_enforcer._web_dataset import build_web_dataset, dataset_to_payload
from steam_backlog_enforcer.config import Config, State, load_snapshot
@ -330,6 +340,89 @@ def block_gaming(days: int, *, confirm: bool = False) -> dict[str, Any]:
return {"ok": True, "applied": True, "action": "block_gaming", "days": days}
@mcp.tool()
def get_gaming_time() -> dict[str, Any]:
"""Return today's gaming-budget usage and whether the cutoff has engaged.
The gaming day starts at 06:00 local time, so a session played after
midnight still counts against the day it began in. Reads on-disk state and
the live mount table only — no network, no Steam API key, no config secrets.
"""
config = Config.load()
rules = rules_for(config, demo=False)
stored = load_state(demo=False)
masked = sorted(str(path) for path in mounted_targets())
if stored is None:
return {
"ok": True,
"recorded": False,
"budget_seconds": rules.budget_seconds,
"enforcement": rules.enforcement,
"masked_launchers": masked,
}
return {
"ok": True,
"recorded": True,
"gaming_day": stored.day_key,
"day_starts_at": "06:00 local",
"seconds_used": round(stored.seconds, 1),
"budget_seconds": rules.budget_seconds,
"seconds_remaining": round(max(0.0, rules.budget_seconds - stored.seconds), 1),
"blocked": stored.is_blocked(),
"enforcement": rules.enforcement,
"counts_launchers": rules.count_launchers,
"masked_launchers": masked,
}
@mcp.tool()
def reset_gaming_time(*, confirm: bool = False) -> dict[str, Any]:
"""Reset today's gaming counter and lift the block (gated, privileged write).
With ``confirm=False`` (the default) returns a preview and mutates nothing.
With ``confirm=True`` it attempts the reset, which **requires root** (it
releases bind mounts over ``/usr/bin/steam`` and friends). The MCP server
runs unprivileged, so this will normally return ``{"ok": false, "reason":
"requires elevated privileges"}`` rather than succeeding.
Args:
confirm: Set ``True`` to attempt the reset; otherwise preview.
"""
if not confirm:
return {
"ok": True,
"preview": True,
"action": "reset_gaming_time",
"effect": (
"Zeroes today's gaming counter and unmasks every launcher "
"binary, restoring a full daily budget immediately."
),
"requires_root": True,
"confirm_required": True,
}
if os.geteuid() != 0:
return {"ok": False, "reason": "requires elevated privileges"}
try:
released = release_block()
now = datetime.now(timezone.utc).astimezone()
save_state(
PlaytimeState(day_key=gaming_day_key(now), last_tick_at=now.timestamp()),
demo=False,
)
except (OSError, ValueError) as exc: # never crash the server on a write
logger.warning("reset_gaming_time failed: %s", exc)
return {"ok": False, "reason": "requires elevated privileges"}
logger.info("reset_gaming_time applied; released %d mount(s)", len(released))
return {
"ok": True,
"applied": True,
"action": "reset_gaming_time",
"released": [str(path) for path in released],
}
def main() -> None:
"""Run the MCP server over stdio (STDOUT = JSON-RPC, STDERR = logs)."""
logger.info(

View File

@ -0,0 +1,617 @@
"""Daily gaming-time budget: accounting, day rollover and cutoff policy.
Tracks how long the user has spent gaming during the current *gaming day* and,
once the budget is exhausted, drives the cutoff sequence implemented in
:mod:`steam_backlog_enforcer._playtime_block`.
Three design points that are load-bearing:
**The gaming day starts at 06:00 local, not midnight.** A session that runs
past midnight belongs to the day it started in; a midnight boundary would hand
out a fresh 8-hour budget in the middle of a late session. Rollover is checked
lazily on every tick against a stored day key rather than scheduled, so it
survives suspend, reboot and daemon restarts with no timer of its own.
**Elapsed time is measured from wall-clock deltas, clamped.** Counting "one
tick = one interval" silently under-counts whenever the loop is slow and
over-counts nothing; measuring the raw delta over-counts wildly across a
suspend. Clamping each delta to a small multiple of the tick interval makes
suspends, missed ticks, daemon restarts and NTP steps all behave.
**Launchers are matched by cmdline, not by ``comm``.** ``/usr/bin/lutris`` is a
``#!/usr/bin/env python3`` script, so the kernel records its ``comm`` as
``python3`` and the ``comm``-based scan in :mod:`enforcer` never sees it. That
is why :func:`get_pids_by_cmdline_names` exists alongside
``enforcer.get_pids_by_process_names`` rather than replacing it.
"""
from __future__ import annotations
from dataclasses import dataclass, field, replace
from datetime import datetime, timedelta, timezone
import json
import logging
import os
from pathlib import Path
from typing import TYPE_CHECKING
from steam_backlog_enforcer._playtime_block import (
kill_gaming_processes,
mounted_targets,
reconcile,
request_steam_shutdown,
steam_and_launcher_pids,
)
from steam_backlog_enforcer._total_block import (
LAUNCHER_PROCESS_NAMES,
is_total_block_active,
)
from steam_backlog_enforcer._whitelist import _try_set_immutable, unlock_for_write
from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
from steam_backlog_enforcer.enforcer import (
get_pids_by_process_names,
get_running_steam_game_pids,
)
from steam_backlog_enforcer.library_hider import _resolve_desktop_user, _run_as_user
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config
logger = logging.getLogger(__name__)
PLAYTIME_STATE_FILE = CONFIG_DIR / "playtime_state.json"
PLAYTIME_DEMO_STATE_FILE = CONFIG_DIR / "playtime_demo_state.json"
_PROC = Path("/proc")
_SCHEMA_VERSION = 1
# The gaming day starts here, local time. 05:59 belongs to the previous day.
_DAY_BOUNDARY_HOURS = 6
# Each tick's measured delta is capped at this many tick intervals. Two is the
# smallest value that tolerates one skipped tick without inflating the count.
_DELTA_CLAMP_FACTOR = 2.0
_DEFAULT_BUDGET_SECONDS = 8 * 3600.0
_DEMO_BUDGET_SECONDS = 60.0
# Seconds remaining at which to warn, descending.
_WARN_AT: tuple[int, ...] = (3600, 1800, 600, 300)
_DEMO_WARN_AT: tuple[int, ...] = (30, 10)
# Grace between "budget exhausted" and masking the Steam binaries. Masking
# makes `steam -shutdown` a no-op, so Steam must be given a moment to close
# its games cleanly first — otherwise every single cutoff is a hard SIGTERM.
_SHUTDOWN_GRACE_SECONDS = 3.0
_SIGKILL_AFTER_SECONDS = 30.0
_DEMO_SIGKILL_AFTER_SECONDS = 10.0
# argv[0] basenames that tell us nothing: the real program is argv[1].
_INTERPRETERS = frozenset({"bash", "env", "java", "perl", "python", "python3", "sh"})
# An interpreter invocation needs at least `<interp> <script>` to name a program.
_MIN_ARGV_FOR_INTERPRETER = 2
_SECONDS_PER_MINUTE = 60
_MINUTES_PER_HOUR = 60
@dataclass
class PlaytimeState:
"""Per-gaming-day playtime accounting, persisted between ticks."""
schema_version: int = _SCHEMA_VERSION
day_key: str = ""
"""Gaming day this record covers, ``YYYY-MM-DD``, 06:00-shifted."""
seconds: float = 0.0
"""Qualifying seconds accrued during ``day_key``."""
last_tick_at: float = 0.0
"""Epoch seconds of the last accounting tick; ``0.0`` means never ticked."""
blocked_at: float = 0.0
"""Epoch seconds when the cutoff engaged; ``0.0`` means not blocked."""
warned_seconds: list[int] = field(default_factory=list)
"""Warning thresholds already fired during ``day_key``."""
def is_blocked(self) -> bool:
"""Whether the cutoff has engaged for this gaming day.
Deliberately a method rather than a field: ``save`` serialises
``self.__dict__``, so a derived field would be persisted and could
drift out of step with ``blocked_at``.
"""
return self.blocked_at > 0.0
@dataclass(frozen=True)
class PlaytimeRules:
"""Resolved budget policy for one tick.
Frozen and passed explicitly so that every decision function stays pure
and testable without touching ``Config`` or the clock.
"""
budget_seconds: float
warn_at: tuple[int, ...]
sigkill_after: float
count_launchers: bool
enforcement: bool
demo: bool
def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
"""Build the policy for this tick from *config*.
Demo mode changes the budget and the warning thresholds and nothing else —
in particular it uses the same qualifying-process predicate as production,
because a demo that exercises different logic proves nothing about it.
Args:
config: Loaded user configuration.
demo: Whether this is a short-budget demo run.
Returns:
The rules governing this tick.
"""
return PlaytimeRules(
budget_seconds=_DEMO_BUDGET_SECONDS
if demo
else float(config.daily_gaming_seconds),
warn_at=_DEMO_WARN_AT if demo else _WARN_AT,
sigkill_after=(_DEMO_SIGKILL_AFTER_SECONDS if demo else _SIGKILL_AFTER_SECONDS),
count_launchers=config.count_launcher_processes,
enforcement=config.playtime_enforcement,
demo=demo,
)
def gaming_day_key(now: datetime) -> str:
"""Return the ``YYYY-MM-DD`` gaming day containing *now*.
The boundary is 06:00 local, so 02:00 on the 5th belongs to the 4th.
Args:
now: Timezone-aware local timestamp.
Returns:
The gaming day key.
"""
return (now - timedelta(hours=_DAY_BOUNDARY_HOURS)).date().isoformat()
def state_path(*, demo: bool) -> Path:
"""Return the state file for a demo or production run.
Demo runs use a separate file so a demo can never consume or clobber the
real day's counter.
Args:
demo: Whether this is a demo run.
Returns:
Path to the state file.
"""
return PLAYTIME_DEMO_STATE_FILE if demo else PLAYTIME_STATE_FILE
def load_state(*, demo: bool) -> PlaytimeState | None:
"""Load persisted state, or ``None`` if it is absent or unusable.
``None`` deliberately covers "missing", "corrupt" and "written by a
different schema" alike: the caller's recovery path is the same for all
three, and treating an unreadable file as a fresh day is exactly the
reset-by-deletion the immutable flag exists to prevent.
Args:
demo: Whether this is a demo run.
Returns:
The stored state, or ``None`` if it could not be read.
"""
path = state_path(demo=demo)
if not path.exists():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
logger.warning("Playtime state at %s is unreadable; assuming tampered.", path)
return None
if not isinstance(data, dict) or data.get("schema_version") != _SCHEMA_VERSION:
logger.warning("Playtime state at %s has an unknown schema.", path)
return None
fields = PlaytimeState.__dataclass_fields__
return PlaytimeState(**{k: v for k, v in data.items() if k in fields})
def save_state(state: PlaytimeState, *, demo: bool) -> None:
"""Persist *state*, re-applying the immutable flag for production runs.
``_atomic_write`` finishes with ``rename(2)``, which fails ``EPERM`` against
an immutable destination — so clearing the flag first is mandatory, not
decorative. Demo state is left mutable: it carries no enforcement weight and
an immutable copy could not be deleted during cleanup, even by root.
Args:
state: State to write.
demo: Whether this is a demo run.
"""
path = state_path(demo=demo)
if not demo:
unlock_for_write(path)
_atomic_write(path, json.dumps(state.__dict__, indent=2) + "\n")
if not demo:
_try_set_immutable(path, immutable=True)
def get_pids_by_cmdline_names(names: frozenset[str]) -> dict[int, str]:
"""Scan ``/proc/*/cmdline`` for processes whose program name is in *names*.
Complements ``enforcer.get_pids_by_process_names``, which matches on
``comm`` and therefore cannot see interpreter-launched programs: the kernel
records ``/usr/bin/lutris`` as ``python3``. When ``argv[0]``'s basename is a
known interpreter this falls through to ``argv[1]``.
The daemon itself runs under ``python3``, so its own PID is skipped.
Args:
names: Program basenames to match.
Returns:
Mapping of PID to the matched name.
"""
own_pid = os.getpid()
found: dict[int, str] = {}
for entry in _PROC.iterdir():
if not entry.name.isdigit():
continue
pid = int(entry.name)
if pid == own_pid:
continue
matched = _match_cmdline(entry, names)
if matched is not None:
found[pid] = matched
return found
def _match_cmdline(entry: Path, names: frozenset[str]) -> str | None:
"""Return the name in *names* that *entry*'s cmdline runs, if any.
Args:
entry: A ``/proc/<pid>`` directory.
names: Program basenames to match.
Returns:
The matched name, or ``None``.
"""
try:
raw = (entry / "cmdline").read_bytes()
except (OSError, ValueError):
return None
argv = [
part for part in raw.decode("utf-8", errors="replace").split("\x00") if part
]
if not argv:
return None
first = Path(argv[0]).name
if first in names:
return first
if first not in _INTERPRETERS or len(argv) < _MIN_ARGV_FOR_INTERPRETER:
return None
second = Path(argv[1]).name
return second if second in names else None
def qualifying_pids(rules: PlaytimeRules) -> set[int]:
"""Return PIDs whose runtime counts against the daily budget.
Steam games are identified by the ``SteamAppId`` environment variable, using
the same ``!= 0`` predicate ``enforcer.enforce_allowed_game`` uses to exclude
the Steam client tree — browsing the store is not gaming.
Args:
rules: Policy for this tick.
Returns:
The set of qualifying PIDs.
"""
pids = {pid for pid, app_id in get_running_steam_game_pids().items() if app_id != 0}
if rules.count_launchers:
pids |= set(get_pids_by_process_names(LAUNCHER_PROCESS_NAMES))
pids |= set(get_pids_by_cmdline_names(LAUNCHER_PROCESS_NAMES))
return pids
def accumulate(
state: PlaytimeState,
*,
now: datetime,
qualifying: set[int],
interval: float,
) -> PlaytimeState:
"""Add this tick's elapsed gaming time to *state*.
Pure. Every argument is mandatory, including the clock — a defaulted ``now``
would add a branch that exists only to be covered.
The delta is clamped to ``_DELTA_CLAMP_FACTOR`` intervals so that a suspend,
a skipped tick or a forward NTP step cannot inflate the count, and floored
at zero so a backward step cannot refund time.
Args:
state: Current accounting state.
now: Timezone-aware local timestamp for this tick.
qualifying: PIDs currently counting against the budget.
interval: Nominal seconds between ticks.
Returns:
The updated state.
"""
stamp = now.timestamp()
if state.last_tick_at <= 0.0:
return replace(state, last_tick_at=stamp)
delta = stamp - state.last_tick_at
if delta <= 0.0 or not qualifying:
return replace(state, last_tick_at=stamp)
credited = min(delta, interval * _DELTA_CLAMP_FACTOR)
return replace(state, last_tick_at=stamp, seconds=state.seconds + credited)
def roll_over(state: PlaytimeState, *, day_key: str) -> PlaytimeState:
"""Reset *state* if the gaming day has changed.
Pure. ``last_tick_at`` carries across the boundary so the first tick of the
new day measures a sane delta rather than restarting from zero.
Args:
state: Current accounting state.
day_key: The gaming day *now* falls in.
Returns:
Either *state* unchanged, or a fresh record for *day_key*.
"""
if state.day_key == day_key:
return state
return PlaytimeState(day_key=day_key, last_tick_at=state.last_tick_at)
def pending_warning(state: PlaytimeState, rules: PlaytimeRules) -> int | None:
"""Return the largest un-fired warning threshold *state* has crossed.
Pure. Returning only the largest matters when a tick is skipped: crossing
two thresholds at once should warn once, not twice.
Args:
state: Current accounting state.
rules: Policy for this tick.
Returns:
The threshold in seconds-remaining, or ``None`` if none is due.
"""
remaining = rules.budget_seconds - state.seconds
for threshold in rules.warn_at:
if remaining <= threshold and threshold not in state.warned_seconds:
return threshold
return None
def notify_desktop_user(title: str, body: str) -> None:
"""Send a desktop notification into the real user's session.
``enforcer.send_notification`` runs ``notify-send`` as root with no
``DBUS_SESSION_BUS_ADDRESS``, which cannot reach the user's session bus from
a system service. This routes through the same ``sudo -u <user> env ...``
mechanism that ``library_hider`` uses to launch Steam.
Args:
title: Notification title.
body: Notification body.
"""
user = _resolve_desktop_user()
try:
_run_as_user(["notify-send", title, body, "--icon=dialog-warning"], user)
except (OSError, ValueError):
logger.debug("Could not send desktop notification.")
def playtime_tick(config: Config, *, interval: float, 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.
Args:
config: Loaded user configuration.
interval: Nominal seconds between enforce-loop ticks.
demo: Whether this is a short-budget demo run.
"""
now = datetime.now(timezone.utc).astimezone()
rules = rules_for(config, demo=demo)
state = roll_over(_state_or_recover(rules, now=now), day_key=gaming_day_key(now))
if is_total_block_active():
# The total block runs `pacman -R steam` every tick; our bind mounts
# would make that fail EBUSY. It is strictly stronger — get out of its
# way, and stop accruing against a budget nobody can spend.
reconcile(should_block=False)
save_state(state, demo=demo)
return
state = accumulate(
state,
now=now,
qualifying=qualifying_pids(rules),
interval=interval,
)
save_state(_policy(state, rules, now=now), demo=demo)
def _state_or_recover(rules: PlaytimeRules, *, now: datetime) -> PlaytimeState:
"""Load state, synthesising a fail-closed record if it is unusable.
If the state file is gone but the binaries are still masked, the only safe
reading is that the file was deleted to lift the block. Synthesising an
exhausted budget keeps it in force; the normal 06:00 rollover then lifts it
with no special case.
Args:
rules: Policy for this tick.
now: Timezone-aware local timestamp.
Returns:
Loaded or synthesised state.
"""
loaded = load_state(demo=rules.demo)
if loaded is not None:
return loaded
stamp = now.timestamp()
fresh = PlaytimeState(day_key=gaming_day_key(now), last_tick_at=stamp)
if not mounted_targets():
return fresh
logger.warning("Playtime state missing while blocked; failing closed.")
return replace(fresh, seconds=rules.budget_seconds, blocked_at=stamp)
def _policy(
state: PlaytimeState,
rules: PlaytimeRules,
*,
now: datetime,
) -> PlaytimeState:
"""Apply warnings, the cutoff, and block upkeep to *state*.
Args:
state: Accounting state after this tick's accumulation.
rules: Policy for this tick.
now: Timezone-aware local timestamp.
Returns:
The updated state.
"""
if not rules.enforcement:
# A kill switch must still release: leaving live mounts behind would
# make "disabled" mean "permanently blocked".
reconcile(should_block=False)
return state
if state.seconds < rules.budget_seconds:
reconcile(should_block=False)
return _warn(state, rules)
if not state.is_blocked():
return _begin_cutoff(state, rules, now=now)
return _sustain_block(state, rules, now=now)
def _warn(state: PlaytimeState, rules: PlaytimeRules) -> PlaytimeState:
"""Fire the due warning, if any, and record that it fired.
Args:
state: Current accounting state.
rules: Policy for this tick.
Returns:
The updated state.
"""
threshold = pending_warning(state, rules)
if threshold is None:
return state
notify_desktop_user(
"Gaming time running out",
f"{_humanise(threshold)} of today's gaming budget left.",
)
return replace(state, warned_seconds=[*state.warned_seconds, threshold])
def _begin_cutoff(
state: PlaytimeState,
rules: PlaytimeRules,
*,
now: datetime,
) -> PlaytimeState:
"""Start the cutoff: ask Steam to close cleanly, then SIGTERM the games.
No mount happens on this tick. Masking the Steam binaries turns
``steam -shutdown`` into a no-op, which would cost a clean close — and
therefore a cloud-save flush — on every single cutoff.
Args:
state: Current accounting state.
rules: Policy for this tick.
now: Timezone-aware local timestamp.
Returns:
The updated state.
"""
logger.warning("Daily gaming budget exhausted; shutting Steam down.")
notify_desktop_user(
"Gaming time is up",
"Daily budget used up. Steam and games are shutting down. Unblocks at 06:00.",
)
request_steam_shutdown()
kill_gaming_processes(_kill_set(rules), force=False)
return replace(state, blocked_at=now.timestamp())
def _sustain_block(
state: PlaytimeState,
rules: PlaytimeRules,
*,
now: datetime,
) -> PlaytimeState:
"""Hold the block: mask the binaries once the grace has passed, keep killing.
Args:
state: Current accounting state.
rules: Policy for this tick.
now: Timezone-aware local timestamp.
Returns:
*state*, unchanged — upkeep mutates the system, not the record.
"""
elapsed = now.timestamp() - state.blocked_at
if elapsed >= _SHUTDOWN_GRACE_SECONDS:
reconcile(should_block=True)
kill_gaming_processes(_kill_set(rules), force=elapsed >= rules.sigkill_after)
return state
def _kill_set(rules: PlaytimeRules) -> set[int]:
"""Return every PID the cutoff should terminate.
Wider than :func:`qualifying_pids` on purpose. A Lutris-launched Wine game
has no ``SteamAppId`` and its ``comm`` is the game's own binary, so neither
matcher sees it — only walking down from the launcher does.
Args:
rules: Policy for this tick.
Returns:
PIDs to signal.
"""
return qualifying_pids(rules) | steam_and_launcher_pids()
def _humanise(seconds: int) -> str:
"""Render a warning threshold as a short human phrase.
Args:
seconds: Seconds remaining.
Returns:
A phrase such as ``"30 minutes"`` or ``"10 seconds"``.
"""
if seconds < _SECONDS_PER_MINUTE:
return f"{seconds} seconds"
minutes = seconds // _SECONDS_PER_MINUTE
return "1 hour" if minutes == _MINUTES_PER_HOUR else f"{minutes} minutes"

View File

@ -0,0 +1,462 @@
"""Make Steam and game launchers unlaunchable once the daily budget is spent.
The block is a **read-only bind mount** of a refusal stub over every launcher
entry point. Nothing else here is negotiable, and future changes must not
substitute an alternative:
* ``chmod 000`` and rename-aside both mutate persistent on-disk state that
outlives this process. A crash mid-block leaves Steam unexecutable with no
automatic recovery.
* ``guardctl file-guard`` is an anti-*drift* keeper, not a swapper: it snapshots
the target as canonical and copies canonical back over the target. Pointing it
at a stub would permanently overwrite ``/usr/bin/steam``.
A bind mount mutates zero bytes of the real binary, ``umount`` is a perfect
restore, and a reboot clears every mount unconditionally. That last property is
the whole safety argument: the worst case is "reboot fixes it".
Four targets, because the first three are bypasses of one another —
``/usr/bin/steam`` is a two-line ``exec /usr/lib/steam/steam "$@"`` wrapper, and
Steam ships its own executable bootstrap inside the user's home directory.
``/proc/self/mountinfo`` is the authority on what is currently masked, never the
saved state. That is what makes the block self-healing: a stale mount is
released even if the state file was deleted, and a mount lost to a reboot is
re-applied on the next tick.
**The daemon must keep a shared mount namespace.** Adding ``PrivateMounts=``,
``PrivateTmp=`` or ``ProtectSystem=`` to the systemd unit would make these
mounts invisible outside the service, silently turning the block into a no-op
while the counter kept running. :func:`mounts_are_visible` checks for exactly
that at runtime.
"""
from __future__ import annotations
import contextlib
import logging
import os
from pathlib import Path
import shutil
import signal
import subprocess
from steam_backlog_enforcer._total_block import (
LAUNCHER_PROCESS_NAMES,
STEAM_CLIENT_PROCESS_NAMES,
)
from steam_backlog_enforcer.enforcer import get_pids_by_process_names
logger = logging.getLogger(__name__)
BLOCK_TARGETS: tuple[Path, ...] = (
# Every steam.desktop action uses this path; it is only a wrapper.
Path("/usr/bin/steam"),
# The real bootstrap. Named directly rather than via the
# /usr/lib/steam/steam symlink because mount(2) resolves symlinks, so
# mounting the link would mount here anyway under a misleading name.
Path("/usr/lib/steam/bin_steam.sh"),
# Steam's self-updating bootstrap, owned by the user and independently
# executable.
Path.home() / ".local" / "share" / "Steam" / "steam.sh",
Path("/usr/bin/lutris"),
)
# The stub lives on root-owned tmpfs, NOT in CONFIG_DIR. A bind mount shares the
# inode: a stub in a user-writable directory could be rewritten to exec the real
# bootstrap, and the block would evaporate without a single mount changing.
_STUB_DIR = Path("/run/steam-backlog-enforcer")
STUB_PATH = _STUB_DIR / "gaming-blocked"
MOUNTINFO_PATH = Path("/proc/self/mountinfo")
_INIT_MOUNTINFO_PATH = Path("/proc/1/mountinfo")
# pacman holds this for the whole transaction. Re-mounting mid-transaction is
# what would actually cause the EBUSY the PreTransaction hook exists to avoid.
PACMAN_LOCK = Path("/var/lib/pacman/db.lck")
_PROC = Path("/proc")
# Bind mounts stack, so one umount need not be enough. guard-lib's
# collapse_bind_mount() uses the same bounded loop for the same reason.
_MAX_UMOUNT_PASSES = 20
_MOUNTINFO_MIN_FIELDS = 5
_MOUNTPOINT_FIELD = 4
_STAT_PPID_INDEX = 1
_MAX_PROCESS_TREE_DEPTH = 32
_STUB_MODE = 0o755
_STUB_SCRIPT = """#!/bin/sh
# Installed by steam-backlog-enforcer. Bind-mounted over the real launcher
# while the daily gaming budget is exhausted; removed automatically at 06:00.
echo "Gaming blocked: daily budget used up. Unblocks at 06:00." >&2
exit 1
"""
def block_targets() -> tuple[Path, ...]:
"""Return the launcher paths the block masks.
An accessor rather than a direct import of ``BLOCK_TARGETS``: module-level
constants are bound by value at import time, so a second binding elsewhere
would need its own test patch to stay isolated. Routing every reader through
one function keeps that to a single patch point.
Returns:
The masked launcher paths.
"""
return BLOCK_TARGETS
def _run(cmd: list[str]) -> bool:
"""Run a privileged command, reporting success.
The single subprocess seam in this module, so tests can assert on exact
argv without ever executing ``mount``.
Args:
cmd: Argument vector to execute.
Returns:
True if the command exited zero.
"""
try:
result = subprocess.run(cmd, capture_output=True, check=False, timeout=15)
except (OSError, subprocess.SubprocessError):
logger.exception("Playtime block command failed: %s", cmd)
return False
if result.returncode != 0:
logger.warning("Playtime block command exited %d: %s", result.returncode, cmd)
return result.returncode == 0
def _unescape_mountinfo(field: str) -> str:
"""Decode the octal escapes the kernel writes into mountinfo paths.
Args:
field: A raw mountinfo field.
Returns:
The decoded field.
"""
return (
field.replace(r"\040", " ")
.replace(r"\011", "\t")
.replace(r"\012", "\n")
.replace(r"\134", "\\")
)
def _mountpoints(path: Path) -> set[Path]:
"""Parse *path* as mountinfo and return the mount points it lists.
Args:
path: A mountinfo file.
Returns:
Every mount point named in the file.
"""
try:
raw = path.read_text(encoding="utf-8")
except (OSError, ValueError):
logger.warning("Could not read mountinfo at %s.", path)
return set()
points: set[Path] = set()
for line in raw.splitlines():
fields = line.split(" ")
if len(fields) < _MOUNTINFO_MIN_FIELDS:
continue
points.add(Path(_unescape_mountinfo(fields[_MOUNTPOINT_FIELD])))
return points
def mounted_targets() -> set[Path]:
"""Return the block targets currently masked by a mount.
Read in-process from mountinfo rather than shelling out to ``findmnt``:
this runs on every enforce tick, and a fork per tick is exactly the polling
cost this codebase avoids elsewhere.
Anything mounted on a block target is treated as ours — nothing else on this
system mounts over those paths.
Returns:
The masked targets.
"""
return _mountpoints(MOUNTINFO_PATH) & set(block_targets())
def mounts_are_visible() -> bool:
"""Whether mounts made here are visible to the rest of the system.
A private mount namespace (``PrivateMounts=``, ``ProtectSystem=`` and
friends) would keep every mount inside the service, so the block would do
nothing while the counter kept running — a silent fail-open. Compared
against PID 1's namespace, which is the one that matters.
Returns:
True if no mount we made is missing from PID 1's view.
"""
ours = mounted_targets()
if not ours:
return True
return bool(ours & _mountpoints(_INIT_MOUNTINFO_PATH))
def _ensure_stub() -> bool:
"""Write the refusal stub, replacing any existing copy.
Returns:
True if the stub is present and executable.
"""
try:
_STUB_DIR.mkdir(parents=True, exist_ok=True)
STUB_PATH.write_text(_STUB_SCRIPT, encoding="utf-8")
STUB_PATH.chmod(_STUB_MODE)
except OSError:
logger.exception("Could not write the playtime refusal stub.")
return False
return True
def apply_block() -> list[Path]:
"""Mask every unmasked block target with the refusal stub.
Skipped entirely while pacman holds its database lock: re-mounting during a
transaction is what makes package extraction fail ``EBUSY``. The kill loop is
not gated this way, so enforcement continues regardless.
Returns:
The targets newly masked by this call.
"""
if PACMAN_LOCK.exists():
logger.info("pacman transaction in progress; deferring playtime mounts.")
return []
if not _ensure_stub():
return []
already = mounted_targets()
newly: list[Path] = []
for target in block_targets():
if target in already or not target.exists():
continue
if not _run([_mount_bin(), "--bind", str(STUB_PATH), str(target)]):
continue
if _run([_mount_bin(), "-o", "remount,ro,bind", str(target)]):
newly.append(target)
return newly
def release_block() -> list[Path]:
"""Unmask every masked block target.
Loops per target because bind mounts stack; bounded so a target that cannot
be unmounted cannot spin forever.
Returns:
The targets released by this call.
"""
released: list[Path] = []
for target in mounted_targets():
for _ in range(_MAX_UMOUNT_PASSES):
if target not in mounted_targets():
released.append(target)
break
if not _run([_umount_bin(), "-l", str(target)]):
logger.warning("Could not release playtime mount on %s.", target)
break
return released
def reconcile(*, should_block: bool) -> tuple[list[Path], list[Path]]:
"""Drive the on-disk mount state toward *should_block*.
Idempotent and driven by mountinfo, so it costs one file read per tick when
nothing needs changing. Because it reads the live mount table rather than
saved state, it releases a stale block even if the state file was deleted,
and re-applies a block that a reboot cleared.
Args:
should_block: Whether the launchers should currently be masked.
Returns:
A ``(newly_masked, newly_released)`` pair.
"""
if not should_block:
return [], release_block()
masked = apply_block()
if masked and not mounts_are_visible():
logger.error(
"Playtime mounts are not visible outside this service — the block "
"is not in force. Check the unit for PrivateMounts/ProtectSystem.",
)
return masked, []
def request_steam_shutdown() -> None:
"""Ask Steam to close cleanly, without waiting for it.
``library_hider._shutdown_steam`` polls for up to 30 seconds, which would
stall the 3-second enforce loop for ten ticks. The cutoff sequence gives
Steam its grace period across ticks instead.
"""
steam = shutil.which("steam")
if steam is None:
logger.info("Steam binary not on PATH; skipping clean shutdown.")
return
_run([steam, "-shutdown"])
def steam_and_launcher_pids() -> set[int]:
"""Return PIDs of the Steam client and every known game launcher.
Returns:
The launcher and client PIDs.
"""
names = STEAM_CLIENT_PROCESS_NAMES | LAUNCHER_PROCESS_NAMES
return set(get_pids_by_process_names(names))
def _read_ppid(pid: int) -> int | None:
"""Return the parent PID of *pid*, or ``None`` if it cannot be read.
``/proc/<pid>/stat`` embeds ``comm`` in parentheses and ``comm`` may contain
spaces and parentheses, so the fields after it are found by splitting on the
last ``") "`` rather than by whitespace position.
Args:
pid: Process to inspect.
Returns:
The parent PID, or ``None``.
"""
try:
raw = (_PROC / str(pid) / "stat").read_text(encoding="utf-8")
except (OSError, ValueError):
return None
_, _, rest = raw.rpartition(") ")
fields = rest.split()
if len(fields) <= _STAT_PPID_INDEX:
return None
try:
return int(fields[_STAT_PPID_INDEX])
except ValueError:
return None
def _child_map() -> dict[int, list[int]]:
"""Build a parent-PID to child-PIDs map from ``/proc``.
Returns:
Mapping of parent PID to its direct children.
"""
children: dict[int, list[int]] = {}
try:
entries = list(_PROC.iterdir())
except OSError:
return children
for entry in entries:
if not entry.name.isdigit():
continue
pid = int(entry.name)
ppid = _read_ppid(pid)
if ppid is not None:
children.setdefault(ppid, []).append(pid)
return children
def descendant_pids(roots: set[int]) -> set[int]:
"""Return every process descended from *roots*.
A Lutris-launched Wine game carries no ``SteamAppId`` and its ``comm`` is
the game's own binary, so neither of the budget matchers can see it. Walking
down from the launcher is the only thing that does.
The daemon's own process and its ancestors are excluded, so a misidentified
root cannot make the enforcer kill itself.
Args:
roots: Processes to walk down from.
Returns:
The transitive descendants, excluding *roots* themselves.
"""
children = _child_map()
protected = _own_process_chain()
found: set[int] = set()
frontier = [pid for pid in roots if pid not in protected]
for _ in range(_MAX_PROCESS_TREE_DEPTH):
if not frontier:
break
nxt: list[int] = []
for pid in frontier:
for child in children.get(pid, []):
if child in found or child in protected:
continue
found.add(child)
nxt.append(child)
frontier = nxt
return found
def _own_process_chain() -> set[int]:
"""Return this process and its ancestors.
Returns:
PIDs that must never be signalled.
"""
chain: set[int] = set()
pid: int | None = os.getpid()
for _ in range(_MAX_PROCESS_TREE_DEPTH):
if pid is None or pid <= 0 or pid in chain:
break
chain.add(pid)
pid = _read_ppid(pid)
return chain
def kill_gaming_processes(pids: set[int], *, force: bool) -> list[int]:
"""Signal every process in *pids*, plus everything descended from them.
Args:
pids: Processes to terminate.
force: Send ``SIGKILL`` instead of ``SIGTERM``.
Returns:
The PIDs actually signalled.
"""
protected = _own_process_chain()
targets = (pids | descendant_pids(pids)) - protected
sig = signal.SIGKILL if force else signal.SIGTERM
signalled: list[int] = []
for pid in sorted(targets):
with contextlib.suppress(ProcessLookupError, PermissionError):
os.kill(pid, sig)
signalled.append(pid)
return signalled
def _mount_bin() -> str:
"""Return the ``mount`` executable path.
Returns:
Absolute path to ``mount``.
"""
return shutil.which("mount") or "/usr/bin/mount"
def _umount_bin() -> str:
"""Return the ``umount`` executable path.
Returns:
Absolute path to ``umount``.
"""
return shutil.which("umount") or "/usr/bin/umount"

View File

@ -66,6 +66,25 @@ class Config:
All active picks stay installed and visible; the enforcer treats them as
one allowed set. Raising this weakens enforcement proportionally.
"""
daily_gaming_seconds: int = 8 * 3600
"""Gaming seconds allowed per gaming day, which starts at 06:00 local.
Spending this budget shuts Steam down and masks the launcher binaries until
the next 06:00. See :mod:`steam_backlog_enforcer._playtime`.
"""
count_launcher_processes: bool = True
"""Whether time spent in non-Steam launchers counts against the budget.
Set False if a launcher legitimately idles in the tray all day. Leaving it
True is the safer default: under-counting is the failure mode that quietly
defeats the budget entirely.
"""
playtime_enforcement: bool = True
"""Master switch for the daily gaming budget.
Disabling stops the cutoff but never strands a live block — the release
path runs regardless, so "disabled" cannot come to mean "blocked forever".
"""
def save(self) -> None:
"""Persist config to disk."""

View File

@ -32,8 +32,13 @@ from steam_backlog_enforcer._allowed_games import (
MANUAL_LOCK_DAYS as _MANUAL_LOCK_DAYS,
)
from steam_backlog_enforcer._cmd_done import cmd_done
from steam_backlog_enforcer._cmd_playtime import (
cmd_enforce,
cmd_gaming_reset,
cmd_gaming_status,
cmd_gaming_unblock,
)
from steam_backlog_enforcer._enforce_loop import (
do_enforce,
get_all_owned_app_ids,
)
from steam_backlog_enforcer._hltb_types import load_hltb_cache
@ -109,13 +114,25 @@ _MANUAL_LOCK_EXEMPT_COMMANDS = frozenset(
# Allowed so a second game can be locked in alongside the first; the
# cap inside cmd_pick_manual is what stops this being a way out.
"pick-manual",
# The daily gaming budget is orthogonal to which game is assigned, and
# gaming-unblock is a recovery hatch for a stuck bind mount - locking
# it behind a manual pick would leave Steam masked with no way back.
"gaming-status",
"gaming-unblock",
"gaming-reset",
}
)
# Commands that remain usable while a total gaming block is active. Far
# stricter than _MANUAL_LOCK_EXEMPT_COMMANDS: no done/pick/reset/
# add-exception - there is no in-app way to shorten a total block.
_TOTAL_BLOCK_EXEMPT_COMMANDS = frozenset({"status", "enforce"})
#
# gaming-unblock is included because a playtime bind mount makes the total
# block's own `pacman -R steam` fail EBUSY - it must stay reachable exactly
# when the two collide. gaming-reset is NOT included: it shortens enforcement.
_TOTAL_BLOCK_EXEMPT_COMMANDS = frozenset(
{"status", "enforce", "gaming-status", "gaming-unblock"}
)
# ──────────────────────────────────────────────────────────────
@ -905,7 +922,6 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
"check": ("Check assigned game completion", do_check),
"status": ("Show current status", cmd_status),
"list": ("List games from snapshot", cmd_list),
"enforce": ("Run enforcer: block, uninstall, kill, hide", do_enforce),
"install": ("Install the assigned game", cmd_install),
"hide": ("Hide all non-assigned games in library", cmd_hide),
"unhide": ("Unhide all games in library", cmd_unhide),
@ -919,6 +935,8 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
"pick": ("Manually pick your next game from candidates", cmd_pick),
"stats": ("Show backlog completion-time estimates", cmd_stats),
"serve": ("Start the interactive web UI (browser) server", cmd_serve),
"gaming-status": ("Show today's gaming time and block state", cmd_gaming_status),
"gaming-reset": ("Reset today's gaming counter (root + YES)", cmd_gaming_reset),
}
# Extra commands with non-standard arg handling (shown in help but not in COMMANDS).
@ -929,6 +947,8 @@ _EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = {
f"Undo a manual pick within {_MANUAL_GRACE_DAYS} days (needs app_id)"
),
"block-gaming": "Block ALL gaming for <days> days, no in-app undo",
"enforce": "Run enforcer: block, uninstall, kill, hide (--demo for a 60s budget)",
"gaming-unblock": "Force-release playtime bind mounts (root; recovery hatch)",
}
_ALL_COMMANDS: dict[str, str] = {
@ -981,6 +1001,40 @@ def _print_usage(unknown: str | None = None) -> None:
_echo(f" {name:<14s} {desc}")
def _dispatch_extra_command(command: str, config: Config, state: State) -> bool:
"""Dispatch commands whose arguments do not fit the ``COMMANDS`` signature.
Split out of :func:`main` to keep its branch count within the complexity
limit; every entry here takes raw ``sys.argv`` tail arguments rather than
the ``(config, state)`` pair ``COMMANDS`` callables receive.
Args:
command: Canonical command name.
config: Loaded configuration.
state: Loaded state.
Returns:
True if *command* was handled here.
"""
if command == "add-exception":
cmd_add_exception(sys.argv[2:])
return True
if command == "block-gaming":
cmd_block_gaming(sys.argv[2:])
return True
if command == "pick-manual":
cmd_pick_manual(config, state, sys.argv[2:])
return True
if command == "abandon-pick":
cmd_abandon_pick(config, state, sys.argv[2:])
return True
if command == "enforce":
sys.exit(cmd_enforce(config, state, sys.argv[2:]))
if command == "gaming-unblock":
sys.exit(cmd_gaming_unblock(sys.argv[2:]))
return False
def main() -> None:
"""CLI entry point."""
if len(sys.argv) < _MIN_CLI_ARGS:
@ -1011,22 +1065,7 @@ def main() -> None:
# This also covers add-exception (previously dispatched before state load).
_enforce_manual_pick_lock(command, state)
# add-exception, pick-manual, and block-gaming have non-standard
# argument structures.
if command == "add-exception":
cmd_add_exception(sys.argv[2:])
return
if command == "block-gaming":
cmd_block_gaming(sys.argv[2:])
return
if command == "pick-manual":
cmd_pick_manual(config, state, sys.argv[2:])
return
if command == "abandon-pick":
cmd_abandon_pick(config, state, sys.argv[2:])
if _dispatch_extra_command(command, config, state):
return
_, func = COMMANDS[command]

View File

@ -137,6 +137,74 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
yield
@pytest.fixture(autouse=True)
def _isolate_playtime(tmp_path: Path) -> Iterator[None]:
"""Redirect the daily-gaming-budget paths to a temporary directory.
Separate from :func:`_isolate_filesystem` only because CPython caps a
``with`` statement at 20 statically nested blocks and that fixture is
already at the limit.
The ``BLOCK_TARGETS`` patch is the load-bearing one: unpatched, that tuple
names the real ``/usr/bin/steam``, and a test reaching the mount code would
mask the user's actual Steam install behind a refusal stub.
"""
# Namespaced names: an autouse fixture that creates bare "proc"/"targets"
# inside tmp_path would claim those names for every other test in the
# suite (test_enforcer builds its own tmp_path/"proc" with mkdir()).
fake_config = tmp_path / "config"
fake_config.mkdir(exist_ok=True)
fake_proc = tmp_path / "playtime_proc"
fake_proc.mkdir(exist_ok=True)
fake_targets = tmp_path / "playtime_targets"
fake_targets.mkdir(exist_ok=True)
with (
# Computed at import time from CONFIG_DIR, so patching CONFIG_DIR
# alone does not redirect them.
patch(
"steam_backlog_enforcer._playtime.PLAYTIME_STATE_FILE",
fake_config / "playtime_state.json",
),
patch(
"steam_backlog_enforcer._playtime.PLAYTIME_DEMO_STATE_FILE",
fake_config / "playtime_demo_state.json",
),
patch("steam_backlog_enforcer._playtime._PROC", fake_proc),
# Never let a test chattr +i a file inside tmp_path — pytest could
# then not clean the directory up.
patch("steam_backlog_enforcer._playtime._try_set_immutable", MagicMock()),
patch("steam_backlog_enforcer._playtime.unlock_for_write", MagicMock()),
patch(
"steam_backlog_enforcer._playtime_block.BLOCK_TARGETS",
(
fake_targets / "steam",
fake_targets / "bin_steam.sh",
fake_targets / "steam.sh",
fake_targets / "lutris",
),
),
patch("steam_backlog_enforcer._playtime_block._STUB_DIR", tmp_path / "run"),
patch(
"steam_backlog_enforcer._playtime_block.STUB_PATH",
tmp_path / "run" / "gaming-blocked",
),
patch(
"steam_backlog_enforcer._playtime_block.MOUNTINFO_PATH",
tmp_path / "mountinfo",
),
patch(
"steam_backlog_enforcer._playtime_block._INIT_MOUNTINFO_PATH",
tmp_path / "init_mountinfo",
),
patch(
"steam_backlog_enforcer._playtime_block.PACMAN_LOCK", tmp_path / "db.lck"
),
patch("steam_backlog_enforcer._playtime_block._PROC", fake_proc),
):
yield
@pytest.fixture(autouse=True)
def _block_real_subprocesses() -> Iterator[None]:
"""Block subprocess calls that could launch real Steam or modify system.
@ -177,6 +245,14 @@ def _block_real_subprocesses() -> Iterator[None]:
"steam_backlog_enforcer.library_hider.subprocess.Popen",
noop_popen,
),
# The single most important patch in this file. _playtime_block._run
# executes `mount --bind` against paths that, unpatched, are the real
# /usr/bin/steam. A test that reached the real subprocess would mask
# the user's actual Steam install with a refusal stub.
patch(
"steam_backlog_enforcer._playtime_block.subprocess.run",
noop_run,
),
):
yield

View File

@ -0,0 +1,215 @@
"""Tests for the daily-gaming-budget CLI handlers."""
from __future__ import annotations
from unittest.mock import patch
from steam_backlog_enforcer._cmd_playtime import (
cmd_enforce,
cmd_gaming_reset,
cmd_gaming_status,
cmd_gaming_unblock,
)
from steam_backlog_enforcer._playtime import PlaytimeState, load_state, save_state
from steam_backlog_enforcer.config import Config, State
PKG = "steam_backlog_enforcer._cmd_playtime"
def _echoed(mock_echo: object) -> str:
return " ".join(str(c) for c in mock_echo.call_args_list)
class TestCmdEnforce:
def test_runs_in_production_by_default(self) -> None:
with patch(f"{PKG}.do_enforce") as mock_run, patch(f"{PKG}._echo"):
assert cmd_enforce(Config(), State(), []) == 0
assert mock_run.call_args.kwargs["demo"] is False
def test_demo_flag_enables_demo(self) -> None:
with patch(f"{PKG}.do_enforce") as mock_run, patch(f"{PKG}._echo"):
assert cmd_enforce(Config(), State(), ["--demo"]) == 0
assert mock_run.call_args.kwargs["demo"] is True
def test_demo_announces_itself(self) -> None:
with patch(f"{PKG}.do_enforce"), patch(f"{PKG}._echo") as mock_echo:
cmd_enforce(Config(), State(), ["--demo"])
assert "DEMO MODE" in _echoed(mock_echo)
def test_unknown_flag_exits_nonzero_without_running(self) -> None:
"""A mistyped --Demo must not silently run the real 8-hour budget."""
with patch(f"{PKG}.do_enforce") as mock_run, patch(f"{PKG}._echo") as mock_echo:
assert cmd_enforce(Config(), State(), ["--Demo"]) == 2
mock_run.assert_not_called()
assert "Unknown argument" in _echoed(mock_echo)
def test_unknown_flag_alongside_demo_still_rejected(self) -> None:
with patch(f"{PKG}.do_enforce") as mock_run, patch(f"{PKG}._echo"):
assert cmd_enforce(Config(), State(), ["--demo", "--bogus"]) == 2
mock_run.assert_not_called()
class TestCmdGamingStatus:
def test_reports_no_state_yet(self) -> None:
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value=set()),
):
cmd_gaming_status(Config(), State())
assert "no state recorded yet" in _echoed(mock_echo)
def test_reports_usage_and_remaining(self) -> None:
save_state(PlaytimeState(day_key="2026-07-27", seconds=100.0), demo=False)
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value=set()),
):
cmd_gaming_status(Config(daily_gaming_seconds=500), State())
output = _echoed(mock_echo)
assert "used:" in output
assert "100s" in output
assert "400s" in output
def test_hides_the_demo_section_when_unused(self) -> None:
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value=set()),
):
cmd_gaming_status(Config(), State())
assert "Demo budget" not in _echoed(mock_echo)
def test_shows_the_demo_section_when_present(self) -> None:
save_state(PlaytimeState(day_key="2026-07-27", seconds=5.0), demo=True)
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value=set()),
):
cmd_gaming_status(Config(), State())
assert "Demo budget" in _echoed(mock_echo)
def test_lists_masked_launchers(self) -> None:
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value={"/usr/bin/steam"}),
):
cmd_gaming_status(Config(), State())
output = _echoed(mock_echo)
assert "1/4" in output
assert "/usr/bin/steam" in output
def test_reports_blocked_flag(self) -> None:
save_state(
PlaytimeState(day_key="2026-07-27", seconds=1.0, blocked_at=5.0),
demo=False,
)
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value=set()),
):
cmd_gaming_status(Config(), State())
assert "blocked: True" in _echoed(mock_echo)
class TestCmdGamingReset:
def test_requires_root(self) -> None:
with (
patch(f"{PKG}.os.geteuid", return_value=1000),
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.release_block") as mock_release,
):
assert cmd_gaming_reset(Config(), State()) == 1
mock_release.assert_not_called()
assert "requires root" in _echoed(mock_echo)
def test_aborts_without_the_typed_confirmation(self) -> None:
with (
patch(f"{PKG}.os.geteuid", return_value=0),
patch(f"{PKG}._echo") as mock_echo,
patch("builtins.input", return_value="yes"),
patch(f"{PKG}.release_block") as mock_release,
):
assert cmd_gaming_reset(Config(), State()) == 1
mock_release.assert_not_called()
assert "Aborted" in _echoed(mock_echo)
def test_resets_counter_and_releases_mounts(self) -> None:
save_state(
PlaytimeState(day_key="2026-07-26", seconds=999.0, blocked_at=5.0),
demo=False,
)
with (
patch(f"{PKG}.os.geteuid", return_value=0),
patch(f"{PKG}._echo"),
patch("builtins.input", return_value="YES"),
patch(f"{PKG}.release_block", return_value=["/usr/bin/steam"]),
):
assert cmd_gaming_reset(Config(), State()) == 0
stored = load_state(demo=False)
assert stored.seconds == 0.0
assert stored.is_blocked() is False
def test_reports_how_many_mounts_were_released(self) -> None:
with (
patch(f"{PKG}.os.geteuid", return_value=0),
patch(f"{PKG}._echo") as mock_echo,
patch("builtins.input", return_value="YES"),
patch(f"{PKG}.release_block", return_value=["/a", "/b"]),
):
cmd_gaming_reset(Config(), State())
assert "Released 2 mount(s)" in _echoed(mock_echo)
class TestCmdGamingUnblock:
def test_rejects_arguments(self) -> None:
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.release_block") as mock_release,
):
assert cmd_gaming_unblock(["--force"]) == 2
mock_release.assert_not_called()
assert "Unknown argument" in _echoed(mock_echo)
def test_requires_root(self) -> None:
with (
patch(f"{PKG}.os.geteuid", return_value=1000),
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.release_block") as mock_release,
):
assert cmd_gaming_unblock([]) == 1
mock_release.assert_not_called()
assert "requires root" in _echoed(mock_echo)
def test_reports_when_nothing_was_mounted(self) -> None:
with (
patch(f"{PKG}.os.geteuid", return_value=0),
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.release_block", return_value=[]),
):
assert cmd_gaming_unblock([]) == 0
assert "No playtime mounts" in _echoed(mock_echo)
def test_lists_released_mounts(self) -> None:
with (
patch(f"{PKG}.os.geteuid", return_value=0),
patch(f"{PKG}._echo") as mock_echo,
patch(
f"{PKG}.release_block",
return_value=["/usr/bin/steam", "/usr/bin/lutris"],
),
):
assert cmd_gaming_unblock([]) == 0
output = _echoed(mock_echo)
assert "/usr/bin/steam" in output
assert "/usr/bin/lutris" in output
def test_consults_no_state_at_all(self) -> None:
"""The recovery hatch must work with an unreadable state file."""
with (
patch(f"{PKG}.os.geteuid", return_value=0),
patch(f"{PKG}._echo"),
patch(
f"{PKG}.load_state", side_effect=AssertionError("must not be called")
),
patch(f"{PKG}.release_block", return_value=[]),
):
assert cmd_gaming_unblock([]) == 0

View File

@ -158,10 +158,26 @@ class TestDoEnforce:
with patch(f"{PKG}.steam_is_installed", return_value=True):
yield
def test_no_game(self) -> None:
with patch(f"{PKG}._echo") as mock_echo:
do_enforce(Config(), State())
assert any("No game" in str(c) for c in mock_echo.call_args_list)
def test_no_game_says_so_but_keeps_looping(self) -> None:
"""No assignment is reported, but the loop still runs.
The daily gaming budget is accounted for from inside this loop, so
returning here would stop enforcing it whenever a game is finished
but not yet rescanned.
"""
state = State()
with (
patch(f"{PKG}._echo") as mock_echo,
patch.object(State, "load", return_value=state),
patch(
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
):
do_enforce(Config(), state)
assert any("No game" in str(c) for c in mock_echo.call_args_list)
mock_iter.assert_called_once()
def test_steam_absent_idles_without_setup(self) -> None:
"""With Steam gone, say so and keep looping - never exit.

View File

@ -87,8 +87,7 @@ class TestDoEnforceTotalBlock:
mock_setup.assert_not_called()
def test_active_with_no_game_does_not_early_return(self) -> None:
"""Without total block, no assigned game means do_enforce returns
immediately. With it active, the loop must still run."""
"""A total block with no assigned game must still run the loop."""
state = State()
config = Config()
with (
@ -105,11 +104,82 @@ class TestDoEnforceTotalBlock:
do_enforce(config, state)
mock_iter.assert_called_once()
def test_inactive_with_no_game_returns_early(self) -> None:
def test_inactive_with_no_game_still_enters_the_loop(self) -> None:
"""No assigned game must NOT exit the process.
It used to return here, which silently stopped enforcing the daily
gaming budget for as long as a game was finished but not yet
rescanned - the budget is accounted for from inside this loop.
"""
state = State()
with (
patch(f"{PKG}.is_total_block_active", return_value=False),
patch(f"{PKG}._echo") as mock_echo,
patch.object(State, "load", return_value=state),
patch(
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
):
do_enforce(Config(), State())
do_enforce(Config(), state)
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "No game" in output
assert "budget is still enforced" in output
mock_iter.assert_called_once()
class TestPlaytimeIsEnforcedFirst:
"""The gaming budget must run before every other guard in the tick."""
def test_runs_even_when_total_block_short_circuits(self) -> None:
"""Time is still spent, and a 06:00 release still has to happen."""
config = Config()
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.playtime_tick") as mock_playtime,
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.enforce_total_block_tick"),
):
_enforce_loop_iteration(config, state)
mock_playtime.assert_called_once()
def test_receives_the_loop_interval(self) -> None:
config = Config()
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.playtime_tick") as mock_playtime,
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.enforce_total_block_tick"),
):
_enforce_loop_iteration(config, state)
assert mock_playtime.call_args.kwargs["interval"] == 3
assert mock_playtime.call_args.kwargs["demo"] is False
def test_demo_flag_is_plumbed_through(self) -> None:
config = Config()
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.playtime_tick") as mock_playtime,
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.enforce_total_block_tick"),
):
_enforce_loop_iteration(config, state, demo=True)
assert mock_playtime.call_args.kwargs["demo"] is True
def test_do_enforce_plumbs_demo_into_the_iteration(self) -> None:
state = State(current_app_id=1, current_game_name="G")
with (
patch(f"{PKG}.is_total_block_active", return_value=False),
patch(f"{PKG}.steam_is_installed", return_value=True),
patch(f"{PKG}._enforce_setup"),
patch(f"{PKG}._echo"),
patch.object(State, "load", return_value=state),
patch(
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
):
do_enforce(Config(), state, demo=True)
assert mock_iter.call_args.kwargs["demo"] is True

View File

@ -106,7 +106,18 @@ class TestEnforceTotalBlockLock:
_enforce_total_block_lock("add-exception")
def test_exempt_set_is_stricter_than_manual_pick(self) -> None:
assert frozenset({"status", "enforce"}) == _TOTAL_BLOCK_EXEMPT_COMMANDS
"""Pinned exactly: widening this set weakens the total block.
gaming-unblock is present because a playtime bind mount makes the
total block's own `pacman -R steam` fail EBUSY - it has to stay
reachable exactly when the two collide. gaming-reset is deliberately
absent: it would shorten enforcement.
"""
assert (
frozenset({"status", "enforce", "gaming-status", "gaming-unblock"})
== _TOTAL_BLOCK_EXEMPT_COMMANDS
)
assert "gaming-reset" not in _TOTAL_BLOCK_EXEMPT_COMMANDS
# ──────────────────────────────────────────────────────────────
@ -258,3 +269,66 @@ class TestCmdStatusTotalBlock:
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "TOTAL GAMING BLOCK ACTIVE" in output
assert "Days remaining" not in output
# ──────────────────────────────────────────────────────────────
# main() dispatch to the daily-gaming-budget commands
# ──────────────────────────────────────────────────────────────
class TestMainDispatchGamingCommands:
def test_enforce_passes_its_tail_args_and_exits_with_the_code(self) -> None:
"""`enforce` left COMMANDS because its signature cannot carry --demo."""
argv = ["prog", "enforce", "--demo"]
with (
patch.object(sys, "argv", argv),
patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
patch(f"{PKG}.State.load", return_value=State()),
patch(f"{PKG}.is_total_block_active", return_value=False),
patch(f"{PKG}.cmd_enforce", return_value=0) as mock_cmd,
pytest.raises(SystemExit) as exc_info,
):
main()
assert exc_info.value.code == 0
assert mock_cmd.call_args.args[2] == ["--demo"]
def test_enforce_propagates_a_nonzero_exit(self) -> None:
argv = ["prog", "enforce", "--bogus"]
with (
patch.object(sys, "argv", argv),
patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
patch(f"{PKG}.State.load", return_value=State()),
patch(f"{PKG}.is_total_block_active", return_value=False),
patch(f"{PKG}.cmd_enforce", return_value=2),
pytest.raises(SystemExit) as exc_info,
):
main()
assert exc_info.value.code == 2
def test_gaming_unblock_dispatches(self) -> None:
argv = ["prog", "gaming-unblock"]
with (
patch.object(sys, "argv", argv),
patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
patch(f"{PKG}.State.load", return_value=State()),
patch(f"{PKG}.is_total_block_active", return_value=False),
patch(f"{PKG}.cmd_gaming_unblock", return_value=0) as mock_cmd,
pytest.raises(SystemExit) as exc_info,
):
main()
assert exc_info.value.code == 0
mock_cmd.assert_called_once_with([])
def test_gaming_unblock_reachable_during_a_total_block(self) -> None:
"""A playtime mount makes the total block's own pacman -R fail EBUSY."""
argv = ["prog", "gaming-unblock"]
with (
patch.object(sys, "argv", argv),
patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
patch(f"{PKG}.State.load", return_value=State()),
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.cmd_gaming_unblock", return_value=0) as mock_cmd,
pytest.raises(SystemExit),
):
main()
mock_cmd.assert_called_once_with([])

View File

@ -281,6 +281,99 @@ class TestBlockGamingGate:
assert "privileges" in out["reason"]
class TestGetGamingTime:
def test_reports_unrecorded_state(self) -> None:
with (
patch.object(_mcp, "load_state", return_value=None),
patch.object(_mcp, "mounted_targets", return_value=set()),
):
out = _mcp.get_gaming_time()
assert out["recorded"] is False
assert out["budget_seconds"] == 8 * 3600
def test_reports_usage(self) -> None:
stored = _mcp.PlaytimeState(day_key="2026-07-27", seconds=100.0)
with (
patch.object(_mcp, "load_state", return_value=stored),
patch.object(_mcp, "mounted_targets", return_value=set()),
):
out = _mcp.get_gaming_time()
assert out["recorded"] is True
assert out["gaming_day"] == "2026-07-27"
assert out["seconds_used"] == 100.0
assert out["seconds_remaining"] == 8 * 3600 - 100.0
assert out["blocked"] is False
def test_remaining_never_goes_negative(self) -> None:
stored = _mcp.PlaytimeState(day_key="d", seconds=10**9)
with (
patch.object(_mcp, "load_state", return_value=stored),
patch.object(_mcp, "mounted_targets", return_value=set()),
):
out = _mcp.get_gaming_time()
assert out["seconds_remaining"] == 0.0
def test_lists_masked_launchers(self) -> None:
stored = _mcp.PlaytimeState(day_key="d", seconds=1.0, blocked_at=2.0)
with (
patch.object(_mcp, "load_state", return_value=stored),
patch.object(_mcp, "mounted_targets", return_value={"/usr/bin/steam"}),
):
out = _mcp.get_gaming_time()
assert out["blocked"] is True
assert out["masked_launchers"] == ["/usr/bin/steam"]
def test_leaks_no_secret(self) -> None:
"""No Config secret may cross the MCP boundary."""
with (
patch.object(_mcp, "load_state", return_value=None),
patch.object(_mcp, "mounted_targets", return_value=set()),
):
out = _mcp.get_gaming_time()
assert "steam_api_key" not in out
assert "steam_id" not in out
class TestResetGamingTime:
def test_preview_by_default(self) -> None:
with patch.object(_mcp, "release_block") as mock_release:
out = _mcp.reset_gaming_time()
mock_release.assert_not_called()
assert out["preview"] is True
assert out["confirm_required"] is True
assert out["requires_root"] is True
def test_confirm_unprivileged_returns_gracefully(self) -> None:
with (
patch.object(_mcp.os, "geteuid", return_value=1000),
patch.object(_mcp, "release_block") as mock_release,
):
out = _mcp.reset_gaming_time(confirm=True)
mock_release.assert_not_called()
assert out["ok"] is False
assert "privileges" in out["reason"]
def test_confirm_applies_as_root(self) -> None:
with (
patch.object(_mcp.os, "geteuid", return_value=0),
patch.object(_mcp, "release_block", return_value=["/usr/bin/steam"]),
patch.object(_mcp, "save_state") as mock_save,
):
out = _mcp.reset_gaming_time(confirm=True)
mock_save.assert_called_once()
assert out["applied"] is True
assert out["released"] == ["/usr/bin/steam"]
def test_confirm_oserror_returns_gracefully(self) -> None:
with (
patch.object(_mcp.os, "geteuid", return_value=0),
patch.object(_mcp, "release_block", side_effect=OSError("boom")),
):
out = _mcp.reset_gaming_time(confirm=True)
assert out["ok"] is False
assert "privileges" in out["reason"]
def test_main_runs_stdio_server() -> None:
with patch.object(_mcp.mcp, "run") as run:
_mcp.main()

View File

@ -0,0 +1,197 @@
"""Tests for playtime day keys, state persistence and rule resolution."""
from __future__ import annotations
from dataclasses import FrozenInstanceError
from datetime import datetime, timedelta, timezone
import json
from unittest.mock import patch
import pytest
from steam_backlog_enforcer._playtime import (
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,
state_path,
)
from steam_backlog_enforcer.config import Config
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
def _at(year: int, month: int, day: int, hour: int, minute: int = 0) -> datetime:
return datetime(year, month, day, hour, minute, tzinfo=LOCAL)
class TestGamingDayKey:
def test_just_before_the_boundary_is_the_previous_day(self) -> None:
assert gaming_day_key(_at(2026, 7, 27, 5, 59)) == "2026-07-26"
def test_exactly_at_the_boundary_starts_the_new_day(self) -> None:
assert gaming_day_key(_at(2026, 7, 27, 6, 0)) == "2026-07-27"
def test_evening_is_the_same_day(self) -> None:
assert gaming_day_key(_at(2026, 7, 27, 23, 30)) == "2026-07-27"
def test_after_midnight_belongs_to_the_day_it_started(self) -> None:
assert gaming_day_key(_at(2026, 7, 28, 1, 15)) == "2026-07-27"
def test_exact_midnight(self) -> None:
assert gaming_day_key(_at(2026, 7, 28, 0, 0)) == "2026-07-27"
def test_month_boundary(self) -> None:
assert gaming_day_key(_at(2026, 8, 1, 3, 0)) == "2026-07-31"
def test_dst_transition_day_still_yields_a_key(self) -> None:
"""The gaming day is 23h or 25h across DST; the budget is not rescaled."""
warsaw_spring_forward = datetime(
2026, 3, 29, 12, 0, tzinfo=timezone(timedelta(hours=2))
)
assert gaming_day_key(warsaw_spring_forward) == "2026-03-29"
class TestStatePath:
def test_production_and_demo_differ(self) -> None:
assert state_path(demo=False) != state_path(demo=True)
def test_demo_path_is_named_demo(self) -> None:
assert "demo" in state_path(demo=True).name
class TestPlaytimeState:
def test_not_blocked_by_default(self) -> None:
assert PlaytimeState().is_blocked() is False
def test_blocked_when_stamped(self) -> None:
assert PlaytimeState(blocked_at=1.0).is_blocked() is True
def test_is_blocked_is_not_serialised(self) -> None:
"""A derived field would be persisted and could drift from blocked_at."""
assert "is_blocked" not in PlaytimeState().__dict__
class TestSaveLoadRoundTrip:
def test_round_trip(self) -> None:
original = PlaytimeState(
day_key="2026-07-27",
seconds=123.5,
last_tick_at=99.0,
blocked_at=42.0,
warned_seconds=[600, 300],
)
save_state(original, demo=False)
loaded = load_state(demo=False)
assert loaded == original
def test_demo_and_production_are_independent(self) -> None:
save_state(PlaytimeState(day_key="prod", seconds=10.0), demo=False)
save_state(PlaytimeState(day_key="demo", seconds=20.0), demo=True)
assert load_state(demo=False).day_key == "prod"
assert load_state(demo=True).day_key == "demo"
def test_missing_file_is_none(self) -> None:
assert load_state(demo=False) is None
def test_corrupt_json_is_none(self) -> None:
state_path(demo=False).write_text("{not json", encoding="utf-8")
assert load_state(demo=False) is None
def test_unreadable_file_is_none(self) -> None:
save_state(PlaytimeState(day_key="x"), demo=False)
with patch(f"{PKG}.Path.read_text", side_effect=OSError("nope")):
assert load_state(demo=False) is None
def test_wrong_schema_version_is_none(self) -> None:
state_path(demo=False).write_text(
json.dumps({"schema_version": 99, "day_key": "x"}), encoding="utf-8"
)
assert load_state(demo=False) is None
def test_non_dict_json_is_none(self) -> None:
state_path(demo=False).write_text("[1, 2, 3]", encoding="utf-8")
assert load_state(demo=False) is None
def test_unknown_keys_are_dropped(self) -> None:
state_path(demo=False).write_text(
json.dumps({"schema_version": 1, "day_key": "d", "bogus": 1}),
encoding="utf-8",
)
loaded = load_state(demo=False)
assert loaded is not None
assert loaded.day_key == "d"
class TestSaveStateImmutability:
def test_production_save_unlocks_then_relocks(self) -> None:
"""rename(2) onto an immutable file is EPERM, so the unlock is required."""
with (
patch(f"{PKG}.unlock_for_write") as mock_unlock,
patch(f"{PKG}._try_set_immutable") as mock_lock,
):
save_state(PlaytimeState(day_key="x"), demo=False)
mock_unlock.assert_called_once_with(state_path(demo=False))
mock_lock.assert_called_once_with(state_path(demo=False), immutable=True)
def test_demo_save_is_left_mutable(self) -> None:
"""An immutable demo file could not be deleted during cleanup."""
with (
patch(f"{PKG}.unlock_for_write") as mock_unlock,
patch(f"{PKG}._try_set_immutable") as mock_lock,
):
save_state(PlaytimeState(day_key="x"), demo=True)
mock_unlock.assert_not_called()
mock_lock.assert_not_called()
class TestRulesFor:
def test_production_budget_comes_from_config(self) -> None:
rules = rules_for(Config(daily_gaming_seconds=100), demo=False)
assert rules.budget_seconds == 100.0
assert rules.demo is False
def test_demo_budget_is_sixty_seconds(self) -> None:
rules = rules_for(Config(daily_gaming_seconds=28800), demo=True)
assert rules.budget_seconds == 60.0
assert rules.demo is True
def test_demo_warns_in_seconds_not_minutes(self) -> None:
assert rules_for(Config(), demo=True).warn_at == (30, 10)
assert rules_for(Config(), demo=False).warn_at == (3600, 1800, 600, 300)
def test_demo_escalates_to_sigkill_sooner(self) -> None:
assert rules_for(Config(), demo=True).sigkill_after == 10.0
assert rules_for(Config(), demo=False).sigkill_after == 30.0
def test_launcher_toggle_is_carried_through(self) -> None:
assert (
rules_for(
Config(count_launcher_processes=False), demo=False
).count_launchers
is False
)
assert (
rules_for(Config(count_launcher_processes=True), demo=False).count_launchers
is True
)
def test_enforcement_toggle_is_carried_through(self) -> None:
assert (
rules_for(Config(playtime_enforcement=False), demo=False).enforcement
is False
)
def test_demo_uses_the_same_qualifying_predicate(self) -> None:
"""Demo may differ in budget and warnings only."""
prod = rules_for(Config(count_launcher_processes=True), demo=False)
demo = rules_for(Config(count_launcher_processes=True), demo=True)
assert prod.count_launchers == demo.count_launchers
def test_rules_are_frozen(self) -> None:
"""Frozen so the decision helpers stay pure and cheap to reason about."""
rules = rules_for(Config(), demo=False)
with pytest.raises(FrozenInstanceError):
rules.budget_seconds = 1.0

View File

@ -0,0 +1,182 @@
"""Tests for the playtime block's mount discovery and stub handling."""
from __future__ import annotations
import subprocess
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer import _playtime_block as blk
from steam_backlog_enforcer._playtime_block import (
_ensure_stub,
_mount_bin,
_mountpoints,
_run,
_umount_bin,
_unescape_mountinfo,
block_targets,
mounted_targets,
mounts_are_visible,
)
if TYPE_CHECKING:
from pathlib import Path
PKG = "steam_backlog_enforcer._playtime_block"
def _mountinfo_line(mount_point: str) -> str:
return f"36 25 0:32 / {mount_point} rw,relatime shared:5 - tmpfs tmpfs rw"
def _write_mountinfo(tmp_path: Path, points: list[str]) -> None:
(tmp_path / "mountinfo").write_text(
"\n".join(_mountinfo_line(p) for p in points) + "\n" if points else "",
encoding="utf-8",
)
class TestUnescapeMountinfo:
def test_space(self) -> None:
assert _unescape_mountinfo(r"/mnt/my\040dir") == "/mnt/my dir"
def test_tab_newline_backslash(self) -> None:
assert _unescape_mountinfo(r"a\011b") == "a\tb"
assert _unescape_mountinfo(r"a\012b") == "a\nb"
assert _unescape_mountinfo(r"a\134b") == "a\\b"
def test_plain_string_untouched(self) -> None:
assert _unescape_mountinfo("/usr/bin/steam") == "/usr/bin/steam"
class TestMountpoints:
def test_unreadable_file_returns_empty(self, tmp_path: Path) -> None:
assert _mountpoints(tmp_path / "does-not-exist") == set()
def test_empty_file(self, tmp_path: Path) -> None:
path = tmp_path / "mi"
path.write_text("", encoding="utf-8")
assert _mountpoints(path) == set()
def test_skips_short_lines(self, tmp_path: Path) -> None:
path = tmp_path / "mi"
path.write_text("1 2 3\n" + _mountinfo_line("/a") + "\n", encoding="utf-8")
assert {str(p) for p in _mountpoints(path)} == {"/a"}
def test_decodes_escapes(self, tmp_path: Path) -> None:
path = tmp_path / "mi"
path.write_text(_mountinfo_line(r"/mnt/a\040b") + "\n", encoding="utf-8")
assert {str(p) for p in _mountpoints(path)} == {"/mnt/a b"}
class TestMountedTargets:
def test_none_mounted(self, tmp_path: Path) -> None:
_write_mountinfo(tmp_path, [])
assert mounted_targets() == set()
def test_ignores_unrelated_mounts(self, tmp_path: Path) -> None:
_write_mountinfo(tmp_path, ["/", "/home", "/etc/hosts"])
assert mounted_targets() == set()
def test_detects_a_target(self, tmp_path: Path) -> None:
target = block_targets()[0]
_write_mountinfo(tmp_path, ["/", str(target)])
assert mounted_targets() == {target}
def test_detects_all_targets(self, tmp_path: Path) -> None:
_write_mountinfo(tmp_path, [str(t) for t in block_targets()])
assert mounted_targets() == set(block_targets())
class TestMountsAreVisible:
def test_true_when_nothing_mounted(self, tmp_path: Path) -> None:
_write_mountinfo(tmp_path, [])
assert mounts_are_visible() is True
def test_true_when_init_sees_the_mount(self, tmp_path: Path) -> None:
target = block_targets()[0]
_write_mountinfo(tmp_path, [str(target)])
(tmp_path / "init_mountinfo").write_text(
_mountinfo_line(str(target)) + "\n", encoding="utf-8"
)
assert mounts_are_visible() is True
def test_false_when_init_cannot_see_it(self, tmp_path: Path) -> None:
"""A private mount namespace would look exactly like this."""
target = block_targets()[0]
_write_mountinfo(tmp_path, [str(target)])
(tmp_path / "init_mountinfo").write_text(
_mountinfo_line("/unrelated") + "\n", encoding="utf-8"
)
assert mounts_are_visible() is False
class TestEnsureStub:
def test_writes_executable_stub(self, tmp_path: Path) -> None:
assert _ensure_stub() is True
stub = tmp_path / "run" / "gaming-blocked"
assert stub.exists()
assert stub.stat().st_mode & 0o111
assert "Gaming blocked" in stub.read_text(encoding="utf-8")
def test_idempotent(self, tmp_path: Path) -> None:
assert _ensure_stub() is True
assert _ensure_stub() is True
def test_returns_false_on_oserror(self) -> None:
with patch(f"{PKG}.Path.mkdir", side_effect=OSError("boom")):
assert _ensure_stub() is False
class TestRun:
def test_true_on_zero_exit(self) -> None:
with patch(
f"{PKG}.subprocess.run",
return_value=MagicMock(returncode=0),
) as mock_run:
assert _run(["mount", "--bind", "a", "b"]) is True
mock_run.assert_called_once()
def test_false_on_nonzero_exit(self) -> None:
with patch(f"{PKG}.subprocess.run", return_value=MagicMock(returncode=1)):
assert _run(["mount"]) is False
def test_false_on_oserror(self) -> None:
with patch(f"{PKG}.subprocess.run", side_effect=OSError("nope")):
assert _run(["mount"]) is False
def test_false_on_timeout(self) -> None:
with patch(
f"{PKG}.subprocess.run",
side_effect=subprocess.TimeoutExpired("mount", 15),
):
assert _run(["mount"]) is False
class TestBinaryResolution:
def test_mount_bin_found(self) -> None:
with patch(f"{PKG}.shutil.which", return_value="/bin/mount"):
assert _mount_bin() == "/bin/mount"
def test_mount_bin_fallback(self) -> None:
with patch(f"{PKG}.shutil.which", return_value=None):
assert _mount_bin() == "/usr/bin/mount"
def test_umount_bin_found(self) -> None:
with patch(f"{PKG}.shutil.which", return_value="/bin/umount"):
assert _umount_bin() == "/bin/umount"
def test_umount_bin_fallback(self) -> None:
with patch(f"{PKG}.shutil.which", return_value=None):
assert _umount_bin() == "/usr/bin/umount"
class TestBlockTargets:
def test_accessor_returns_the_constant(self) -> None:
assert block_targets() == blk.BLOCK_TARGETS
def test_four_targets_are_isolated_in_tests(self) -> None:
"""Guards the conftest patch: these must never be the real binaries."""
assert len(block_targets()) == 4
for target in block_targets():
assert not str(target).startswith("/usr/")

View File

@ -0,0 +1,438 @@
"""Tests for applying/releasing the playtime block and the kill walk."""
from __future__ import annotations
import os
import signal
from typing import TYPE_CHECKING
from unittest.mock import patch
from steam_backlog_enforcer._playtime_block import (
_child_map,
_own_process_chain,
_read_ppid,
apply_block,
block_targets,
descendant_pids,
kill_gaming_processes,
reconcile,
release_block,
request_steam_shutdown,
steam_and_launcher_pids,
)
if TYPE_CHECKING:
from pathlib import Path
PKG = "steam_backlog_enforcer._playtime_block"
def _line(mount_point: str) -> str:
return f"36 25 0:32 / {mount_point} rw,relatime shared:5 - tmpfs tmpfs rw"
def _set_mounted(tmp_path: Path, points: list[Path]) -> None:
text = "".join(f"{_line(str(p))}\n" for p in points)
(tmp_path / "mountinfo").write_text(text, encoding="utf-8")
def _make_targets_exist() -> None:
for target in block_targets():
target.write_text("#!/bin/sh\nexec real\n", encoding="utf-8")
class TestApplyBlock:
def test_deferred_while_pacman_holds_the_lock(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
(tmp_path / "db.lck").write_text("", encoding="utf-8")
with patch(f"{PKG}._run") as mock_run:
assert apply_block() == []
mock_run.assert_not_called()
def test_returns_empty_when_stub_cannot_be_written(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
with (
patch(f"{PKG}._ensure_stub", return_value=False),
patch(f"{PKG}._run") as mock_run,
):
assert apply_block() == []
mock_run.assert_not_called()
def test_mounts_every_existing_target(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
with patch(f"{PKG}._run", return_value=True):
assert apply_block() == list(block_targets())
def test_skips_missing_targets(self, tmp_path: Path) -> None:
_set_mounted(tmp_path, [])
block_targets()[0].write_text("x", encoding="utf-8")
with patch(f"{PKG}._run", return_value=True):
assert apply_block() == [block_targets()[0]]
def test_skips_already_mounted(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [block_targets()[0]])
with patch(f"{PKG}._run", return_value=True):
assert apply_block() == list(block_targets()[1:])
def test_bind_failure_skips_remount(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
with patch(f"{PKG}._run", return_value=False) as mock_run:
assert apply_block() == []
# One --bind attempt per target, and no remount followed any of them.
assert all("--bind" in call.args[0] for call in mock_run.call_args_list)
def test_remount_failure_is_not_counted(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
# First call (--bind) succeeds, second (remount) fails, repeatedly.
with patch(f"{PKG}._run", side_effect=[True, False] * 4):
assert apply_block() == []
def test_uses_mount_and_remount_argv(self, tmp_path: Path) -> None:
_set_mounted(tmp_path, [])
block_targets()[0].write_text("x", encoding="utf-8")
with (
patch(f"{PKG}._run", return_value=True) as mock_run,
patch(f"{PKG}._mount_bin", return_value="/bin/mount"),
):
apply_block()
argvs = [call.args[0] for call in mock_run.call_args_list]
assert argvs[0][:2] == ["/bin/mount", "--bind"]
assert argvs[1][:3] == ["/bin/mount", "-o", "remount,ro,bind"]
class TestReleaseBlock:
def test_nothing_mounted(self, tmp_path: Path) -> None:
_set_mounted(tmp_path, [])
with patch(f"{PKG}._run") as mock_run:
assert release_block() == []
mock_run.assert_not_called()
def test_releases_a_single_mount(self, tmp_path: Path) -> None:
target = block_targets()[0]
_set_mounted(tmp_path, [target])
def fake_run(_cmd: list[str]) -> bool:
_set_mounted(tmp_path, [])
return True
with patch(f"{PKG}._run", side_effect=fake_run):
assert release_block() == [target]
def test_loops_for_stacked_mounts(self, tmp_path: Path) -> None:
target = block_targets()[0]
_set_mounted(tmp_path, [target])
remaining = {"n": 3}
def fake_run(_cmd: list[str]) -> bool:
remaining["n"] -= 1
_set_mounted(tmp_path, [target] if remaining["n"] > 0 else [])
return True
with patch(f"{PKG}._run", side_effect=fake_run) as mock_run:
assert release_block() == [target]
assert mock_run.call_count == 3
def test_breaks_when_umount_fails(self, tmp_path: Path) -> None:
target = block_targets()[0]
_set_mounted(tmp_path, [target])
with patch(f"{PKG}._run", return_value=False) as mock_run:
assert release_block() == []
assert mock_run.call_count == 1
def test_bounded_when_umount_never_takes_effect(self, tmp_path: Path) -> None:
"""Always-succeeds-but-never-unmounts must not spin forever."""
target = block_targets()[0]
_set_mounted(tmp_path, [target])
with patch(f"{PKG}._run", return_value=True) as mock_run:
assert release_block() == []
assert mock_run.call_count == 20
def test_uses_lazy_umount_argv(self, tmp_path: Path) -> None:
target = block_targets()[0]
_set_mounted(tmp_path, [target])
with (
patch(f"{PKG}._run", return_value=False) as mock_run,
patch(f"{PKG}._umount_bin", return_value="/bin/umount"),
):
release_block()
assert mock_run.call_args_list[0].args[0] == ["/bin/umount", "-l", str(target)]
class TestReconcile:
def test_release_path(self, tmp_path: Path) -> None:
target = block_targets()[0]
_set_mounted(tmp_path, [target])
def fake_run(_cmd: list[str]) -> bool:
_set_mounted(tmp_path, [])
return True
with patch(f"{PKG}._run", side_effect=fake_run):
masked, released = reconcile(should_block=False)
assert masked == []
assert released == [target]
def test_apply_path(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
with (
patch(f"{PKG}._run", return_value=True),
patch(f"{PKG}.mounts_are_visible", return_value=True),
):
masked, released = reconcile(should_block=True)
assert masked == list(block_targets())
assert released == []
def test_logs_when_mounts_are_invisible(self, tmp_path: Path) -> None:
_make_targets_exist()
_set_mounted(tmp_path, [])
with (
patch(f"{PKG}._run", return_value=True),
patch(f"{PKG}.mounts_are_visible", return_value=False),
patch(f"{PKG}.logger.error") as mock_error,
):
reconcile(should_block=True)
mock_error.assert_called_once()
def test_no_visibility_check_when_nothing_was_masked(self, tmp_path: Path) -> None:
_set_mounted(tmp_path, [])
with (
patch(f"{PKG}._run", return_value=True),
patch(f"{PKG}.mounts_are_visible") as mock_visible,
):
reconcile(should_block=True)
mock_visible.assert_not_called()
class TestRequestSteamShutdown:
def test_skips_when_steam_is_absent(self) -> None:
with (
patch(f"{PKG}.shutil.which", return_value=None),
patch(f"{PKG}._run") as mock_run,
):
request_steam_shutdown()
mock_run.assert_not_called()
def test_sends_shutdown(self) -> None:
with (
patch(f"{PKG}.shutil.which", return_value="/usr/bin/steam"),
patch(f"{PKG}._run", return_value=True) as mock_run,
):
request_steam_shutdown()
mock_run.assert_called_once_with(["/usr/bin/steam", "-shutdown"])
class TestSteamAndLauncherPids:
def test_unions_both_name_sets(self) -> None:
with patch(
f"{PKG}.get_pids_by_process_names",
return_value={11: "steam", 22: "lutris"},
) as mock_get:
assert steam_and_launcher_pids() == {11, 22}
names = mock_get.call_args.args[0]
assert "steam" in names
assert "lutris" in names
class TestReadPpid:
def _write_stat(self, tmp_path: Path, pid: int, comm: str, ppid: int) -> None:
proc = tmp_path / "playtime_proc" / str(pid)
proc.mkdir(parents=True, exist_ok=True)
(proc / "stat").write_text(
f"{pid} ({comm}) S {ppid} {pid} 0 0 -1 0", encoding="utf-8"
)
def test_reads_simple_comm(self, tmp_path: Path) -> None:
self._write_stat(tmp_path, 100, "steam", 1)
assert _read_ppid(100) == 1
def test_handles_comm_with_spaces_and_parens(self, tmp_path: Path) -> None:
"""comm can contain both, so field position alone cannot be trusted."""
self._write_stat(tmp_path, 101, "my game (x) 2", 42)
assert _read_ppid(101) == 42
def test_missing_file(self) -> None:
assert _read_ppid(999999) is None
def test_malformed_stat_too_few_fields(self, tmp_path: Path) -> None:
proc = tmp_path / "playtime_proc" / "102"
proc.mkdir(parents=True, exist_ok=True)
(proc / "stat").write_text("102 (x) S", encoding="utf-8")
assert _read_ppid(102) is None
def test_non_integer_ppid(self, tmp_path: Path) -> None:
proc = tmp_path / "playtime_proc" / "103"
proc.mkdir(parents=True, exist_ok=True)
(proc / "stat").write_text("103 (x) S notanumber 1", encoding="utf-8")
assert _read_ppid(103) is None
class TestChildMap:
def _write(self, tmp_path: Path, pid: int, ppid: int) -> None:
proc = tmp_path / "playtime_proc" / str(pid)
proc.mkdir(parents=True, exist_ok=True)
(proc / "stat").write_text(f"{pid} (p) S {ppid} 1 0", encoding="utf-8")
def test_builds_mapping(self, tmp_path: Path) -> None:
self._write(tmp_path, 10, 1)
self._write(tmp_path, 11, 10)
self._write(tmp_path, 12, 10)
mapping = _child_map()
assert sorted(mapping[10]) == [11, 12]
def test_skips_non_digit_entries(self, tmp_path: Path) -> None:
(tmp_path / "playtime_proc" / "cpuinfo").touch()
(tmp_path / "playtime_proc" / "self").mkdir()
self._write(tmp_path, 20, 1)
assert 1 in _child_map()
def test_unreadable_proc_returns_empty(self) -> None:
with patch(f"{PKG}.Path.iterdir", side_effect=OSError("nope")):
assert _child_map() == {}
def test_entry_without_stat_is_skipped(self, tmp_path: Path) -> None:
(tmp_path / "playtime_proc" / "30").mkdir()
self._write(tmp_path, 31, 1)
assert 30 not in {c for kids in _child_map().values() for c in kids}
class TestDescendantPids:
def test_walks_the_tree(self) -> None:
tree = {100: [101, 102], 101: [103]}
with (
patch(f"{PKG}._child_map", return_value=tree),
patch(f"{PKG}._own_process_chain", return_value=set()),
):
assert descendant_pids({100}) == {101, 102, 103}
def test_excludes_the_roots_themselves(self) -> None:
with (
patch(f"{PKG}._child_map", return_value={100: [101]}),
patch(f"{PKG}._own_process_chain", return_value=set()),
):
assert 100 not in descendant_pids({100})
def test_never_returns_our_own_process(self) -> None:
with (
patch(f"{PKG}._child_map", return_value={100: [101, 102]}),
patch(f"{PKG}._own_process_chain", return_value={101}),
):
assert descendant_pids({100}) == {102}
def test_protected_root_is_not_walked(self) -> None:
with (
patch(f"{PKG}._child_map", return_value={100: [101]}),
patch(f"{PKG}._own_process_chain", return_value={100}),
):
assert descendant_pids({100}) == set()
def test_no_children(self) -> None:
with (
patch(f"{PKG}._child_map", return_value={}),
patch(f"{PKG}._own_process_chain", return_value=set()),
):
assert descendant_pids({100}) == set()
def test_cycle_terminates(self) -> None:
with (
patch(f"{PKG}._child_map", return_value={1: [2], 2: [1]}),
patch(f"{PKG}._own_process_chain", return_value=set()),
):
assert descendant_pids({1}) == {2, 1}
def test_depth_is_bounded(self) -> None:
chain = {i: [i + 1] for i in range(1, 100)}
with (
patch(f"{PKG}._child_map", return_value=chain),
patch(f"{PKG}._own_process_chain", return_value=set()),
):
assert len(descendant_pids({1})) == 32
class TestOwnProcessChain:
def test_includes_self(self) -> None:
with patch(f"{PKG}._read_ppid", return_value=None):
assert _own_process_chain() == {os.getpid()}
def test_walks_to_init(self) -> None:
with patch(f"{PKG}._read_ppid", side_effect=[500, 1, 0]):
chain = _own_process_chain()
assert chain == {os.getpid(), 500, 1}
def test_stops_on_repeat(self) -> None:
with patch(f"{PKG}._read_ppid", return_value=os.getpid()):
assert _own_process_chain() == {os.getpid()}
def test_depth_is_bounded(self) -> None:
"""An unbroken ancestor chain must stop at the depth cap, not spin."""
counter = iter(range(10_000, 20_000))
with patch(f"{PKG}._read_ppid", side_effect=lambda _pid: next(counter)):
assert len(_own_process_chain()) == 32
class TestKillGamingProcesses:
def test_sigterm_by_default(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value=set()),
patch(f"{PKG}._own_process_chain", return_value=set()),
patch(f"{PKG}.os.kill") as mock_kill,
):
assert kill_gaming_processes({7}, force=False) == [7]
mock_kill.assert_called_once_with(7, signal.SIGTERM)
def test_sigkill_when_forced(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value=set()),
patch(f"{PKG}._own_process_chain", return_value=set()),
patch(f"{PKG}.os.kill") as mock_kill,
):
kill_gaming_processes({7}, force=True)
mock_kill.assert_called_once_with(7, signal.SIGKILL)
def test_includes_descendants(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value={8, 9}),
patch(f"{PKG}._own_process_chain", return_value=set()),
patch(f"{PKG}.os.kill"),
):
assert kill_gaming_processes({7}, force=False) == [7, 8, 9]
def test_never_signals_our_own_chain(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value=set()),
patch(f"{PKG}._own_process_chain", return_value={7}),
patch(f"{PKG}.os.kill") as mock_kill,
):
assert kill_gaming_processes({7, 8}, force=False) == [8]
mock_kill.assert_called_once_with(8, signal.SIGTERM)
def test_process_already_gone(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value=set()),
patch(f"{PKG}._own_process_chain", return_value=set()),
patch(f"{PKG}.os.kill", side_effect=ProcessLookupError),
):
assert kill_gaming_processes({7}, force=False) == []
def test_permission_denied(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value=set()),
patch(f"{PKG}._own_process_chain", return_value=set()),
patch(f"{PKG}.os.kill", side_effect=PermissionError),
):
assert kill_gaming_processes({7}, force=False) == []
def test_empty_set(self) -> None:
with (
patch(f"{PKG}.descendant_pids", return_value=set()),
patch(f"{PKG}._own_process_chain", return_value=set()),
patch(f"{PKG}.os.kill") as mock_kill,
):
assert kill_gaming_processes(set(), force=False) == []
mock_kill.assert_not_called()

View File

@ -0,0 +1,283 @@
"""Tests for playtime accumulation, rollover, warnings and /proc scanning."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import os
from typing import TYPE_CHECKING
from unittest.mock import patch
from steam_backlog_enforcer._playtime import (
PlaytimeState,
_humanise,
_match_cmdline,
accumulate,
get_pids_by_cmdline_names,
pending_warning,
qualifying_pids,
roll_over,
rules_for,
)
from steam_backlog_enforcer.config import Config
if TYPE_CHECKING:
from pathlib import Path
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
NOW = datetime(2026, 7, 27, 20, 0, tzinfo=LOCAL)
INTERVAL = 3.0
def _state_ticked(seconds_ago: float, **kwargs: object) -> PlaytimeState:
return PlaytimeState(last_tick_at=NOW.timestamp() - seconds_ago, **kwargs)
class TestAccumulate:
def test_first_ever_tick_credits_nothing(self) -> None:
out = accumulate(
PlaytimeState(last_tick_at=0.0),
now=NOW,
qualifying={1},
interval=INTERVAL,
)
assert out.seconds == 0.0
assert out.last_tick_at == NOW.timestamp()
def test_normal_tick_credits_the_delta(self) -> None:
out = accumulate(_state_ticked(3.0), now=NOW, qualifying={1}, interval=INTERVAL)
assert out.seconds == 3.0
def test_idle_tick_credits_nothing_but_advances_the_clock(self) -> None:
out = accumulate(
_state_ticked(3.0), now=NOW, qualifying=set(), interval=INTERVAL
)
assert out.seconds == 0.0
assert out.last_tick_at == NOW.timestamp()
def test_long_gap_is_clamped(self) -> None:
"""A suspend must not credit hours of gaming that never happened."""
out = accumulate(
_state_ticked(3600.0), now=NOW, qualifying={1}, interval=INTERVAL
)
assert out.seconds == 6.0
def test_gap_exactly_at_the_clamp(self) -> None:
out = accumulate(_state_ticked(6.0), now=NOW, qualifying={1}, interval=INTERVAL)
assert out.seconds == 6.0
def test_backward_clock_step_credits_nothing(self) -> None:
"""An NTP step backwards must never refund time already spent."""
out = accumulate(
_state_ticked(-10.0),
now=NOW,
qualifying={1},
interval=INTERVAL,
)
assert out.seconds == 0.0
assert out.last_tick_at == NOW.timestamp()
def test_zero_delta(self) -> None:
out = accumulate(_state_ticked(0.0), now=NOW, qualifying={1}, interval=INTERVAL)
assert out.seconds == 0.0
def test_accrues_across_ticks(self) -> None:
state = PlaytimeState(seconds=100.0, last_tick_at=NOW.timestamp() - 3.0)
out = accumulate(state, now=NOW, qualifying={1}, interval=INTERVAL)
assert out.seconds == 103.0
def test_is_pure(self) -> None:
state = _state_ticked(3.0)
accumulate(state, now=NOW, qualifying={1}, interval=INTERVAL)
assert state.seconds == 0.0
class TestRollOver:
def test_same_day_is_unchanged(self) -> None:
state = PlaytimeState(day_key="2026-07-27", seconds=50.0, blocked_at=9.0)
assert roll_over(state, day_key="2026-07-27") is state
def test_new_day_resets_counter_and_block(self) -> None:
state = PlaytimeState(
day_key="2026-07-27",
seconds=50.0,
blocked_at=9.0,
warned_seconds=[600],
)
out = roll_over(state, day_key="2026-07-28")
assert out.day_key == "2026-07-28"
assert out.seconds == 0.0
assert out.is_blocked() is False
assert out.warned_seconds == []
def test_last_tick_carries_across_the_boundary(self) -> None:
"""Otherwise the first tick of the new day measures a bogus delta."""
state = PlaytimeState(day_key="2026-07-27", last_tick_at=1234.0)
assert roll_over(state, day_key="2026-07-28").last_tick_at == 1234.0
def test_empty_day_key_rolls_over(self) -> None:
assert roll_over(PlaytimeState(), day_key="2026-07-28").day_key == "2026-07-28"
class TestPendingWarning:
def _rules(self, *, demo: bool = False) -> object:
return rules_for(Config(daily_gaming_seconds=8 * 3600), demo=demo)
def test_none_when_far_from_the_budget(self) -> None:
state = PlaytimeState(seconds=0.0)
assert pending_warning(state, self._rules()) is None
def test_fires_at_the_first_threshold(self) -> None:
state = PlaytimeState(seconds=8 * 3600 - 3600)
assert pending_warning(state, self._rules()) == 3600
def test_skips_already_fired_thresholds(self) -> None:
state = PlaytimeState(seconds=8 * 3600 - 1800, warned_seconds=[3600])
assert pending_warning(state, self._rules()) == 1800
def test_returns_largest_uncrossed_when_two_are_skipped(self) -> None:
"""A missed tick crossing two thresholds must warn once, not twice."""
state = PlaytimeState(seconds=8 * 3600 - 400)
assert pending_warning(state, self._rules()) == 3600
def test_none_when_all_fired(self) -> None:
state = PlaytimeState(
seconds=8 * 3600 - 100, warned_seconds=[3600, 1800, 600, 300]
)
assert pending_warning(state, self._rules()) is None
def test_demo_thresholds(self) -> None:
state = PlaytimeState(seconds=35.0)
assert pending_warning(state, self._rules(demo=True)) == 30
def test_over_budget_still_reports(self) -> None:
state = PlaytimeState(seconds=8 * 3600 + 500, warned_seconds=[3600, 1800, 600])
assert pending_warning(state, self._rules()) == 300
class TestMatchCmdline:
def _proc(self, tmp_path: Path, pid: int, argv: list[str]) -> Path:
entry = tmp_path / "playtime_proc" / str(pid)
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(
b"\x00".join(a.encode() for a in argv) + b"\x00"
)
return entry
def test_direct_argv0_match(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 10, ["/usr/bin/lutris", "--foo"])
assert _match_cmdline(entry, frozenset({"lutris"})) == "lutris"
def test_interpreter_falls_through_to_argv1(self, tmp_path: Path) -> None:
"""The real case: /usr/bin/lutris is a python3 script."""
entry = self._proc(tmp_path, 11, ["/usr/bin/python3", "/usr/bin/lutris"])
assert _match_cmdline(entry, frozenset({"lutris"})) == "lutris"
def test_interpreter_with_no_argv1(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 12, ["python3"])
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_interpreter_running_something_else(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 13, ["python3", "/usr/bin/other"])
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_non_interpreter_no_match(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 14, ["/usr/bin/firefox"])
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_empty_cmdline(self, tmp_path: Path) -> None:
entry = tmp_path / "playtime_proc" / "15"
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(b"")
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_unreadable_cmdline(self, tmp_path: Path) -> None:
entry = tmp_path / "playtime_proc" / "16"
entry.mkdir(parents=True, exist_ok=True)
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_undecodable_bytes_do_not_raise(self, tmp_path: Path) -> None:
entry = tmp_path / "playtime_proc" / "17"
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(b"\xff\xfe\x00")
assert _match_cmdline(entry, frozenset({"lutris"})) is None
class TestGetPidsByCmdlineNames:
def _proc(self, tmp_path: Path, pid: int, argv: list[str]) -> None:
entry = tmp_path / "playtime_proc" / str(pid)
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(
b"\x00".join(a.encode() for a in argv) + b"\x00"
)
def test_finds_matches(self, tmp_path: Path) -> None:
self._proc(tmp_path, 100, ["python3", "/usr/bin/lutris"])
self._proc(tmp_path, 101, ["/usr/bin/firefox"])
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {100: "lutris"}
def test_skips_non_digit_entries(self, tmp_path: Path) -> None:
(tmp_path / "playtime_proc" / "cpuinfo").touch()
(tmp_path / "playtime_proc" / "self").mkdir(exist_ok=True)
self._proc(tmp_path, 102, ["python3", "/usr/bin/lutris"])
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {102: "lutris"}
def test_never_matches_our_own_pid(self, tmp_path: Path) -> None:
"""The daemon is itself python3 - it must not count as a launcher."""
self._proc(tmp_path, os.getpid(), ["python3", "/usr/bin/lutris"])
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {}
def test_empty_proc(self) -> None:
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {}
class TestQualifyingPids:
def test_steam_games_only_when_launchers_are_off(self) -> None:
rules = rules_for(Config(count_launcher_processes=False), demo=False)
with (
patch(f"{PKG}.get_running_steam_game_pids", return_value={5: 440, 6: 0}),
patch(f"{PKG}.get_pids_by_process_names") as mock_comm,
):
assert qualifying_pids(rules) == {5}
mock_comm.assert_not_called()
def test_steam_client_appid_zero_does_not_count(self) -> None:
"""Browsing the store is not gaming."""
rules = rules_for(Config(count_launcher_processes=False), demo=False)
with patch(f"{PKG}.get_running_steam_game_pids", return_value={9: 0}):
assert qualifying_pids(rules) == set()
def test_unions_launchers_from_both_scanners(self) -> None:
rules = rules_for(Config(count_launcher_processes=True), demo=False)
with (
patch(f"{PKG}.get_running_steam_game_pids", return_value={5: 440}),
patch(f"{PKG}.get_pids_by_process_names", return_value={7: "heroic"}),
patch(f"{PKG}.get_pids_by_cmdline_names", return_value={8: "lutris"}),
):
assert qualifying_pids(rules) == {5, 7, 8}
def test_no_processes_at_all(self) -> None:
rules = rules_for(Config(), demo=False)
with (
patch(f"{PKG}.get_running_steam_game_pids", return_value={}),
patch(f"{PKG}.get_pids_by_process_names", return_value={}),
patch(f"{PKG}.get_pids_by_cmdline_names", return_value={}),
):
assert qualifying_pids(rules) == set()
class TestHumanise:
def test_seconds(self) -> None:
assert _humanise(30) == "30 seconds"
def test_exact_minute_boundary(self) -> None:
assert _humanise(60) == "1 minutes"
def test_minutes(self) -> None:
assert _humanise(300) == "5 minutes"
def test_one_hour_is_special_cased(self) -> None:
assert _humanise(3600) == "1 hour"
def test_thirty_minutes(self) -> None:
assert _humanise(1800) == "30 minutes"

View File

@ -0,0 +1,350 @@
"""Tests for the playtime tick state machine and cutoff sequence."""
from __future__ import annotations
from contextlib import ExitStack
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
import pytest
from steam_backlog_enforcer._playtime import (
PlaytimeState,
_begin_cutoff,
_kill_set,
_policy,
_state_or_recover,
_sustain_block,
_warn,
load_state,
notify_desktop_user,
playtime_tick,
rules_for,
save_state,
)
from steam_backlog_enforcer.config import Config
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
NOW = datetime(2026, 7, 27, 20, 0, tzinfo=LOCAL)
TODAY = "2026-07-27"
@pytest.fixture
def quiet_tick() -> object:
"""Neutralise every side effect the tick can produce."""
with ExitStack() as stack:
mocks = {
name: stack.enter_context(patch(f"{PKG}.{name}"))
for name in (
"reconcile",
"request_steam_shutdown",
"kill_gaming_processes",
"notify_desktop_user",
"mounted_targets",
)
}
mocks["mounted_targets"].return_value = set()
stack.enter_context(patch(f"{PKG}.is_total_block_active", return_value=False))
yield mocks
def _rules(*, demo: bool = False, **cfg: object) -> object:
return rules_for(Config(**cfg), demo=demo)
class TestStateOrRecover:
def test_loads_existing_state(self) -> None:
save_state(PlaytimeState(day_key=TODAY, seconds=42.0), demo=False)
with patch(f"{PKG}.mounted_targets", return_value=set()):
out = _state_or_recover(_rules(), now=NOW)
assert out.seconds == 42.0
def test_fresh_state_when_nothing_stored_and_nothing_mounted(self) -> None:
with patch(f"{PKG}.mounted_targets", return_value=set()):
out = _state_or_recover(_rules(), now=NOW)
assert out.seconds == 0.0
assert out.is_blocked() is False
assert out.day_key == TODAY
def test_fails_closed_when_state_is_gone_but_mounts_remain(self) -> None:
"""Deleting the state file must not be a way to lift the block."""
rules = _rules(daily_gaming_seconds=100)
with patch(f"{PKG}.mounted_targets", return_value={"/usr/bin/steam"}):
out = _state_or_recover(rules, now=NOW)
assert out.seconds == 100.0
assert out.is_blocked() is True
def test_fails_closed_on_corrupt_state(self) -> None:
from steam_backlog_enforcer._playtime import state_path
state_path(demo=False).write_text("{bad", encoding="utf-8")
with patch(f"{PKG}.mounted_targets", return_value={"/usr/bin/steam"}):
out = _state_or_recover(_rules(daily_gaming_seconds=50), now=NOW)
assert out.is_blocked() is True
class TestPolicyBelowBudget:
def test_releases_and_warns(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=8 * 3600 - 3600)
with (
patch(f"{PKG}.reconcile") as mock_rec,
patch(f"{PKG}.notify_desktop_user") as mock_notify,
):
out = _policy(state, _rules(), now=NOW)
mock_rec.assert_called_once_with(should_block=False)
mock_notify.assert_called_once()
assert out.warned_seconds == [3600]
def test_no_warning_when_far_from_budget(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=0.0)
with (
patch(f"{PKG}.reconcile"),
patch(f"{PKG}.notify_desktop_user") as mock_notify,
):
out = _policy(state, _rules(), now=NOW)
mock_notify.assert_not_called()
assert out.warned_seconds == []
class TestPolicyAtOrOverBudget:
def test_engages_the_cutoff_when_not_yet_blocked(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=10**6)
with (
patch(f"{PKG}._begin_cutoff") as mock_begin,
patch(f"{PKG}._sustain_block") as mock_sustain,
):
_policy(state, _rules(), now=NOW)
mock_begin.assert_called_once()
mock_sustain.assert_not_called()
def test_sustains_the_block_once_engaged(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=10**6, blocked_at=1.0)
with (
patch(f"{PKG}._begin_cutoff") as mock_begin,
patch(f"{PKG}._sustain_block") as mock_sustain,
):
_policy(state, _rules(), now=NOW)
mock_begin.assert_not_called()
mock_sustain.assert_called_once()
class TestPolicyEnforcementDisabled:
def test_releases_but_does_not_block(self) -> None:
"""Disabling must never come to mean 'blocked forever'."""
state = PlaytimeState(day_key=TODAY, seconds=10**6)
with (
patch(f"{PKG}.reconcile") as mock_rec,
patch(f"{PKG}.request_steam_shutdown") as mock_shutdown,
):
out = _policy(state, _rules(playtime_enforcement=False), now=NOW)
mock_rec.assert_called_once_with(should_block=False)
mock_shutdown.assert_not_called()
assert out.is_blocked() is False
class TestBeginCutoff:
def test_asks_steam_to_close_before_masking_anything(self) -> None:
"""Masking makes `steam -shutdown` a no-op, so order is load-bearing."""
state = PlaytimeState(day_key=TODAY, seconds=60.0)
with (
patch(f"{PKG}.reconcile") as mock_rec,
patch(f"{PKG}.request_steam_shutdown") as mock_shutdown,
patch(f"{PKG}.kill_gaming_processes") as mock_kill,
patch(f"{PKG}.notify_desktop_user") as mock_notify,
patch(f"{PKG}._kill_set", return_value={7}),
):
out = _begin_cutoff(state, _rules(demo=True), now=NOW)
mock_shutdown.assert_called_once()
mock_rec.assert_not_called()
mock_kill.assert_called_once_with({7}, force=False)
mock_notify.assert_called_once()
assert out.blocked_at == NOW.timestamp()
class TestSustainBlock:
def _run(
self, elapsed: float, *, demo: bool = False
) -> tuple[MagicMock, MagicMock]:
state = PlaytimeState(
day_key=TODAY, seconds=10**6, blocked_at=NOW.timestamp() - elapsed
)
with (
patch(f"{PKG}.reconcile") as mock_rec,
patch(f"{PKG}.kill_gaming_processes") as mock_kill,
patch(f"{PKG}._kill_set", return_value={7}),
):
_sustain_block(state, _rules(demo=demo), now=NOW)
return mock_rec, mock_kill
def test_grace_period_defers_the_mount(self) -> None:
mock_rec, mock_kill = self._run(1.0)
mock_rec.assert_not_called()
mock_kill.assert_called_once_with({7}, force=False)
def test_mounts_once_the_grace_has_passed(self) -> None:
mock_rec, _ = self._run(3.0)
mock_rec.assert_called_once_with(should_block=True)
def test_escalates_to_sigkill(self) -> None:
_, mock_kill = self._run(31.0)
mock_kill.assert_called_once_with({7}, force=True)
def test_no_escalation_just_before_the_threshold(self) -> None:
_, mock_kill = self._run(29.0)
mock_kill.assert_called_once_with({7}, force=False)
def test_demo_escalates_sooner(self) -> None:
_, mock_kill = self._run(11.0, demo=True)
mock_kill.assert_called_once_with({7}, force=True)
def test_state_is_unchanged(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=99.0, blocked_at=1.0)
with (
patch(f"{PKG}.reconcile"),
patch(f"{PKG}.kill_gaming_processes"),
patch(f"{PKG}._kill_set", return_value=set()),
):
out = _sustain_block(state, _rules(), now=NOW)
assert out is state
class TestKillSet:
def test_is_wider_than_the_budget_predicate(self) -> None:
"""A Lutris Wine game is invisible to both budget matchers."""
with (
patch(f"{PKG}.qualifying_pids", return_value={1}),
patch(f"{PKG}.steam_and_launcher_pids", return_value={2}),
):
assert _kill_set(_rules()) == {1, 2}
class TestWarn:
def test_records_the_threshold(self) -> None:
state = PlaytimeState(seconds=8 * 3600 - 300, warned_seconds=[3600, 1800, 600])
with patch(f"{PKG}.notify_desktop_user") as mock_notify:
out = _warn(state, _rules())
assert out.warned_seconds == [3600, 1800, 600, 300]
assert "5 minutes" in mock_notify.call_args.args[1]
def test_no_op_when_nothing_is_due(self) -> None:
state = PlaytimeState(seconds=0.0)
with patch(f"{PKG}.notify_desktop_user") as mock_notify:
out = _warn(state, _rules())
mock_notify.assert_not_called()
assert out is state
class TestNotifyDesktopUser:
def test_routes_through_the_desktop_session(self) -> None:
"""notify-send as root with no DBUS address never reaches the user."""
with (
patch(f"{PKG}._resolve_desktop_user", return_value="kuhy"),
patch(f"{PKG}._run_as_user") as mock_run,
):
notify_desktop_user("T", "B")
cmd, user = mock_run.call_args.args
assert cmd[0] == "notify-send"
assert user == "kuhy"
def test_swallows_errors(self) -> None:
with (
patch(f"{PKG}._resolve_desktop_user", return_value="kuhy"),
patch(f"{PKG}._run_as_user", side_effect=OSError("no session")),
):
notify_desktop_user("T", "B")
class TestPlaytimeTick:
def test_accumulates_while_a_game_runs(self, quiet_tick: dict) -> None:
save_state(
PlaytimeState(day_key=TODAY, last_tick_at=NOW.timestamp() - 3.0),
demo=False,
)
with (
patch(f"{PKG}.qualifying_pids", return_value={7}),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
assert load_state(demo=False).seconds == 3.0
def test_total_block_releases_and_stops_accruing(self, quiet_tick: dict) -> None:
"""Our mounts would make the total block's `pacman -R steam` fail."""
save_state(
PlaytimeState(
day_key=TODAY, seconds=5.0, last_tick_at=NOW.timestamp() - 3.0
),
demo=False,
)
with (
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.qualifying_pids", return_value={7}),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
quiet_tick["reconcile"].assert_called_once_with(should_block=False)
assert load_state(demo=False).seconds == 5.0
def test_rollover_clears_a_stale_block(self, quiet_tick: dict) -> None:
save_state(
PlaytimeState(
day_key="2026-07-26",
seconds=10**6,
blocked_at=1.0,
last_tick_at=NOW.timestamp() - 3.0,
),
demo=False,
)
with (
patch(f"{PKG}.qualifying_pids", return_value=set()),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
stored = load_state(demo=False)
assert stored.day_key == TODAY
assert stored.seconds == 0.0
assert stored.is_blocked() is False
quiet_tick["reconcile"].assert_called_with(should_block=False)
def test_cutoff_engages_at_the_budget(self, quiet_tick: dict) -> None:
save_state(
PlaytimeState(
day_key=TODAY, seconds=59.0, last_tick_at=NOW.timestamp() - 3.0
),
demo=True,
)
with (
patch(f"{PKG}.qualifying_pids", return_value={7}),
patch(f"{PKG}._kill_set", return_value={7}),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0, demo=True)
stored = load_state(demo=True)
assert stored.is_blocked() is True
quiet_tick["request_steam_shutdown"].assert_called_once()
def test_demo_uses_a_separate_state_file(self, quiet_tick: dict) -> None:
"""A demo run must never consume the real day's budget."""
with (
patch(f"{PKG}.qualifying_pids", return_value={7}),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0, demo=True)
assert load_state(demo=True) is not None
assert load_state(demo=False) is None
def test_first_tick_writes_state(self, quiet_tick: dict) -> None:
with (
patch(f"{PKG}.qualifying_pids", return_value=set()),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
stored = load_state(demo=False)
assert stored is not None
assert stored.day_key == TODAY