steam-backlog-enforcer/steam_backlog_enforcer/tests/test_serve_stale.py
Krzysztof kuhy Rudnicki 433a1d6c28
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
pre-commit / pre-commit (push) Waiting to run
Tests / test (push) Waiting to run
Stop the web UI serving numbers the enforcer is not applying
A server started as `python -c "from ..._web_server import serve; serve(
port=8000)"` ran for five hours after the code beneath it changed, reporting
an 8h gaming budget while the daemon enforced 6h and had already cut gaming
off. Two separate holes let that happen.

The staleness check only ran from cmd_serve, so a server started any other
way never met it. And is_our_server matched argv containing
"steam_backlog_enforcer.main", which a `python -c` process does not have --
so that server could not even be identified as ours: `run.sh serve` would
have refused the port and named it as a stranger rather than replacing it.

_serve_stale puts the check inside the server, where no launch path can skip
it. Every route, static included, returns 503 once any package .py is newer
than this process, because a fresh page fetching stale numbers is the same
lie with extra steps. The answer is memoised for five seconds so it does not
stat the package per request; that bounds the staleness window instead of
leaving it open forever.

Refusing is only half a fix -- a process that 503s forever is honest and
useless -- so a refused request also stands the server down. The new
steam-backlog-enforcer-web.service supplies the other half with
Restart=always: it dies on stale code and returns on current code with
nobody having to remember anything. A user unit, not a system one; it serves
read-only views and needs no privilege. install.sh installs and enables it
rather than printing instructions.

is_our_server now substring-matches the joined argv, so run.sh serve,
python -m ...main serve and a bare python -c are all recognised.

newest_py_since moves into _serve_stale and _serve_guard delegates to it.
Both needed the same scan, and two copies would have been free to disagree
about whether a server is current -- which is the entire failure being fixed.

Verified end to end: healthy 200, touch _budget_view.py, next request 503,
systemd replaces the process, back to 200 reporting 6.0h and blocked=True.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Fcwu78wk17FWMHbbfbyicc
2026-08-29 23:39:40 +02:00

138 lines
4.6 KiB
Python

"""Tests for _serve_stale: a server must not outlive its own source.
On 2026-08-29 a hand-started ``python -c`` server stayed up for hours after the
code beneath it changed, reporting an 8h gaming budget while the daemon
enforced 6h. The launcher guard could not have helped: it only runs from
``cmd_serve``, which that process never went through. So the check lives in the
server, where no launch path can skip it.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from steam_backlog_enforcer._serve_stale import (
_RECHECK_INTERVAL_SECONDS,
outdated_source,
reset_cache,
)
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
_PKG = "steam_backlog_enforcer._serve_stale"
@pytest.fixture(autouse=True)
def _clear() -> Iterator[None]:
"""Keep the memo from leaking answers between tests.
Yields:
None, with the cache empty before and after.
"""
reset_cache()
yield
reset_cache()
@pytest.fixture
def fake_package(tmp_path: Path) -> Path:
"""Point the module at a throwaway package tree.
Args:
tmp_path: pytest temporary directory.
Returns:
The fake package root.
"""
root = tmp_path / "pkg"
root.mkdir()
(root / "module.py").write_text("x = 1\n", encoding="utf-8")
return root
def _touch(path: Path, mtime: float) -> None:
"""Set a file's mtime.
Args:
path: File to stamp.
mtime: Epoch seconds to set.
"""
path.write_text("x = 2\n", encoding="utf-8")
import os
os.utime(path, (mtime, mtime))
class TestOutdatedSource:
"""The check answers "has my own code moved on since I started"."""
def test_current_code_is_not_stale(self, fake_package: Path) -> None:
"""A tree older than the process start is fine."""
_touch(fake_package / "module.py", 1000.0)
with patch(f"{_PKG}._PACKAGE_ROOT", fake_package):
assert outdated_source(2000.0, now=0.0) is None
def test_a_newer_file_is_named(self, fake_package: Path) -> None:
"""The offending path is returned, so the refusal is auditable."""
changed = fake_package / "module.py"
_touch(changed, 3000.0)
with patch(f"{_PKG}._PACKAGE_ROOT", fake_package):
assert outdated_source(2000.0, now=0.0) == changed
def test_staleness_is_logged_at_error(
self,
fake_package: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Refusing to serve is loud: it needs to reach the journal."""
_touch(fake_package / "module.py", 3000.0)
with (
caplog.at_level(logging.ERROR),
patch(f"{_PKG}._PACKAGE_ROOT", fake_package),
):
outdated_source(2000.0, now=0.0)
assert "outdated code" in caplog.text
assert "run.sh serve" in caplog.text
def test_a_file_removed_mid_scan_is_skipped(self, fake_package: Path) -> None:
"""A vanished file cannot make the process stale."""
_touch(fake_package / "module.py", 3000.0)
with (
patch(f"{_PKG}._PACKAGE_ROOT", fake_package),
patch(
"steam_backlog_enforcer._serve_stale.Path.stat",
side_effect=OSError("gone"),
),
):
assert outdated_source(2000.0, now=0.0) is None
def test_the_answer_is_memoised(self, fake_package: Path) -> None:
"""Statting the whole package on every request would be wasteful."""
_touch(fake_package / "module.py", 1000.0)
with patch(f"{_PKG}._PACKAGE_ROOT", fake_package) as root:
outdated_source(2000.0, now=0.0)
# A change inside the memo window is deliberately not seen yet.
_touch(root / "module.py", 3000.0)
assert outdated_source(2000.0, now=1.0) is None
def test_the_memo_expires(self, fake_package: Path) -> None:
"""The staleness window is bounded, unlike the staleness it replaces."""
_touch(fake_package / "module.py", 1000.0)
with patch(f"{_PKG}._PACKAGE_ROOT", fake_package) as root:
outdated_source(2000.0, now=0.0)
_touch(root / "module.py", 3000.0)
later = _RECHECK_INTERVAL_SECONDS + 1
assert outdated_source(2000.0, now=later) is not None
def test_it_reads_the_clock_when_not_told_one(self, fake_package: Path) -> None:
"""The production path takes no `now`; exercise that branch too."""
_touch(fake_package / "module.py", 1000.0)
with patch(f"{_PKG}._PACKAGE_ROOT", fake_package):
assert outdated_source(2000.0) is None