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: resolve gaming time from the shared earned_time registry
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
Details
md-naming / markdown naming convention (push) Waiting to run
Details
pre-commit / pre-commit (${{ matrix.group }}) (mypy, mypy, ) (push) Waiting to run
Details
pre-commit / pre-commit (${{ matrix.group }}) (pylint, pylint, ) (push) Waiting to run
Details
pre-commit / pre-commit (${{ matrix.group }}) (rest, , mypy,pylint) (push) Waiting to run
Details
Tests / test (push) Waiting to run
Details
Tests / web-coverage (push) Waiting to run
Details
repo-contract / repo contract (push) Waiting to run
Details

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-10-04 19:01:27 +02:00
parent 19c9f8d24f
commit c8bd9b21ff
20 changed files with 288 additions and 463 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -110,6 +110,7 @@ repos:
- --disable-error-code=list-item
additional_dependencies:
- types-requests
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time
# ===========================================================================
# PYLINT
@ -129,6 +130,7 @@ repos:
# (steam_backlog_enforcer._mcp_server).
additional_dependencies:
- aiohttp==3.14.3
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time
- howlongtobeatpy==1.0.23
- mcp==2.3.0
- pytest

3
pyproject.toml
View File

@ -14,6 +14,9 @@ dependencies = [
# are declared here so a clean install imports the package.
"aiohttp==3.14.3",
"websockets==17.2",
# The earner registry (gaming budget terms); stdlib-only, so the root
# daemon imports it from the desktop user's site-packages.
"earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time",
]
# Optional extras. The `mcp` group is ONLY needed by the MCP server

1
requirements.txt
View File

@ -5,6 +5,7 @@ aiohttp==3.14.3
bandit==1.9.4
codespell==2.4.3
coverage==7.16.2
earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time
howlongtobeatpy==1.0.24
mcp==2.3.0
mypy==2.4.0

194
steam_backlog_enforcer/_budget_resolve.py
View File

@ -1,29 +1,32 @@
"""Today's gaming budget: a floor, plus whatever today earned.
The budget is a sum, not a choice between values, capped at
``max_gaming_seconds`` (8h):
Every number lives in the shared earner registry (``earned_time``,
kuhyx/utils), which screen-locker reads for the shutdown time too. The budget
is its sum, capped at the gaming ceiling (8h):
base (4h) + workout bonus (2h) + LeetCode bonus (1h) + reading bonus (1h)
base (4h) + workout (2h) + LeetCode (1h) + reading (1h) + ...
so a day earns anything from 4h to 8h. The earners are read **independently**
-- :mod:`steam_backlog_enforcer._workout_budget`,
:mod:`steam_backlog_enforcer._leetcode_bonus` and
:mod:`steam_backlog_enforcer._reading_bonus` share no state and none can fail
in a way that changes another's term. That is what "the LeetCode bonus
must not interfere with the workout" means in code.
The base is 5h minus every penalised earner's cut once its day arrives
(book-guard's reading hour, from 2026-10-01). The earners are read
**independently** -- :mod:`steam_backlog_enforcer._workout_budget`,
:mod:`steam_backlog_enforcer._leetcode_bonus`,
:mod:`steam_backlog_enforcer._reading_bonus` and, for any earner registered
later, :mod:`steam_backlog_enforcer._ledger_earners` share no state and none
can fail in a way that changes another's term. That is what "the LeetCode
bonus must not interfere with the workout" means in code.
**Fail closed.** An answer that could not be obtained contributes nothing, the
same as a "no". The difference is only in what gets reported: an unreadable
LeetCode ledger raises an incident, an honest "not solved yet" does not.
**Rising, in normal use -- but not guaranteed.** Both earners only ever go
**Rising, in normal use -- but not guaranteed.** Every earner only ever goes
false->true within a day, so in the ordinary case the budget starts at the floor
and rises. That is a property of the *inputs*, not something enforced here:
nothing persists a per-day high-water mark, so anything that changes a resolved
answer mid-day -- editing the config, or deploying a change onto a day already
in progress -- lowers the budget immediately and re-prices time already spent.
Seconds already accrued are then measured against the new, smaller budget, and
the cutoff fires on the next tick with no warning first, because
answer mid-day -- changing the registry, or deploying a change onto a day
already in progress -- lowers the budget immediately and re-prices time already
spent. Seconds already accrued are then measured against the new, smaller
budget, and the cutoff fires on the next tick with no warning first, because
``warned_seconds`` records thresholds by seconds *remaining* and remaining has
already gone negative.
@ -34,20 +37,20 @@ these numbers.
**One seam.** :func:`resolve_budget` is called from ``rules_for`` and nowhere
else, so the enforcing daemon and the read-only HTTP/MCP views resolve the same
number. The daemon holds its ``Config`` for the process lifetime while
``_budget_view`` reloads it per request, so anything resolved at only one of
those two sites would let the UI report a budget the daemon was not enforcing.
Callers that want the breakdown read it off ``PlaytimeRules``, never by
number. Callers that want the breakdown read it off ``PlaytimeRules``, never by
resolving a second time.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
from dataclasses import dataclass, field
from datetime import datetime
import logging
from typing import TYPE_CHECKING, Final
import earned_time
from steam_backlog_enforcer._ledger_earners import ledger_answer
from steam_backlog_enforcer._leetcode_bonus import leetcode_solved_today
from steam_backlog_enforcer._reading_bonus import read_today
from steam_backlog_enforcer._workout_budget import workout_logged_today
@ -57,18 +60,28 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
_SECONDS_PER_MINUTE: Final = 60.0
_SECONDS_PER_HOUR: Final = 3600.0
# The 5h -> 4h base cut belongs to book-guard's reading hour, and book-guard's
# gate starts on 2026-10-01. Until then the old 5h floor stands: cutting an
# hour a reader cannot yet earn back was a same-day loss (2026-09-26).
READING_BASE_FROM: Final = date(2026, 10, 1)
_PRE_READING_EXTRA: Final = 3600.0
def base_for(configured: float, today: date) -> float:
"""The floor for ``today``: the configured base, +1h before the cut."""
return configured + (_PRE_READING_EXTRA if today < READING_BASE_FROM else 0.0)
# How each earner reads in the reason string: (earned, missed, unknown).
# An earner registered later without an entry here gets a generic phrase.
_PHRASES: Final[dict[str, tuple[str, str, str]]] = {
"workout": (
"workout counted",
"no counted workout",
"workout unknown (screen-locker unreachable)",
),
"leetcode": (
"LeetCode solve recorded",
"no LeetCode solve recorded",
"LeetCode unknown (ledger and status API both unreadable)",
),
"reading": (
"reading credited",
"no reading credited",
"reading unknown (book-guard ledger unreadable)",
),
}
@dataclass(frozen=True)
@ -82,6 +95,8 @@ class BudgetResolution:
leetcode_seconds: Seconds added by a LeetCode solve, or 0.
reason: A human-readable account, for the journal and ``/api/budget``.
reading_seconds: Seconds added by a credited reading session, or 0.
earned_seconds: Seconds added per earner name, every registered earner
included -- the generic form of the three fields above.
"""
seconds: float
@ -90,112 +105,61 @@ class BudgetResolution:
leetcode_seconds: float
reason: str
reading_seconds: float = 0.0
earned_seconds: dict[str, float] = field(default_factory=dict)
def _bonus_seconds(configured: int, label: str) -> float:
"""A configured bonus, refusing a negative one.
Args:
configured: The configured value.
label: The config field name, for the error message.
Returns:
The bonus in seconds, clamped at zero. A negative bonus would mean
earning something *cost* time, which is never what was meant.
"""
if configured < 0:
logger.error(
"%s is negative (%d); clamping to 0 so earning something can never "
"cost gaming time.",
label,
configured,
)
return 0.0
return float(configured)
def _describe(*, answer: bool | None, earned: str, missed: str, unknown: str) -> str:
"""Render one earner's answer for the reason string.
Args:
answer: True, False, or None for "could not check".
earned: Phrase for True.
missed: Phrase for False.
unknown: Phrase for None.
Returns:
The matching phrase.
"""
def _describe(name: str, label: str, answer: int | None) -> str:
"""Render one earner's answer for the reason string."""
earned, missed, unknown = _PHRASES.get(
name, (f"{label} credited", f"no {label} credited", f"{label} unknown")
)
if answer is None:
return unknown
return earned if answer else missed
def _answers(config: Config) -> dict[str, bool | None]:
"""Every registered earner's answer for today, each read independently."""
answers: dict[str, bool | None] = {
"workout": workout_logged_today(config),
"leetcode": leetcode_solved_today(config),
"reading": read_today(config),
}
for earner in earned_time.EARNERS:
if earner.name not in answers:
answers[earner.name] = ledger_answer(earner)
return answers
def resolve_budget(config: Config) -> BudgetResolution:
"""Return today's gaming budget, and what earned it.
Args:
config: Loaded user configuration.
config: Loaded user configuration (for the earners' transports).
Returns:
The floor plus a bonus for each of today's earners. An answer that
could not be obtained contributes nothing, exactly as a "no" does --
the difference is only in what gets logged and reported.
"""
base = base_for(
_bonus_seconds(config.base_gaming_seconds, "base_gaming_seconds"),
datetime.now().astimezone().date(),
# The registry is passed explicitly: _answers iterated this same tuple, so
# the earners asked and the earners summed can never differ.
day = earned_time.resolve(
_answers(config), datetime.now().astimezone().date(), earned_time.EARNERS
)
workout = workout_logged_today(config)
leetcode = leetcode_solved_today(config)
reading = read_today(config)
workout_seconds = (
_bonus_seconds(config.workout_bonus_seconds, "workout_bonus_seconds")
if workout
else 0.0
)
leetcode_seconds = (
_bonus_seconds(config.leetcode_bonus_seconds, "leetcode_bonus_seconds")
if leetcode
else 0.0
)
reading_seconds = (
_bonus_seconds(config.reading_bonus_seconds, "reading_bonus_seconds")
if reading
else 0.0
)
ceiling = _bonus_seconds(config.max_gaming_seconds, "max_gaming_seconds")
total = min(ceiling, base + workout_seconds + leetcode_seconds + reading_seconds)
workout_part = _describe(
answer=workout,
earned="workout counted",
missed="no counted workout",
unknown="workout unknown (screen-locker unreachable)",
)
leetcode_part = _describe(
answer=leetcode,
earned="LeetCode solve recorded",
missed="no LeetCode solve recorded",
unknown="LeetCode unknown (ledger and status API both unreadable)",
)
reading_part = _describe(
answer=reading,
earned="reading credited",
missed="no reading credited",
unknown="reading unknown (book-guard ledger unreadable)",
)
reason = (
f"{total / _SECONDS_PER_HOUR:.1f}h: {workout_part}, {leetcode_part}, "
f"{reading_part}"
earned = {t.earner.name: t.gaming_minutes * _SECONDS_PER_MINUTE for t in day.terms}
total = day.gaming_minutes * _SECONDS_PER_MINUTE
parts = ", ".join(
_describe(t.earner.name, t.earner.label, t.answer) for t in day.terms
)
reason = f"{total / _SECONDS_PER_HOUR:.1f}h: {parts}"
logger.info("Gaming budget %s", reason)
return BudgetResolution(
seconds=total,
base_seconds=base,
workout_seconds=workout_seconds,
leetcode_seconds=leetcode_seconds,
base_seconds=day.base.gaming_minutes * _SECONDS_PER_MINUTE,
workout_seconds=earned.get("workout", 0.0),
leetcode_seconds=earned.get("leetcode", 0.0),
reason=reason,
reading_seconds=reading_seconds,
reading_seconds=earned.get("reading", 0.0),
earned_seconds=earned,
)

113
steam_backlog_enforcer/_leetcode_ledger.py
View File

@ -1,10 +1,9 @@
"""Reading leetcode-guard's ledger for today's accepted submissions.
Credits are verified against the shared HMAC key, reproducing gatelock's
canonicalisation in eight lines rather than importing it: gatelock is installed
into the user's site-packages and this daemon runs as root, so the import is
not available. ``test_leetcode_ledger.py`` pins the canonicalisation to a fixed
vector so the duplication cannot drift silently.
Reading and verifying is the shared reader in ``earned_time`` (kuhyx/utils),
the same one screen-locker uses; ``test_leetcode_ledger.py`` pins its HMAC
canonicalisation to a fixed vector. gatelock itself is not imported: it pulls
in tkinter, and this daemon runs headless as root.
Only ``credit`` entries count. A ``seen`` entry is first-run seeding worth
zero, and a ``charge`` proves only that the day was *settled* -- which also
@ -21,83 +20,17 @@ Every unreadable state returns ``None``, never ``False``. "Cannot check" is not
from __future__ import annotations
from datetime import datetime, time
import hashlib
import hmac
import json
import logging
from pathlib import Path
from typing import Any, Final
from typing import Final
import earned_time
logger = logging.getLogger(__name__)
# The key every locker signs its state with. Root-owned, world-readable.
HMAC_KEY_FILE: Final = Path("/etc/workout-locker/hmac.key")
_CREDIT: Final = "credit"
def _verified(entry: dict[str, Any], key: bytes) -> bool:
"""Whether a ledger entry's HMAC signature is genuine.
Args:
entry: The full entry, including its ``hmac`` field.
key: The shared signing key.
Returns:
Whether the signature matches. An unverified credit is worth an hour of
gaming time, so a forged one must not count.
"""
stored = entry.get("hmac")
if not isinstance(stored, str):
return False
body = {k: v for k, v in entry.items() if k != "hmac"}
payload = json.dumps(body, sort_keys=True, separators=(",", ":"))
expected = hmac.new(key, payload.encode(), hashlib.sha256).hexdigest()
return hmac.compare_digest(stored, expected)
def _landed_today(entry: dict[str, Any], *, window: tuple[float, float]) -> bool:
"""Whether this credit's accepted submission happened today, locally.
Args:
entry: One verified credit entry.
window: Local midnight and now, as unix seconds.
Returns:
Whether the solve falls inside today. ``detail.submitted_at`` is
LeetCode's own timestamp; ``day`` is the *harvest* day and runs late, so
it is only a fallback -- it can miss a late solve, never invent one.
"""
detail = entry.get("detail")
raw = detail.get("submitted_at") if isinstance(detail, dict) else None
if raw is not None:
try:
stamp = float(str(raw))
except ValueError:
logger.warning(
"LeetCode credit %r has an unparsable submitted_at (%r); "
"falling back to its day key",
entry.get("entry_id"),
raw,
)
else:
start, end = window
return start <= stamp <= end
today = datetime.now().astimezone().date().isoformat()
return entry.get("day") == today
def _today_window() -> tuple[float, float]:
"""Local midnight and now, as unix seconds.
Returns:
The inclusive window a solve must fall in to count for today.
"""
moment = datetime.now().astimezone()
midnight = datetime.combine(moment.date(), time.min, moment.tzinfo)
return midnight.timestamp(), moment.timestamp()
def read_ledger_solved_today(path: Path) -> bool | None:
"""Read leetcode-guard's ledger and count today's verified credits.
@ -109,34 +42,4 @@ def read_ledger_solved_today(path: Path) -> bool | None:
True or False when the ledger could be read and verified, ``None`` when
it could not -- which is never "nothing was solved".
"""
try:
key = HMAC_KEY_FILE.read_bytes().strip()
except OSError as exc:
logger.warning("Cannot read the integrity key at %s (%s)", HMAC_KEY_FILE, exc)
return None
if not key:
logger.warning("Integrity key at %s is empty", HMAC_KEY_FILE)
return None
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except OSError as exc:
logger.warning("Cannot read the LeetCode ledger at %s (%s)", path, exc)
return None
except ValueError as exc:
logger.warning("LeetCode ledger at %s is not valid JSON (%s)", path, exc)
return None
rows = raw.get("entries") if isinstance(raw, dict) else None
if not isinstance(rows, list):
logger.warning("LeetCode ledger at %s has no entries array", path)
return None
window = _today_window()
return any(
isinstance(row, dict)
and row.get("kind") == _CREDIT
and _verified(row, key)
and _landed_today(row, window=window)
for row in rows
)
return earned_time.done_today(earned_time.LEETCODE, path, HMAC_KEY_FILE)

73
steam_backlog_enforcer/_reading_bonus.py
View File

@ -5,7 +5,7 @@ the same shape as leetcode-guard's, and screen-locker reads the same fact for a
shutdown hour. Shaped like :mod:`steam_backlog_enforcer._leetcode_bonus` where
it matters:
**Fail closed.** A missing ledger, bad JSON or an unreadable key yields
**Fail closed.** Bad JSON, a missing entries array or an unreadable key yields
``None`` -- no bonus, logged at warning. Failing open would let the coupling be
defeated by deleting a file.
@ -23,16 +23,13 @@ session -- dated by ``detail["ended_at"]``, when the reading happened.
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
from steam_backlog_enforcer._leetcode_ledger import (
HMAC_KEY_FILE,
_today_window,
_verified,
)
import earned_time
from steam_backlog_enforcer._leetcode_ledger import HMAC_KEY_FILE
from steam_backlog_enforcer._status_api import AnswerCache
if TYPE_CHECKING:
@ -49,69 +46,15 @@ def reset_cache() -> None:
_cache.clear()
def _read_today(entry: dict[str, Any], *, window: tuple[float, float]) -> bool:
"""Whether this verified credit earns the hour and was read today."""
detail = entry.get("detail")
if not isinstance(detail, dict) or detail.get("bonus") != "1":
return False
try:
ended = float(str(detail.get("ended_at")))
except ValueError:
logger.warning(
"reading credit %r has no usable ended_at", entry.get("entry_id")
)
return False
start, end = window
return start <= ended <= end
def _rows(path: Path) -> list[Any] | None:
"""The ledger's entry array, or ``None`` when it cannot be read."""
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
# No ledger = book-guard has never registered a book: an honest "no
# reading", not a fault -- and not worth a warning every 3 seconds.
return []
except OSError as exc:
logger.warning("Cannot read the book-guard ledger at %s (%s)", path, exc)
return None
except ValueError as exc:
logger.warning("book-guard ledger at %s is not valid JSON (%s)", path, exc)
return None
rows = raw.get("entries") if isinstance(raw, dict) else None
if not isinstance(rows, list):
logger.warning("book-guard ledger at %s has no entries array", path)
return None
return rows
def read_ledger_read_today(path: Path) -> bool | None:
"""Whether a verified, bonus-eligible reading session ended today.
Returns:
True or False when the ledger could be read, ``None`` when it could
not -- which is never "did not read".
not -- which is never "did not read". A ledger that does not exist
yet (no book registered) is an honest, silent "no".
"""
try:
key = HMAC_KEY_FILE.read_bytes().strip()
except OSError as exc:
logger.warning("Cannot read the integrity key at %s (%s)", HMAC_KEY_FILE, exc)
return None
if not key:
logger.warning("Integrity key at %s is empty", HMAC_KEY_FILE)
return None
rows = _rows(path)
if rows is None:
return None
window = _today_window()
return any(
isinstance(row, dict)
and row.get("kind") == "credit"
and _verified(row, key)
and _read_today(row, window=window)
for row in rows
)
return earned_time.done_today(earned_time.READING, path, HMAC_KEY_FILE)
def read_today(config: Config) -> bool | None:

12
steam_backlog_enforcer/config.py
View File

@ -92,18 +92,6 @@ class Config:
All active picks stay installed and visible; the enforcer treats them as
one allowed set. Raising this weakens enforcement proportionally.
"""
base_gaming_seconds: int = 4 * 3600
"""Floor for a day that earned nothing; also the fail-closed answer.
5h -> 4h when book-guard's reading hour arrives (from 2026-10-01; until
then ``_budget_resolve.base_for`` adds the hour back): the 8h best case is
unchanged, a day without reading is an hour shorter."""
workout_bonus_seconds: int = 2 * 3600
leetcode_bonus_seconds: int = 1 * 3600
reading_bonus_seconds: int = 1 * 3600
"""The three bonuses, added to the floor and to each other. Read
independently, so a failing LeetCode read cannot disturb the workout."""
max_gaming_seconds: int = 8 * 3600
"""Hard ceiling on the resolved budget, whatever the terms add up to."""
book_ledger_path: str = "~/.local/share/book_guard/ledger.json"
"""book-guard's signed ledger: the reading hour's only transport."""
workout_status_url: str = "http://127.0.0.1:8770/api/status"

16
steam_backlog_enforcer/tests/_budget_dates.py
View File

@ -1,8 +1,9 @@
"""Pin the day ``resolve_budget`` believes it is, around the 5h -> 4h base cut.
``_budget_resolve.base_for`` adds an hour to the floor before
``READING_BASE_FROM``. A budget test that reads the real clock would change its
answer on 2026-10-01, so every such test pins the date on one side of the cut.
``earned_time`` takes the reading hour off the floor from
``READING.penalty_from`` on. A budget test that reads the real clock would
change its answer on 2026-10-01, so every such test pins the date on one side
of the cut.
The module asks ``datetime.now().astimezone().date()``, so that chain is what
the pin answers.
"""
@ -13,14 +14,17 @@ from datetime import timedelta
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer._budget_resolve import READING_BASE_FROM
import earned_time
if TYPE_CHECKING:
from contextlib import AbstractContextManager
from datetime import date
AFTER_CUT = READING_BASE_FROM
BEFORE_CUT = READING_BASE_FROM - timedelta(days=1)
_CUT = earned_time.READING.penalty_from
assert _CUT is not None, "earned_time.READING lost its penalty_from date"
AFTER_CUT = _CUT
BEFORE_CUT = _CUT - timedelta(days=1)
def pin_today(day: date) -> AbstractContextManager[object]:

9
steam_backlog_enforcer/tests/_isolate_reading.py
View File

@ -12,6 +12,10 @@ overrides it, and the module's key path points at a file that does not exist,
so even a direct call without the test's own key fixture can never reach the
real ledger. The module memo is cleared around every test.
Any other registered earner (:mod:`steam_backlog_enforcer._ledger_earners`)
gets the same two layers: its memo is cleared and its key binding is hidden,
while ``conftest`` redirects its ``LEDGER_HOME``.
Split out of ``conftest.py`` to keep every file inside the 250-line cap;
``conftest`` imports the fixture by name, which is what registers it.
"""
@ -23,7 +27,7 @@ from unittest.mock import patch
import pytest
from steam_backlog_enforcer import _reading_bonus
from steam_backlog_enforcer import _ledger_earners, _reading_bonus
if TYPE_CHECKING:
from collections.abc import Iterator
@ -41,12 +45,15 @@ def _isolate_reading(tmp_path: Path) -> Iterator[None]:
None, with the reading earner isolated for the whole test.
"""
_reading_bonus.reset_cache()
_ledger_earners.reset_cache()
with (
patch(
"steam_backlog_enforcer._budget_resolve.read_today",
return_value=False,
),
patch.object(_reading_bonus, "HMAC_KEY_FILE", tmp_path / "no-reading-hmac.key"),
patch.object(_ledger_earners, "HMAC_KEY_FILE", tmp_path / "no-ledger-hmac.key"),
):
yield
_reading_bonus.reset_cache()
_ledger_earners.reset_cache()

8
steam_backlog_enforcer/tests/_ledger_fixtures.py
View File

@ -7,12 +7,11 @@ is what registers it there.
from __future__ import annotations
import hashlib
import hmac
import json
from typing import TYPE_CHECKING, Any
from unittest.mock import patch
import earned_time
import pytest
if TYPE_CHECKING:
@ -25,7 +24,7 @@ KEY = b"0123456789abcdef0123456789abcdef"
def sign(entry: dict[str, Any], key: bytes = KEY) -> dict[str, Any]:
"""Attach a genuine signature to an entry.
"""Attach a genuine signature to an entry, signed as every locker signs.
Args:
entry: The entry to sign.
@ -34,8 +33,7 @@ def sign(entry: dict[str, Any], key: bytes = KEY) -> dict[str, Any]:
Returns:
The same entry with an ``hmac`` field.
"""
payload = json.dumps(entry, sort_keys=True, separators=(",", ":"))
entry["hmac"] = hmac.new(key, payload.encode(), hashlib.sha256).hexdigest()
entry["hmac"] = earned_time.entry_signature(entry, key)
return entry

69
steam_backlog_enforcer/tests/_no_workout_http.py
View File

@ -9,10 +9,12 @@ those services happen to be running and whether the machine's owner has
trained or solved today. Tests would pass in the morning and fail in the
evening.
The stub returns the *fully earned* budget (base + all three bonuses = 8h),
which is what every pre-coupling test was written against. Tests that care about a
coupling patch a level below this, at ``_workout_budget._fetch_workout_today``
or ``_leetcode_bonus.read_ledger_solved_today``.
The stub returns the *fully earned* budget (base + every registered earner,
capped at the 8h ceiling), which is what every pre-coupling test was written
against. Tests that need an exact budget use :func:`fixed_budget`; tests that
care about a coupling patch a level below this, at
``_workout_budget._fetch_workout_today`` or
``_leetcode_bonus.read_ledger_solved_today``.
Split out of ``conftest.py`` to keep every file inside the 250-line cap;
``conftest`` imports the fixture by name, which is what registers it.
@ -20,42 +22,74 @@ Split out of ``conftest.py`` to keep every file inside the 250-line cap;
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from unittest.mock import patch
import earned_time
import pytest
from steam_backlog_enforcer._budget_resolve import BudgetResolution
if TYPE_CHECKING:
from collections.abc import Iterator
from contextlib import AbstractContextManager
from steam_backlog_enforcer.config import Config
_RESOLVER = "steam_backlog_enforcer._playtime_state.resolve_budget"
_MINUTE = 60.0
def _earned_budget(config: Config) -> BudgetResolution:
def earned_budget(_config: Config) -> BudgetResolution:
"""Return the fully earned budget, as a day with every bonus would.
The numbers come from the ``earned_time`` registry: today's base plus every
registered earner's term, capped at the gaming ceiling.
Args:
config: Loaded user configuration.
_config: Loaded user configuration (unused; the registry decides).
Returns:
A resolution whose seconds are base plus every bonus.
"""
base = float(config.base_gaming_seconds)
workout = float(config.workout_bonus_seconds)
leetcode = float(config.leetcode_bonus_seconds)
reading = float(config.reading_bonus_seconds)
base = earned_time.base_for(datetime.now().astimezone().date()).gaming_minutes
earned = {e.name: e.gaming_minutes * _MINUTE for e in earned_time.EARNERS}
ceiling = earned_time.GAMING_CEILING_MINUTES * _MINUTE
total = min(ceiling, base * _MINUTE + sum(earned.values()))
return BudgetResolution(
seconds=base + workout + leetcode + reading,
base_seconds=base,
workout_seconds=workout,
leetcode_seconds=leetcode,
seconds=total,
base_seconds=base * _MINUTE,
workout_seconds=earned.get("workout", 0.0),
leetcode_seconds=earned.get("leetcode", 0.0),
reason="stubbed: fully earned",
reading_seconds=reading,
reading_seconds=earned.get("reading", 0.0),
earned_seconds=earned,
)
def fixed_budget(seconds: float) -> AbstractContextManager[object]:
"""Make ``rules_for`` resolve exactly ``seconds``, all of it base.
For tests that need a budget the registry's whole minutes cannot express
(100s, 0s, ...).
Args:
seconds: The budget to resolve.
Returns:
A patch context manager replacing the resolver ``rules_for`` calls.
"""
resolution = BudgetResolution(
seconds=seconds,
base_seconds=seconds,
workout_seconds=0.0,
leetcode_seconds=0.0,
reason="stubbed: fixed",
)
return patch(_RESOLVER, return_value=resolution)
@pytest.fixture(autouse=True)
def _no_workout_http() -> Iterator[None]:
"""Stop rules_for from making a real HTTP call to the screen locker.
@ -63,8 +97,5 @@ def _no_workout_http() -> Iterator[None]:
Yields:
None, with the budget resolver stubbed for the whole test.
"""
with patch(
"steam_backlog_enforcer._playtime_state.resolve_budget",
side_effect=_earned_budget,
):
with patch(_RESOLVER, side_effect=earned_budget):
yield

2
steam_backlog_enforcer/tests/conftest.py
View File

@ -173,6 +173,8 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
"steam_backlog_enforcer._enforce_loop.CONFIG_FILE",
fake_config / "config.json",
),
# Generic earners resolve their ledgers under the real home directory.
patch("steam_backlog_enforcer._ledger_earners.LEDGER_HOME", tmp_path),
):
yield

36
steam_backlog_enforcer/tests/test_budget_resolve.py
View File

@ -107,6 +107,20 @@ class TestTheEightDays:
== resolved.seconds
)
def test_every_earner_is_in_the_generic_breakdown(self) -> None:
"""``earned_seconds`` carries the same terms as the named fields."""
with (
patch(_WORKOUT, return_value=True),
patch(_LEETCODE, return_value=False),
patch(_READING, return_value=True),
):
resolved = resolve_budget(Config())
assert resolved.earned_seconds == {
"workout": 2 * _HOUR,
"leetcode": 0.0,
"reading": _HOUR,
}
class TestFailingClosed:
"""An answer that could not be obtained is worth nothing, and says so."""
@ -159,26 +173,8 @@ class TestFailingClosed:
assert "unknown" not in answered
class TestMisconfiguration:
"""Config that would invert the incentive is refused, not enforced."""
def test_a_negative_bonus_is_clamped(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Earning something must never *cost* gaming time.
Args:
caplog: pytest's log capture.
"""
config = Config(workout_bonus_seconds=-3600)
with (
caplog.at_level(logging.ERROR),
patch(_WORKOUT, return_value=True),
patch(_LEETCODE, return_value=False),
):
resolved = resolve_budget(config)
assert resolved.seconds == 4.0 * _HOUR
assert "workout_bonus_seconds is negative" in caplog.text
class TestTheExplanation:
"""The reason string and the per-earner breakdown."""
def test_the_budget_is_explained(self, caplog: pytest.LogCaptureFixture) -> None:
"""A three-hour swing between days should never be silent.

66
steam_backlog_enforcer/tests/test_budget_resolve_reading.py
View File

@ -11,14 +11,13 @@ from datetime import timedelta
from typing import TYPE_CHECKING
from unittest.mock import patch
import earned_time
import pytest
from steam_backlog_enforcer import _budget_view as view
from steam_backlog_enforcer import _leetcode_bonus, _workout_budget
from steam_backlog_enforcer._budget_resolve import (
READING_BASE_FROM,
BudgetResolution,
base_for,
resolve_budget,
)
from steam_backlog_enforcer._playtime_state import rules_for
@ -53,16 +52,11 @@ def _clear_caches() -> Iterator[None]:
_leetcode_bonus.reset_cache()
def _resolve(
*,
earners: tuple[bool, bool, bool | None],
config: Config | None = None,
) -> BudgetResolution:
def _resolve(*, earners: tuple[bool, bool, bool | None]) -> BudgetResolution:
"""Resolve with the workout, LeetCode and reading answers fixed.
Args:
earners: The workout, LeetCode and reading answers.
config: The configuration to resolve; defaults to ``Config()``.
Returns:
The resolved budget.
@ -73,24 +67,29 @@ def _resolve(
patch(_LEETCODE, return_value=leetcode),
patch(_READING, return_value=reading),
):
return resolve_budget(config or Config())
return resolve_budget(Config())
class TestBaseFor:
class TestTheBase:
"""The floor carries the old extra hour until book-guard's gate starts."""
def test_the_day_before_the_cut_keeps_the_old_floor(self) -> None:
"""2026-09-30 still gets the hour that reading cannot yet earn back."""
assert base_for(4 * _HOUR, BEFORE_CUT) == 5 * _HOUR
with pin_today(BEFORE_CUT):
resolved = _resolve(earners=(False, False, False))
assert resolved.base_seconds == 5 * _HOUR
def test_the_cut_day_itself_uses_the_configured_floor(self) -> None:
"""The boundary is exclusive: 2026-10-01 is the first 4h day."""
assert base_for(4 * _HOUR, READING_BASE_FROM) == 4 * _HOUR
def test_the_cut_day_itself_uses_the_lower_floor(self) -> None:
"""The boundary is inclusive: 2026-10-01 is the first 4h day."""
with pin_today(AFTER_CUT):
resolved = _resolve(earners=(False, False, False))
assert resolved.base_seconds == 4 * _HOUR
def test_every_later_day_uses_the_configured_floor(self) -> None:
def test_every_later_day_uses_the_lower_floor(self) -> None:
"""No drift back to 5h once the cut has happened."""
later = READING_BASE_FROM + timedelta(days=400)
assert base_for(4 * _HOUR, later) == 4 * _HOUR
with pin_today(AFTER_CUT + timedelta(days=400)):
resolved = _resolve(earners=(False, False, False))
assert resolved.base_seconds == 4 * _HOUR
class TestTheReadingTerm:
@ -120,22 +119,6 @@ class TestTheReadingTerm:
assert resolved.seconds == 7 * _HOUR
assert "reading unknown (book-guard ledger unreadable)" in resolved.reason
def test_a_negative_reading_bonus_is_clamped(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Reading must never *cost* gaming time.
Args:
caplog: pytest's log capture.
"""
with pin_today(AFTER_CUT):
resolved = _resolve(
earners=(False, False, True),
config=Config(reading_bonus_seconds=-3600),
)
assert resolved.seconds == 4 * _HOUR
assert "reading_bonus_seconds is negative" in caplog.text
class TestTheCeiling:
"""Whatever the terms add up to, the budget never exceeds the maximum."""
@ -153,18 +136,21 @@ class TestTheCeiling:
resolved = _resolve(earners=(False, False, False))
assert resolved.seconds == 5 * _HOUR
def test_a_lower_configured_ceiling_binds(self) -> None:
with pin_today(AFTER_CUT):
resolved = _resolve(
earners=(True, True, True),
config=Config(max_gaming_seconds=6 * 3600),
)
def test_a_lower_registry_ceiling_binds(self) -> None:
"""The ceiling is the registry's, read when the budget resolves."""
with (
pin_today(AFTER_CUT),
# resolve() binds the ceiling from _policy by name, so patch there.
patch("earned_time._resolve.GAMING_CEILING_MINUTES", 6 * 60),
):
resolved = _resolve(earners=(True, True, True))
assert resolved.seconds == 6 * _HOUR
def test_after_the_cut_a_full_day_is_exactly_the_ceiling(self) -> None:
with pin_today(AFTER_CUT):
resolved = _resolve(earners=(True, True, True))
assert resolved.seconds == Config().max_gaming_seconds == 8 * _HOUR
ceiling = earned_time.GAMING_CEILING_MINUTES * 60
assert resolved.seconds == ceiling == 8 * _HOUR
class TestReadingReachesTheRulesAndTheView:

25
steam_backlog_enforcer/tests/test_budget_view.py
View File

@ -8,7 +8,6 @@ from unittest.mock import patch
from steam_backlog_enforcer import _budget_view as view
from steam_backlog_enforcer import _playtime_state as state_mod
from steam_backlog_enforcer._budget_log_tail import RunningGame, SessionView
from steam_backlog_enforcer._budget_resolve import BudgetResolution
from steam_backlog_enforcer._budget_view import (
CORRUPT,
DENIED,
@ -23,6 +22,7 @@ from steam_backlog_enforcer._budget_view import (
from steam_backlog_enforcer._playtime_history import HistoryDay
from steam_backlog_enforcer._playtime_state import PlaytimeState, rules_for
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.tests._no_workout_http import earned_budget, fixed_budget
# Built at import time, so the autouse _no_workout_http fixture is not in
# effect yet -- without this patch, collecting this module makes a real HTTP
@ -35,17 +35,7 @@ with (
patch("steam_backlog_enforcer._playtime_state.carry_into", return_value=0.0),
patch(
"steam_backlog_enforcer._playtime_state.resolve_budget",
side_effect=lambda config: BudgetResolution(
seconds=float(config.base_gaming_seconds)
+ float(config.workout_bonus_seconds)
+ float(config.leetcode_bonus_seconds)
+ float(config.reading_bonus_seconds),
base_seconds=float(config.base_gaming_seconds),
workout_seconds=float(config.workout_bonus_seconds),
leetcode_seconds=float(config.leetcode_bonus_seconds),
reason="stubbed: fully earned",
reading_seconds=float(config.reading_bonus_seconds),
),
side_effect=earned_budget,
),
):
RULES = rules_for(Config(), demo=False)
@ -112,15 +102,8 @@ class TestBuildToday:
assert today["blocked_at"] == 2.0
def test_a_zero_budget_reads_as_fully_spent(self) -> None:
rules = rules_for(
Config(
base_gaming_seconds=0,
workout_bonus_seconds=0,
leetcode_bonus_seconds=0,
reading_bonus_seconds=0,
),
demo=False,
)
with fixed_budget(0.0):
rules = rules_for(Config(), demo=False)
today = build_today(PlaytimeState(day_key="d", seconds=0.0), rules)
assert today["fraction_used"] == 1.0

12
steam_backlog_enforcer/tests/test_cmd_playtime.py
View File

@ -16,6 +16,7 @@ from steam_backlog_enforcer._playtime_state import (
save_state,
)
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.tests._no_workout_http import fixed_budget
PKG = "steam_backlog_enforcer._cmd_playtime"
@ -67,16 +68,9 @@ class TestCmdGamingStatus:
with (
patch(f"{PKG}._echo") as mock_echo,
patch(f"{PKG}.mounted_targets", return_value=set()),
fixed_budget(500.0),
):
cmd_gaming_status(
Config(
base_gaming_seconds=500,
workout_bonus_seconds=0,
leetcode_bonus_seconds=0,
reading_bonus_seconds=0,
),
State(),
)
cmd_gaming_status(Config(), State())
output = _echoed(mock_echo)
assert "used:" in output
assert "100s" in output

22
steam_backlog_enforcer/tests/test_config.py
View File

@ -71,6 +71,28 @@ class TestConfig:
cfg = Config.load()
assert cfg.steam_api_key == "k"
def test_load_drops_the_retired_budget_fields(self, tmp_path: Path) -> None:
"""A config.json written before earned_time still loads.
The budget numbers moved to the registry; a live config that still
carries them must not crash the daemon, and must not set anything.
"""
retired = {
"base_gaming_seconds": 1,
"workout_bonus_seconds": 2,
"leetcode_bonus_seconds": 3,
"reading_bonus_seconds": 4,
"max_gaming_seconds": 5,
}
config_file = tmp_path / "config.json"
config_file.write_text(
json.dumps({"steam_api_key": "k", **retired}), encoding="utf-8"
)
with patch("steam_backlog_enforcer.config.CONFIG_FILE", config_file):
cfg = Config.load()
assert cfg.steam_api_key == "k"
assert not any(hasattr(cfg, name) for name in retired)
class TestState:
"""Tests for State dataclass."""

49
steam_backlog_enforcer/tests/test_leetcode_ledger.py
View File

@ -1,11 +1,11 @@
"""Tests for _leetcode_ledger: which entries count as a solve today.
The signature check is load-bearing. That module reproduces gatelock's
canonicalisation rather than importing it, because gatelock lives in the user's
site-packages and the enforcer runs as root. ``TestSignatureCanonicalisation``
pins it to a fixed vector so the duplication cannot drift silently -- if
leetcode-guard ever changes how it signs, that test fails instead of every
credit quietly ceasing to count.
The signature check is load-bearing. The shared reader in ``earned_time``
reproduces gatelock's canonicalisation rather than importing it (gatelock pulls
in tkinter, and the enforcer runs headless as root).
``TestSignatureCanonicalisation`` pins it to a fixed vector so the duplication
cannot drift silently -- if leetcode-guard ever changes how it signs, that test
fails instead of every credit quietly ceasing to count.
"""
from __future__ import annotations
@ -13,10 +13,9 @@ from __future__ import annotations
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from steam_backlog_enforcer._leetcode_ledger import (
_verified,
read_ledger_solved_today,
)
import earned_time
from steam_backlog_enforcer._leetcode_ledger import read_ledger_solved_today
from steam_backlog_enforcer.tests._ledger_fixtures import (
KEY,
credit,
@ -34,7 +33,7 @@ __all__ = ["key_file"]
class TestSignatureCanonicalisation:
"""The duplicated eight lines must keep matching gatelock's."""
"""The shared canonicalisation must keep matching gatelock's."""
def test_a_known_vector_still_verifies(self) -> None:
"""Pinned so a change in how entries are signed fails loudly here."""
@ -53,17 +52,31 @@ class TestSignatureCanonicalisation:
},
"hmac": "4f6986bdb019bf28f607918ffb1b5bc79d252f887ade2e25aa83a31d1e49423e",
}
assert _verified(entry, KEY) is True
assert earned_time.entry_signature(entry, KEY) == entry["hmac"]
def test_a_tampered_entry_fails(self) -> None:
"""Changing any field must invalidate the signature."""
def test_a_tampered_entry_fails(self, tmp_path: Path, key_file: Path) -> None:
"""Changing any field must invalidate the signature.
Args:
tmp_path: pytest's temporary directory.
key_file: The patched signing key.
"""
assert key_file.exists()
entry = credit(when=datetime.now().astimezone())
entry["amount"] = 99
assert _verified(entry, KEY) is False
assert read_ledger_solved_today(write_ledger(tmp_path, [entry])) is False
def test_a_missing_signature_fails(self) -> None:
"""An entry with no hmac field is not a signed entry."""
assert _verified({"kind": "credit"}, KEY) is False
def test_a_missing_signature_fails(self, tmp_path: Path, key_file: Path) -> None:
"""An entry with no hmac field is not a signed entry.
Args:
tmp_path: pytest's temporary directory.
key_file: The patched signing key.
"""
assert key_file.exists()
entry = credit(when=datetime.now().astimezone())
del entry["hmac"]
assert read_ledger_solved_today(write_ledger(tmp_path, [entry])) is False
class TestReadingTheLedger:

14
steam_backlog_enforcer/tests/test_playtime.py
View File

@ -18,6 +18,7 @@ from steam_backlog_enforcer._playtime_state import (
state_path,
)
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.tests._no_workout_http import fixed_budget
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
@ -162,16 +163,9 @@ class TestSaveStateImmutability:
class TestRulesFor:
def test_production_budget_comes_from_config(self) -> None:
rules = rules_for(
Config(
base_gaming_seconds=100,
workout_bonus_seconds=0,
leetcode_bonus_seconds=0,
reading_bonus_seconds=0,
),
demo=False,
)
def test_production_budget_comes_from_the_resolver(self) -> None:
with fixed_budget(100.0):
rules = rules_for(Config(), demo=False)
assert rules.budget_seconds == 100.0
assert rules.demo is False

25
steam_backlog_enforcer/tests/test_playtime_tick.py
View File

@ -22,6 +22,7 @@ from steam_backlog_enforcer._playtime_state import (
)
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.tests._fake_session import fake_session
from steam_backlog_enforcer.tests._no_workout_http import fixed_budget
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
@ -51,8 +52,11 @@ def quiet_tick() -> object:
yield mocks
def _rules(*, demo: bool = False, **cfg: object) -> object:
return rules_for(Config(**cfg), demo=demo)
def _rules(*, demo: bool = False, budget: float | None = None) -> object:
if budget is None:
return rules_for(Config(), demo=demo)
with fixed_budget(budget):
return rules_for(Config(), demo=demo)
class TestStateOrRecover:
@ -71,12 +75,7 @@ class TestStateOrRecover:
def test_fails_closed_when_state_is_gone_but_mounts_remain(self) -> None:
"""Deleting the state file must not be a way to lift the block."""
rules = _rules(
base_gaming_seconds=100,
workout_bonus_seconds=0,
leetcode_bonus_seconds=0,
reading_bonus_seconds=0,
)
rules = _rules(budget=100.0)
with patch(f"{PKG}.mounted_targets", return_value={"/usr/bin/steam"}):
out = _state_or_recover(rules, now=NOW)
assert out.seconds == 100.0
@ -87,15 +86,7 @@ class TestStateOrRecover:
state_path(demo=False).write_text("{bad", encoding="utf-8")
with patch(f"{PKG}.mounted_targets", return_value={"/usr/bin/steam"}):
out = _state_or_recover(
_rules(
base_gaming_seconds=50,
workout_bonus_seconds=0,
leetcode_bonus_seconds=0,
reading_bonus_seconds=0,
),
now=NOW,
)
out = _state_or_recover(_rules(budget=50.0), now=NOW)
assert out.is_blocked() is True