mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-10 19:27:04 +01:00
Make serve re-runnable, and cover it to 100%
The serve rework was sitting untracked: a /proc-based port guard, the start-or-defer decision, and a stale-bundle rebuild. It had no tests, and it changed cmd_serve(_config, state) to cmd_serve(argv) without updating the one test that called it, so the suite was red and coverage was 251 lines short. Tests, all against fakes: - _fake_proc builds a synthetic /proc. _NET_TCP is derived from _PROC at import time, so patch_proc redirects both; patching _PROC alone would have had these tests parse the host's real process table. - _serve_guard is split across two test files to stay under the 250-line cap. os is replaced wholesale in every termination test, so no signal can reach a real process even if the guard's self-protection were removed. - _web_build mocks subprocess entirely and redirects every path constant into tmp_path. _no_subprocess now blocks _web_build.subprocess.run too: an unguarded test would fire a real `npm run build`, rewriting web/dist and perturbing frontend_is_stale() for every later test in the run. Source changes, both required rather than incidental: - _serve_guard's wildcard match was a `# noqa: S104` on a "0.0.0.0" literal, which the no-noqa hook rejects outright. It is now the all-zero hex column /proc actually prints, so the dotted-quad literal never appears. - main/__init__.py's serve dispatch arm was itself uncovered; it now has a test asserting the raw argv passes through unparsed. Also drops the document.hidden guard from useBudgetPoll: it skipped the very first load, so the budget panel sat on "Reading budget state…" forever in a background tab, which is every tab under browser automation. Verified: 1712 python tests, 100% branch coverage; 128 vitest, 100%. The guard's /proc parser was checked read-only against a real listener on :8000 and resolved the same pid as `ss`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MKFzwzwRU1JD7SFLHtQBsz
This commit is contained in:
parent
1faf211767
commit
47f77f5f1e
248
steam_backlog_enforcer/_serve_guard.py
Normal file
248
steam_backlog_enforcer/_serve_guard.py
Normal file
@ -0,0 +1,248 @@
|
||||
"""Guard that makes ``serve`` safe to run twice.
|
||||
|
||||
Binding a second web server to the same port used to raise a bare
|
||||
``OSError: [Errno 98]`` traceback. Instead of failing, this resolves *who*
|
||||
owns the port and decides between three outcomes: an already-running server on
|
||||
current code is reported and the caller exits 0; one running stale code is
|
||||
terminated so a fresh process can take the port; anything else is a hard error
|
||||
naming the owner.
|
||||
|
||||
Ownership is resolved through ``/proc`` rather than ``ss`` so the whole thing
|
||||
stays importable, dependency-free and unit-testable against a fake ``/proc``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import socket
|
||||
import time
|
||||
|
||||
_PROC = Path("/proc")
|
||||
_NET_TCP = (_PROC / "net" / "tcp", _PROC / "net" / "tcp6")
|
||||
|
||||
# /proc/net/tcp state column: 0A is TCP_LISTEN. Anything else is a connection,
|
||||
# not a bind, and must not be mistaken for the process owning the port.
|
||||
_LISTEN_STATE = "0A"
|
||||
|
||||
# argv fragments that identify one of our own web servers.
|
||||
_OWN_MODULE = "steam_backlog_enforcer.main"
|
||||
_OWN_COMMAND = "serve"
|
||||
|
||||
# Package tree whose .py mtimes decide whether a running server is stale.
|
||||
_PACKAGE_ROOT = Path(__file__).resolve().parent
|
||||
|
||||
# A wildcard IPv4 bind renders as an all-zero address column. Spelled as hex
|
||||
# rather than _hex_v4("0.0.0.0") so the dotted-quad literal never appears: this
|
||||
# module only ever *matches* that address, it must never bind it.
|
||||
_WILDCARD_V4 = "0" * 8
|
||||
|
||||
# /proc/<pid>/stat after the comm field: fields[0] is state, fields[1] ppid.
|
||||
_MIN_STAT_FIELDS_AFTER_COMM = 2
|
||||
_TERM_WAIT_SECONDS = 5.0
|
||||
_POLL_INTERVAL = 0.1
|
||||
|
||||
|
||||
def _hex_port(port: int) -> str:
|
||||
"""Render *port* the way /proc/net/tcp does (upper-case, 4 hex digits)."""
|
||||
return f"{port:04X}"
|
||||
|
||||
|
||||
def _hex_v4(host: str) -> str:
|
||||
"""Render an IPv4 address the way /proc/net/tcp does (little-endian hex)."""
|
||||
packed = socket.inet_pton(socket.AF_INET, host)
|
||||
return f"{int.from_bytes(packed, 'little'):08X}"
|
||||
|
||||
|
||||
def _hex_v6(host: str) -> str:
|
||||
"""Render an IPv6 address the way /proc/net/tcp6 does.
|
||||
|
||||
The 16 bytes are four 32-bit words, each byte-swapped independently.
|
||||
"""
|
||||
packed = socket.inet_pton(socket.AF_INET6, host)
|
||||
words = (packed[i : i + 4] for i in range(0, 16, 4))
|
||||
return "".join(f"{int.from_bytes(word, 'little'):08X}" for word in words)
|
||||
|
||||
|
||||
def _conflicting_addresses(host: str) -> frozenset[str]:
|
||||
"""Address columns whose listener would block a bind on *host*.
|
||||
|
||||
Matching on the port alone is wrong: a listener on ``192.168.1.5:8000``
|
||||
does not conflict with ``127.0.0.1:8000``, and reporting it as the owner
|
||||
would name an innocent process. Conversely a wildcard bind (``0.0.0.0``,
|
||||
``::``) or a v4-mapped one does conflict, and lives in tcp6 even when the
|
||||
bind we want is IPv4 - which is why both files are scanned.
|
||||
"""
|
||||
return frozenset(
|
||||
{
|
||||
_hex_v4(host),
|
||||
_WILDCARD_V4,
|
||||
_hex_v6("::"),
|
||||
_hex_v6(f"::ffff:{host}"),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _listening_inodes(host: str, port: int) -> set[int]:
|
||||
"""Collect socket inodes listening on an address that blocks *host*:*port*."""
|
||||
suffix = ":" + _hex_port(port)
|
||||
addresses = _conflicting_addresses(host)
|
||||
inodes: set[int] = set()
|
||||
for path in _NET_TCP:
|
||||
try:
|
||||
lines = path.read_text(encoding="utf-8").splitlines()[1:]
|
||||
except OSError:
|
||||
continue
|
||||
for line in lines:
|
||||
fields = line.split()
|
||||
min_fields = 10
|
||||
if len(fields) <= min_fields:
|
||||
continue
|
||||
local = fields[1]
|
||||
if not local.endswith(suffix) or fields[3] != _LISTEN_STATE:
|
||||
continue
|
||||
if local[: -len(suffix)] not in addresses:
|
||||
continue
|
||||
with contextlib.suppress(ValueError):
|
||||
inodes.add(int(fields[9]))
|
||||
return inodes
|
||||
|
||||
|
||||
def _pid_owns_inode(pid_dir: Path, targets: set[int]) -> bool:
|
||||
"""Check whether any file descriptor of *pid_dir* is one of *targets*."""
|
||||
wanted = {f"socket:[{inode}]" for inode in targets}
|
||||
try:
|
||||
fds = list((pid_dir / "fd").iterdir())
|
||||
except OSError:
|
||||
# A process we cannot inspect (other user, or exited mid-scan).
|
||||
return False
|
||||
for fd in fds:
|
||||
try:
|
||||
if fd.readlink().as_posix() in wanted:
|
||||
return True
|
||||
except OSError:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
def find_port_owner(host: str, port: int) -> int | None:
|
||||
"""Return the PID listening on *port*, or None if nothing is.
|
||||
|
||||
Only processes readable by this user can be attributed; a listener owned
|
||||
by another user resolves to None, which callers must treat as "occupied by
|
||||
something I cannot identify", never as "free".
|
||||
"""
|
||||
inodes = _listening_inodes(host, port)
|
||||
if not inodes:
|
||||
return None
|
||||
for entry in _PROC.iterdir():
|
||||
if not entry.name.isdigit():
|
||||
continue
|
||||
if _pid_owns_inode(entry, inodes):
|
||||
return int(entry.name)
|
||||
return None
|
||||
|
||||
|
||||
def read_cmdline(pid: int) -> list[str]:
|
||||
"""Return argv for *pid*, or an empty list if it cannot be read."""
|
||||
try:
|
||||
raw = (_PROC / str(pid) / "cmdline").read_bytes()
|
||||
except OSError:
|
||||
return []
|
||||
return [part for part in raw.decode("utf-8", "replace").split("\0") if part]
|
||||
|
||||
|
||||
def is_our_server(pid: int) -> bool:
|
||||
"""Check whether *pid* is one of our own ``serve`` processes."""
|
||||
argv = read_cmdline(pid)
|
||||
return _OWN_MODULE in argv and _OWN_COMMAND in argv
|
||||
|
||||
|
||||
def process_started_at(pid: int) -> float | None:
|
||||
"""Return the start time of *pid* as an epoch float.
|
||||
|
||||
``/proc/<pid>`` itself is stamped with the process start time, which is
|
||||
exactly the boundary we need: any source file newer than it was not the
|
||||
file that process loaded.
|
||||
"""
|
||||
try:
|
||||
return (_PROC / str(pid)).stat().st_mtime
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def newest_py_since(started_at: float) -> Path | None:
|
||||
"""Return the newest ``.py`` in the package modified after *started_at*.
|
||||
|
||||
Returning the path rather than a bool is deliberate: the restart message
|
||||
names the file, so the decision to kill a server is always auditable.
|
||||
"""
|
||||
newest: Path | None = None
|
||||
newest_mtime = started_at
|
||||
for path in _PACKAGE_ROOT.rglob("*.py"):
|
||||
try:
|
||||
mtime = path.stat().st_mtime
|
||||
except OSError:
|
||||
continue
|
||||
if mtime > newest_mtime:
|
||||
newest, newest_mtime = path, mtime
|
||||
return newest
|
||||
|
||||
|
||||
def _own_pid_chain() -> set[int]:
|
||||
"""Return this process and its ancestors.
|
||||
|
||||
The process running this check matches :func:`is_our_server` by
|
||||
construction, so a wrong owner lookup would otherwise have us kill
|
||||
ourselves mid-startup.
|
||||
"""
|
||||
chain: set[int] = set()
|
||||
pid: int | None = os.getpid()
|
||||
while pid is not None and pid > 0 and pid not in chain:
|
||||
chain.add(pid)
|
||||
pid = _read_ppid(pid)
|
||||
return chain
|
||||
|
||||
|
||||
def _read_ppid(pid: int) -> int | None:
|
||||
"""Return the parent PID of *pid* from ``/proc/<pid>/stat``."""
|
||||
try:
|
||||
stat = (_PROC / str(pid) / "stat").read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
# comm sits in parentheses and may contain spaces; ppid follows it.
|
||||
_, _, rest = stat.partition(") ")
|
||||
fields = rest.split()
|
||||
if len(fields) < _MIN_STAT_FIELDS_AFTER_COMM:
|
||||
return None
|
||||
with contextlib.suppress(ValueError):
|
||||
return int(fields[1])
|
||||
return None
|
||||
|
||||
|
||||
def port_is_free(host: str, port: int) -> bool:
|
||||
"""Check whether *host*:*port* can actually be bound right now."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
|
||||
probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
try:
|
||||
probe.bind((host, port))
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def terminate(pid: int, host: str, port: int) -> bool:
|
||||
"""Stop *pid*, escalating to SIGKILL, and wait for the port to free up."""
|
||||
if pid in _own_pid_chain():
|
||||
return False
|
||||
for sig in (signal.SIGTERM, signal.SIGKILL):
|
||||
with contextlib.suppress(ProcessLookupError, PermissionError):
|
||||
os.kill(pid, sig)
|
||||
deadline = time.monotonic() + _TERM_WAIT_SECONDS
|
||||
while time.monotonic() < deadline:
|
||||
if port_is_free(host, port):
|
||||
return True
|
||||
time.sleep(_POLL_INTERVAL)
|
||||
return port_is_free(host, port)
|
||||
151
steam_backlog_enforcer/_serve_startup.py
Normal file
151
steam_backlog_enforcer/_serve_startup.py
Normal file
@ -0,0 +1,151 @@
|
||||
"""Argument parsing and the start-or-defer decision for ``serve``.
|
||||
|
||||
:mod:`steam_backlog_enforcer._serve_guard` answers *who owns the port*; this
|
||||
module decides *what to do about it*, so ``cmd_serve`` stays a few lines.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import sys
|
||||
from typing import NoReturn
|
||||
|
||||
from steam_backlog_enforcer._serve_guard import (
|
||||
find_port_owner,
|
||||
is_our_server,
|
||||
newest_py_since,
|
||||
port_is_free,
|
||||
process_started_at,
|
||||
read_cmdline,
|
||||
terminate,
|
||||
)
|
||||
from steam_backlog_enforcer._web_server import DEFAULT_HOST, DEFAULT_PORT
|
||||
from steam_backlog_enforcer.game_install import _echo
|
||||
|
||||
# The dataset API is unauthenticated, so a bind must never leave this machine.
|
||||
# IPv4 only: ThreadingHTTPServer's address_family is AF_INET and nothing
|
||||
# overrides it, so accepting "::1" here would ship a flag that always crashes.
|
||||
_LOOPBACK_ALIAS = "localhost"
|
||||
|
||||
_IPV4 = 4
|
||||
_MIN_PORT = 1024
|
||||
_MAX_PORT = 65535
|
||||
|
||||
_SERVE_USAGE = (
|
||||
"Usage: serve [--host HOST] [--port PORT]\n"
|
||||
f" --host : IPv4 loopback only (127.0.0.1 or {_LOOPBACK_ALIAS})\n"
|
||||
f" --port : {_MIN_PORT}-{_MAX_PORT}, default {DEFAULT_PORT}\n\n"
|
||||
"Re-running serve is safe: it rebuilds the frontend when stale, reports an\n"
|
||||
"already-running server, and replaces one running outdated code."
|
||||
)
|
||||
|
||||
|
||||
def _fail_usage(message: str) -> NoReturn:
|
||||
"""Print *message* followed by the usage block, then exit 1."""
|
||||
_echo(message)
|
||||
_echo(_SERVE_USAGE)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _parse_port(raw: str) -> int:
|
||||
"""Parse and range-check a ``--port`` value."""
|
||||
try:
|
||||
port = int(raw)
|
||||
except ValueError:
|
||||
_fail_usage(f"Not a number: {raw}")
|
||||
if not _MIN_PORT <= port <= _MAX_PORT:
|
||||
_fail_usage(f"Port out of range: {port}")
|
||||
return port
|
||||
|
||||
|
||||
def _parse_host(raw: str) -> str:
|
||||
"""Validate a ``--host`` value and canonicalise it to a literal address."""
|
||||
if raw == _LOOPBACK_ALIAS:
|
||||
return DEFAULT_HOST
|
||||
try:
|
||||
addr = ipaddress.ip_address(raw)
|
||||
except ValueError:
|
||||
_fail_usage(f"Not an IP address: {raw}")
|
||||
if not addr.is_loopback or addr.version != _IPV4:
|
||||
_fail_usage(f"Refusing to bind non-IPv4-loopback host: {raw}")
|
||||
return raw
|
||||
|
||||
|
||||
def parse_serve_args(argv: list[str]) -> tuple[str, int]:
|
||||
"""Parse ``serve``'s flags into a (host, port) pair.
|
||||
|
||||
Args:
|
||||
argv: CLI arguments after the command name.
|
||||
|
||||
Returns:
|
||||
The host and port to bind.
|
||||
"""
|
||||
host, port = DEFAULT_HOST, DEFAULT_PORT
|
||||
rest = list(argv)
|
||||
while rest:
|
||||
flag = rest.pop(0)
|
||||
if flag not in {"--host", "--port"}:
|
||||
_fail_usage(f"Unknown argument: {flag}")
|
||||
if not rest:
|
||||
_fail_usage(f"{flag} needs a value.")
|
||||
value = rest.pop(0)
|
||||
if flag == "--port":
|
||||
port = _parse_port(value)
|
||||
else:
|
||||
host = _parse_host(value)
|
||||
return host, port
|
||||
|
||||
|
||||
def _describe(pid: int) -> str:
|
||||
"""Render a short, human-readable identification of *pid*."""
|
||||
argv = read_cmdline(pid)
|
||||
return f"pid {pid} ({' '.join(argv)})" if argv else f"pid {pid}"
|
||||
|
||||
|
||||
def _exit_foreign(port: int, owner: str) -> NoReturn:
|
||||
"""Report a port held by something that is not ours, then exit 1."""
|
||||
_echo(f"Port {port} is held by {owner}.", flush=True)
|
||||
_echo(f"Free it, or choose another port: serve --port {port + 1}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def ensure_port_available(host: str, port: int) -> None:
|
||||
"""Make *host*:*port* bindable, or exit with a clear explanation.
|
||||
|
||||
Returns normally when the port is free or was freed by stopping a stale
|
||||
server. Exits 0 when an up-to-date server of ours already holds it - that
|
||||
is success, not failure: the UI the caller asked for is already up. Exits
|
||||
1 when the port belongs to something we must not touch.
|
||||
"""
|
||||
if port_is_free(host, port):
|
||||
return
|
||||
|
||||
pid = find_port_owner(host, port)
|
||||
if pid is None:
|
||||
_exit_foreign(port, "a process this user cannot inspect")
|
||||
elif not is_our_server(pid):
|
||||
_exit_foreign(port, f"{_describe(pid)}, which is not our web UI")
|
||||
else:
|
||||
_replace_or_defer(host, port, pid)
|
||||
|
||||
|
||||
def _replace_or_defer(host: str, port: int, pid: int) -> None:
|
||||
"""Exit 0 if the server at *pid* is current, else terminate it."""
|
||||
started_at = process_started_at(pid)
|
||||
stale = None if started_at is None else newest_py_since(started_at)
|
||||
if started_at is not None and stale is None:
|
||||
_echo(
|
||||
"Steam Backlog Enforcer web UI already running: "
|
||||
f"http://{host}:{port} (pid {pid})",
|
||||
flush=True,
|
||||
)
|
||||
sys.exit(0)
|
||||
|
||||
reason = "start time unreadable" if started_at is None else f"{stale} is newer"
|
||||
_echo(f"Restarting stale web UI (pid {pid}): {reason}.", flush=True)
|
||||
if not is_our_server(pid):
|
||||
# The PID was recycled between the scan and here; never signal blind.
|
||||
_exit_foreign(port, f"{_describe(pid)}, which is not our web UI")
|
||||
if not terminate(pid, host, port):
|
||||
_echo(f"Could not free port {port} from pid {pid}.", flush=True)
|
||||
sys.exit(1)
|
||||
94
steam_backlog_enforcer/_web_build.py
Normal file
94
steam_backlog_enforcer/_web_build.py
Normal file
@ -0,0 +1,94 @@
|
||||
"""Rebuild the web bundle when its sources are newer than ``web/dist``.
|
||||
|
||||
``_web_server`` reads ``web/dist`` per request, so a rebuild takes effect
|
||||
without restarting anything - but nothing ever triggered one, which meant
|
||||
``serve`` could quietly hand the browser a bundle older than the working tree.
|
||||
This closes that gap: a stale bundle is rebuilt, and a build that fails stops
|
||||
the launch rather than serving the old files under a new-looking process.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
from steam_backlog_enforcer.game_install import _echo
|
||||
|
||||
WEB_DIR = (Path(__file__).resolve().parent.parent / "web").resolve()
|
||||
_BUILT_MARKER = WEB_DIR / "dist" / "index.html"
|
||||
|
||||
# Everything whose change invalidates the built bundle. Anything under src,
|
||||
# plus the build configuration itself - a vite/tsconfig edit changes the
|
||||
# output even when not one line of src moved.
|
||||
_SOURCE_DIR = WEB_DIR / "src"
|
||||
_SOURCE_FILES = (
|
||||
WEB_DIR / "index.html",
|
||||
WEB_DIR / "package.json",
|
||||
WEB_DIR / "vite.config.ts",
|
||||
WEB_DIR / "tsconfig.json",
|
||||
WEB_DIR / "tsconfig.app.json",
|
||||
WEB_DIR / "tsconfig.node.json",
|
||||
)
|
||||
|
||||
_BUILD_TIMEOUT_SECONDS = 600
|
||||
|
||||
_NPM_MISSING = (
|
||||
"npm was not found on PATH, so the stale web bundle cannot be rebuilt.\n"
|
||||
"npm is installed via nvm here, which only puts it on an interactive\n"
|
||||
"shell's PATH - a systemd unit or cron job will not see it. Either run\n"
|
||||
"serve from a normal shell, or build once by hand: cd web && npm run build"
|
||||
)
|
||||
|
||||
|
||||
def _mtime(path: Path) -> float:
|
||||
"""Return the mtime of *path*, or 0.0 when it does not exist."""
|
||||
try:
|
||||
return path.stat().st_mtime
|
||||
except OSError:
|
||||
return 0.0
|
||||
|
||||
|
||||
def frontend_is_stale() -> bool:
|
||||
"""Check whether any build input is newer than the built bundle.
|
||||
|
||||
A missing ``dist/index.html`` counts as stale so the first ``serve`` in a
|
||||
fresh clone builds instead of serving the "not built" placeholder.
|
||||
"""
|
||||
built = _mtime(_BUILT_MARKER)
|
||||
if built == 0.0:
|
||||
return WEB_DIR.is_dir()
|
||||
sources = list(_SOURCE_FILES)
|
||||
if _SOURCE_DIR.is_dir():
|
||||
sources.extend(path for path in _SOURCE_DIR.rglob("*") if path.is_file())
|
||||
return any(_mtime(path) > built for path in sources)
|
||||
|
||||
|
||||
def build_frontend() -> bool:
|
||||
"""Run ``npm run build`` in ``web/``.
|
||||
|
||||
Returns:
|
||||
True if the bundle was rebuilt, False if the build failed (the
|
||||
captured output is printed either way).
|
||||
"""
|
||||
npm = shutil.which("npm")
|
||||
if npm is None:
|
||||
_echo(_NPM_MISSING)
|
||||
return False
|
||||
|
||||
_echo("Frontend sources changed - rebuilding web/dist ...", flush=True)
|
||||
result = subprocess.run(
|
||||
[npm, "run", "build"],
|
||||
cwd=WEB_DIR,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=_BUILD_TIMEOUT_SECONDS,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
_echo("Frontend build FAILED - refusing to serve a stale bundle.\n")
|
||||
_echo(result.stdout.strip())
|
||||
_echo(result.stderr.strip())
|
||||
return False
|
||||
_echo("Frontend build complete.", flush=True)
|
||||
return True
|
||||
@ -134,7 +134,6 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
|
||||
"done": ("Finish game, open HLTB, pick next", cmd_done),
|
||||
"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),
|
||||
}
|
||||
@ -142,6 +141,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
|
||||
# Extra commands with non-standard arg handling (shown in help but not in COMMANDS).
|
||||
_EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = {
|
||||
"add-exception": "Request 24h-locked whitelist exception (use --reason)",
|
||||
"serve": "Start the web UI (--port N; replaces a stale server)",
|
||||
"pick-manual": f"Pick a game by app_id, lock enforcer for {_MANUAL_LOCK_DAYS} days",
|
||||
"abandon-pick": "Undo a manual pick at any time (needs app_id)",
|
||||
"block-gaming": "Block ALL gaming for <days> days, no in-app undo",
|
||||
@ -185,6 +185,9 @@ def _dispatch_extra_command(command: str, config: Config, state: State) -> bool:
|
||||
if command == "block-gaming":
|
||||
cmd_block_gaming(sys.argv[2:])
|
||||
return True
|
||||
if command == "serve":
|
||||
cmd_serve(sys.argv[2:])
|
||||
return True
|
||||
if command == "pick-manual":
|
||||
cmd_pick_manual(config, state, sys.argv[2:])
|
||||
return True
|
||||
|
||||
@ -6,12 +6,18 @@ total gaming block.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import sys
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from steam_backlog_enforcer._config_setup import interactive_setup
|
||||
from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
|
||||
from steam_backlog_enforcer._serve_startup import (
|
||||
ensure_port_available,
|
||||
parse_serve_args,
|
||||
)
|
||||
from steam_backlog_enforcer._total_block import start_total_block
|
||||
from steam_backlog_enforcer._web_build import build_frontend, frontend_is_stale
|
||||
from steam_backlog_enforcer._web_server import serve
|
||||
from steam_backlog_enforcer._whitelist import (
|
||||
add_pending_exception,
|
||||
@ -108,9 +114,32 @@ def cmd_setup(_config: Config, _state: State) -> None:
|
||||
interactive_setup()
|
||||
|
||||
|
||||
def cmd_serve(_config: Config, _state: State) -> None:
|
||||
"""Start the interactive web UI server (read-only, localhost only)."""
|
||||
serve()
|
||||
def cmd_serve(args: list[str]) -> None:
|
||||
"""Start the interactive web UI server (read-only, localhost only).
|
||||
|
||||
Re-running this is safe rather than fatal: an already-running server on
|
||||
current code is reported, one on outdated code is replaced, and a stale
|
||||
frontend bundle is rebuilt before anything is served.
|
||||
|
||||
Args:
|
||||
args: CLI argument list after the command name.
|
||||
"""
|
||||
host, port = parse_serve_args(args)
|
||||
# Build BEFORE touching the port. A failed build must not cost you the
|
||||
# server that was already running, and rebuilding first is what lets the
|
||||
# "already running" path hand the browser a fresh bundle on its next
|
||||
# request - _web_server reads web/dist per request, not at startup.
|
||||
if frontend_is_stale() and not build_frontend():
|
||||
sys.exit(1)
|
||||
ensure_port_available(host, port)
|
||||
try:
|
||||
serve(host, port)
|
||||
except OSError as exc:
|
||||
# Last-ditch cover for the gap between the /proc check and the bind.
|
||||
if exc.errno != errno.EADDRINUSE:
|
||||
raise
|
||||
_echo(f"Port {port} was taken while starting up. Try again.")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def cmd_add_exception(args: list[str]) -> None:
|
||||
|
||||
108
steam_backlog_enforcer/tests/_fake_proc.py
Normal file
108
steam_backlog_enforcer/tests/_fake_proc.py
Normal file
@ -0,0 +1,108 @@
|
||||
"""Builders for a fake ``/proc`` tree used by the _serve_guard tests.
|
||||
|
||||
``_serve_guard`` was written to resolve port ownership through ``/proc`` rather
|
||||
than ``ss`` precisely so it could be graded against a synthetic tree instead of
|
||||
the host's real process table. This module is that tree.
|
||||
|
||||
Note that ``_NET_TCP`` is derived from ``_PROC`` at import time, so redirecting
|
||||
the guard needs *both* patched — see :func:`patch_proc`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import contextmanager
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer import _serve_guard
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator, Sequence
|
||||
from pathlib import Path
|
||||
|
||||
# A real /proc/net/tcp row has 17 whitespace-separated columns. The guard reads
|
||||
# only 1 (local address), 3 (state) and 9 (inode), but rejects short rows, so
|
||||
# the surrounding columns have to be present and correctly positioned.
|
||||
_REM_ADDRESS = "00000000:0000"
|
||||
_QUEUES = "00000000:00000000 00:00000000 00000000"
|
||||
_UID_TIMEOUT = "1000 0"
|
||||
_ROW_TAIL = "1 0000000000000000 100 0 0 10 0"
|
||||
|
||||
|
||||
def make_proc(root: Path) -> Path:
|
||||
"""Create an empty fake ``/proc`` under *root* and return it."""
|
||||
proc = root / "proc"
|
||||
(proc / "net").mkdir(parents=True)
|
||||
for name in ("tcp", "tcp6"):
|
||||
write_net_tcp(proc, [], v6=name == "tcp6")
|
||||
return proc
|
||||
|
||||
|
||||
def tcp_row(local: str, inode: object = 4242, state: str = "0A") -> str:
|
||||
"""Render one ``/proc/net/tcp`` row for a socket at *local*.
|
||||
|
||||
*inode* is deliberately untyped so a test can plant a non-numeric value in
|
||||
the column the guard parses with ``int()``.
|
||||
"""
|
||||
return (
|
||||
f" 0: {local} {_REM_ADDRESS} {state} {_QUEUES} "
|
||||
f"{_UID_TIMEOUT} {inode} {_ROW_TAIL}"
|
||||
)
|
||||
|
||||
|
||||
def write_net_tcp(proc: Path, rows: Sequence[str], *, v6: bool = False) -> None:
|
||||
"""Write *rows* into ``net/tcp`` (or ``net/tcp6``) with a header line."""
|
||||
name = "tcp6" if v6 else "tcp"
|
||||
header = " sl local_address rem_address st ...\n"
|
||||
(proc / "net" / name).write_text(header + "\n".join(rows), encoding="utf-8")
|
||||
|
||||
|
||||
def add_pid(
|
||||
proc: Path,
|
||||
pid: int,
|
||||
*,
|
||||
inodes: Sequence[int] = (),
|
||||
cmdline: Sequence[str] = (),
|
||||
ppid: int | None = None,
|
||||
) -> Path:
|
||||
"""Create ``/proc/<pid>`` with the requested sockets, argv and parent."""
|
||||
pid_dir = proc / str(pid)
|
||||
pid_dir.mkdir()
|
||||
fd_dir = pid_dir / "fd"
|
||||
fd_dir.mkdir()
|
||||
for index, inode in enumerate(inodes):
|
||||
(fd_dir / str(index)).symlink_to(f"socket:[{inode}]")
|
||||
if cmdline:
|
||||
(pid_dir / "cmdline").write_bytes("\0".join(cmdline).encode("utf-8") + b"\0")
|
||||
if ppid is not None:
|
||||
# comm may contain spaces and parentheses; the guard splits on ") ".
|
||||
(pid_dir / "stat").write_text(
|
||||
f"{pid} (py thon) S {ppid} 1 1 0", encoding="utf-8"
|
||||
)
|
||||
return pid_dir
|
||||
|
||||
|
||||
@contextmanager
|
||||
def patch_proc(proc: Path) -> Iterator[None]:
|
||||
"""Point the guard at *proc*, including the derived _NET_TCP tuple."""
|
||||
with (
|
||||
patch.object(_serve_guard, "_PROC", proc),
|
||||
patch.object(
|
||||
_serve_guard,
|
||||
"_NET_TCP",
|
||||
(proc / "net" / "tcp", proc / "net" / "tcp6"),
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
def add_opaque_pid(proc: Path, pid: int) -> Path:
|
||||
"""Create a ``/proc/<pid>`` whose ``fd/`` cannot be opened.
|
||||
|
||||
This is what another user's process looks like from here: it holds the
|
||||
port, but nothing about it can be attributed.
|
||||
"""
|
||||
pid_dir = proc / str(pid)
|
||||
pid_dir.mkdir()
|
||||
(pid_dir / "fd").write_text("", encoding="utf-8")
|
||||
return pid_dir
|
||||
@ -91,5 +91,11 @@ def _block_real_subprocesses() -> Iterator[None]:
|
||||
"steam_backlog_enforcer._playtime_run.subprocess.run",
|
||||
noop_run,
|
||||
),
|
||||
# A real `npm run build` here would rewrite web/dist/index.html, which
|
||||
# then perturbs frontend_is_stale() for every later test in the run.
|
||||
patch(
|
||||
"steam_backlog_enforcer._web_build.subprocess.run",
|
||||
noop_run,
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
99
steam_backlog_enforcer/tests/test_cmd_serve.py
Normal file
99
steam_backlog_enforcer/tests/test_cmd_serve.py
Normal file
@ -0,0 +1,99 @@
|
||||
"""Tests for the main.cmd_serve wiring — 100% branch coverage.
|
||||
|
||||
Split from test_web_server.py to stay inside the 250-line file cap: that file
|
||||
covers the HTTP server itself, this one covers the command that starts it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from steam_backlog_enforcer import main as main_mod
|
||||
from steam_backlog_enforcer._web_server import DEFAULT_HOST
|
||||
from steam_backlog_enforcer.main import misc as misc_mod
|
||||
|
||||
_CALLS: list[str] = []
|
||||
|
||||
|
||||
def _record_build() -> bool:
|
||||
"""Stand in for a successful build, recording when it ran."""
|
||||
_CALLS.append("build")
|
||||
return True
|
||||
|
||||
|
||||
def _record_guard(*_args: object) -> None:
|
||||
"""Stand in for the port guard, recording when it ran."""
|
||||
_CALLS.append("guard")
|
||||
|
||||
|
||||
class TestCmdServe:
|
||||
"""Tests for the main.cmd_serve wiring.
|
||||
|
||||
Every patch targets the module that *defines* cmd_serve, not the main
|
||||
package that re-exports it: cmd_serve resolves these names from its own
|
||||
globals, so patching the re-export would not bite.
|
||||
"""
|
||||
|
||||
def test_invokes_serve_with_parsed_args(self) -> None:
|
||||
with (
|
||||
patch.object(misc_mod, "frontend_is_stale", return_value=False),
|
||||
patch.object(misc_mod, "ensure_port_available") as guard,
|
||||
patch.object(misc_mod, "serve") as mock_serve,
|
||||
):
|
||||
main_mod.cmd_serve(["--port", "8123"])
|
||||
guard.assert_called_once_with(DEFAULT_HOST, 8123)
|
||||
mock_serve.assert_called_once_with(DEFAULT_HOST, 8123)
|
||||
|
||||
def test_stale_frontend_is_rebuilt_before_binding(self) -> None:
|
||||
# Order matters: a failed build must not cost the caller the server
|
||||
# that was already running, so the build happens before the port check.
|
||||
_CALLS.clear()
|
||||
with (
|
||||
patch.object(misc_mod, "frontend_is_stale", return_value=True),
|
||||
patch.object(misc_mod, "build_frontend", side_effect=_record_build),
|
||||
patch.object(misc_mod, "ensure_port_available", side_effect=_record_guard),
|
||||
patch.object(misc_mod, "serve"),
|
||||
):
|
||||
main_mod.cmd_serve([])
|
||||
assert _CALLS == ["build", "guard"]
|
||||
|
||||
def test_failed_build_exits_1(self) -> None:
|
||||
with (
|
||||
patch.object(misc_mod, "frontend_is_stale", return_value=True),
|
||||
patch.object(misc_mod, "build_frontend", return_value=False),
|
||||
patch.object(misc_mod, "ensure_port_available") as guard,
|
||||
patch.object(misc_mod, "serve") as mock_serve,
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
main_mod.cmd_serve([])
|
||||
assert exc.value.code == 1
|
||||
guard.assert_not_called()
|
||||
mock_serve.assert_not_called()
|
||||
|
||||
def test_port_stolen_after_the_check_exits_1(self) -> None:
|
||||
# Covers the race between the /proc scan and the actual bind.
|
||||
taken = OSError(errno.EADDRINUSE, "Address already in use")
|
||||
with (
|
||||
patch.object(misc_mod, "frontend_is_stale", return_value=False),
|
||||
patch.object(misc_mod, "ensure_port_available"),
|
||||
patch.object(misc_mod, "serve", side_effect=taken),
|
||||
patch.object(misc_mod, "_echo") as echo,
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
main_mod.cmd_serve([])
|
||||
assert exc.value.code == 1
|
||||
assert "was taken while starting up" in echo.call_args[0][0]
|
||||
|
||||
def test_unrelated_oserror_propagates(self) -> None:
|
||||
# Swallowing this would turn a real fault into a bare "try again".
|
||||
denied = OSError(errno.EACCES, "Permission denied")
|
||||
with (
|
||||
patch.object(misc_mod, "frontend_is_stale", return_value=False),
|
||||
patch.object(misc_mod, "ensure_port_available"),
|
||||
patch.object(misc_mod, "serve", side_effect=denied),
|
||||
pytest.raises(OSError, match="Permission denied"),
|
||||
):
|
||||
main_mod.cmd_serve([])
|
||||
@ -32,6 +32,21 @@ class TestMainDispatchAddException:
|
||||
mock_cmd.assert_called_once_with(["440", "--reason", VALID_REASON])
|
||||
|
||||
|
||||
class TestMainDispatchServe:
|
||||
def test_dispatches_serve_with_raw_argv(self) -> None:
|
||||
# serve parses its own flags, so the dispatcher must hand the argv
|
||||
# through untouched rather than pre-parsing it.
|
||||
argv = ["prog", "serve", "--port", "8123"]
|
||||
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}.cmd_serve") as mock_cmd,
|
||||
):
|
||||
main()
|
||||
mock_cmd.assert_called_once_with(["--port", "8123"])
|
||||
|
||||
|
||||
class TestMainDispatchPickManual:
|
||||
def test_dispatches_pick_manual(self) -> None:
|
||||
argv = ["prog", "pick-manual", "489830"]
|
||||
|
||||
222
steam_backlog_enforcer/tests/test_serve_guard.py
Normal file
222
steam_backlog_enforcer/tests/test_serve_guard.py
Normal file
@ -0,0 +1,222 @@
|
||||
"""Tests for _serve_guard's /proc parsing and port-ownership resolution.
|
||||
|
||||
Split from test_serve_guard_lifecycle.py to stay inside the 250-line file cap;
|
||||
this half covers "who holds the port", that half covers "what to do about it".
|
||||
Every test runs against a fake /proc built in tmp_path — none reads the host's
|
||||
real process table, and none signals a real pid.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from steam_backlog_enforcer import _serve_guard
|
||||
from steam_backlog_enforcer.tests._fake_proc import (
|
||||
add_opaque_pid,
|
||||
add_pid,
|
||||
make_proc,
|
||||
patch_proc,
|
||||
tcp_row,
|
||||
write_net_tcp,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
_HOST = "127.0.0.1"
|
||||
_PORT = 8000
|
||||
_LOCAL_V4 = f"{_serve_guard._hex_v4(_HOST)}:{_serve_guard._hex_port(_PORT)}"
|
||||
|
||||
|
||||
class TestHexRendering:
|
||||
"""Tests for the /proc address column encoders."""
|
||||
|
||||
def test_port_is_upper_hex(self) -> None:
|
||||
assert _serve_guard._hex_port(8000) == "1F40"
|
||||
|
||||
def test_ipv4_is_little_endian(self) -> None:
|
||||
assert _serve_guard._hex_v4("127.0.0.1") == "0100007F"
|
||||
|
||||
def test_ipv6_swaps_each_word(self) -> None:
|
||||
assert _serve_guard._hex_v6("::1") == "0" * 24 + "01000000"
|
||||
|
||||
def test_v4_mapped_v6_is_rendered(self) -> None:
|
||||
mapped = _serve_guard._hex_v6("::ffff:127.0.0.1")
|
||||
assert mapped.endswith("0100007F")
|
||||
|
||||
|
||||
class TestConflictingAddresses:
|
||||
"""Tests for which listeners actually block a bind."""
|
||||
|
||||
def test_includes_self_wildcard_and_mapped(self) -> None:
|
||||
addrs = _serve_guard._conflicting_addresses(_HOST)
|
||||
assert _serve_guard._hex_v4(_HOST) in addrs
|
||||
assert _serve_guard._WILDCARD_V4 in addrs
|
||||
assert _serve_guard._hex_v6("::") in addrs
|
||||
|
||||
def test_excludes_an_unrelated_interface(self) -> None:
|
||||
# A listener on a LAN address must never be reported as the owner of a
|
||||
# loopback bind; naming it would blame an innocent process.
|
||||
assert _serve_guard._hex_v4(
|
||||
"192.168.1.5"
|
||||
) not in _serve_guard._conflicting_addresses(_HOST)
|
||||
|
||||
|
||||
class TestListeningInodes:
|
||||
"""Tests for parsing /proc/net/tcp."""
|
||||
|
||||
def test_finds_matching_listener(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [tcp_row(_LOCAL_V4, 4242)])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == {4242}
|
||||
|
||||
def test_finds_wildcard_listener(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
wildcard = f"{_serve_guard._WILDCARD_V4}:{_serve_guard._hex_port(_PORT)}"
|
||||
write_net_tcp(proc, [tcp_row(wildcard, 7)])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == {7}
|
||||
|
||||
def test_finds_v6_wildcard_listener(self, tmp_path: Path) -> None:
|
||||
# A dual-stack bind lives in tcp6 even when our bind is IPv4, which is
|
||||
# why both files are scanned.
|
||||
proc = make_proc(tmp_path)
|
||||
local = f"{_serve_guard._hex_v6('::')}:{_serve_guard._hex_port(_PORT)}"
|
||||
write_net_tcp(proc, [tcp_row(local, 9)], v6=True)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == {9}
|
||||
|
||||
def test_ignores_non_listening_state(self, tmp_path: Path) -> None:
|
||||
# 01 is ESTABLISHED: an outbound connection, not a bind.
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [tcp_row(_LOCAL_V4, 4242, state="01")])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == set()
|
||||
|
||||
def test_ignores_other_port(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
other = f"{_serve_guard._hex_v4(_HOST)}:{_serve_guard._hex_port(9999)}"
|
||||
write_net_tcp(proc, [tcp_row(other, 4242)])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == set()
|
||||
|
||||
def test_ignores_other_interface(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
lan = f"{_serve_guard._hex_v4('192.168.1.5')}:{_serve_guard._hex_port(_PORT)}"
|
||||
write_net_tcp(proc, [tcp_row(lan, 4242)])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == set()
|
||||
|
||||
def test_ignores_truncated_row(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [f" 0: {_LOCAL_V4} 0A 4242"])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == set()
|
||||
|
||||
def test_ignores_non_numeric_inode(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [tcp_row(_LOCAL_V4, inode="notanumber")])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == set()
|
||||
|
||||
def test_unreadable_file_is_skipped(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [tcp_row(_LOCAL_V4, 4242)])
|
||||
(proc / "net" / "tcp6").unlink()
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._listening_inodes(_HOST, _PORT) == {4242}
|
||||
|
||||
|
||||
class TestPidOwnsInode:
|
||||
"""Tests for matching a socket inode to a process."""
|
||||
|
||||
def test_matching_fd_is_found(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
pid_dir = add_pid(proc, 100, inodes=[1, 4242])
|
||||
assert _serve_guard._pid_owns_inode(pid_dir, {4242}) is True
|
||||
|
||||
def test_unrelated_fds_do_not_match(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
pid_dir = add_pid(proc, 100, inodes=[1, 2])
|
||||
assert _serve_guard._pid_owns_inode(pid_dir, {4242}) is False
|
||||
|
||||
def test_unreadable_fd_dir_is_not_an_owner(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
pid_dir = add_opaque_pid(proc, 100)
|
||||
assert _serve_guard._pid_owns_inode(pid_dir, {4242}) is False
|
||||
|
||||
def test_unreadable_single_fd_is_skipped(self, tmp_path: Path) -> None:
|
||||
# A plain file in fd/ cannot be readlink()ed; the scan must continue
|
||||
# to the sibling entry rather than abandoning the process.
|
||||
proc = make_proc(tmp_path)
|
||||
pid_dir = add_pid(proc, 100, inodes=[4242])
|
||||
(pid_dir / "fd" / "broken").write_text("", encoding="utf-8")
|
||||
assert _serve_guard._pid_owns_inode(pid_dir, {4242}) is True
|
||||
|
||||
|
||||
class TestFindPortOwner:
|
||||
"""Tests for end-to-end owner resolution."""
|
||||
|
||||
def test_no_listener_returns_none(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.find_port_owner(_HOST, _PORT) is None
|
||||
|
||||
def test_resolves_the_listening_pid(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [tcp_row(_LOCAL_V4, 4242)])
|
||||
add_pid(proc, 100, inodes=[9])
|
||||
add_pid(proc, 200, inodes=[4242])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.find_port_owner(_HOST, _PORT) == 200
|
||||
|
||||
def test_uninspectable_owner_resolves_to_none(self, tmp_path: Path) -> None:
|
||||
# Another user's listener: occupied, but not attributable. Callers must
|
||||
# read this as "cannot identify", never as "free".
|
||||
proc = make_proc(tmp_path)
|
||||
write_net_tcp(proc, [tcp_row(_LOCAL_V4, 4242)])
|
||||
add_opaque_pid(proc, 100)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.find_port_owner(_HOST, _PORT) is None
|
||||
|
||||
|
||||
class TestReadCmdline:
|
||||
"""Tests for argv reading."""
|
||||
|
||||
def test_returns_argv(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(proc, 100, cmdline=["python", "-m", "steam_backlog_enforcer.main"])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.read_cmdline(100)[0] == "python"
|
||||
|
||||
def test_missing_process_is_empty(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.read_cmdline(4321) == []
|
||||
|
||||
|
||||
class TestIsOurServer:
|
||||
"""Tests for recognising one of our own serve processes."""
|
||||
|
||||
def test_our_serve_process_matches(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(
|
||||
proc, 100, cmdline=["python", "-m", "steam_backlog_enforcer.main", "serve"]
|
||||
)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.is_our_server(100) is True
|
||||
|
||||
def test_other_subcommand_does_not_match(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(
|
||||
proc, 100, cmdline=["python", "-m", "steam_backlog_enforcer.main", "status"]
|
||||
)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.is_our_server(100) is False
|
||||
|
||||
def test_unrelated_process_does_not_match(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(proc, 100, cmdline=["nginx", "serve"])
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.is_our_server(100) is False
|
||||
225
steam_backlog_enforcer/tests/test_serve_guard_lifecycle.py
Normal file
225
steam_backlog_enforcer/tests/test_serve_guard_lifecycle.py
Normal file
@ -0,0 +1,225 @@
|
||||
"""Tests for _serve_guard's staleness check and process termination.
|
||||
|
||||
The other half of the guard's tests lives in test_serve_guard.py; this file was
|
||||
split off to stay inside the 250-line cap.
|
||||
|
||||
``os.kill`` is mocked in every termination test, and the one test that passes a
|
||||
real pid passes this process's own — which the guard is required to refuse.
|
||||
No test here can signal anything.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import os
|
||||
import signal
|
||||
import socket
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer import _serve_guard
|
||||
from steam_backlog_enforcer.tests._fake_proc import add_pid, make_proc, patch_proc
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
_HOST = "127.0.0.1"
|
||||
|
||||
|
||||
class TestProcessStartedAt:
|
||||
"""Tests for reading a process start time."""
|
||||
|
||||
def test_returns_directory_mtime(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
pid_dir = add_pid(proc, 100)
|
||||
os.utime(pid_dir, (1000.0, 1000.0))
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.process_started_at(100) == 1000.0
|
||||
|
||||
def test_missing_process_returns_none(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard.process_started_at(4321) is None
|
||||
|
||||
|
||||
class TestNewestPySince:
|
||||
"""Tests for deciding whether a running server loaded outdated code."""
|
||||
|
||||
def test_no_newer_file_returns_none(self, tmp_path: Path) -> None:
|
||||
(tmp_path / "a.py").write_text("x", encoding="utf-8")
|
||||
os.utime(tmp_path / "a.py", (1000.0, 1000.0))
|
||||
with patch.object(_serve_guard, "_PACKAGE_ROOT", tmp_path):
|
||||
assert _serve_guard.newest_py_since(2000.0) is None
|
||||
|
||||
def test_newer_file_is_named(self, tmp_path: Path) -> None:
|
||||
# The path is returned rather than a bool so the restart message can
|
||||
# name the file: killing a server is always auditable.
|
||||
nested = tmp_path / "pkg"
|
||||
nested.mkdir()
|
||||
(nested / "b.py").write_text("x", encoding="utf-8")
|
||||
os.utime(nested / "b.py", (3000.0, 3000.0))
|
||||
with patch.object(_serve_guard, "_PACKAGE_ROOT", tmp_path):
|
||||
assert _serve_guard.newest_py_since(2000.0) == nested / "b.py"
|
||||
|
||||
def test_the_newest_of_several_wins(self, tmp_path: Path) -> None:
|
||||
for name, mtime in (("a.py", 3000.0), ("b.py", 4000.0), ("c.py", 3500.0)):
|
||||
(tmp_path / name).write_text("x", encoding="utf-8")
|
||||
os.utime(tmp_path / name, (mtime, mtime))
|
||||
with patch.object(_serve_guard, "_PACKAGE_ROOT", tmp_path):
|
||||
assert _serve_guard.newest_py_since(2000.0) == tmp_path / "b.py"
|
||||
|
||||
def test_vanished_file_is_skipped(self, tmp_path: Path) -> None:
|
||||
(tmp_path / "gone.py").symlink_to(tmp_path / "missing.py")
|
||||
with patch.object(_serve_guard, "_PACKAGE_ROOT", tmp_path):
|
||||
assert _serve_guard.newest_py_since(1000.0) is None
|
||||
|
||||
|
||||
class TestReadPpid:
|
||||
"""Tests for parsing the parent pid out of /proc/<pid>/stat."""
|
||||
|
||||
def test_reads_ppid_past_a_comm_with_spaces(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(proc, 100, ppid=42)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._read_ppid(100) == 42
|
||||
|
||||
def test_missing_stat_returns_none(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(proc, 100)
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._read_ppid(100) is None
|
||||
|
||||
def test_truncated_stat_returns_none(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(proc, 100)
|
||||
(proc / "100" / "stat").write_text("100 (py) S", encoding="utf-8")
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._read_ppid(100) is None
|
||||
|
||||
def test_non_numeric_ppid_returns_none(self, tmp_path: Path) -> None:
|
||||
proc = make_proc(tmp_path)
|
||||
add_pid(proc, 100)
|
||||
(proc / "100" / "stat").write_text("100 (py) S bogus 1", encoding="utf-8")
|
||||
with patch_proc(proc):
|
||||
assert _serve_guard._read_ppid(100) is None
|
||||
|
||||
|
||||
class TestOwnPidChain:
|
||||
"""Tests for the self-protection ancestry walk."""
|
||||
|
||||
def test_includes_self_and_ancestors(self) -> None:
|
||||
with patch.object(_serve_guard, "_read_ppid", side_effect=[7, None]):
|
||||
chain = _serve_guard._own_pid_chain()
|
||||
assert os.getpid() in chain
|
||||
assert 7 in chain
|
||||
|
||||
def test_stops_at_pid_zero(self) -> None:
|
||||
# init's parent is 0, which is not a process and must not be walked.
|
||||
with patch.object(_serve_guard, "_read_ppid", return_value=0):
|
||||
chain = _serve_guard._own_pid_chain()
|
||||
assert chain == {os.getpid()}
|
||||
|
||||
def test_a_cycle_terminates(self) -> None:
|
||||
# A stat file claiming a process is its own parent must not hang the
|
||||
# walk; the guard runs before the server starts, so a spin is a freeze.
|
||||
with patch.object(_serve_guard, "_read_ppid", return_value=os.getpid()):
|
||||
assert _serve_guard._own_pid_chain() == {os.getpid()}
|
||||
|
||||
|
||||
class TestPortIsFree:
|
||||
"""Tests for the real bind probe."""
|
||||
|
||||
def test_unused_port_is_free(self) -> None:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
|
||||
probe.bind((_HOST, 0))
|
||||
port = probe.getsockname()[1]
|
||||
assert _serve_guard.port_is_free(_HOST, port) is True
|
||||
|
||||
def test_bound_port_is_not_free(self) -> None:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as held:
|
||||
held.bind((_HOST, 0))
|
||||
held.listen(1)
|
||||
port = held.getsockname()[1]
|
||||
assert _serve_guard.port_is_free(_HOST, port) is False
|
||||
|
||||
|
||||
class TestTerminate:
|
||||
"""Tests for stopping a stale server.
|
||||
|
||||
``_serve_guard.os`` is replaced wholesale in every test, so no signal can
|
||||
reach a real process even if the guard's own protections were removed.
|
||||
"""
|
||||
|
||||
def test_refuses_to_signal_self(self) -> None:
|
||||
# The process running this check matches is_our_server by construction,
|
||||
# so a wrong owner lookup would otherwise have it kill itself.
|
||||
with (
|
||||
patch.object(_serve_guard, "os") as fake_os,
|
||||
patch.object(_serve_guard, "_read_ppid", return_value=None),
|
||||
):
|
||||
fake_os.getpid.return_value = os.getpid()
|
||||
assert _serve_guard.terminate(os.getpid(), _HOST, 8000) is False
|
||||
fake_os.kill.assert_not_called()
|
||||
|
||||
def test_sigterm_frees_the_port(self) -> None:
|
||||
with (
|
||||
patch.object(_serve_guard, "_own_pid_chain", return_value=set()),
|
||||
patch.object(_serve_guard, "_TERM_WAIT_SECONDS", 0.05),
|
||||
patch.object(_serve_guard, "_POLL_INTERVAL", 0.0),
|
||||
patch.object(_serve_guard, "port_is_free", return_value=True),
|
||||
patch.object(_serve_guard, "os") as fake_os,
|
||||
):
|
||||
assert _serve_guard.terminate(999, _HOST, 8000) is True
|
||||
fake_os.kill.assert_called_once_with(999, signal.SIGTERM)
|
||||
|
||||
def test_escalates_to_sigkill(self) -> None:
|
||||
# The port frees only once SIGKILL lands, driven off the recorded
|
||||
# signals rather than a call count, so the SIGTERM window's length
|
||||
# cannot make this flaky.
|
||||
sent: list[int] = []
|
||||
|
||||
def freed(_host: str, _port: int) -> bool:
|
||||
return signal.SIGKILL in sent
|
||||
|
||||
with (
|
||||
patch.object(_serve_guard, "_own_pid_chain", return_value=set()),
|
||||
patch.object(_serve_guard, "_TERM_WAIT_SECONDS", 0.01),
|
||||
patch.object(_serve_guard, "_POLL_INTERVAL", 0.0),
|
||||
patch.object(_serve_guard, "port_is_free", side_effect=freed),
|
||||
patch.object(_serve_guard, "os") as fake_os,
|
||||
):
|
||||
fake_os.kill.side_effect = lambda _pid, sig: sent.append(sig)
|
||||
assert _serve_guard.terminate(999, _HOST, 8000) is True
|
||||
assert sent == [signal.SIGTERM, signal.SIGKILL]
|
||||
|
||||
def test_gives_up_when_the_port_never_frees(self) -> None:
|
||||
with (
|
||||
patch.object(_serve_guard, "_own_pid_chain", return_value=set()),
|
||||
patch.object(_serve_guard, "_TERM_WAIT_SECONDS", 0.0),
|
||||
patch.object(_serve_guard, "_POLL_INTERVAL", 0.0),
|
||||
patch.object(_serve_guard, "port_is_free", return_value=False),
|
||||
patch.object(_serve_guard, "os"),
|
||||
):
|
||||
assert _serve_guard.terminate(999, _HOST, 8000) is False
|
||||
|
||||
def test_already_exited_process_is_not_an_error(self) -> None:
|
||||
with (
|
||||
patch.object(_serve_guard, "_own_pid_chain", return_value=set()),
|
||||
patch.object(_serve_guard, "_TERM_WAIT_SECONDS", 0.0),
|
||||
patch.object(_serve_guard, "_POLL_INTERVAL", 0.0),
|
||||
patch.object(_serve_guard, "port_is_free", return_value=True),
|
||||
patch.object(_serve_guard, "os") as fake_os,
|
||||
):
|
||||
fake_os.kill.side_effect = ProcessLookupError
|
||||
assert _serve_guard.terminate(999, _HOST, 8000) is True
|
||||
|
||||
def test_permission_denied_is_not_an_error(self) -> None:
|
||||
with (
|
||||
patch.object(_serve_guard, "_own_pid_chain", return_value=set()),
|
||||
patch.object(_serve_guard, "_TERM_WAIT_SECONDS", 0.0),
|
||||
patch.object(_serve_guard, "_POLL_INTERVAL", 0.0),
|
||||
patch.object(_serve_guard, "port_is_free", return_value=False),
|
||||
patch.object(_serve_guard, "os") as fake_os,
|
||||
):
|
||||
fake_os.kill.side_effect = PermissionError(errno.EPERM, "not yours")
|
||||
assert _serve_guard.terminate(999, _HOST, 8000) is False
|
||||
198
steam_backlog_enforcer/tests/test_serve_startup.py
Normal file
198
steam_backlog_enforcer/tests/test_serve_startup.py
Normal file
@ -0,0 +1,198 @@
|
||||
"""Tests for _serve_startup — 100% branch coverage.
|
||||
|
||||
The guard functions are imported into this module's namespace, so every patch
|
||||
targets ``_serve_startup.<name>``: patching ``_serve_guard`` would not bite.
|
||||
No test here signals a real process — ``terminate`` is always a mock.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from steam_backlog_enforcer import _serve_startup
|
||||
from steam_backlog_enforcer._web_server import DEFAULT_HOST, DEFAULT_PORT
|
||||
|
||||
_PKG = "steam_backlog_enforcer._serve_startup"
|
||||
|
||||
|
||||
def _exits_1(argv: list[str]) -> None:
|
||||
"""Assert that parsing *argv* fails with exit code 1."""
|
||||
with patch(f"{_PKG}._echo"), pytest.raises(SystemExit) as exc:
|
||||
_serve_startup.parse_serve_args(argv)
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
class TestParseServeArgs:
|
||||
"""Tests for argument parsing."""
|
||||
|
||||
def test_defaults_when_no_flags(self) -> None:
|
||||
assert _serve_startup.parse_serve_args([]) == (DEFAULT_HOST, DEFAULT_PORT)
|
||||
|
||||
def test_port_flag(self) -> None:
|
||||
assert _serve_startup.parse_serve_args(["--port", "8123"]) == (
|
||||
DEFAULT_HOST,
|
||||
8123,
|
||||
)
|
||||
|
||||
def test_host_flag_accepts_literal_loopback(self) -> None:
|
||||
assert _serve_startup.parse_serve_args(["--host", "127.0.0.2"]) == (
|
||||
"127.0.0.2",
|
||||
DEFAULT_PORT,
|
||||
)
|
||||
|
||||
def test_host_alias_canonicalises(self) -> None:
|
||||
host, _ = _serve_startup.parse_serve_args(["--host", "localhost"])
|
||||
assert host == DEFAULT_HOST
|
||||
|
||||
def test_both_flags_together(self) -> None:
|
||||
assert _serve_startup.parse_serve_args(
|
||||
["--host", "localhost", "--port", "9000"]
|
||||
) == (DEFAULT_HOST, 9000)
|
||||
|
||||
def test_unknown_argument_exits(self) -> None:
|
||||
_exits_1(["--wat"])
|
||||
|
||||
def test_flag_without_value_exits(self) -> None:
|
||||
_exits_1(["--port"])
|
||||
|
||||
def test_non_numeric_port_exits(self) -> None:
|
||||
_exits_1(["--port", "eight"])
|
||||
|
||||
def test_privileged_port_exits(self) -> None:
|
||||
_exits_1(["--port", "80"])
|
||||
|
||||
def test_port_above_range_exits(self) -> None:
|
||||
_exits_1(["--port", "99999"])
|
||||
|
||||
def test_non_address_host_exits(self) -> None:
|
||||
_exits_1(["--host", "example.com"])
|
||||
|
||||
def test_routable_host_exits(self) -> None:
|
||||
# The dataset API is unauthenticated; a bind must never leave the box.
|
||||
_exits_1(["--host", "8.8.8.8"])
|
||||
|
||||
def test_ipv6_loopback_exits(self) -> None:
|
||||
# ThreadingHTTPServer is AF_INET, so ::1 would always crash at bind.
|
||||
_exits_1(["--host", "::1"])
|
||||
|
||||
|
||||
class TestDescribe:
|
||||
"""Tests for the pid description helper."""
|
||||
|
||||
def test_includes_argv_when_readable(self) -> None:
|
||||
with patch(f"{_PKG}.read_cmdline", return_value=["python", "-m", "x"]):
|
||||
assert _serve_startup._describe(42) == "pid 42 (python -m x)"
|
||||
|
||||
def test_falls_back_to_bare_pid(self) -> None:
|
||||
with patch(f"{_PKG}.read_cmdline", return_value=[]):
|
||||
assert _serve_startup._describe(42) == "pid 42"
|
||||
|
||||
|
||||
class TestEnsurePortAvailable:
|
||||
"""Tests for the start-or-defer decision."""
|
||||
|
||||
def test_free_port_returns(self) -> None:
|
||||
# Returning normally *is* the assertion: every other path exits.
|
||||
with patch(f"{_PKG}.port_is_free", return_value=True):
|
||||
_serve_startup.ensure_port_available(DEFAULT_HOST, 8000)
|
||||
|
||||
def test_uninspectable_owner_exits_1(self) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.port_is_free", return_value=False),
|
||||
patch(f"{_PKG}.find_port_owner", return_value=None),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
_serve_startup.ensure_port_available(DEFAULT_HOST, 8000)
|
||||
assert exc.value.code == 1
|
||||
assert "cannot inspect" in echo.call_args_list[0][0][0]
|
||||
|
||||
def test_foreign_owner_exits_1_and_is_named(self) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.port_is_free", return_value=False),
|
||||
patch(f"{_PKG}.find_port_owner", return_value=99),
|
||||
patch(f"{_PKG}.is_our_server", return_value=False),
|
||||
patch(f"{_PKG}.read_cmdline", return_value=["nginx"]),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
_serve_startup.ensure_port_available(DEFAULT_HOST, 8000)
|
||||
assert exc.value.code == 1
|
||||
assert "pid 99 (nginx)" in echo.call_args_list[0][0][0]
|
||||
|
||||
def test_our_server_delegates_to_replace_or_defer(self) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.port_is_free", return_value=False),
|
||||
patch(f"{_PKG}.find_port_owner", return_value=99),
|
||||
patch(f"{_PKG}.is_our_server", return_value=True),
|
||||
patch(f"{_PKG}._replace_or_defer") as replace,
|
||||
):
|
||||
_serve_startup.ensure_port_available(DEFAULT_HOST, 8000)
|
||||
replace.assert_called_once_with(DEFAULT_HOST, 8000, 99)
|
||||
|
||||
|
||||
class TestReplaceOrDefer:
|
||||
"""Tests for the stale-or-current decision about our own server."""
|
||||
|
||||
def test_current_server_exits_0_with_url(self) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.process_started_at", return_value=1000.0),
|
||||
patch(f"{_PKG}.newest_py_since", return_value=None),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
_serve_startup._replace_or_defer(DEFAULT_HOST, 8000, 99)
|
||||
assert exc.value.code == 0
|
||||
assert f"http://{DEFAULT_HOST}:8000" in echo.call_args_list[0][0][0]
|
||||
|
||||
def test_stale_server_is_terminated(self, tmp_path: object) -> None:
|
||||
newer = "steam_backlog_enforcer/main/misc.py"
|
||||
with (
|
||||
patch(f"{_PKG}.process_started_at", return_value=1000.0),
|
||||
patch(f"{_PKG}.newest_py_since", return_value=newer),
|
||||
patch(f"{_PKG}.is_our_server", return_value=True),
|
||||
patch(f"{_PKG}.terminate", return_value=True) as term,
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
):
|
||||
_serve_startup._replace_or_defer(DEFAULT_HOST, 8000, 99)
|
||||
term.assert_called_once_with(99, DEFAULT_HOST, 8000)
|
||||
assert f"{newer} is newer" in echo.call_args_list[0][0][0]
|
||||
|
||||
def test_unreadable_start_time_restarts(self) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.process_started_at", return_value=None),
|
||||
patch(f"{_PKG}.is_our_server", return_value=True),
|
||||
patch(f"{_PKG}.terminate", return_value=True),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
):
|
||||
_serve_startup._replace_or_defer(DEFAULT_HOST, 8000, 99)
|
||||
assert "start time unreadable" in echo.call_args_list[0][0][0]
|
||||
|
||||
def test_recycled_pid_is_never_signalled(self) -> None:
|
||||
# Between the port scan and here the process can exit and its pid be
|
||||
# reused. Signalling on the stale answer would kill a bystander.
|
||||
with (
|
||||
patch(f"{_PKG}.process_started_at", return_value=None),
|
||||
patch(f"{_PKG}.is_our_server", return_value=False),
|
||||
patch(f"{_PKG}.read_cmdline", return_value=["someone-else"]),
|
||||
patch(f"{_PKG}.terminate") as term,
|
||||
patch(f"{_PKG}._echo"),
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
_serve_startup._replace_or_defer(DEFAULT_HOST, 8000, 99)
|
||||
assert exc.value.code == 1
|
||||
term.assert_not_called()
|
||||
|
||||
def test_failed_termination_exits_1(self) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.process_started_at", return_value=None),
|
||||
patch(f"{_PKG}.is_our_server", return_value=True),
|
||||
patch(f"{_PKG}.terminate", return_value=False),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
pytest.raises(SystemExit) as exc,
|
||||
):
|
||||
_serve_startup._replace_or_defer(DEFAULT_HOST, 8000, 99)
|
||||
assert exc.value.code == 1
|
||||
assert "Could not free port 8000" in echo.call_args_list[-1][0][0]
|
||||
158
steam_backlog_enforcer/tests/test_web_build.py
Normal file
158
steam_backlog_enforcer/tests/test_web_build.py
Normal file
@ -0,0 +1,158 @@
|
||||
"""Tests for _web_build — 100% branch coverage.
|
||||
|
||||
Every module-level path constant is derived from the real repo ``web/`` at
|
||||
import time, so each test redirects all of them into ``tmp_path``. Patching
|
||||
only ``WEB_DIR`` would leave ``_BUILT_MARKER`` and friends pointing at the
|
||||
working tree, and the tests would silently grade the developer's own bundle.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from steam_backlog_enforcer import _web_build
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
_PKG = "steam_backlog_enforcer._web_build"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def web(tmp_path: Path) -> Iterator[Path]:
|
||||
"""Redirect the whole module's path constants into a fake web/ tree."""
|
||||
root = tmp_path / "web"
|
||||
(root / "src").mkdir(parents=True)
|
||||
(root / "dist").mkdir()
|
||||
with (
|
||||
patch.object(_web_build, "WEB_DIR", root),
|
||||
patch.object(_web_build, "_BUILT_MARKER", root / "dist" / "index.html"),
|
||||
patch.object(_web_build, "_SOURCE_DIR", root / "src"),
|
||||
patch.object(
|
||||
_web_build,
|
||||
"_SOURCE_FILES",
|
||||
(root / "index.html", root / "package.json"),
|
||||
),
|
||||
):
|
||||
yield root
|
||||
|
||||
|
||||
def _touch(path: Path, mtime: float) -> None:
|
||||
"""Create *path* with a fixed mtime."""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("x", encoding="utf-8")
|
||||
import os
|
||||
|
||||
os.utime(path, (mtime, mtime))
|
||||
|
||||
|
||||
class TestMtime:
|
||||
"""Tests for the _mtime helper."""
|
||||
|
||||
def test_missing_path_is_zero(self, tmp_path: Path) -> None:
|
||||
assert _web_build._mtime(tmp_path / "nope") == 0.0
|
||||
|
||||
def test_existing_path_returns_mtime(self, tmp_path: Path) -> None:
|
||||
target = tmp_path / "f"
|
||||
_touch(target, 1000.0)
|
||||
assert _web_build._mtime(target) == pytest.approx(1000.0)
|
||||
|
||||
|
||||
class TestFrontendIsStale:
|
||||
"""Tests for frontend_is_stale."""
|
||||
|
||||
def test_unbuilt_but_present_web_dir_is_stale(self, web: Path) -> None:
|
||||
# No dist/index.html at all: a fresh clone must build, not serve the
|
||||
# "not built" placeholder.
|
||||
assert web.is_dir()
|
||||
assert _web_build.frontend_is_stale() is True
|
||||
|
||||
def test_missing_web_dir_is_not_stale(self, web: Path) -> None:
|
||||
# Nothing to build from — claiming staleness would make serve exit 1
|
||||
# on an install that legitimately ships no frontend sources.
|
||||
with patch.object(_web_build, "WEB_DIR", web / "absent"):
|
||||
assert _web_build.frontend_is_stale() is False
|
||||
|
||||
def test_newer_source_file_is_stale(self, web: Path) -> None:
|
||||
_touch(web / "dist" / "index.html", 1000.0)
|
||||
_touch(web / "package.json", 2000.0)
|
||||
assert _web_build.frontend_is_stale() is True
|
||||
|
||||
def test_newer_src_tree_entry_is_stale(self, web: Path) -> None:
|
||||
_touch(web / "dist" / "index.html", 1000.0)
|
||||
_touch(web / "src" / "nested" / "app.tsx", 2000.0)
|
||||
assert _web_build.frontend_is_stale() is True
|
||||
|
||||
def test_all_sources_older_is_current(self, web: Path) -> None:
|
||||
_touch(web / "package.json", 1000.0)
|
||||
_touch(web / "src" / "app.tsx", 1000.0)
|
||||
_touch(web / "dist" / "index.html", 2000.0)
|
||||
assert _web_build.frontend_is_stale() is False
|
||||
|
||||
def test_directories_in_src_are_skipped(self, web: Path) -> None:
|
||||
# A directory's own mtime bumps whenever a file inside it is added or
|
||||
# removed, so counting it would report staleness after a build.
|
||||
_touch(web / "src" / "old.tsx", 1000.0)
|
||||
_touch(web / "dist" / "index.html", 2000.0)
|
||||
(web / "src" / "assets").mkdir()
|
||||
assert _web_build.frontend_is_stale() is False
|
||||
|
||||
def test_missing_src_dir_still_checks_flat_sources(self, web: Path) -> None:
|
||||
_touch(web / "dist" / "index.html", 1000.0)
|
||||
_touch(web / "index.html", 2000.0)
|
||||
with patch.object(_web_build, "_SOURCE_DIR", web / "absent"):
|
||||
assert _web_build.frontend_is_stale() is True
|
||||
|
||||
|
||||
class TestBuildFrontend:
|
||||
"""Tests for build_frontend. subprocess is never actually executed."""
|
||||
|
||||
def test_missing_npm_explains_and_fails(self, web: Path) -> None:
|
||||
with (
|
||||
patch(f"{_PKG}.shutil.which", return_value=None),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
):
|
||||
assert _web_build.build_frontend() is False
|
||||
assert "npm was not found" in echo.call_args[0][0]
|
||||
|
||||
def test_successful_build_returns_true(self, web: Path) -> None:
|
||||
run = MagicMock(return_value=MagicMock(returncode=0))
|
||||
with (
|
||||
patch(f"{_PKG}.shutil.which", return_value="/usr/bin/npm"),
|
||||
patch(f"{_PKG}.subprocess.run", run),
|
||||
patch(f"{_PKG}._echo"),
|
||||
):
|
||||
assert _web_build.build_frontend() is True
|
||||
assert run.call_args[0][0] == ["/usr/bin/npm", "run", "build"]
|
||||
assert run.call_args[1]["cwd"] == web
|
||||
|
||||
def test_failed_build_prints_output_and_returns_false(self, web: Path) -> None:
|
||||
result = MagicMock(returncode=1, stdout=" out \n", stderr=" err \n")
|
||||
with (
|
||||
patch(f"{_PKG}.shutil.which", return_value="/usr/bin/npm"),
|
||||
patch(f"{_PKG}.subprocess.run", return_value=result),
|
||||
patch(f"{_PKG}._echo") as echo,
|
||||
):
|
||||
assert _web_build.build_frontend() is False
|
||||
printed = [call[0][0] for call in echo.call_args_list]
|
||||
assert any("FAILED" in line for line in printed)
|
||||
assert "out" in printed
|
||||
assert "err" in printed
|
||||
|
||||
def test_timeout_propagates(self, web: Path) -> None:
|
||||
# A hung build must not be swallowed into "bundle is fine".
|
||||
with (
|
||||
patch(f"{_PKG}.shutil.which", return_value="/usr/bin/npm"),
|
||||
patch(
|
||||
f"{_PKG}.subprocess.run",
|
||||
side_effect=subprocess.TimeoutExpired("npm", 1),
|
||||
),
|
||||
patch(f"{_PKG}._echo"),
|
||||
pytest.raises(subprocess.TimeoutExpired),
|
||||
):
|
||||
_web_build.build_frontend()
|
||||
@ -9,10 +9,7 @@ import threading
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from steam_backlog_enforcer import main as main_mod
|
||||
from steam_backlog_enforcer._web_server import create_server, serve
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
from steam_backlog_enforcer.main import misc as misc_mod
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
@ -206,15 +203,3 @@ class TestServe:
|
||||
):
|
||||
serve()
|
||||
fake.server_close.assert_called_once()
|
||||
|
||||
|
||||
class TestCmdServe:
|
||||
"""Tests for the main.cmd_serve wiring."""
|
||||
|
||||
def test_invokes_serve(self) -> None:
|
||||
# Patch the module that *defines* cmd_serve, not the main package that
|
||||
# re-exports it: cmd_serve resolves ``serve`` from its own globals, so
|
||||
# patching the re-export would not bite.
|
||||
with patch.object(misc_mod, "serve") as mock_serve:
|
||||
main_mod.cmd_serve(Config(), State())
|
||||
mock_serve.assert_called_once()
|
||||
|
||||
@ -69,7 +69,9 @@ describe('useBudgetPoll', () => {
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('does not poll while the tab is hidden', async () => {
|
||||
// Browser automation always drives a hidden tab, so this is the case that
|
||||
// used to leave the panel stuck on its loading text forever.
|
||||
it('polls while the tab is hidden', async () => {
|
||||
vi.useFakeTimers()
|
||||
setHidden(true)
|
||||
const fetchMock = vi
|
||||
@ -77,10 +79,11 @@ describe('useBudgetPoll', () => {
|
||||
.mockResolvedValue({ ok: true, json: async () => makeBudget() })
|
||||
vi.stubGlobal('fetch', fetchMock)
|
||||
renderHook(() => useBudgetPoll(false, 1000))
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1)
|
||||
await act(async () => {
|
||||
vi.advanceTimersByTime(5000)
|
||||
vi.advanceTimersByTime(2000)
|
||||
})
|
||||
expect(fetchMock).not.toHaveBeenCalled()
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('stops polling after unmount', async () => {
|
||||
|
||||
@ -11,11 +11,15 @@ export interface BudgetPollState {
|
||||
export const POLL_MS = 5000
|
||||
|
||||
/**
|
||||
* Poll `/api/budget` while mounted, pausing whenever the tab is hidden.
|
||||
* Poll `/api/budget` for as long as the panel is mounted.
|
||||
*
|
||||
* The hook is only mounted by the budget tab, so switching away stops the
|
||||
* requests entirely; the `document.hidden` check covers the other case, a tab
|
||||
* left open in the background for hours.
|
||||
* Visibility is deliberately not consulted. A `document.hidden` guard used to
|
||||
* skip the request, but it also swallowed the very first load, so a panel that
|
||||
* mounted in a background tab sat on "Reading budget state…" with no data and
|
||||
* no error. Every tab driven by browser automation is hidden, which made the
|
||||
* panel unreadable that way. Switching tabs already unmounts the hook, so the
|
||||
* only cost of dropping the guard is one request per interval in a background
|
||||
* tab left open.
|
||||
*/
|
||||
export function useBudgetPoll(demo: boolean, intervalMs: number = POLL_MS): BudgetPollState {
|
||||
const [data, setData] = useState<BudgetSnapshot | null>(null)
|
||||
@ -25,7 +29,6 @@ export function useBudgetPoll(demo: boolean, intervalMs: number = POLL_MS): Budg
|
||||
let cancelled = false
|
||||
|
||||
const load = () => {
|
||||
if (document.hidden) return
|
||||
fetchBudget(demo)
|
||||
.then((snapshot) => {
|
||||
// A response can arrive after unmount; committing it would warn and,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user