kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-11-10 19:27:04 +01:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

budget: wake on screen-locker's log.json; stat stamps drop stale earner answers

Browse Source 
The enforce loop now waits on an inotify watch of log.json's directory
(_file_wake, ctypes, no new dependency) instead of a plain sleep, so a
credited workout moves the gaming budget within the tick. Earner answers
(workout, LeetCode, reading) are memoised with the source file's
(inode, mtime_ns, size) stamp, so a rewritten file re-asks at once instead
of after the 60 s TTL. Any inotify setup or read failure warns and falls
back to the plain sleep; the stamp still catches the change next tick.

Tests: conftest stubs the wake factory and points workout_log_path at a
tmp path, so no test ever watches the real screen-locker tree.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018RHwVPyiiQMPh4M3JQwQku
This commit is contained in:
Krzysztof Rudnicki 2026-10-09 18:52:58 +02:00
parent 9af0d76b4f
commit eb47067617
14 changed files with 721 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
steam_backlog_enforcer/_enforce_loop.py
View File

@ -4,7 +4,6 @@ from __future__ import annotations
import json
import logging
import time
from steam_backlog_enforcer._actions import allowed_app_ids, allowed_games
from steam_backlog_enforcer._echo import _echo
@ -12,6 +11,7 @@ from steam_backlog_enforcer._enforce_steps import (
_enforce_setup,
_reinstall_missing_allowed,
)
from steam_backlog_enforcer._file_wake import open_file_wake
# Re-exported: five modules and the test suite import get_all_owned_app_ids
# from here, so splitting it into _owned_apps_cache stays invisible to them.
@ -30,6 +30,7 @@ from steam_backlog_enforcer._total_block import (
total_block_needs_cleanup,
)
from steam_backlog_enforcer._whitelist_locking import lock_enforcement_files
from steam_backlog_enforcer._workout_budget import workout_log_path
from steam_backlog_enforcer.config import (
CONFIG_FILE,
Config,
@ -222,6 +223,8 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
# One session for the whole daemon: the audit journal needs to see the
# previous tick's record, so it cannot be rebuilt per iteration.
session = new_session(demo=demo)
# A credited workout wakes the wait, so the budget moves within the tick.
wake = open_file_wake([workout_log_path(config)])
try:
while True:
# Reload state from disk so CLI changes (e.g. new game
@ -231,7 +234,7 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
fresh = State.load()
except (json.JSONDecodeError, OSError, ValueError) as exc:
logger.warning("Failed to reload state: %s", exc)
time.sleep(ENFORCE_INTERVAL)
wake.wait(ENFORCE_INTERVAL)
continue
# Every field, not a hand-picked few: the daemon's pick sweep saves
# this object, and any field left stale here (cooldowns, pick
@ -242,6 +245,6 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
vars(state).update(vars(fresh))
_enforce_loop_iteration(config, state, session=session, demo=demo)
time.sleep(ENFORCE_INTERVAL)
wake.wait(ENFORCE_INTERVAL)
except KeyboardInterrupt:
_echo("\nEnforcer stopped.")

168
steam_backlog_enforcer/_file_wake.py Normal file
View File

@ -0,0 +1,168 @@
"""Wake the enforce loop early when a watched file is rewritten.
The loop ticks every few seconds. A workout credited on the PC should move the
gaming budget well inside one tick, so the wait between ticks returns the
moment screen-locker rewrites its ``log.json``; without an event it waits the
full interval, exactly like the plain sleep it replaces.
Watches the *directory*, filtered to the file's name, never the file itself:
writers replace these files atomically (temp file + rename), which swaps the
inode, and a watch on the file dies after the first write.
inotify through ctypes rather than a dependency -- it is three libc calls. Any
failure logs a warning and degrades to a plain sleep: the answer cache's stat
stamp still picks the change up on the next tick, so degraded means "within a
tick", never "missed".
"""
from __future__ import annotations
import ctypes
import logging
import os
import select
import struct
import time
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from collections.abc import Iterable
from pathlib import Path
logger = logging.getLogger(__name__)
# <sys/inotify.h>
_IN_CLOSE_WRITE: Final = 0x00000008
_IN_MOVED_TO: Final = 0x00000080
_IN_CREATE: Final = 0x00000100
_IN_Q_OVERFLOW: Final = 0x00004000
_IN_IGNORED: Final = 0x00008000
_IN_ONLYDIR: Final = 0x01000000
_WATCH_MASK: Final = _IN_CLOSE_WRITE | _IN_MOVED_TO | _IN_CREATE | _IN_ONLYDIR
# struct inotify_event { int wd; uint32_t mask, cookie, len; char name[]; }
_EVENT_HEADER: Final = struct.Struct("iIII")
_READ_SIZE: Final = 64 * 1024
class FileWake:
"""An interruptible inter-tick wait. Build once; never per tick.
A per-tick watcher would leak descriptors and miss every write landing
between two ticks, which is the whole window this exists to close.
"""
def __init__(self, fd: int | None, names: dict[int, frozenset[str]]) -> None:
"""Wrap an inotify ``fd`` whose watch descriptors map to file names.
Args:
fd: Non-blocking inotify descriptor, or ``None`` for plain sleep.
names: Watch descriptor -> the file names that count in that dir.
"""
self._fd = fd
self._names = names
self._poll = select.poll()
if fd is not None:
self._poll.register(fd, select.POLLIN)
def wait(self, timeout: float) -> bool:
"""Sleep up to ``timeout`` seconds, returning early on a watched write.
Returns:
True when a watched file changed, False when the time ran out.
"""
deadline = time.monotonic() + timeout
while (remaining := deadline - time.monotonic()) > 0:
if self._fd is None:
time.sleep(remaining)
return False
if self._poll.poll(remaining * 1000) and self._drain():
return True
return False
def close(self) -> None:
"""Release the inotify descriptor; later waits are plain sleeps."""
if self._fd is not None:
self._poll.unregister(self._fd)
os.close(self._fd)
self._fd = None
def _degrade(self, reason: str) -> None:
"""Drop to plain sleep, saying why, once."""
logger.warning(
"Workout wake-up disabled (%s) -- the budget now updates on the "
"next enforce tick instead of instantly.",
reason,
)
self.close()
def _drain(self) -> bool:
"""Read every queued event; whether any of them is a watched file."""
try:
buf = os.read(self._fd, _READ_SIZE) if self._fd is not None else b""
except BlockingIOError:
return False
except OSError as exc:
self._degrade(f"reading inotify events failed: {exc}")
return False
changed = False
offset = 0
while offset + _EVENT_HEADER.size <= len(buf):
wd, mask, _cookie, length = _EVENT_HEADER.unpack_from(buf, offset)
start = offset + _EVENT_HEADER.size
name = os.fsdecode(buf[start : start + length].split(b"\0", 1)[0])
offset = start + length
if mask & _IN_Q_OVERFLOW or name in self._names.get(wd, ()):
changed = True
elif mask & _IN_IGNORED:
self._names.pop(wd, None)
if not self._names:
self._degrade("the watched directory was removed")
return changed
return changed
def open_file_wake(paths: Iterable[Path]) -> FileWake:
"""Watch the parent directory of each path for writes to that path.
Args:
paths: Files to wake on; their directories must exist.
Returns:
A :class:`FileWake`; a plain-sleep one, warned about, when inotify or
every watch could not be set up.
"""
by_dir: dict[Path, set[str]] = {}
for path in paths:
by_dir.setdefault(path.parent, set()).add(path.name)
try:
libc = ctypes.CDLL(None, use_errno=True)
libc.inotify_add_watch.argtypes = [
ctypes.c_int,
ctypes.c_char_p,
ctypes.c_uint32,
]
fd = libc.inotify_init1(os.O_NONBLOCK | os.O_CLOEXEC)
except (OSError, AttributeError) as exc:
logger.warning("inotify unavailable (%s) -- waking on the tick only.", exc)
return FileWake(None, {})
if fd < 0:
reason = os.strerror(ctypes.get_errno())
logger.warning("inotify_init1 failed (%s) -- waking on the tick only.", reason)
return FileWake(None, {})
names: dict[int, frozenset[str]] = {}
for directory, files in by_dir.items():
wd = libc.inotify_add_watch(fd, os.fsencode(directory), _WATCH_MASK)
if wd < 0:
logger.warning(
"Cannot watch %s (%s) -- %s changes wake only on the tick.",
directory,
os.strerror(ctypes.get_errno()),
", ".join(sorted(files)),
)
continue
names[wd] = frozenset(files)
if not names:
os.close(fd)
return FileWake(None, {})
return FileWake(fd, names)

10
steam_backlog_enforcer/_leetcode_bonus.py
View File

@ -38,7 +38,7 @@ from typing import TYPE_CHECKING
from steam_backlog_enforcer._bonus_incident import report_leetcode_incident
from steam_backlog_enforcer._leetcode_ledger import read_ledger_solved_today
from steam_backlog_enforcer._status_api import AnswerCache, get_status
from steam_backlog_enforcer._status_api import AnswerCache, file_stamp, get_status
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config
@ -89,18 +89,20 @@ def leetcode_solved_today(config: Config) -> bool | None:
incident while an honest "not solved yet" does not.
"""
url = config.leetcode_status_url
cached = _cache.fresh(url)
path = Path(config.leetcode_ledger_path).expanduser()
# A new credit row rewrites the ledger; its stamp drops the memo at once.
stamp = file_stamp(path)
cached = _cache.fresh(url, stamp=stamp)
if cached is not None:
return cached
path = Path(config.leetcode_ledger_path).expanduser()
answer = read_ledger_solved_today(path)
if answer is None:
answer = _fallback_to_endpoint(url, path)
if answer is None:
return None
return _cache.store(url, answer=answer)
return _cache.store(url, answer=answer, stamp=stamp)
def _fallback_to_endpoint(url: str, path: Path) -> bool | None:

11
steam_backlog_enforcer/_reading_bonus.py
View File

@ -30,7 +30,7 @@ from typing import TYPE_CHECKING, Final
import earned_time
from steam_backlog_enforcer._leetcode_ledger import HMAC_KEY_FILE
from steam_backlog_enforcer._status_api import AnswerCache
from steam_backlog_enforcer._status_api import AnswerCache, file_stamp
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config
@ -59,10 +59,13 @@ def read_ledger_read_today(path: Path) -> bool | None:
def read_today(config: Config) -> bool | None:
"""Today's reading answer, memoised like the other earners."""
cached = _cache.fresh(_CACHE_KEY)
path = Path(config.book_ledger_path).expanduser()
# Stat before reading, so a credit landing mid-read is re-read next tick.
stamp = file_stamp(path)
cached = _cache.fresh(_CACHE_KEY, stamp=stamp)
if cached is not None:
return cached
answer = read_ledger_read_today(Path(config.book_ledger_path).expanduser())
answer = read_ledger_read_today(path)
if answer is None:
return None
return _cache.store(_CACHE_KEY, answer=answer)
return _cache.store(_CACHE_KEY, answer=answer, stamp=stamp)

74
steam_backlog_enforcer/_status_api.py
View File

@ -15,10 +15,16 @@ from __future__ import annotations
import http.client
import json
import logging
import time
from typing import Any, Final
from typing import TYPE_CHECKING, Any, Final
from urllib.parse import urlsplit
if TYPE_CHECKING:
from pathlib import Path
logger = logging.getLogger(__name__)
# Loopback, so a short timeout is generous. The daemon ticks frequently and
# must never block on this.
_TIMEOUT_SECONDS: Final = 2.0
@ -26,9 +32,49 @@ _TIMEOUT_SECONDS: Final = 2.0
# A successful answer is reused for this long. The daemon would otherwise make
# one HTTP call per tick, and the answer changes at most a few times a day. The
# cost is that a workout logged now raises the budget within a minute rather
# than instantly -- invisible next to a two-hour swing.
# than instantly -- invisible next to a two-hour swing. A caller that knows
# which file the answer is derived from also passes its stat stamp, and a
# changed stamp drops the answer at once (see :func:`file_stamp`).
_CACHE_TTL_SECONDS: Final = 60.0
# ``(st_ino, st_mtime_ns, st_size)`` of a source file; ``None`` when absent.
type Stamp = tuple[int, int, int] | None
# Last stat error per path, so an unreadable file warns once, not every tick.
_stat_errors: dict[Path, str] = {}
def file_stamp(path: Path) -> Stamp:
"""Cheap identity of ``path``'s current contents, for cache invalidation.
The inode is part of it because writers replace these files atomically:
a rename within the same mtime tick still changes the inode.
Returns:
The stamp, or ``None`` when the file does not exist -- an ordinary
state (no workout log yet, no book registered), compared like any
other stamp. Any other stat failure also yields ``None``, warned once
per distinct error so a 3 s loop cannot flood the journal.
"""
try:
st = path.stat()
except FileNotFoundError:
_stat_errors.pop(path, None)
return None
except OSError as exc:
if _stat_errors.get(path) != str(exc):
_stat_errors[path] = str(exc)
logger.warning(
"Cannot stat %s (%s) -- its cached answer now expires only "
"on the %ds TTL instead of on change.",
path,
exc,
int(_CACHE_TTL_SECONDS),
)
return None
_stat_errors.pop(path, None)
return (st.st_ino, st.st_mtime_ns, st.st_size)
def get_status(url: str) -> dict[str, Any]:
"""GET a status endpoint and return its JSON body.
@ -64,25 +110,33 @@ class AnswerCache:
Only *successes* are stored. A failure is retried on the very next tick, so
a restarted status server takes effect immediately instead of serving the
floor for another minute.
floor for another minute. An answer stored with a :data:`Stamp` is also
dropped as soon as the caller presents a different one.
"""
def __init__(self) -> None:
"""Start empty."""
self._entries: dict[str, tuple[float, bool]] = {}
self._entries: dict[str, tuple[float, bool, Stamp]] = {}
def clear(self) -> None:
"""Drop every cached answer, forcing the next call to re-ask."""
self._entries.clear()
def fresh(self, url: str) -> bool | None:
"""The cached answer for ``url`` if it is still within its TTL."""
def fresh(self, url: str, *, stamp: Stamp = None) -> bool | None:
"""The cached answer for ``url`` if within its TTL and ``stamp``."""
cached = self._entries.get(url)
if cached is not None and time.monotonic() - cached[0] < _CACHE_TTL_SECONDS:
if cached is None or cached[2] != stamp:
return None
if time.monotonic() - cached[0] < _CACHE_TTL_SECONDS:
return cached[1]
return None
def store(self, url: str, *, answer: bool) -> bool:
"""Remember ``answer`` for ``url`` as of now, and hand it back."""
self._entries[url] = (time.monotonic(), answer)
def store(self, url: str, *, answer: bool, stamp: Stamp = None) -> bool:
"""Remember ``answer`` for ``url`` as of now, and hand it back.
``stamp`` must be taken *before* the answer was fetched: a write that
lands mid-fetch then leaves the stored stamp stale, so the next call
re-asks instead of trusting a pre-write answer for a whole TTL.
"""
self._entries[url] = (time.monotonic(), answer, stamp)
return answer

15
steam_backlog_enforcer/_workout_budget.py
View File

@ -16,9 +16,10 @@ would be indistinguishable from an earned one.
from __future__ import annotations
import logging
from pathlib import Path
from typing import TYPE_CHECKING
from steam_backlog_enforcer._status_api import AnswerCache, get_status
from steam_backlog_enforcer._status_api import AnswerCache, file_stamp, get_status
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config
@ -33,6 +34,11 @@ def reset_cache() -> None:
_cache.clear()
def workout_log_path(config: Config) -> Path:
"""screen-locker's ``log.json``, which ``gaming.workout_today`` derives from."""
return Path(config.workout_log_path).expanduser()
def _fetch_workout_today(url: str) -> bool:
"""Read ``gaming.workout_today`` from the screen-locker status API.
@ -62,7 +68,10 @@ def workout_logged_today(config: Config) -> bool | None:
distinct from ``False`` so the caller can log which one happened.
"""
url = config.workout_status_url
cached = _cache.fresh(url)
# Stat before asking: a credited workout rewrites log.json, and the new
# stamp re-asks on the very next tick instead of after the cache TTL.
stamp = file_stamp(workout_log_path(config))
cached = _cache.fresh(url, stamp=stamp)
if cached is not None:
return cached
@ -86,4 +95,4 @@ def workout_logged_today(config: Config) -> bool | None:
)
return None
return _cache.store(url, answer=answer)
return _cache.store(url, answer=answer, stamp=stamp)

2
steam_backlog_enforcer/config.py
View File

@ -96,6 +96,8 @@ class Config:
"""book-guard's signed ledger: the reading hour's only transport."""
workout_status_url: str = "http://127.0.0.1:8770/api/status"
"""Where to ask whether today has a counted workout. Loopback only."""
workout_log_path: str = "~/src/screen-locker/screen_locker/log.json"
"""screen-locker's log.json; a change re-asks that URL at once."""
leetcode_ledger_path: str = "~/.local/share/leetcode_guard/ledger.json"
leetcode_status_url: str = "http://127.0.0.1:8771/api/status"
"""LeetCode's transports: ledger first, this loopback URL as fallback."""

19
steam_backlog_enforcer/tests/_no_workout_http.py
View File

@ -34,10 +34,13 @@ from steam_backlog_enforcer._budget_resolve import BudgetResolution
if TYPE_CHECKING:
from collections.abc import Iterator
from contextlib import AbstractContextManager
from pathlib import Path
from steam_backlog_enforcer.config import Config
_RESOLVER = "steam_backlog_enforcer._playtime_state.resolve_budget"
_WORKOUT = "steam_backlog_enforcer._workout_budget"
_LOOP = "steam_backlog_enforcer._enforce_loop"
_MINUTE = 60.0
@ -91,11 +94,23 @@ def fixed_budget(seconds: float) -> AbstractContextManager[object]:
@pytest.fixture(autouse=True)
def _no_workout_http() -> Iterator[None]:
def _no_workout_http(tmp_path: Path) -> Iterator[None]:
"""Stop rules_for from making a real HTTP call to the screen locker.
Also points screen-locker's ``log.json`` at a tmp path, at both bindings:
the cache stamp must not stat the real log, and no test may ever put an
inotify watch on the real ``~/src/screen-locker`` directory.
Args:
tmp_path: pytest's temporary directory.
Yields:
None, with the budget resolver stubbed for the whole test.
"""
with patch(_RESOLVER, side_effect=earned_budget):
log = tmp_path / "screen-locker" / "log.json"
with (
patch(_RESOLVER, side_effect=earned_budget),
patch(f"{_WORKOUT}.workout_log_path", return_value=log),
patch(f"{_LOOP}.workout_log_path", return_value=log),
):
yield

3
steam_backlog_enforcer/tests/conftest.py
View File

@ -194,7 +194,8 @@ def _no_real_sleep() -> Iterator[None]:
patch("steam_backlog_enforcer._steam_client.time.sleep", noop),
patch("steam_backlog_enforcer._steam_launch.time.sleep", noop),
patch("steam_backlog_enforcer._steam_api_client.time.sleep", noop),
patch("steam_backlog_enforcer._enforce_loop.time.sleep", noop),
# The loop waits on an inotify wake, not time.sleep: stub its factory.
patch("steam_backlog_enforcer._enforce_loop.open_file_wake", noop),
):
yield

12
steam_backlog_enforcer/tests/test_do_enforce.py
View File

@ -52,7 +52,7 @@ class TestDoEnforce:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(Config(), state)
assert any("No game" in str(c) for c in mock_echo.call_args_list)
@ -76,7 +76,7 @@ class TestDoEnforce:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
),
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(Config(), state)
@ -95,7 +95,7 @@ class TestDoEnforce:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
),
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(config, state)
@ -119,7 +119,7 @@ class TestDoEnforce:
f"{PKG}._enforce_loop_iteration",
side_effect=side_effect,
),
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(config, state)
assert call_count == 2
@ -147,7 +147,7 @@ class TestDoEnforce:
patch(f"{PKG}._echo"),
patch.object(State, "load", side_effect=load_side_effect),
patch(f"{PKG}._enforce_loop_iteration") as mock_iter,
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(config, state)
mock_iter.assert_not_called()
@ -175,7 +175,7 @@ class TestStateReload:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
),
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(Config(), state)
assert state.manual_picks == fresh.manual_picks

10
steam_backlog_enforcer/tests/test_enforce_loop.py
View File

@ -17,6 +17,8 @@ if TYPE_CHECKING:
from pathlib import Path
PKG = "steam_backlog_enforcer._enforce_loop"
# The owned-ids cache reads its own clock; _enforce_loop no longer imports time.
_CLOCK = "steam_backlog_enforcer._owned_apps_cache.time.time"
ENFORCE_STEPS_PKG = "steam_backlog_enforcer._enforce_steps"
OWNED_APPS_CACHE_PKG = "steam_backlog_enforcer._owned_apps_cache"
@ -187,7 +189,7 @@ class TestOwnedIdsCacheHelpers:
)
with (
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_FILE", cache_file),
patch(f"{PKG}.time.time", return_value=10_000.0),
patch(_CLOCK, return_value=10_000.0),
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_TTL_SECONDS", 60),
):
assert _load_owned_app_ids_cache("sid") is None
@ -200,7 +202,7 @@ class TestOwnedIdsCacheHelpers:
)
with (
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_FILE", cache_file),
patch(f"{PKG}.time.time", return_value=10_010.0),
patch(_CLOCK, return_value=10_010.0),
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_TTL_SECONDS", 60),
):
assert _load_owned_app_ids_cache("sid") is None
@ -215,7 +217,7 @@ class TestOwnedIdsCacheHelpers:
)
with (
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_FILE", cache_file),
patch(f"{PKG}.time.time", return_value=10_010.0),
patch(_CLOCK, return_value=10_010.0),
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_TTL_SECONDS", 60),
):
assert _load_owned_app_ids_cache("sid") == [1, 2]
@ -224,7 +226,7 @@ class TestOwnedIdsCacheHelpers:
cache_file = tmp_path / "owned.json"
with (
patch(f"{OWNED_APPS_CACHE_PKG}._OWNED_IDS_CACHE_FILE", cache_file),
patch(f"{PKG}.time.time", return_value=123.0),
patch(_CLOCK, return_value=123.0),
patch(f"{OWNED_APPS_CACHE_PKG}._atomic_write") as mock_atomic,
):
_save_owned_app_ids_cache("sid", [10, 20])

8
steam_backlog_enforcer/tests/test_enforce_loop_part3.py
View File

@ -86,7 +86,7 @@ class TestDoEnforceTotalBlock:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
),
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(config, state)
mock_setup.assert_not_called()
@ -104,7 +104,7 @@ class TestDoEnforceTotalBlock:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(config, state)
mock_iter.assert_called_once()
@ -125,7 +125,7 @@ class TestDoEnforceTotalBlock:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(Config(), state)
output = " ".join(str(c) for c in mock_echo.call_args_list)
@ -184,7 +184,7 @@ class TestPlaytimeIsEnforcedFirst:
f"{PKG}._enforce_loop_iteration",
side_effect=KeyboardInterrupt,
) as mock_iter,
patch(f"{PKG}.time.sleep"),
patch(f"{PKG}.open_file_wake"),
):
do_enforce(Config(), state, demo=True)
assert mock_iter.call_args.kwargs["demo"] is True

249
steam_backlog_enforcer/tests/test_file_wake.py Normal file
View File

@ -0,0 +1,249 @@
"""Tests for _file_wake: the inotify wait that wakes the enforce loop early.
The real-kernel tests watch a fresh ``tmp_path`` directory, never the real
screen-locker tree. The parsing branches the kernel will not produce on demand
(queue overflow, a watch torn down) are fed through a pipe carrying hand-packed
``inotify_event`` records, which is byte-for-byte what ``os.read`` returns.
"""
from __future__ import annotations
import logging
import os
import time
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
import pytest
from steam_backlog_enforcer import _file_wake
from steam_backlog_enforcer._file_wake import FileWake, open_file_wake
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
from pathlib import Path
_PKG = "steam_backlog_enforcer._file_wake"
# Far above real latency so CI cannot flake; elapsed asserts pin "early".
_LONG = 5.0
_SHORT = 0.05
@pytest.fixture
def log(tmp_path: Path) -> Path:
"""A ``log.json`` path inside a directory of its own."""
directory = tmp_path / "screen_locker"
directory.mkdir()
return directory / "log.json"
@pytest.fixture
def wake(log: Path) -> Iterator[FileWake]:
"""A live watcher on ``log``, closed afterwards."""
watcher = open_file_wake([log])
yield watcher
watcher.close()
def _timed_wait(watcher: FileWake, timeout: float) -> tuple[bool, float]:
"""Run ``watcher.wait`` and measure how long it took."""
start = time.monotonic()
woke = watcher.wait(timeout)
return woke, time.monotonic() - start
def _event(wd: int, mask: int, name: bytes = b"") -> bytes:
"""Pack one ``struct inotify_event`` with a NUL-padded name."""
padded = name + b"\0" * (16 - len(name) % 16) if name else b""
return _file_wake._EVENT_HEADER.pack(wd, mask, 0, len(padded)) + padded
type Piped = Callable[[dict[int, frozenset[str]]], tuple[FileWake, int]]
@pytest.fixture
def piped() -> Iterator[Piped]:
"""Build FileWakes reading a non-blocking pipe instead of inotify.
Teardown goes through ``FileWake.close`` (idempotent), never a raw
``os.close`` of the read end: a watcher that degraded already closed it,
and the number may since belong to some unrelated descriptor.
"""
made: list[tuple[FileWake, int]] = []
def make(names: dict[int, frozenset[str]]) -> tuple[FileWake, int]:
read_fd, write_fd = os.pipe()
os.set_blocking(read_fd, False)
made.append((FileWake(read_fd, names), write_fd))
return made[-1]
yield make
for watcher, write_fd in made:
watcher.close()
os.close(write_fd)
class TestRealInotify:
"""Against the kernel, in a tmp directory."""
def test_atomic_rename_wakes(self, log: Path, wake: FileWake) -> None:
tmp = log.with_name("log.json.tmp")
tmp.write_text("{}")
# Drain the temp file's own events: they must not count as a wake.
assert wake.wait(_SHORT) is False
tmp.replace(log)
woke, elapsed = _timed_wait(wake, _LONG)
assert woke is True
assert elapsed < 1.0
def test_in_place_write_wakes(self, log: Path, wake: FileWake) -> None:
log.write_text("{}")
woke, elapsed = _timed_wait(wake, _LONG)
assert woke is True
assert elapsed < 1.0
def test_other_file_does_not_wake(self, log: Path, wake: FileWake) -> None:
log.with_name("other.json").write_text("{}")
woke, elapsed = _timed_wait(wake, _SHORT)
assert woke is False
assert elapsed >= _SHORT
def test_times_out_without_events(self, wake: FileWake) -> None:
woke, elapsed = _timed_wait(wake, _SHORT)
assert woke is False
assert elapsed >= _SHORT
def test_two_files_in_one_directory_share_a_watch(self, log: Path) -> None:
other = log.with_name("ledger.json")
watcher = open_file_wake([log, other])
try:
other.write_text("{}")
assert watcher.wait(_LONG) is True
finally:
watcher.close()
def test_one_missing_directory_keeps_the_other_watch(
self, log: Path, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
missing = tmp_path / "gone" / "log.json"
with caplog.at_level(logging.WARNING, logger=_PKG):
watcher = open_file_wake([missing, log])
try:
assert "Cannot watch" in caplog.text
log.write_text("{}")
assert watcher.wait(_LONG) is True
finally:
watcher.close()
def test_close_twice_is_harmless(self, wake: FileWake) -> None:
wake.close()
wake.close()
# conftest no-ops time.sleep process-wide, so assert the call instead.
with patch(f"{_PKG}.time.sleep") as sleep:
assert wake.wait(_SHORT) is False
sleep.assert_called_once()
class TestSetupFallback:
"""Every setup failure warns and degrades to a plain sleep."""
def _assert_plain_sleep(self, watcher: FileWake) -> None:
with patch(f"{_PKG}.time.sleep") as sleep:
assert watcher.wait(_LONG) is False
sleep.assert_called_once()
assert sleep.call_args.args[0] == pytest.approx(_LONG, abs=0.5)
def test_unwatchable_directory(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level(logging.WARNING, logger=_PKG):
watcher = open_file_wake([tmp_path / "missing" / "log.json"])
assert "Cannot watch" in caplog.text
assert "log.json" in caplog.text
self._assert_plain_sleep(watcher)
def test_no_libc(self, log: Path, caplog: pytest.LogCaptureFixture) -> None:
with (
caplog.at_level(logging.WARNING, logger=_PKG),
patch(f"{_PKG}.ctypes.CDLL", side_effect=OSError("no libc")),
):
watcher = open_file_wake([log])
assert "inotify unavailable (no libc)" in caplog.text
self._assert_plain_sleep(watcher)
def test_init_fails(self, log: Path, caplog: pytest.LogCaptureFixture) -> None:
libc = MagicMock()
libc.inotify_init1.return_value = -1
with (
caplog.at_level(logging.WARNING, logger=_PKG),
patch(f"{_PKG}.ctypes.CDLL", return_value=libc),
):
watcher = open_file_wake([log])
assert "inotify_init1 failed" in caplog.text
libc.inotify_add_watch.assert_not_called()
self._assert_plain_sleep(watcher)
class TestEventParsing:
"""Branches fed through a pipe of hand-packed events."""
def test_queue_overflow_wakes(self, piped: Piped) -> None:
watcher, write_fd = piped({1: frozenset({"log.json"})})
# wd -1 with no name: an overflow says nothing about *which* file, so
# it must count as "maybe changed" rather than be dropped.
os.write(write_fd, _event(-1, _file_wake._IN_Q_OVERFLOW))
woke, elapsed = _timed_wait(watcher, _LONG)
assert woke is True
assert elapsed < 1.0
def test_unrelated_event_on_unknown_wd_is_ignored(self, piped: Piped) -> None:
watcher, write_fd = piped({1: frozenset({"log.json"})})
os.write(write_fd, _event(7, _file_wake._IN_CLOSE_WRITE, b"log.json"))
assert watcher.wait(_SHORT) is False
def test_removed_watch_degrades_with_warning(
self, piped: Piped, caplog: pytest.LogCaptureFixture
) -> None:
watcher, write_fd = piped({1: frozenset({"log.json"})})
os.write(write_fd, _event(1, _file_wake._IN_IGNORED))
with (
caplog.at_level(logging.WARNING, logger=_PKG),
patch(f"{_PKG}.time.sleep") as sleep,
):
assert watcher.wait(_LONG) is False
assert "the watched directory was removed" in caplog.text
sleep.assert_called_once()
def test_one_removed_watch_of_two_keeps_watching(
self, piped: Piped, caplog: pytest.LogCaptureFixture
) -> None:
names = {1: frozenset({"a.json"}), 2: frozenset({"b.json"})}
watcher, write_fd = piped(names)
os.write(write_fd, _event(1, _file_wake._IN_IGNORED))
with caplog.at_level(logging.WARNING, logger=_PKG):
assert watcher.wait(_SHORT) is False
assert caplog.text == ""
os.write(write_fd, _event(2, _file_wake._IN_MOVED_TO, b"b.json"))
assert watcher.wait(_LONG) is True
def test_spurious_readiness_keeps_waiting(self, piped: Piped) -> None:
watcher, write_fd = piped({1: frozenset({"log.json"})})
os.write(write_fd, b"x")
with patch(f"{_PKG}.os.read", side_effect=BlockingIOError):
woke, elapsed = _timed_wait(watcher, _SHORT)
assert woke is False
assert elapsed >= _SHORT
def test_read_error_degrades_with_warning(
self, piped: Piped, caplog: pytest.LogCaptureFixture
) -> None:
watcher, write_fd = piped({1: frozenset({"log.json"})})
os.write(write_fd, b"x")
with (
caplog.at_level(logging.WARNING, logger=_PKG),
patch(f"{_PKG}.os.read", side_effect=OSError("EIO")),
patch(f"{_PKG}.time.sleep") as sleep,
):
assert watcher.wait(_LONG) is False
assert "reading inotify events failed: EIO" in caplog.text
sleep.assert_called_once()

172
steam_backlog_enforcer/tests/test_status_api_stamp.py Normal file
View File

@ -0,0 +1,172 @@
"""Tests for stat-stamp invalidation of the earner answer cache.
A credited workout (or LeetCode / reading credit) rewrites a file the answer
is derived from. The stamp makes the very next tick re-ask instead of serving
the memoised pre-credit answer for the rest of the 60 s TTL.
"""
from __future__ import annotations
import logging
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from steam_backlog_enforcer import _status_api, _workout_budget
from steam_backlog_enforcer._status_api import AnswerCache, file_stamp
# Bound at import, before conftest swaps the module attribute for a tmp path.
from steam_backlog_enforcer._workout_budget import workout_log_path
from steam_backlog_enforcer.config import Config
_API = "steam_backlog_enforcer._status_api"
_URL = "http://127.0.0.1:8770/api/status"
def _unstatable(error: OSError) -> MagicMock:
"""A path whose ``stat`` fails with ``error`` (hashable, like a Path)."""
path = MagicMock(spec=Path)
path.stat.side_effect = error
return path
class TestFileStamp:
"""Identity of a file's contents, or None."""
def test_missing_file_is_none(self, tmp_path: Path) -> None:
assert file_stamp(tmp_path / "absent.json") is None
def test_same_contents_same_stamp(self, tmp_path: Path) -> None:
path = tmp_path / "log.json"
path.write_text("{}")
assert file_stamp(path) == file_stamp(path)
def test_atomic_replace_changes_the_stamp(self, tmp_path: Path) -> None:
path = tmp_path / "log.json"
path.write_text("{}")
before = file_stamp(path)
tmp = tmp_path / "log.json.tmp"
tmp.write_text("{}")
tmp.replace(path)
# Same size and possibly the same mtime tick: the inode tells them apart.
assert file_stamp(path) != before
def test_in_place_growth_changes_the_stamp(self, tmp_path: Path) -> None:
path = tmp_path / "log.json"
path.write_text("{}")
before = file_stamp(path)
path.write_text('{"a": 1}')
assert file_stamp(path) != before
def test_stat_error_warns_once_per_distinct_error(
self, caplog: pytest.LogCaptureFixture
) -> None:
path = _unstatable(PermissionError("denied"))
with caplog.at_level(logging.WARNING, logger=_API):
assert file_stamp(path) is None
assert file_stamp(path) is None
path.stat.side_effect = PermissionError("still denied")
assert file_stamp(path) is None
warnings = [r for r in caplog.records if "Cannot stat" in r.getMessage()]
assert len(warnings) == 2
assert "60s TTL" in warnings[0].getMessage()
_status_api._stat_errors.pop(path, None)
def test_recovery_re_arms_the_warning(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
path = tmp_path / "log.json"
path.write_text("{}")
with caplog.at_level(logging.WARNING, logger=_API):
with patch.object(Path, "stat", side_effect=PermissionError("x")):
assert file_stamp(path) is None
assert file_stamp(path) is not None
assert path not in _status_api._stat_errors
with patch.object(Path, "stat", side_effect=PermissionError("x")):
assert file_stamp(path) is None
assert caplog.text.count("Cannot stat") == 2
_status_api._stat_errors.pop(path, None)
def test_vanishing_clears_a_recorded_error(self, tmp_path: Path) -> None:
path = tmp_path / "log.json"
_status_api._stat_errors[path] = "old"
assert file_stamp(path) is None
assert path not in _status_api._stat_errors
class TestAnswerCacheStamp:
"""A stored stamp must match the caller's for the answer to be reused."""
def test_matching_stamp_hits(self) -> None:
cache = AnswerCache()
cache.store(_URL, answer=True, stamp=(1, 2, 3))
assert cache.fresh(_URL, stamp=(1, 2, 3)) is True
def test_changed_stamp_misses(self) -> None:
cache = AnswerCache()
cache.store(_URL, answer=False, stamp=(1, 2, 3))
assert cache.fresh(_URL, stamp=(1, 2, 4)) is None
def test_file_appearing_misses(self) -> None:
cache = AnswerCache()
cache.store(_URL, answer=False, stamp=None)
assert cache.fresh(_URL, stamp=(1, 2, 3)) is None
def test_unstamped_callers_still_hit(self) -> None:
cache = AnswerCache()
cache.store(_URL, answer=True)
assert cache.fresh(_URL) is True
def test_entries_carry_the_stamp(self) -> None:
cache = AnswerCache()
cache.store(_URL, answer=True, stamp=(9, 9, 9))
_when, answer, stamp = cache._entries[_URL]
assert (answer, stamp) == (True, (9, 9, 9))
def test_ttl_still_applies_with_a_matching_stamp(self) -> None:
cache = AnswerCache()
with patch(f"{_API}.time.monotonic", return_value=0.0):
cache.store(_URL, answer=True, stamp=(1, 2, 3))
with patch(f"{_API}.time.monotonic", return_value=61.0):
assert cache.fresh(_URL, stamp=(1, 2, 3)) is None
class TestWorkoutLogPath:
"""Where screen-locker's log.json is read from."""
def test_expands_the_home_directory(self) -> None:
config = Config()
config.workout_log_path = "~/locker/log.json"
assert workout_log_path(config) == Path.home() / "locker" / "log.json"
def test_default_points_at_screen_locker(self) -> None:
expected = Path.home() / "src/screen-locker/screen_locker/log.json"
assert workout_log_path(Config()) == expected
class TestWorkoutAnswerFollowsTheLog:
"""A rewritten log.json re-asks the locker inside the TTL."""
@pytest.fixture(autouse=True)
def _clear_cache(self) -> None:
_workout_budget.reset_cache()
def test_credit_is_seen_on_the_next_call(self) -> None:
config = Config()
# conftest redirected this to a tmp path; that is the file we rewrite.
log = _workout_budget.workout_log_path(config)
log.parent.mkdir(parents=True)
log.write_text("[]")
with patch(
f"{_workout_budget.__name__}._fetch_workout_today",
side_effect=[False, True],
) as fetch:
assert _workout_budget.workout_logged_today(config) is False
assert _workout_budget.workout_logged_today(config) is False
assert fetch.call_count == 1
tmp = log.with_name("log.json.tmp")
tmp.write_text('[{"credited": true}]')
tmp.replace(log)
assert _workout_budget.workout_logged_today(config) is True
assert fetch.call_count == 2