diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 55c435a..dcec611 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -134,6 +134,7 @@ repos: # than merely an old one. - gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.9.0#subdirectory=gatelock - crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.1#subdirectory=crdt-sync + - earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time - freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.0#subdirectory=freedays - mcp>=2,<3 - pydantic @@ -154,6 +155,7 @@ repos: # than merely an old one. - gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.9.0#subdirectory=gatelock - crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.1#subdirectory=crdt-sync + - earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time - freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.0#subdirectory=freedays - mcp>=2,<3 - pydantic diff --git a/AGENTS.md b/AGENTS.md index 4730f19..d25c6fd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,29 +35,31 @@ weeks because a missing token `return`ed `[]` and real errors only logged at ## The shutdown hour is derived, never incremented into `/etc/shutdown-schedule.conf` is recomputed from scratch by -`_shutdown_base.reset_to_base_if_new_day` on each new day: -`BASE_HOUR (19) + today's workout hours + the LeetCode hour + the reading hour`, -capped at 23. +`_shutdown_base.reset_to_base_if_new_day` on each new day: the base plus +every earned term, capped at 23. The numbers are NOT in this repo: they live +in the shared earner registry `earned_time` (kuhyx/utils, `earned_time/`), +which steam-backlog-enforcer reads for gaming time too. Base = 20:00 minus +each earner's cut once its `penalty_from` day arrives (reading, from +2026-10-01 -> 19:00); terms = workout 2h + 1h per further session, LeetCode +1h, reading 1h, plus whatever is registered later. Anything that pushes the hour with a plain read-add-write and is *not* a term of that derivation gets wiped by the next reset -- that is how a 00:02 workout -lost its hours on 2026-09-13. Add a new bonus source as a term first, then -(optionally) as a live pass stamped in `shutdown_base.json` so the two do not -double-apply. The base is the constant, not the state file: the file once -persisted `base_*_hour`, which made the code's default dead. +lost its hours on 2026-09-13. A new bonus source is an `Earner` in the +registry, never code here: the reset includes it as a term, and the live pass +(`apply_flat_bonuses_if_new`, stamped `_bonus_date` in +`shutdown_base.json`) picks it up for flat earners. The base is the registry +constant, not the state file: the file once persisted `base_*_hour`, which +made the code's default dead. -The LeetCode hour is read from leetcode-guard's ledger -(`_leetcode_bonus.py`): HMAC-verified `credit` entries, keyed on LeetCode's -own `submitted_at`, flat +1h/day, and *fail closed* -- an unreadable ledger or -key earns nothing. leetcode-guard itself stays read-only; it never writes -the config. +Flat earners are read from their gate's ledger by `earned_time.done_today` +via `_earned.py`: HMAC-verified `credit` rows matched by the earner's own +rule (LeetCode's `submitted_at`, book-guard's `detail.bonus == "1"` + +`ended_at`), and *fail closed* -- an unreadable ledger or key earns nothing, +logged. The gates stay read-only; none writes the config. -The reading hour is the same shape, from book-guard's ledger -(`_reading_bonus.py`): HMAC-verified `credit` entries with `detail.bonus == -"1"`, dated by `detail.ended_at` (when the reading happened, not when the quiz -was passed), flat +1h/day, fail closed. The base moves 20 -> 19 on -2026-10-01 (`base_hour()`, `READING_BASE_FROM`) -- book-guard's start date, so -the cut never lands before the hour that pays for it; the 23:00 best case is -unchanged. +The schedule stores whole hours; `_earned.to_hours` raises on a remainder +instead of flooring, so a 30-minute earner needs the schedule to learn +minutes first. ## The morning session is the carrot — and the early-bird window is it diff --git a/install_systemd.sh b/install_systemd.sh index 15eb302..0585a2b 100755 --- a/install_systemd.sh +++ b/install_systemd.sh @@ -89,7 +89,7 @@ fi rm -f "$USER_SERVICE_DIR/workout-locker.timer" # Seed shutdown_base.json if not already present. It holds only date stamps; -# the base hour itself is BASE_HOUR in screen_locker/_shutdown_base.py. +# the base hour itself comes from the earned_time registry (kuhyx/utils). SHUTDOWN_BASE="$SCRIPT_DIR/screen_locker/shutdown_base.json" if [[ ! -f "$SHUTDOWN_BASE" ]]; then printf '{\n "last_reset_date": ""\n}\n' > "$SHUTDOWN_BASE" diff --git a/pyproject.toml b/pyproject.toml index db8c50e..9f25aaf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.9.0#subdirectory=gatelock", "crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.1#subdirectory=crdt-sync", "freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.0#subdirectory=freedays", + "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time", ] [project.scripts] diff --git a/requirements.txt b/requirements.txt index fd21249..514f670 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,6 +4,7 @@ bandit==1.9.4 codespell==2.4.3 coverage==7.16.2 crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.1#subdirectory=crdt-sync +earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.0#subdirectory=freedays gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.9.0#subdirectory=gatelock mcp==2.3.0 diff --git a/screen_locker/_earned.py b/screen_locker/_earned.py new file mode 100644 index 0000000..8025d60 --- /dev/null +++ b/screen_locker/_earned.py @@ -0,0 +1,90 @@ +"""Shutdown hours from the shared earner registry (``earned_time``, kuhyx/utils). + +Every gate's reward -- the workout, the LeetCode hour, the reading hour and +whatever is registered next -- is defined once in ``earned_time``, which +steam-backlog-enforcer reads for gaming time too. This module is the +screen-locker side of it: where the ledgers live, and the hour conversion. + +The gates publish facts only; none writes the shutdown config. Their ledgers +are HMAC-verified against :data:`HMAC_KEY_FILE` and *fail closed* -- an +unreadable ledger or key earns nothing, and says so. + +**Whole hours only, for now.** ``/etc/shutdown-schedule.conf`` stores hours, +while the registry is in minutes. :func:`to_hours` refuses a remainder rather +than flooring it, so a 30-minute earner cannot silently become 0 here before +the schedule learns minutes. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import TYPE_CHECKING, Final + +import earned_time + +from screen_locker._constants import HMAC_KEY_FILE + +if TYPE_CHECKING: + from datetime import datetime + +_logger: Final = logging.getLogger(__name__) + +# Every earner's ``ledger`` is relative to this. Read-only from here; resolved +# at call time so the test suite's redirect (tests/_isolated_state.py) applies. +LEDGER_HOME: Final = Path.home() + +_MINUTES_PER_HOUR: Final = 60 + + +def to_hours(minutes: int) -> int: + """Convert a registry amount to the schedule's whole hours. + + Raises: + ValueError: ``minutes`` is not a whole number of hours. The shutdown + schedule cannot express it, and flooring would quietly drop it. + """ + hours, remainder = divmod(minutes, _MINUTES_PER_HOUR) + if remainder: + msg = ( + f"{minutes} minutes is not a whole hour; /etc/shutdown-schedule.conf " + "stores hours only" + ) + raise ValueError(msg) + return hours + + +def ledger_file(earner: earned_time.Earner) -> Path: + """Where ``earner``'s gate keeps its ledger.""" + if earner.ledger is None: + msg = f"earner {earner.name!r} has no ledger" + raise ValueError(msg) + return LEDGER_HOME / earner.ledger + + +def flat_earners() -> tuple[earned_time.Earner, ...]: + """The once-per-day, ledger-backed earners (the workout is counted here).""" + return tuple( + e for e in earned_time.EARNERS if e.kind == "flat" and e.ledger is not None + ) + + +def earned_today( + earner: earned_time.Earner, now: datetime | None = None +) -> bool | None: + """Whether ``earner``'s gate recorded today's credit; ``None`` if unknown.""" + return earned_time.done_today(earner, ledger_file(earner), HMAC_KEY_FILE, now=now) + + +def flat_answers() -> dict[str, bool | None]: + """Today's answer from every flat earner, an unknown one logged.""" + answers: dict[str, bool | None] = {} + for earner in flat_earners(): + answer = earned_today(earner) + if answer is None: + _logger.warning( + "%s state could not be checked; no shutdown time for it", + earner.label, + ) + answers[earner.name] = answer + return answers diff --git a/screen_locker/_leetcode_bonus.py b/screen_locker/_leetcode_bonus.py deleted file mode 100644 index 0da1d68..0000000 --- a/screen_locker/_leetcode_bonus.py +++ /dev/null @@ -1,183 +0,0 @@ -"""One extra shutdown hour on a day with an accepted LeetCode submission. - -leetcode-guard never writes the shutdown config itself; it only publishes the -fact through its ledger, the same way ``steam-backlog-enforcer`` reads it for -an hour of gaming budget. The write has to live here because -:func:`~screen_locker._shutdown_base.reset_to_base_if_new_day` recomputes the -config from scratch on every new day -- a bonus written by another program -would be wiped by that reset, exactly as a pre-tick workout credit was on -2026-09-13. So the hour is a *term* of the derived target, not an increment -that survives by luck. - -Three things are pinned deliberately, mirroring leetcode-guard's own -``solved_today`` contract: - -**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 -happens from banked credit, from the escape hatch and from a classified -outage, so it can never stand in for a solve. - -**The solve time is ``detail["submitted_at"]``, not ``day``.** ``day`` is the -*harvesting* run's date, and a problem solved at 23:50 and harvested the next -morning carries the next day's key. It earns for the day it was solved on, -which by then is over -- never for the next one. - -**"Cannot check" is not "not solved".** An unreadable ledger or key yields -``None``, and the caller fails closed to no hour. Failing open would let the -coupling be defeated by deleting a file. -""" - -from __future__ import annotations - -from datetime import UTC, datetime, time -import json -import logging -from pathlib import Path -from typing import Final - -from gatelock.log_integrity import verify_entry_hmac - -from screen_locker._constants import HMAC_KEY_FILE - -_logger: Final = logging.getLogger(__name__) - -# Flat, once per day, regardless of how many problems were solved -- the -# workout is the harder thing and keeps the bigger, per-session reward. -LEETCODE_BONUS_HOURS: Final = 1 - -# leetcode-guard's real ledger (its DATA_DIR / "ledger.json"). Read-only from -# here; resolved at call time so the test suite's path redirect applies. -LEETCODE_LEDGER_FILE: Final = Path.home() / ".local/share/leetcode_guard/ledger.json" - -_CREDIT: Final = "credit" - - -def _today_window(now: datetime | None = None) -> tuple[float, float]: - """Local midnight and now, as unix seconds. - - Args: - now: Stand-in for the current local time, for tests. - - Returns: - The inclusive window a solve must fall in to count for today. - """ - moment = (now or datetime.now(tz=UTC)).astimezone() - midnight = datetime.combine(moment.date(), time.min, moment.tzinfo) - return midnight.timestamp(), moment.timestamp() - - -def _landed_today(entry: dict[str, object], *, window: tuple[float, float]) -> bool: - """Whether this verified credit's accepted submission happened today. - - Args: - entry: One ledger entry, already known to be a verified credit. - window: Local midnight and now, as unix seconds. - - Returns: - Whether the solve falls inside today. Without a usable - ``submitted_at`` the harvest ``day`` is the fallback: it runs late, - never early, so it can miss a late solve but 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 - start, _ = window - today = datetime.fromtimestamp(start).astimezone().date().isoformat() - return entry.get("day") == today - - -def _read_entries(ledger_path: Path, label: str = "LeetCode") -> list[object] | None: - """Read the ledger's entry list, or ``None`` when it cannot be read. - - Args: - ledger_path: leetcode-guard's (or book-guard's) ledger file. - label: Whose ledger, for the log line. - - Returns: - The raw entries, or ``None`` -- which is never "no entries". - """ - try: - raw = json.loads(ledger_path.read_text(encoding="utf-8")) - except OSError as exc: - _logger.warning("Cannot read the %s ledger at %s (%s)", label, ledger_path, exc) - return None - except ValueError as exc: - _logger.warning( - "%s ledger at %s is not valid JSON (%s)", label, ledger_path, exc - ) - return None - rows = raw.get("entries") if isinstance(raw, dict) else None - if not isinstance(rows, list): - _logger.warning("%s ledger at %s has no entries array", label, ledger_path) - return None - return rows - - -def leetcode_solved_today( - ledger_path: Path | None = None, - *, - now: datetime | None = None, -) -> bool | None: - """Whether a verified accepted submission landed today, locally. - - Args: - ledger_path: The ledger to read. None resolves - :data:`LEETCODE_LEDGER_FILE` at call time. - now: Stand-in for the current local time, for tests. - - Returns: - True or False when the ledger could be read and verified, ``None`` - when it could not -- which is never "nothing was solved". - """ - target = LEETCODE_LEDGER_FILE if ledger_path is None else ledger_path - try: - key_usable = bool(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_usable: - _logger.warning("Integrity key at %s is empty", HMAC_KEY_FILE) - return None - - rows = _read_entries(target) - if rows is None: - return None - - window = _today_window(now) - return any( - isinstance(row, dict) - and row.get("kind") == _CREDIT - and verify_entry_hmac(row, key_file=HMAC_KEY_FILE) - and _landed_today(row, window=window) - for row in rows - ) - - -def leetcode_bonus_hours(ledger_path: Path | None = None) -> int: - """The shutdown hours today's LeetCode solve has earned, failing closed. - - Args: - ledger_path: The ledger to read. None resolves the real one. - - Returns: - :data:`LEETCODE_BONUS_HOURS` on a day with a verified solve, else 0. - A ledger that cannot be checked is 0 too, and logged -- it must never - be mistaken for a confident "not solved", but it earns nothing either. - """ - solved = leetcode_solved_today(ledger_path) - if solved is None: - _logger.warning("LeetCode solve state could not be checked; no shutdown hour") - return 0 - return LEETCODE_BONUS_HOURS if solved else 0 diff --git a/screen_locker/_reading_bonus.py b/screen_locker/_reading_bonus.py deleted file mode 100644 index da5c4a6..0000000 --- a/screen_locker/_reading_bonus.py +++ /dev/null @@ -1,106 +0,0 @@ -"""One extra shutdown hour on a day with a credited paper-book reading session. - -book-guard (``~/src/book-guard``) publishes reading through its ledger, the -same way leetcode-guard publishes solves, and steam-backlog-enforcer reads the -same fact for an hour of gaming. Like the LeetCode hour it is a *term* of the -daily derivation in :mod:`screen_locker._shutdown_base`, never an increment -another program writes into the config. - -The contract, pinned in book-guard's ``_quiz.record_verdict``: - -**Only ``credit`` rows count**, and only those with ``detail["bonus"] == -"1"`` -- book-guard sets that when the session passed its quiz *and* was big -enough (20+ pages, 20+ minutes). A ``reject`` is a failed quiz; an -``escape`` forgives a lock and earns nothing. - -**The reading time is ``detail["ended_at"]``**, not the row's creation: a -session read at 22:30 and quizzed next morning earns for the evening it was -read, which by then is over -- never for the quiz day. - -**"Cannot check" is not "did not read"**: an unreadable ledger or key yields -``None`` and the caller fails closed to no hour. -""" - -from __future__ import annotations - -from datetime import UTC, datetime -import logging -from pathlib import Path -from typing import Final - -from gatelock.log_integrity import verify_entry_hmac - -from screen_locker._constants import HMAC_KEY_FILE -from screen_locker._leetcode_bonus import _read_entries, _today_window - -_logger: Final = logging.getLogger(__name__) - -READING_BONUS_HOURS: Final = 1 - -# book-guard's real ledger. Read-only from here; resolved at call time so the -# test suite's path redirect (tests/_isolated_state.py) applies. -READING_LEDGER_FILE: Final = Path.home() / ".local/share/book_guard/ledger.json" - - -def _read_today(entry: dict[str, object], *, window: tuple[float, float]) -> bool: - """Whether this verified credit is bonus-eligible 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 read_today( - ledger_path: Path | None = None, - *, - now: datetime | None = None, -) -> bool | None: - """Whether a verified, bonus-eligible reading session ended today, locally. - - Returns: - True or False when the ledger could be read and verified, ``None`` - when it could not -- which is never "did not read". - """ - target = READING_LEDGER_FILE if ledger_path is None else ledger_path - try: - key_usable = bool(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_usable: - _logger.warning("Integrity key at %s is empty", HMAC_KEY_FILE) - return None - if not target.exists(): - return False # book-guard never registered a book: no reading, no fault - rows = _read_entries(target, "book-guard") - if rows is None: - return None - window = _today_window(now or datetime.now(tz=UTC)) - return any( - isinstance(row, dict) - and row.get("kind") == "credit" - and verify_entry_hmac(row, key_file=HMAC_KEY_FILE) - and _read_today(row, window=window) - for row in rows - ) - - -def reading_bonus_hours(ledger_path: Path | None = None) -> int: - """The shutdown hours today's reading has earned, failing closed. - - A ledger that does not exist yet (book-guard never ran) is a plain 0, - logged by :func:`_read_entries` like any unreadable one. - """ - done = read_today(ledger_path) - if done is None: - _logger.warning("Reading state could not be checked; no shutdown hour") - return 0 - return READING_BONUS_HOURS if done else 0 diff --git a/screen_locker/_shutdown_base.py b/screen_locker/_shutdown_base.py index c8a7b45..a2e18ed 100644 --- a/screen_locker/_shutdown_base.py +++ b/screen_locker/_shutdown_base.py @@ -1,23 +1,24 @@ """Daily shutdown-time base reset for the screen locker. -On each new calendar day the shutdown config is reset to the base hour -(:data:`BASE_HOUR`, 19:00) so that the day's bonuses always layer on top of a -known floor rather than accumulating indefinitely across days. +On each new calendar day the shutdown config is reset to the day's base so +that the day's bonuses always layer on top of a known floor rather than +accumulating indefinitely across days. -The base is a constant, not state. It used to be persisted in the state file -and read back on every reset, which made the code's default dead: the file -kept re-writing whatever it was installed with, so changing the constant -changed nothing on the machine. The state file now carries only date stamps. +The base and every bonus come from the shared earner registry +(``earned_time``, see :mod:`screen_locker._earned`): 20:00 before any penalty, +minus each earner's cut once its ``penalty_from`` day arrives (book-guard's +reading hour, from 2026-10-01). Nothing here is state. The base used to be +persisted in the state file and read back on every reset, which made the +code's default dead; the file now carries only date stamps. The reset re-derives what today has ALREADY earned and writes base plus that -in one go: the workout hours from today's log, the LeetCode hour from -leetcode-guard's ledger (:mod:`screen_locker._leetcode_bonus`) and the reading -hour from book-guard's (:mod:`screen_locker._reading_bonus`). Live credits -are read-add-write against the config, so a workout credited between local -midnight and the first locker tick of the day (a manual log at 00:02, say) -had its hours wiped by the reset that followed -- on 2026-09-13 that left the -bar at 21:00 with a counted workout on disk. Any bonus source that is not a -term of this derivation has that bug. +in one go: the workout hours from today's log and every flat earner's hour +from its gate's ledger. Live credits are read-add-write against the config, +so a workout credited between local midnight and the first locker tick of the +day (a manual log at 00:02, say) had its hours wiped by the reset that +followed -- on 2026-09-13 that left the bar at 21:00 with a counted workout on +disk. Any bonus source that is not a term of this derivation has that bug; +registering it in ``earned_time`` makes it one. The sick-day state file is cleared on reset so the sick-restore path cannot overwrite the fresh base when it runs later in the same startup. @@ -30,61 +31,46 @@ import json import logging from typing import TYPE_CHECKING, Any +import earned_time + from screen_locker._day import today_str -from screen_locker._leetcode_bonus import LEETCODE_BONUS_HOURS, leetcode_bonus_hours +from screen_locker._earned import earned_today, flat_answers, flat_earners, to_hours from screen_locker._log_io import load_workout_log -from screen_locker._reading_bonus import READING_BONUS_HOURS, reading_bonus_hours from screen_locker._weekly_check import count_day_credits -from screen_locker._workout_credit import earned_shutdown_bonus_hours if TYPE_CHECKING: - from collections.abc import Callable from pathlib import Path _logger = logging.getLogger(__name__) -# The hour every day starts from before any bonus. Moved 21 -> 20 on -# 2026-09-15, and 20 -> 19 together with book-guard's reading hour: the -# ceiling stays 23:00, so every day's best case (workouts + LeetCode + reading) -# is unchanged and a day without reading ends an hour earlier. The cut waits -# for book-guard's gate (2026-10-01) -- taking an hour the reader cannot yet -# earn back was a same-day loss when it first shipped on 2026-09-26. -BASE_HOUR = 19 -READING_BASE_FROM = date(2026, 10, 1) - def base_hour(day: date | None = None) -> int: - """The base for ``day`` (default today): 20 before the reading cut, then 19.""" + """The base for ``day`` (default today), every penalty in force taken off.""" target = day or datetime.now().astimezone().date() - return BASE_HOUR if target >= READING_BASE_FROM else BASE_HOUR + 1 + return to_hours(earned_time.base_for(target, earned_time.EARNERS).shutdown_minutes) -# Mirrors RESTORE_CEILING in adjust_shutdown_schedule.sh: the script clamps any -# restore above it, so the target is computed against the same ceiling here -# rather than asking for hours the write will silently cut. -_RESTORE_CEILING_HOUR = 23 - -_LEETCODE_STAMP = "leetcode_bonus_date" -_READING_STAMP = "reading_bonus_date" +def _stamp(earner: earned_time.Earner) -> str: + """The once-per-day key of ``earner``'s live pass in the state file.""" + return f"{earner.name}_bonus_date" -def today_earned_bonus_hours(log_file: Path, today: str) -> int: - """Return the shutdown hours today's logged workouts have already earned. +def today_credit_count(log_file: Path, today: str) -> int: + """Return how many distinct workouts today's log counts. Counted with the same :func:`~screen_locker._weekly_check.count_day_credits` rule as the weekly total, so a workout recorded twice earns once. """ entries = load_workout_log(log_file).get(today, []) - credit_count = count_day_credits(today, [e for e in entries if isinstance(e, dict)]) - return earned_shutdown_bonus_hours(credit_count) + return count_day_credits(today, [e for e in entries if isinstance(e, dict)]) def _load_state(state_file: Path) -> dict[str, Any]: """Return the reset state, or an empty dict when it is missing or corrupt. A corrupt file is logged and treated as "nothing stamped": the reset then - runs again, which is idempotent, and the LeetCode hour is re-applied at - worst once, capped by the ceiling. + runs again, which is idempotent, and a flat hour is re-applied at worst + once, capped by the ceiling. """ if not state_file.exists(): return {} @@ -130,9 +116,9 @@ def reset_to_base_if_new_day( ) -> bool: """Reset the shutdown config to the base hour if a new calendar day began. - Writes :data:`BASE_HOUR` plus whatever today has already earned -- the - workout hours in *log_file* (see :func:`today_earned_bonus_hours`) and the - LeetCode hour -- via *mixin*._write_shutdown_config (with restore=True so + Writes the base plus whatever today has already earned -- the workout + hours in *log_file* (see :func:`today_credit_count`) and every flat + earner's hour -- via *mixin*._write_shutdown_config (with restore=True so the script allows moving the time earlier), stamps ``last_reset_date`` in *state_file*, and removes *sick_day_state_file* if it exists so the sick-restore path does not fight with the fresh base on the same startup. @@ -143,11 +129,16 @@ def reset_to_base_if_new_day( if _load_state(state_file).get("last_reset_date") == today: return False - earned = today_earned_bonus_hours(log_file, today) if log_file is not None else 0 - leetcode = leetcode_bonus_hours() - reading = reading_bonus_hours() - base = base_hour() - target = min(_RESTORE_CEILING_HOUR, base + earned + leetcode + reading) + answers: dict[str, int | bool | None] = dict(flat_answers()) + answers[earned_time.WORKOUT.name] = ( + today_credit_count(log_file, today) if log_file is not None else 0 + ) + # The registry is passed explicitly: flat_answers iterated this same + # tuple, so the earners asked and the earners summed can never differ. + day = earned_time.resolve( + answers, datetime.now().astimezone().date(), earned_time.EARNERS + ) + target = to_hours(day.shutdown_minutes) # Preserve the morning-end hour from the live config. config = mixin._read_shutdown_config() @@ -163,20 +154,19 @@ def reset_to_base_if_new_day( # The flat hours are stamped here too: the reset already included them, # so the live pass below must not add them a second time today. new_state: dict[str, Any] = {"last_reset_date": today} - if leetcode: - new_state[_LEETCODE_STAMP] = today - if reading: - new_state[_READING_STAMP] = today + for term in day.terms: + if term.earner.kind == "flat" and term.shutdown_minutes: + new_state[_stamp(term.earner)] = today _save_state(state_file, new_state) + earned = " + ".join( + f"{to_hours(t.shutdown_minutes)}h {t.earner.label}" for t in day.terms + ) _logger.info( - "Daily base reset: %02d:00 (base %d + %dh workout + %dh LeetCode + " - "%dh reading already earned today).", + "Daily base reset: %02d:00 (base %d + %s already earned today).", target, - base, + to_hours(day.base.shutdown_minutes), earned, - leetcode, - reading, ) return True @@ -184,60 +174,46 @@ def reset_to_base_if_new_day( def _apply_flat_bonus( state_file: Path, mixin: object, - *, - stamp: str, - hours: Callable[[], int], - label: str, + earner: earned_time.Earner, ) -> bool: - """Push shutdown later by a once-per-day flat hour, if it is earned. + """Push shutdown later by one earner's once-per-day hour, if it is earned. The daily reset only sees a bonus earned before it ran; this is the live counterpart for the usual case, where the solve or the reading lands hours later and is picked up by the next timer tick. Idempotent through - the ``stamp`` key in *state_file*, which the reset sets as well when it + the earner's stamp in *state_file*, which the reset sets as well when it already included the hour. Returns True if the hour was applied on this call. """ today = today_str() state = _load_state(state_file) + stamp = _stamp(earner) if state.get(stamp) == today: return False - earned = hours() - if earned == 0: + answer = earned_today(earner) + if answer is None: + _logger.warning( + "%s state could not be checked; no shutdown time for it", earner.label + ) return False - if not mixin._adjust_shutdown_time_by(earned): - _logger.warning("%s bonus: failed to write shutdown config.", label) + if not answer: + return False + hours = to_hours(earner.shutdown_minutes) + if not mixin._adjust_shutdown_time_by(hours): + _logger.warning("%s bonus: failed to write shutdown config.", earner.label) return False state[stamp] = today _save_state(state_file, state) - _logger.info("%s bonus: +%dh shutdown time today.", label, earned) + _logger.info("%s bonus: +%dh shutdown time today.", earner.label, hours) return True -def apply_leetcode_bonus_if_new(state_file: Path, mixin: object) -> bool: - """The LeetCode hour, live (see :func:`_apply_flat_bonus`).""" - return _apply_flat_bonus( - state_file, - mixin, - stamp=_LEETCODE_STAMP, - hours=lambda: LEETCODE_BONUS_HOURS if leetcode_bonus_hours() else 0, - label="LeetCode", - ) - - -def apply_reading_bonus_if_new(state_file: Path, mixin: object) -> bool: - """The reading hour, live (see :func:`_apply_flat_bonus`).""" - return _apply_flat_bonus( - state_file, - mixin, - stamp=_READING_STAMP, - hours=lambda: READING_BONUS_HOURS if reading_bonus_hours() else 0, - label="Reading", - ) - - def apply_flat_bonuses_if_new(state_file: Path, mixin: object) -> None: - """Every once-per-day flat hour: LeetCode, then reading.""" - apply_leetcode_bonus_if_new(state_file, mixin) - apply_reading_bonus_if_new(state_file, mixin) + """Every once-per-day flat hour, in registry order (LeetCode, reading, ...). + + The workout is not among them: it is counted, and its live credit is + applied by :class:`~screen_locker._workout_credit.WorkoutCreditMixin`. + """ + for earner in flat_earners(): + _apply_flat_bonus(state_file, mixin, earner) diff --git a/screen_locker/_workout_credit.py b/screen_locker/_workout_credit.py index 9277839..368a4d5 100644 --- a/screen_locker/_workout_credit.py +++ b/screen_locker/_workout_credit.py @@ -13,7 +13,10 @@ from dataclasses import dataclass import logging from typing import TYPE_CHECKING +import earned_time + from screen_locker import _sick_tracker +from screen_locker._earned import to_hours from screen_locker._weekly_check import ( COUNTED_WORKOUT_TYPES, PC_WORKOUT_TYPE, @@ -25,21 +28,19 @@ if TYPE_CHECKING: _logger = logging.getLogger(__name__) -# The shutdown reward for a day's workouts: the first counted workout pushes -# shutdown later by FIRST_WORKOUT_BONUS_HOURS, every further one by -# EXTRA_WORKOUT_BONUS_HOURS. Both the live credit paths below and the daily -# base reset (``_shutdown_base``) derive from these two numbers, so a day's -# earned hours can be recomputed from the log instead of being lost when the -# credit landed before the reset ran. -FIRST_WORKOUT_BONUS_HOURS = 2 -EXTRA_WORKOUT_BONUS_HOURS = 1 +# The shutdown reward for a day's workouts, from the shared registry +# (``earned_time.WORKOUT``): the first counted workout pushes shutdown later by +# FIRST_WORKOUT_BONUS_HOURS, every further one by EXTRA_WORKOUT_BONUS_HOURS. +# The live credit paths below and the daily base reset (``_shutdown_base``) +# both derive from the registry, so a day's earned hours can be recomputed +# from the log instead of being lost when the credit landed before the reset. +FIRST_WORKOUT_BONUS_HOURS = to_hours(earned_time.WORKOUT.shutdown_minutes) +EXTRA_WORKOUT_BONUS_HOURS = to_hours(earned_time.WORKOUT.extra_shutdown_minutes) def earned_shutdown_bonus_hours(credit_count: int) -> int: """Return the shutdown hours a day with *credit_count* distinct workouts earned.""" - if credit_count <= 0: - return 0 - return FIRST_WORKOUT_BONUS_HOURS + (credit_count - 1) * EXTRA_WORKOUT_BONUS_HOURS + return to_hours(earned_time.WORKOUT.shutdown_for(credit_count)) @dataclass(frozen=True) diff --git a/screen_locker/tests/_earned_fixtures.py b/screen_locker/tests/_earned_fixtures.py new file mode 100644 index 0000000..6258dc0 --- /dev/null +++ b/screen_locker/tests/_earned_fixtures.py @@ -0,0 +1,115 @@ +"""Shared helpers for the earner tests: signed ledgers and canned answers. + +Ledgers are written where the code under test reads them -- under +``_earned.LEDGER_HOME``, which the autouse redirect (``_isolated_state.py``) +points at tmp_path -- and signed with ``earned_time.entry_signature`` against +a temp key patched onto ``_earned.HMAC_KEY_FILE``. ``_earned`` holds its own +binding of the key and the suite does not redirect it, so a test that reads a +ledger without :func:`signing_key` would read the host's real key. + +Plain helpers rather than fixtures, so test modules wrap them in their own +fixtures instead of importing fixture names they never reference. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from datetime import datetime +import json +from typing import TYPE_CHECKING, Any, Final +from unittest.mock import MagicMock, patch + +import earned_time + +from screen_locker import _earned, _shutdown_base + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +KEY: Final = b"test-key-bytes" + + +def _any_credit(row: dict[str, object], window: tuple[float, float]) -> bool: + """EXTRA's rule: any verified credit counts, whenever it landed.""" + del row, window + return True + + +# A flat earner the registry does not have yet, for tests that register one +# (``monkeypatch.setattr(earned_time, "EARNERS", (*earned_time.EARNERS, EXTRA))``) +# to pin that nothing in screen-locker needs code for it. +EXTRA: Final = earned_time.Earner( + name="extra", + label="Extra", + gaming_minutes=60, + shutdown_minutes=60, + ledger=".local/share/extra_guard/ledger.json", + match=_any_credit, +) + + +@contextmanager +def signing_key(tmp_path: Path, content: bytes = KEY) -> Iterator[Path]: + """Write ``content`` as the HMAC key and point ``_earned`` at it.""" + key = tmp_path / "hmac.key" + key.write_bytes(content) + with patch.object(_earned, "HMAC_KEY_FILE", key): + yield key + + +def signed(body: dict[str, Any]) -> dict[str, Any]: + """``body`` plus the ``hmac`` a genuine gate would have written.""" + return {**body, "hmac": earned_time.entry_signature(body, KEY)} + + +def write_ledger(earner: earned_time.Earner, entries: list[Any]) -> Path: + """Write ``entries`` as ``earner``'s ledger, where screen-locker reads it.""" + path = _earned.ledger_file(earner) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"entries": entries}), encoding="utf-8") + return path + + +def credit(earner: earned_time.Earner, when: datetime | None = None) -> dict[str, Any]: + """A signed credit row that counts for ``earner`` at ``when`` (default now). + + Whole seconds, floored: a stamp can never land after the reader's "now". + """ + moment = when or datetime.now().astimezone() + stamp = int(moment.timestamp()) + detail: dict[str, Any] = ( + {"bonus": "1", "ended_at": str(stamp)} + if earner.name == earned_time.READING.name + else {"submitted_at": stamp} + ) + return signed( + { + "entry_id": f"{earner.name}:{stamp}", + "kind": "credit", + "day": moment.date().isoformat(), + "detail": detail, + } + ) + + +@contextmanager +def answering(answers: dict[str, bool | None]) -> Iterator[MagicMock]: + """Every flat earner answers from ``answers`` (absent: ``False``). + + ``earned_today`` is bound twice -- the reset reaches it through + ``_earned.flat_answers``, the live pass through ``_shutdown_base`` -- so + both names get the same mock. ``answers`` is read at call time; a test may + change it after entering. + """ + + def _answer(earner: earned_time.Earner, now: datetime | None = None) -> bool | None: + del now + return answers.get(earner.name, False) + + mock = MagicMock(side_effect=_answer) + with ( + patch.object(_earned, "earned_today", mock), + patch.object(_shutdown_base, "earned_today", mock), + ): + yield mock diff --git a/screen_locker/tests/_isolated_state.py b/screen_locker/tests/_isolated_state.py index 67d5fb1..7ebb676 100644 --- a/screen_locker/tests/_isolated_state.py +++ b/screen_locker/tests/_isolated_state.py @@ -72,18 +72,20 @@ ISOLATED_STATE: tuple[tuple[str, tuple[str, ...]], ...] = ( "instance.lock", ("_constants.INSTANCE_LOCK_FILE", "screen_lock.INSTANCE_LOCK_FILE"), ), - # leetcode-guard's ledger. Read-only from here, but a test must never see - # the user's real solves: a real credit would earn a fake shutdown hour - # inside an assertion written against an empty day. + # The home every earner's ledger (leetcode-guard's, book-guard's, ...) is + # resolved under. Read-only from here, but a test must never see the + # user's real credits: one would earn a fake shutdown hour inside an + # assertion written against an empty day. ( - "leetcode_ledger.json", - ("_leetcode_bonus.LEETCODE_LEDGER_FILE",), + "ledger_home", + ("_earned.LEDGER_HOME",), ), - # book-guard's ledger, for the same reason: a real reading credit would - # earn a fake shutdown hour inside an assertion. + # The key those ledgers are verified with, as ``_earned`` binds it. Absent + # by default, so a read no test asked for fails closed instead of using + # the host's real key; earner tests write their own (_earned_fixtures). ( - "reading_ledger.json", - ("_reading_bonus.READING_LEDGER_FILE",), + "earned_hmac.key", + ("_earned.HMAC_KEY_FILE",), ), # wake-alarm's workday-stick file. A test that forgot this would read # the developer's real penalty state. diff --git a/screen_locker/tests/conftest.py b/screen_locker/tests/conftest.py index 0857ba8..9a58083 100644 --- a/screen_locker/tests/conftest.py +++ b/screen_locker/tests/conftest.py @@ -20,6 +20,7 @@ from unittest.mock import MagicMock, patch import pytest from screen_locker._degraded_sources import clear_degraded_sources +from screen_locker._earned import flat_earners from screen_locker.tests._gatelock_fixtures import ( FAKE_OUTPUTS, TWO_OUTPUTS, @@ -181,7 +182,8 @@ def _isolate_state_files(tmp_path: Path) -> Iterator[None]: yield -_SEEDED_STAMPS = ("last_reset_date", "leetcode_bonus_date", "reading_bonus_date") +# Every registered flat earner's live-pass stamp, so a future one is seeded too. +_SEEDED_STAMPS = ("last_reset_date", *(f"{e.name}_bonus_date" for e in flat_earners())) @pytest.fixture(autouse=True) @@ -191,9 +193,9 @@ def _isolate_shutdown_base(tmp_path: Path) -> Iterator[None]: Pre-seeded with today's date so reset_to_base_if_new_day() is a no-op by default (matching the real file's steady state) -- tests that want to exercise the actual reset path patch reset_to_base_if_new_day directly, - same as the rest of the suite already does. The LeetCode and reading - stamps are seeded too, so apply_flat_bonuses_if_new() never reads a - ledger (or the host's real HMAC key) by default. + same as the rest of the suite already does. Every flat earner's stamp is + seeded too, so apply_flat_bonuses_if_new() never reads a ledger (or the + host's real HMAC key) by default. """ target = tmp_path / "shutdown_base.json" today = datetime.now(tz=UTC).strftime("%Y-%m-%d") diff --git a/screen_locker/tests/test_earned.py b/screen_locker/tests/test_earned.py new file mode 100644 index 0000000..90c2994 --- /dev/null +++ b/screen_locker/tests/test_earned.py @@ -0,0 +1,215 @@ +"""Tests for ``_earned``: where screen-locker reads each earner's ledger, and how. + +Which rows count -- forged rows, yesterday's solve, an unparsable stamp -- is +``earned_time``'s rule and is tested there. Pinned here is screen-locker's +side of it: the ledger lives under ``LEDGER_HOME``, the key is +``HMAC_KEY_FILE``, an unreadable one fails closed to ``None`` and is logged, +and the schedule's hours are whole. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +import json +import logging +from typing import TYPE_CHECKING + +import earned_time +import pytest + +from screen_locker import _constants, _earned +from screen_locker._earned import ( + earned_today, + flat_answers, + flat_earners, + ledger_file, + to_hours, +) +from screen_locker.tests._earned_fixtures import ( + EXTRA, + credit, + signing_key, + write_ledger, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_NOW = datetime(2026, 10, 2, 14, 0).astimezone() +_LEDGERED = pytest.mark.parametrize( + "earner", [earned_time.LEETCODE, earned_time.READING], ids=lambda e: e.name +) + + +@pytest.fixture +def key(tmp_path: Path) -> Iterator[Path]: + with signing_key(tmp_path) as path: + yield path + + +class TestToHours: + def test_whole_hours_convert(self) -> None: + assert to_hours(0) == 0 + assert to_hours(120) == 2 + + def test_a_remainder_raises_instead_of_flooring(self) -> None: + """A 30-minute earner must not silently become 0 hours here.""" + with pytest.raises(ValueError, match="not a whole hour"): + to_hours(90) + + +class TestLedgerFile: + @_LEDGERED + def test_resolves_under_ledger_home(self, earner: earned_time.Earner) -> None: + assert earner.ledger is not None + assert ledger_file(earner) == _earned.LEDGER_HOME / earner.ledger + + def test_ledger_home_and_key_are_redirected_for_the_suite( + self, tmp_path: Path + ) -> None: + assert tmp_path / "ledger_home" == _earned.LEDGER_HOME + assert tmp_path / "earned_hmac.key" == _earned.HMAC_KEY_FILE + assert not _earned.HMAC_KEY_FILE.exists() + + def test_suite_never_sees_a_real_ledger(self) -> None: + assert not any(ledger_file(e).exists() for e in flat_earners()) + + def test_earner_without_a_ledger_raises(self) -> None: + with pytest.raises(ValueError, match="has no ledger"): + ledger_file(earned_time.WORKOUT) + + +class TestFlatEarners: + def test_registry_order_without_the_counted_workout(self) -> None: + assert flat_earners() == (earned_time.LEETCODE, earned_time.READING) + + def test_a_newly_registered_earner_is_picked_up( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(earned_time, "EARNERS", (*earned_time.EARNERS, EXTRA)) + assert flat_earners()[-1] is EXTRA + + def test_a_flat_earner_without_a_ledger_is_left_out( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """No ledger means no shared reader: its answer is the consumer's.""" + unledgered = earned_time.Earner( + name="manual", label="manual", gaming_minutes=60, shutdown_minutes=60 + ) + monkeypatch.setattr(earned_time, "EARNERS", (*earned_time.EARNERS, unledgered)) + assert unledgered not in flat_earners() + + def test_conftest_seeds_every_flat_earners_stamp(self) -> None: + """So the live pass never reads a ledger (or the real key) by default.""" + seeded = json.loads(_constants.SHUTDOWN_BASE_FILE.read_text()) + for earner in flat_earners(): + assert f"{earner.name}_bonus_date" in seeded + + +class TestEarnedToday: + @_LEDGERED + def test_signed_credit_today_earns( + self, key: Path, earner: earned_time.Earner + ) -> None: + write_ledger(earner, [credit(earner)]) + assert earned_today(earner) is True + + @_LEDGERED + def test_now_is_passed_through(self, key: Path, earner: earned_time.Earner) -> None: + write_ledger(earner, [credit(earner, _NOW - timedelta(hours=1))]) + assert earned_today(earner, now=_NOW) is True + assert earned_today(earner, now=_NOW - timedelta(hours=2)) is False + assert earned_today(earner, now=_NOW + timedelta(days=1)) is False + + @_LEDGERED + def test_rows_are_verified_against_hmac_key_file( + self, tmp_path: Path, earner: earned_time.Earner + ) -> None: + """Signed with another key than the one on disk: a forgery, no time.""" + write_ledger(earner, [credit(earner)]) + with signing_key(tmp_path, b"another-key"): + assert earned_today(earner) is False + + @_LEDGERED + def test_a_ledger_outside_ledger_home_is_never_read( + self, tmp_path: Path, key: Path, earner: earned_time.Earner + ) -> None: + stray = tmp_path / "ledger.json" + stray.write_text(json.dumps({"entries": [credit(earner)]})) + assert earned_today(earner) is not True + + +class TestCannotCheck: + """Every unreadable state is None -- never a confident False -- and logged.""" + + @_LEDGERED + def test_missing_key( + self, + tmp_path: Path, + earner: earned_time.Earner, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """The key is checked first: without it even "no ledger" is unknown.""" + monkeypatch.setattr(_earned, "HMAC_KEY_FILE", tmp_path / "no.key") + with caplog.at_level(logging.WARNING): + assert earned_today(earner) is None + assert "Cannot read the integrity key" in caplog.text + + def test_empty_key(self, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + write_ledger(earned_time.LEETCODE, [credit(earned_time.LEETCODE)]) + with signing_key(tmp_path, b" \n"), caplog.at_level(logging.WARNING): + assert earned_today(earned_time.LEETCODE) is None + assert "is empty" in caplog.text + + def test_missing_leetcode_ledger_is_unknown( + self, key: Path, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.WARNING): + assert earned_today(earned_time.LEETCODE) is None + assert "Cannot read the LeetCode ledger" in caplog.text + + def test_missing_reading_ledger_is_a_plain_no(self, key: Path) -> None: + """book-guard never ran: no reading, but no fault either.""" + assert earned_today(earned_time.READING) is False + + @_LEDGERED + def test_corrupt_ledger( + self, + key: Path, + earner: earned_time.Earner, + caplog: pytest.LogCaptureFixture, + ) -> None: + path = ledger_file(earner) + path.parent.mkdir(parents=True) + path.write_text("{not json") + with caplog.at_level(logging.WARNING): + assert earned_today(earner) is None + assert "not valid JSON" in caplog.text + + +class TestFlatAnswers: + def test_every_flat_earner_answers_by_name(self, key: Path) -> None: + write_ledger(earned_time.LEETCODE, [credit(earned_time.LEETCODE)]) + assert flat_answers() == {"leetcode": True, "reading": False} + + def test_an_unknown_answer_is_none_and_warned( + self, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.setattr(_earned, "HMAC_KEY_FILE", tmp_path / "no.key") + with caplog.at_level(logging.WARNING): + assert flat_answers() == {"leetcode": None, "reading": None} + assert "LeetCode state could not be checked" in caplog.text + assert "reading state could not be checked" in caplog.text + + def test_a_no_is_not_warned( + self, key: Path, caplog: pytest.LogCaptureFixture + ) -> None: + write_ledger(earned_time.LEETCODE, []) + with caplog.at_level(logging.WARNING): + assert flat_answers() == {"leetcode": False, "reading": False} + assert "could not be checked" not in caplog.text diff --git a/screen_locker/tests/test_leetcode_bonus.py b/screen_locker/tests/test_leetcode_bonus.py deleted file mode 100644 index cdbac59..0000000 --- a/screen_locker/tests/test_leetcode_bonus.py +++ /dev/null @@ -1,195 +0,0 @@ -"""Tests for the LeetCode shutdown hour: reading the ledger, failing closed. - -Entries are signed against a temp key with gatelock's own ``compute_entry_hmac`` -so a forged credit and a genuine one differ in exactly the way they do on the -machine. -""" - -from __future__ import annotations - -from datetime import datetime, timedelta -import json -import logging -from typing import TYPE_CHECKING, Any -from unittest.mock import patch - -from gatelock.log_integrity import compute_entry_hmac -import pytest - -from screen_locker import _leetcode_bonus -from screen_locker._leetcode_bonus import ( - LEETCODE_BONUS_HOURS, - leetcode_bonus_hours, - leetcode_solved_today, -) - -if TYPE_CHECKING: - from pathlib import Path - -_NOW = datetime(2026, 9, 15, 14, 0).astimezone() - - -@pytest.fixture -def key_file(tmp_path: Path) -> Path: - key = tmp_path / "hmac.key" - key.write_bytes(b"test-key-bytes") - with patch.object(_leetcode_bonus, "HMAC_KEY_FILE", key): - yield key - - -def _entry( - kind: str, *, submitted_at: object, key: Path, day: str = "2026-09-15" -) -> dict[str, Any]: - body: dict[str, Any] = { - "entry_id": f"{kind}:{submitted_at}", - "kind": kind, - "day": day, - "detail": {"submitted_at": submitted_at}, - } - return {**body, "hmac": compute_entry_hmac(body, key_file=key)} - - -def _ledger(tmp_path: Path, entries: list[Any]) -> Path: - path = tmp_path / "ledger.json" - path.write_text(json.dumps({"entries": entries})) - return path - - -class TestSolvedToday: - def test_verified_credit_today_is_solved( - self, tmp_path: Path, key_file: Path - ) -> None: - stamp = (_NOW - timedelta(hours=2)).timestamp() - ledger = _ledger(tmp_path, [_entry("credit", submitted_at=stamp, key=key_file)]) - assert leetcode_solved_today(ledger, now=_NOW) is True - - def test_credit_from_yesterday_is_not_solved( - self, tmp_path: Path, key_file: Path - ) -> None: - """A 23:50 solve harvested this morning counts for yesterday, never today.""" - stamp = (_NOW - timedelta(days=1)).timestamp() - ledger = _ledger(tmp_path, [_entry("credit", submitted_at=stamp, key=key_file)]) - assert leetcode_solved_today(ledger, now=_NOW) is False - - def test_seen_and_charge_entries_never_count( - self, tmp_path: Path, key_file: Path - ) -> None: - stamp = _NOW.timestamp() - ledger = _ledger( - tmp_path, - [ - _entry("seen", submitted_at=stamp, key=key_file), - _entry("charge", submitted_at=stamp, key=key_file), - "not-a-dict", - ], - ) - assert leetcode_solved_today(ledger, now=_NOW) is False - - def test_forged_credit_is_ignored(self, tmp_path: Path, key_file: Path) -> None: - entry = _entry("credit", submitted_at=_NOW.timestamp(), key=key_file) - entry["detail"]["submitted_at"] = ( - _NOW.timestamp() - 1 - ) # body no longer matches hmac - ledger = _ledger(tmp_path, [entry]) - assert leetcode_solved_today(ledger, now=_NOW) is False - - def test_unparsable_submitted_at_falls_back_to_day_key( - self, tmp_path: Path, key_file: Path, caplog: pytest.LogCaptureFixture - ) -> None: - ledger = _ledger( - tmp_path, - [ - _entry( - "credit", - submitted_at="soon", - key=key_file, - day=_NOW.date().isoformat(), - ) - ], - ) - with caplog.at_level(logging.WARNING): - assert leetcode_solved_today(ledger, now=_NOW) is True - assert "unparsable submitted_at" in caplog.text - - def test_missing_submitted_at_uses_day_key( - self, tmp_path: Path, key_file: Path - ) -> None: - body: dict[str, Any] = {"entry_id": "x", "kind": "credit", "day": "2000-01-01"} - entry = {**body, "hmac": compute_entry_hmac(body, key_file=key_file)} - assert leetcode_solved_today(_ledger(tmp_path, [entry]), now=_NOW) is False - - def test_resolves_module_path_when_none_given( - self, tmp_path: Path, key_file: Path - ) -> None: - stamp = _NOW.timestamp() - ledger = _ledger(tmp_path, [_entry("credit", submitted_at=stamp, key=key_file)]) - with patch.object(_leetcode_bonus, "LEETCODE_LEDGER_FILE", ledger): - assert leetcode_solved_today(now=_NOW) is True - - -class TestCannotCheck: - """Every unreadable state is None -- never a confident False.""" - - def test_missing_ledger(self, tmp_path: Path, key_file: Path) -> None: - assert leetcode_solved_today(tmp_path / "missing.json") is None - - def test_invalid_json(self, tmp_path: Path, key_file: Path) -> None: - path = tmp_path / "ledger.json" - path.write_text("{not json") - assert leetcode_solved_today(path) is None - - def test_no_entries_array(self, tmp_path: Path, key_file: Path) -> None: - path = tmp_path / "ledger.json" - path.write_text(json.dumps({"entries": "nope"})) - assert leetcode_solved_today(path) is None - - def test_unreadable_key(self, tmp_path: Path) -> None: - with patch.object(_leetcode_bonus, "HMAC_KEY_FILE", tmp_path / "no.key"): - assert leetcode_solved_today(_ledger(tmp_path, [])) is None - - def test_empty_key(self, tmp_path: Path) -> None: - key = tmp_path / "hmac.key" - key.write_bytes(b" \n") - with patch.object(_leetcode_bonus, "HMAC_KEY_FILE", key): - assert leetcode_solved_today(_ledger(tmp_path, [])) is None - - -class TestBonusHours: - def test_solved_earns_the_hour(self) -> None: - with patch.object(_leetcode_bonus, "leetcode_solved_today", return_value=True): - assert leetcode_bonus_hours() == LEETCODE_BONUS_HOURS - - def test_not_solved_earns_nothing(self) -> None: - with patch.object(_leetcode_bonus, "leetcode_solved_today", return_value=False): - assert leetcode_bonus_hours() == 0 - - def test_cannot_check_fails_closed_and_warns( - self, caplog: pytest.LogCaptureFixture - ) -> None: - with ( - patch.object(_leetcode_bonus, "leetcode_solved_today", return_value=None), - caplog.at_level(logging.WARNING), - ): - assert leetcode_bonus_hours() == 0 - assert "could not be checked" in caplog.text - - -class TestReadEntriesLabel: - """_read_entries is shared with book-guard's reader; the label names whose.""" - - def test_default_label_is_leetcode( - self, tmp_path: Path, caplog: pytest.LogCaptureFixture - ) -> None: - with caplog.at_level(logging.WARNING): - assert _leetcode_bonus._read_entries(tmp_path / "missing.json") is None - assert "Cannot read the LeetCode ledger" in caplog.text - - def test_custom_label_names_the_ledger( - self, tmp_path: Path, caplog: pytest.LogCaptureFixture - ) -> None: - path = tmp_path / "ledger.json" - path.write_text(json.dumps({"no": "entries"})) - with caplog.at_level(logging.WARNING): - assert _leetcode_bonus._read_entries(path, "book-guard") is None - assert "book-guard ledger at" in caplog.text - assert "LeetCode" not in caplog.text diff --git a/screen_locker/tests/test_reading_bonus.py b/screen_locker/tests/test_reading_bonus.py deleted file mode 100644 index 6ce0b2a..0000000 --- a/screen_locker/tests/test_reading_bonus.py +++ /dev/null @@ -1,214 +0,0 @@ -"""Tests for the reading shutdown hour: reading book-guard's ledger, failing closed. - -Rows are signed against a temp key with gatelock's own ``compute_entry_hmac`` -so a forged credit and a genuine one differ in exactly the way they do on the -machine. The key is patched on ``_reading_bonus`` itself: the module holds its -own binding of ``HMAC_KEY_FILE``. -""" - -from __future__ import annotations - -from datetime import datetime, timedelta -import json -import logging -from typing import TYPE_CHECKING, Any -from unittest.mock import patch - -from gatelock.log_integrity import compute_entry_hmac -import pytest - -from screen_locker import _reading_bonus -from screen_locker._reading_bonus import ( - READING_BONUS_HOURS, - read_today, - reading_bonus_hours, -) - -if TYPE_CHECKING: - from collections.abc import Iterator - from pathlib import Path - -_NOW = datetime(2026, 10, 2, 21, 0).astimezone() - - -@pytest.fixture -def key_file(tmp_path: Path) -> Iterator[Path]: - key = tmp_path / "hmac.key" - key.write_bytes(b"test-key-bytes") - with patch.object(_reading_bonus, "HMAC_KEY_FILE", key): - yield key - - -def _row( - kind: str = "credit", - *, - key: Path, - ended_at: object = None, - bonus: str = "1", - detail: object = None, -) -> dict[str, Any]: - """One signed ledger row; ``detail`` overrides the default detail body.""" - if detail is None: - stamp = _NOW - timedelta(hours=1) if ended_at is None else ended_at - if isinstance(stamp, datetime): - stamp = str(int(stamp.timestamp())) - detail = {"bonus": bonus, "ended_at": stamp} - body: dict[str, Any] = {"entry_id": f"{kind}:x", "kind": kind, "detail": detail} - return {**body, "hmac": compute_entry_hmac(body, key_file=key)} - - -def _ledger(tmp_path: Path, entries: list[Any]) -> Path: - path = tmp_path / "ledger.json" - path.write_text(json.dumps({"entries": entries})) - return path - - -class TestReadToday: - def test_verified_bonus_credit_read_today( - self, tmp_path: Path, key_file: Path - ) -> None: - ledger = _ledger(tmp_path, [_row(key=key_file)]) - assert read_today(ledger, now=_NOW) is True - - def test_missing_ledger_is_a_plain_false( - self, tmp_path: Path, key_file: Path - ) -> None: - """book-guard never ran: no reading, but no fault either.""" - assert read_today(tmp_path / "missing.json", now=_NOW) is False - - def test_session_ended_yesterday_does_not_count( - self, tmp_path: Path, key_file: Path - ) -> None: - """Read at 22:30, quizzed next morning: earns the evening, not today.""" - row = _row(key=key_file, ended_at=_NOW - timedelta(days=1)) - assert read_today(_ledger(tmp_path, [row]), now=_NOW) is False - - def test_session_ending_after_now_does_not_count( - self, tmp_path: Path, key_file: Path - ) -> None: - row = _row(key=key_file, ended_at=_NOW + timedelta(minutes=5)) - assert read_today(_ledger(tmp_path, [row]), now=_NOW) is False - - def test_small_session_without_bonus_flag_does_not_count( - self, tmp_path: Path, key_file: Path - ) -> None: - row = _row(key=key_file, bonus="0") - assert read_today(_ledger(tmp_path, [row]), now=_NOW) is False - - def test_non_credit_rows_never_count(self, tmp_path: Path, key_file: Path) -> None: - rows = [_row("reject", key=key_file), _row("escape", key=key_file), "junk"] - assert read_today(_ledger(tmp_path, rows), now=_NOW) is False - - def test_non_dict_detail_does_not_count( - self, tmp_path: Path, key_file: Path - ) -> None: - row = _row(key=key_file, detail="not-a-dict") - assert read_today(_ledger(tmp_path, [row]), now=_NOW) is False - - def test_forged_credit_is_ignored(self, tmp_path: Path, key_file: Path) -> None: - row = _row(key=key_file, bonus="0") - row["detail"]["bonus"] = "1" # body no longer matches its hmac - assert read_today(_ledger(tmp_path, [row]), now=_NOW) is False - - @pytest.mark.parametrize("ended_at", ["soon", None]) - def test_unusable_ended_at_does_not_count_and_warns( - self, - tmp_path: Path, - key_file: Path, - caplog: pytest.LogCaptureFixture, - ended_at: object, - ) -> None: - """A missing ended_at stringifies to "None"; neither may invent a day.""" - row = _row(key=key_file, detail={"bonus": "1", "ended_at": ended_at}) - with caplog.at_level(logging.WARNING): - assert read_today(_ledger(tmp_path, [row]), now=_NOW) is False - assert "no usable ended_at" in caplog.text - - def test_one_good_row_among_bad_ones_counts( - self, tmp_path: Path, key_file: Path - ) -> None: - rows = [_row(key=key_file, bonus="0"), _row(key=key_file)] - assert read_today(_ledger(tmp_path, rows), now=_NOW) is True - - def test_defaults_resolve_module_path_and_clock( - self, tmp_path: Path, key_file: Path - ) -> None: - row = _row(key=key_file, ended_at=datetime.now().astimezone()) - ledger = _ledger(tmp_path, [row]) - with patch.object(_reading_bonus, "READING_LEDGER_FILE", ledger): - assert read_today() is True - - -class TestCannotCheck: - """Every unreadable state is None -- never a confident False.""" - - def test_unreadable_key(self, tmp_path: Path) -> None: - with patch.object(_reading_bonus, "HMAC_KEY_FILE", tmp_path / "no.key"): - assert read_today(_ledger(tmp_path, [])) is None - - def test_unreadable_key_wins_over_a_missing_ledger(self, tmp_path: Path) -> None: - """The key is checked first: without it even "no ledger" is unknown.""" - with patch.object(_reading_bonus, "HMAC_KEY_FILE", tmp_path / "no.key"): - assert read_today(tmp_path / "missing.json") is None - - def test_empty_key(self, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: - key = tmp_path / "hmac.key" - key.write_bytes(b" \n") - with ( - patch.object(_reading_bonus, "HMAC_KEY_FILE", key), - caplog.at_level(logging.WARNING), - ): - assert read_today(_ledger(tmp_path, [])) is None - assert "is empty" in caplog.text - - def test_invalid_json_is_logged_as_book_guards( - self, tmp_path: Path, key_file: Path, caplog: pytest.LogCaptureFixture - ) -> None: - path = tmp_path / "ledger.json" - path.write_text("{not json") - with caplog.at_level(logging.WARNING): - assert read_today(path) is None - assert "book-guard ledger at" in caplog.text - assert "not valid JSON" in caplog.text - - def test_no_entries_array(self, tmp_path: Path, key_file: Path) -> None: - path = tmp_path / "ledger.json" - path.write_text(json.dumps({"entries": "nope"})) - assert read_today(path) is None - - def test_unreadable_ledger(self, tmp_path: Path, key_file: Path) -> None: - """A directory where the file should be: exists, but read_text fails.""" - path = tmp_path / "ledger.json" - path.mkdir() - assert read_today(path) is None - - -class TestBonusHours: - def test_read_earns_the_hour(self) -> None: - with patch.object(_reading_bonus, "read_today", return_value=True): - assert reading_bonus_hours() == READING_BONUS_HOURS == 1 - - def test_not_read_earns_nothing(self) -> None: - with patch.object(_reading_bonus, "read_today", return_value=False): - assert reading_bonus_hours() == 0 - - def test_cannot_check_fails_closed_and_warns( - self, caplog: pytest.LogCaptureFixture - ) -> None: - with ( - patch.object(_reading_bonus, "read_today", return_value=None), - caplog.at_level(logging.WARNING), - ): - assert reading_bonus_hours() == 0 - assert "could not be checked" in caplog.text - - def test_passes_the_ledger_path_through(self, tmp_path: Path) -> None: - path = tmp_path / "ledger.json" - with patch.object(_reading_bonus, "read_today", return_value=True) as m: - reading_bonus_hours(path) - m.assert_called_once_with(path) - - def test_isolated_suite_never_sees_a_real_ledger(self, key_file: Path) -> None: - """The conftest redirect points at an absent tmp file, so: 0, not None.""" - assert not _reading_bonus.READING_LEDGER_FILE.exists() - assert reading_bonus_hours() == 0 diff --git a/screen_locker/tests/test_shutdown_base_replay.py b/screen_locker/tests/test_shutdown_base_replay.py index 959b6ba..4b995e6 100644 --- a/screen_locker/tests/test_shutdown_base_replay.py +++ b/screen_locker/tests/test_shutdown_base_replay.py @@ -17,7 +17,7 @@ from screen_locker._day import today_str from screen_locker._shutdown_base import ( base_hour, reset_to_base_if_new_day, - today_earned_bonus_hours, + today_credit_count, ) from screen_locker._weekly_check import ( count_day_credits, @@ -124,23 +124,24 @@ class TestSyncedCopySharesCredit: assert credit_key(_DAY, 0, entry, {"42": 5}) == ("manual_workout", f"{_DAY}#0") -class TestTodayEarnedBonusHours: - """The log → hours derivation the reset relies on.""" +class TestTodayCreditCount: + """The log -> credit count the reset hands to the registry.""" - def test_real_day_earns_two_hours(self, tmp_path: Path) -> None: + def test_real_day_counts_one_workout_worth_two_hours(self, tmp_path: Path) -> None: log = _write_log(tmp_path / "log.json", _DAY, _real_2026_09_13()) - assert today_earned_bonus_hours(log, _DAY) == 2 + assert today_credit_count(log, _DAY) == 1 + assert earned_shutdown_bonus_hours(today_credit_count(log, _DAY)) == 2 - def test_day_absent_from_log_earns_nothing(self, tmp_path: Path) -> None: + def test_day_absent_from_log_counts_nothing(self, tmp_path: Path) -> None: log = _write_log(tmp_path / "log.json", "2026-09-12", _real_2026_09_13()) - assert today_earned_bonus_hours(log, _DAY) == 0 + assert today_credit_count(log, _DAY) == 0 def test_non_dict_entries_are_ignored(self, tmp_path: Path) -> None: log = tmp_path / "log.json" log.write_text( json.dumps({_DAY: ["garbage", _manual("manual:2026-09-13T08:00")]}) ) - assert today_earned_bonus_hours(log, _DAY) == 2 + assert today_credit_count(log, _DAY) == 1 class TestResetReplaysTodaysCredit: diff --git a/screen_locker/tests/test_shutdown_flat_earners.py b/screen_locker/tests/test_shutdown_flat_earners.py new file mode 100644 index 0000000..2fc76b8 --- /dev/null +++ b/screen_locker/tests/test_shutdown_flat_earners.py @@ -0,0 +1,178 @@ +"""Tests for the pass over every flat earner, and for the base it projects. + +The registry, not this repo, lists the earners: a newly registered flat +earner must be a term of the reset and an hour of the live pass with no code +here. Also covers one real signed ledger end to end, and the status +projection following the 2026-10-01 reading cut. +""" + +from __future__ import annotations + +from datetime import UTC, date, datetime +import json +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import earned_time +import pytest + +from screen_locker._day import today_str +from screen_locker._shutdown_base import ( + apply_flat_bonuses_if_new, + base_hour, + reset_to_base_if_new_day, +) +from screen_locker._status_data import gather_status +from screen_locker.tests._earned_fixtures import ( + EXTRA, + answering, + credit, + signing_key, + write_ledger, +) +from screen_locker.tests.test_status_data import _files + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +def _mixin() -> MagicMock: + mixin = MagicMock() + mixin._read_shutdown_config.return_value = (base_hour(), base_hour(), 5) + mixin._write_shutdown_config.return_value = True + mixin._adjust_shutdown_time_by.return_value = True + return mixin + + +@pytest.fixture +def answers() -> Iterator[dict[str, bool | None]]: + """Every flat earner says no until a test sets it.""" + given: dict[str, bool | None] = {"leetcode": False, "reading": False} + with answering(given): + yield given + + +@pytest.fixture +def with_extra(monkeypatch: pytest.MonkeyPatch) -> None: + """Register EXTRA after the existing earners, as a new gate would be.""" + monkeypatch.setattr(earned_time, "EARNERS", (*earned_time.EARNERS, EXTRA)) + + +class TestFlatBonuses: + def test_both_earned_apply_both_and_keep_both_stamps( + self, tmp_path: Path, answers: dict[str, bool | None] + ) -> None: + """The second save must not overwrite the first one's stamp.""" + answers["leetcode"] = True + answers["reading"] = True + state = tmp_path / "state.json" + mixin = _mixin() + apply_flat_bonuses_if_new(state, mixin) + assert mixin._adjust_shutdown_time_by.call_count == 2 + assert json.loads(state.read_text()) == { + "leetcode_bonus_date": today_str(), + "reading_bonus_date": today_str(), + } + apply_flat_bonuses_if_new(state, mixin) + assert mixin._adjust_shutdown_time_by.call_count == 2 + + def test_only_reading_earned_applies_only_reading( + self, tmp_path: Path, answers: dict[str, bool | None] + ) -> None: + answers["reading"] = True + state = tmp_path / "state.json" + mixin = _mixin() + apply_flat_bonuses_if_new(state, mixin) + mixin._adjust_shutdown_time_by.assert_called_once_with(1) + assert json.loads(state.read_text()) == {"reading_bonus_date": today_str()} + + +@pytest.mark.usefixtures("with_extra") +class TestANewlyRegisteredEarner: + def test_reset_includes_and_stamps_it( + self, tmp_path: Path, answers: dict[str, bool | None] + ) -> None: + answers["extra"] = True + state = tmp_path / "state.json" + mixin = _mixin() + assert reset_to_base_if_new_day(state, mixin) is True + mixin._write_shutdown_config.assert_called_once_with( + base_hour() + 1, base_hour() + 1, 5, restore=True + ) + assert json.loads(state.read_text()) == { + "last_reset_date": today_str(), + "extra_bonus_date": today_str(), + } + + def test_live_pass_applies_it_once( + self, tmp_path: Path, answers: dict[str, bool | None] + ) -> None: + answers["extra"] = True + state = tmp_path / "state.json" + mixin = _mixin() + apply_flat_bonuses_if_new(state, mixin) + apply_flat_bonuses_if_new(state, mixin) + mixin._adjust_shutdown_time_by.assert_called_once_with(1) + assert json.loads(state.read_text()) == {"extra_bonus_date": today_str()} + + def test_its_ledger_is_read_like_any_other(self, tmp_path: Path) -> None: + write_ledger(EXTRA, [credit(EXTRA)]) + mixin = _mixin() + with signing_key(tmp_path): + apply_flat_bonuses_if_new(tmp_path / "state.json", mixin) + mixin._adjust_shutdown_time_by.assert_called_once_with(1) + + def test_a_penalised_earner_lowers_the_base_from_its_day( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The base is read from the registry at call time, never a constant.""" + eve, cut = date(2026, 10, 2), date(2026, 10, 3) + penalised = earned_time.Earner( + name="cut", + label="cut", + gaming_minutes=60, + shutdown_minutes=60, + penalty_from=cut, + ) + before = (base_hour(eve), base_hour(cut)) + monkeypatch.setattr(earned_time, "EARNERS", (*earned_time.EARNERS, penalised)) + assert (base_hour(eve), base_hour(cut)) == (before[0], before[1] - 1) + + +class TestRealLedgers: + def test_signed_credits_on_disk_reach_the_reset(self, tmp_path: Path) -> None: + """No mock between the ledgers and the written hour.""" + write_ledger(earned_time.LEETCODE, [credit(earned_time.LEETCODE)]) + write_ledger(earned_time.READING, [credit(earned_time.READING)]) + state = tmp_path / "state.json" + mixin = _mixin() + with signing_key(tmp_path): + assert reset_to_base_if_new_day(state, mixin) is True + mixin._write_shutdown_config.assert_called_once_with( + base_hour() + 2, base_hour() + 2, 5, restore=True + ) + assert json.loads(state.read_text()) == { + "last_reset_date": today_str(), + "leetcode_bonus_date": today_str(), + "reading_bonus_date": today_str(), + } + + +class TestProjectionFollowsTheCut: + @pytest.mark.parametrize( + ("now", "expected"), + [ + (datetime(2026, 9, 30, 12, 0, tzinfo=UTC), 20), + (datetime(2026, 10, 2, 12, 0, tzinfo=UTC), 19), + ], + ) + def test_rest_of_week_uses_the_base_for_that_day( + self, tmp_path: Path, now: datetime, expected: int + ) -> None: + with patch( + "screen_locker._status_data.has_workout_skip_today", return_value=False + ): + snap = gather_status(**_files(tmp_path), now=now) + assert snap.shutdown.rest_of_week[0].hour == expected + assert snap.shutdown.next_week_preview[0].hour == expected diff --git a/screen_locker/tests/test_shutdown_leetcode.py b/screen_locker/tests/test_shutdown_leetcode.py index 62af92b..725f177 100644 --- a/screen_locker/tests/test_shutdown_leetcode.py +++ b/screen_locker/tests/test_shutdown_leetcode.py @@ -9,22 +9,23 @@ from __future__ import annotations import json from typing import TYPE_CHECKING -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock + +import earned_time +import pytest -from screen_locker import _shutdown_base from screen_locker._day import today_str from screen_locker._shutdown_base import ( - apply_leetcode_bonus_if_new, + _apply_flat_bonus, base_hour, reset_to_base_if_new_day, ) +from screen_locker.tests._earned_fixtures import answering if TYPE_CHECKING: from collections.abc import Iterator from pathlib import Path -import pytest - def _mixin(*, adjust_ok: bool = True) -> MagicMock: mixin = MagicMock() @@ -34,22 +35,21 @@ def _mixin(*, adjust_ok: bool = True) -> MagicMock: return mixin -@pytest.fixture(autouse=True) -def _no_reading() -> Iterator[None]: - """Zero the reading hour so the stamps asserted here are LeetCode's only.""" - with patch.object(_shutdown_base, "reading_bonus_hours", return_value=0): - yield +def apply_leetcode_bonus_if_new(state: Path, mixin: MagicMock) -> bool: + """The live pass for the LeetCode earner alone.""" + return _apply_flat_bonus(state, mixin, earned_time.LEETCODE) @pytest.fixture def solved() -> Iterator[MagicMock]: - with patch.object(_shutdown_base, "leetcode_bonus_hours", return_value=1) as m: + """LeetCode solved today; reading not, so the stamps here are LeetCode's.""" + with answering({"leetcode": True, "reading": False}) as m: yield m @pytest.fixture def unsolved() -> Iterator[MagicMock]: - with patch.object(_shutdown_base, "leetcode_bonus_hours", return_value=0) as m: + with answering({"leetcode": False, "reading": False}) as m: yield m diff --git a/screen_locker/tests/test_shutdown_reading.py b/screen_locker/tests/test_shutdown_reading.py index 69b2add..bab8171 100644 --- a/screen_locker/tests/test_shutdown_reading.py +++ b/screen_locker/tests/test_shutdown_reading.py @@ -2,29 +2,29 @@ Covers the base cut it pays for (20 -> 19 from 2026-10-01), the daily reset folding the hour in and stamping it, and the live pass that adds it once for a -reading credit landing after the reset. Mirrors test_shutdown_leetcode.py. +reading credit landing after the reset. Mirrors test_shutdown_leetcode.py; the +pass over every flat earner is in test_shutdown_flat_earners.py. """ from __future__ import annotations -from datetime import UTC, date, datetime +from datetime import date, datetime import json from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch +import earned_time import pytest from screen_locker import _shutdown_base from screen_locker._day import today_str from screen_locker._shutdown_base import ( - READING_BASE_FROM, + _apply_flat_bonus, apply_flat_bonuses_if_new, - apply_reading_bonus_if_new, base_hour, reset_to_base_if_new_day, ) -from screen_locker._status_data import gather_status -from screen_locker.tests.test_status_data import _files +from screen_locker.tests._earned_fixtures import answering if TYPE_CHECKING: from collections.abc import Iterator @@ -39,14 +39,17 @@ def _mixin(*, adjust_ok: bool = True) -> MagicMock: return mixin +def apply_reading_bonus_if_new(state: Path, mixin: MagicMock) -> bool: + """The live pass for the reading earner alone.""" + return _apply_flat_bonus(state, mixin, earned_time.READING) + + @pytest.fixture -def hours() -> Iterator[tuple[MagicMock, MagicMock]]: - """(leetcode, reading) hour sources, both 0 until a test sets them.""" - with ( - patch.object(_shutdown_base, "leetcode_bonus_hours", return_value=0) as lc, - patch.object(_shutdown_base, "reading_bonus_hours", return_value=0) as rd, - ): - yield lc, rd +def answers() -> Iterator[tuple[dict[str, bool | None], MagicMock]]: + """(answers, earned_today mock): every flat earner says no until set.""" + given: dict[str, bool | None] = {"leetcode": False, "reading": False} + with answering(given) as mock: + yield given, mock class TestBaseHour: @@ -54,7 +57,7 @@ class TestBaseHour: assert base_hour(date(2026, 9, 30)) == 20 def test_cut_day_is_nineteen(self) -> None: - assert date(2026, 10, 1) == READING_BASE_FROM + assert earned_time.READING.penalty_from == date(2026, 10, 1) assert base_hour(date(2026, 10, 1)) == 19 def test_long_after_the_cut_stays_nineteen(self) -> None: @@ -66,9 +69,9 @@ class TestBaseHour: class TestResetIncludesReading: def test_reset_writes_base_plus_reading_and_stamps_it( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + self, tmp_path: Path, answers: tuple[dict[str, bool | None], MagicMock] ) -> None: - hours[1].return_value = 1 + answers[0]["reading"] = True state = tmp_path / "state.json" mixin = _mixin() assert reset_to_base_if_new_day(state, mixin) is True @@ -81,10 +84,10 @@ class TestResetIncludesReading: } def test_reset_with_both_flat_hours_stamps_both( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + self, tmp_path: Path, answers: tuple[dict[str, bool | None], MagicMock] ) -> None: - hours[0].return_value = 1 - hours[1].return_value = 1 + answers[0]["leetcode"] = True + answers[0]["reading"] = True state = tmp_path / "state.json" mixin = _mixin() assert reset_to_base_if_new_day(state, mixin) is True @@ -100,23 +103,57 @@ class TestResetIncludesReading: mixin._adjust_shutdown_time_by.assert_not_called() def test_reset_is_still_capped_at_the_ceiling( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + self, tmp_path: Path, answers: tuple[dict[str, bool | None], MagicMock] ) -> None: - hours[0].return_value = 1 - hours[1].return_value = 1 + answers[0]["leetcode"] = True + answers[0]["reading"] = True mixin = _mixin() - with patch.object(_shutdown_base, "today_earned_bonus_hours", return_value=9): + with patch.object(_shutdown_base, "today_credit_count", return_value=9): reset_to_base_if_new_day( tmp_path / "state.json", mixin, log_file=tmp_path / "log.json" ) mixin._write_shutdown_config.assert_called_once_with(23, 23, 5, restore=True) + def test_unknown_reading_earns_nothing_warns_and_is_not_stamped( + self, + tmp_path: Path, + answers: tuple[dict[str, bool | None], MagicMock], + caplog: pytest.LogCaptureFixture, + ) -> None: + """The live pass must ask again: a stamp would forfeit the day's hour.""" + answers[0]["reading"] = None + state = tmp_path / "state.json" + mixin = _mixin() + with caplog.at_level("WARNING"): + assert reset_to_base_if_new_day(state, mixin) is True + assert "reading state could not be checked" in caplog.text + base = base_hour() + mixin._write_shutdown_config.assert_called_once_with( + base, base, 5, restore=True + ) + assert json.loads(state.read_text()) == {"last_reset_date": today_str()} + + def test_reset_logs_every_term( + self, + tmp_path: Path, + answers: tuple[dict[str, bool | None], MagicMock], + caplog: pytest.LogCaptureFixture, + ) -> None: + answers[0]["reading"] = True + with caplog.at_level("INFO"): + reset_to_base_if_new_day(tmp_path / "state.json", _mixin()) + base = base_hour() + assert ( + f"Daily base reset: {base + 1:02d}:00 (base {base} + 0h workout + " + "0h LeetCode + 1h reading already earned today)." + ) in caplog.text + class TestReadingLivePass: def test_applies_once_and_stamps( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + self, tmp_path: Path, answers: tuple[dict[str, bool | None], MagicMock] ) -> None: - hours[1].return_value = 1 + answers[0]["reading"] = True state = tmp_path / "state.json" state.write_text(json.dumps({"last_reset_date": today_str()})) mixin = _mixin() @@ -129,15 +166,15 @@ class TestReadingLivePass: } def test_stamped_today_never_reads_the_ledger( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + self, tmp_path: Path, answers: tuple[dict[str, bool | None], MagicMock] ) -> None: state = tmp_path / "state.json" state.write_text(json.dumps({"reading_bonus_date": today_str()})) assert apply_reading_bonus_if_new(state, _mixin()) is False - hours[1].assert_not_called() + answers[1].assert_not_called() def test_no_reading_no_write( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + self, tmp_path: Path, answers: tuple[dict[str, bool | None], MagicMock] ) -> None: mixin = _mixin() assert apply_reading_bonus_if_new(tmp_path / "state.json", mixin) is False @@ -146,62 +183,41 @@ class TestReadingLivePass: def test_failed_write_leaves_no_stamp_and_warns( self, tmp_path: Path, - hours: tuple[MagicMock, MagicMock], + answers: tuple[dict[str, bool | None], MagicMock], caplog: pytest.LogCaptureFixture, ) -> None: """A stamp without a write would silently forfeit the hour for the day.""" - hours[1].return_value = 1 + answers[0]["reading"] = True state = tmp_path / "state.json" mixin = _mixin(adjust_ok=False) with caplog.at_level("WARNING"): assert apply_reading_bonus_if_new(state, mixin) is False - assert "Reading bonus: failed to write" in caplog.text + assert "reading bonus: failed to write" in caplog.text assert not state.exists() - -class TestFlatBonuses: - def test_both_earned_apply_both_and_keep_both_stamps( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + def test_cannot_check_warns_and_leaves_no_stamp( + self, + tmp_path: Path, + answers: tuple[dict[str, bool | None], MagicMock], + caplog: pytest.LogCaptureFixture, ) -> None: - """The second save must not overwrite the first one's stamp.""" - hours[0].return_value = 1 - hours[1].return_value = 1 + """Unknown is not "no": logged, no hour, and re-asked on the next tick.""" + answers[0]["reading"] = None state = tmp_path / "state.json" mixin = _mixin() - apply_flat_bonuses_if_new(state, mixin) - assert mixin._adjust_shutdown_time_by.call_count == 2 - assert json.loads(state.read_text()) == { - "leetcode_bonus_date": today_str(), - "reading_bonus_date": today_str(), - } - apply_flat_bonuses_if_new(state, mixin) - assert mixin._adjust_shutdown_time_by.call_count == 2 + with caplog.at_level("WARNING"): + assert apply_reading_bonus_if_new(state, mixin) is False + assert "reading state could not be checked" in caplog.text + mixin._adjust_shutdown_time_by.assert_not_called() + assert not state.exists() - def test_only_reading_earned_applies_only_reading( - self, tmp_path: Path, hours: tuple[MagicMock, MagicMock] + def test_applied_hour_is_logged( + self, + tmp_path: Path, + answers: tuple[dict[str, bool | None], MagicMock], + caplog: pytest.LogCaptureFixture, ) -> None: - hours[1].return_value = 1 - state = tmp_path / "state.json" - mixin = _mixin() - apply_flat_bonuses_if_new(state, mixin) - mixin._adjust_shutdown_time_by.assert_called_once_with(1) - assert json.loads(state.read_text()) == {"reading_bonus_date": today_str()} - - -class TestProjectionFollowsTheCut: - @pytest.mark.parametrize( - ("now", "expected"), - [ - (datetime(2026, 9, 30, 12, 0, tzinfo=UTC), 20), - (datetime(2026, 10, 2, 12, 0, tzinfo=UTC), 19), - ], - ) - def test_rest_of_week_uses_the_base_for_that_day( - self, tmp_path: Path, now: datetime, expected: int - ) -> None: - with patch( - "screen_locker._status_data.has_workout_skip_today", return_value=False - ): - snap = gather_status(**_files(tmp_path), now=now) - assert snap.shutdown.rest_of_week[0].hour == expected - assert snap.shutdown.next_week_preview[0].hour == expected + answers[0]["reading"] = True + with caplog.at_level("INFO"): + assert apply_reading_bonus_if_new(tmp_path / "state.json", _mixin()) + assert "reading bonus: +1h shutdown time today." in caplog.text