From 47f77f5f1ece9342c7b321993b1e248d1d00b15b Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 29 Aug 2026 18:52:11 +0200 Subject: [PATCH] Make `serve` re-runnable, and cover it to 100% MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) Claude-Session: https://claude.ai/code/session_01MKFzwzwRU1JD7SFLHtQBsz --- steam_backlog_enforcer/_serve_guard.py | 248 ++++++++++++++++++ steam_backlog_enforcer/_serve_startup.py | 151 +++++++++++ steam_backlog_enforcer/_web_build.py | 94 +++++++ steam_backlog_enforcer/main/__init__.py | 5 +- steam_backlog_enforcer/main/misc.py | 35 ++- steam_backlog_enforcer/tests/_fake_proc.py | 108 ++++++++ .../tests/_no_subprocess.py | 6 + .../tests/test_cmd_serve.py | 99 +++++++ .../tests/test_main_dispatch_extra.py | 15 ++ .../tests/test_serve_guard.py | 222 ++++++++++++++++ .../tests/test_serve_guard_lifecycle.py | 225 ++++++++++++++++ .../tests/test_serve_startup.py | 198 ++++++++++++++ .../tests/test_web_build.py | 158 +++++++++++ .../tests/test_web_server.py | 15 -- web/src/useBudgetPoll.test.ts | 9 +- web/src/useBudgetPoll.ts | 13 +- 16 files changed, 1574 insertions(+), 27 deletions(-) create mode 100644 steam_backlog_enforcer/_serve_guard.py create mode 100644 steam_backlog_enforcer/_serve_startup.py create mode 100644 steam_backlog_enforcer/_web_build.py create mode 100644 steam_backlog_enforcer/tests/_fake_proc.py create mode 100644 steam_backlog_enforcer/tests/test_cmd_serve.py create mode 100644 steam_backlog_enforcer/tests/test_serve_guard.py create mode 100644 steam_backlog_enforcer/tests/test_serve_guard_lifecycle.py create mode 100644 steam_backlog_enforcer/tests/test_serve_startup.py create mode 100644 steam_backlog_enforcer/tests/test_web_build.py diff --git a/steam_backlog_enforcer/_serve_guard.py b/steam_backlog_enforcer/_serve_guard.py new file mode 100644 index 0000000..02162bd --- /dev/null +++ b/steam_backlog_enforcer/_serve_guard.py @@ -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//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/`` 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//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) diff --git a/steam_backlog_enforcer/_serve_startup.py b/steam_backlog_enforcer/_serve_startup.py new file mode 100644 index 0000000..99f6a97 --- /dev/null +++ b/steam_backlog_enforcer/_serve_startup.py @@ -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) diff --git a/steam_backlog_enforcer/_web_build.py b/steam_backlog_enforcer/_web_build.py new file mode 100644 index 0000000..3925d64 --- /dev/null +++ b/steam_backlog_enforcer/_web_build.py @@ -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 diff --git a/steam_backlog_enforcer/main/__init__.py b/steam_backlog_enforcer/main/__init__.py index c9b03ed..89a24b8 100644 --- a/steam_backlog_enforcer/main/__init__.py +++ b/steam_backlog_enforcer/main/__init__.py @@ -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, 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 diff --git a/steam_backlog_enforcer/main/misc.py b/steam_backlog_enforcer/main/misc.py index 2bcdabb..3535a77 100644 --- a/steam_backlog_enforcer/main/misc.py +++ b/steam_backlog_enforcer/main/misc.py @@ -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: diff --git a/steam_backlog_enforcer/tests/_fake_proc.py b/steam_backlog_enforcer/tests/_fake_proc.py new file mode 100644 index 0000000..91e43bb --- /dev/null +++ b/steam_backlog_enforcer/tests/_fake_proc.py @@ -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/`` 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/`` 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 diff --git a/steam_backlog_enforcer/tests/_no_subprocess.py b/steam_backlog_enforcer/tests/_no_subprocess.py index 4bd8453..858eb26 100644 --- a/steam_backlog_enforcer/tests/_no_subprocess.py +++ b/steam_backlog_enforcer/tests/_no_subprocess.py @@ -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 diff --git a/steam_backlog_enforcer/tests/test_cmd_serve.py b/steam_backlog_enforcer/tests/test_cmd_serve.py new file mode 100644 index 0000000..ba50800 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_cmd_serve.py @@ -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([]) diff --git a/steam_backlog_enforcer/tests/test_main_dispatch_extra.py b/steam_backlog_enforcer/tests/test_main_dispatch_extra.py index 422a7fc..9634270 100644 --- a/steam_backlog_enforcer/tests/test_main_dispatch_extra.py +++ b/steam_backlog_enforcer/tests/test_main_dispatch_extra.py @@ -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"] diff --git a/steam_backlog_enforcer/tests/test_serve_guard.py b/steam_backlog_enforcer/tests/test_serve_guard.py new file mode 100644 index 0000000..afbdc3d --- /dev/null +++ b/steam_backlog_enforcer/tests/test_serve_guard.py @@ -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 diff --git a/steam_backlog_enforcer/tests/test_serve_guard_lifecycle.py b/steam_backlog_enforcer/tests/test_serve_guard_lifecycle.py new file mode 100644 index 0000000..dbd6567 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_serve_guard_lifecycle.py @@ -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//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 diff --git a/steam_backlog_enforcer/tests/test_serve_startup.py b/steam_backlog_enforcer/tests/test_serve_startup.py new file mode 100644 index 0000000..98bdcdb --- /dev/null +++ b/steam_backlog_enforcer/tests/test_serve_startup.py @@ -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.``: 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] diff --git a/steam_backlog_enforcer/tests/test_web_build.py b/steam_backlog_enforcer/tests/test_web_build.py new file mode 100644 index 0000000..249142e --- /dev/null +++ b/steam_backlog_enforcer/tests/test_web_build.py @@ -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() diff --git a/steam_backlog_enforcer/tests/test_web_server.py b/steam_backlog_enforcer/tests/test_web_server.py index 94bc964..a6d82c6 100644 --- a/steam_backlog_enforcer/tests/test_web_server.py +++ b/steam_backlog_enforcer/tests/test_web_server.py @@ -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() diff --git a/web/src/useBudgetPoll.test.ts b/web/src/useBudgetPoll.test.ts index 4492847..72c1ee8 100644 --- a/web/src/useBudgetPoll.test.ts +++ b/web/src/useBudgetPoll.test.ts @@ -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 () => { diff --git a/web/src/useBudgetPoll.ts b/web/src/useBudgetPoll.ts index 274da7a..635647e 100644 --- a/web/src/useBudgetPoll.ts +++ b/web/src/useBudgetPoll.ts @@ -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(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,