diff --git a/steam_backlog_enforcer/_playtime.py b/steam_backlog_enforcer/_playtime.py index c96a216..793c96a 100644 --- a/steam_backlog_enforcer/_playtime.py +++ b/steam_backlog_enforcer/_playtime.py @@ -153,6 +153,43 @@ def _state_or_recover(rules: PlaytimeRules, *, now: datetime) -> PlaytimeState: return replace(fresh, seconds=rules.budget_seconds, blocked_at=stamp) +def _release_after_raise( + state: PlaytimeState, + rules: PlaytimeRules, +) -> PlaytimeState: + """Clear block bookkeeping after the day's budget rose past what was spent. + + Only ``roll_over`` at the 06:00 boundary ever reset ``blocked_at`` before + the workout coupling existed, because the budget could not move within a + day. It can now: logging a workout raises it from the unearned floor to the + earned budget, and the mounts come down on the very next tick. Two things + have to come down with them. + + ``blocked_at`` — otherwise ``is_blocked()`` stays true and ``build_today`` + keeps reporting ``"blocked": true``, so the UI would claim you were blocked + while gaming worked fine. That is the exact "state disagrees with reality" + failure the mount-visibility check exists to catch. + + ``warned_seconds`` — the thresholds are keyed on seconds *remaining*, so a + raise makes remaining climb back through values already recorded as fired. + Left alone, the extra two hours would arrive with no warnings at all. + + Args: + state: Accounting state for the current gaming day. + rules: Policy for this tick, holding the newly raised budget. + + Returns: + The state with block bookkeeping cleared. + """ + logger.warning( + "Gaming budget rose to %.1fh past the %.0fs already spent — releasing " + "the block and re-arming warnings for the remaining time.", + rules.budget_seconds / 3600, + state.seconds, + ) + return replace(state, blocked_at=0.0, warned_seconds=[]) + + def _policy( state: PlaytimeState, rules: PlaytimeRules, @@ -177,6 +214,8 @@ def _policy( if state.seconds < rules.budget_seconds: reconcile(should_block=False) + if state.is_blocked(): + state = _release_after_raise(state, rules) return _warn(state, rules) if not state.is_blocked(): diff --git a/steam_backlog_enforcer/_playtime_state.py b/steam_backlog_enforcer/_playtime_state.py index 54ca2bf..ebb795e 100644 --- a/steam_backlog_enforcer/_playtime_state.py +++ b/steam_backlog_enforcer/_playtime_state.py @@ -14,6 +14,7 @@ import logging from typing import TYPE_CHECKING from steam_backlog_enforcer._whitelist import _try_set_immutable, unlock_for_write +from steam_backlog_enforcer._workout_budget import resolve_budget_seconds from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write if TYPE_CHECKING: @@ -138,9 +139,12 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules: The rules governing this tick. """ return PlaytimeRules( - budget_seconds=_DEMO_BUDGET_SECONDS - if demo - else float(config.daily_gaming_seconds), + # Production budget is resolved, not read straight off config: it + # depends on whether a workout was logged today. Resolving it *here* + # rather than at each caller is what stops the daemon (which holds one + # Config for its whole life) and _budget_view (which reloads Config per + # HTTP request) from reporting different budgets. + budget_seconds=_DEMO_BUDGET_SECONDS if demo else resolve_budget_seconds(config), warn_at=_DEMO_WARN_AT if demo else _WARN_AT, sigkill_after=(_DEMO_SIGKILL_AFTER_SECONDS if demo else _SIGKILL_AFTER_SECONDS), count_launchers=config.count_launcher_processes, diff --git a/steam_backlog_enforcer/_workout_budget.py b/steam_backlog_enforcer/_workout_budget.py new file mode 100644 index 0000000..88f3c91 --- /dev/null +++ b/steam_backlog_enforcer/_workout_budget.py @@ -0,0 +1,190 @@ +"""Today's gaming budget, decided by whether a workout was logged. + +The screen locker owns the question "did I work out today"; this module asks it +and turns the answer into seconds. Two values: the *earned* budget +(``daily_gaming_seconds``) when today has a counted workout, and the *unearned* +floor (``unearned_gaming_seconds``) when it does not. + +Three properties this is built around: + +**Fail closed.** Any failure to get an answer -- status server down, timeout, +malformed body, non-200 -- yields the *unearned* floor, logged at warning with +the concrete reason. Failing open would let the whole coupling be defeated by +stopping one user service, and a silent 8h would be indistinguishable from an +earned one. + +**Rising, in normal use -- but not guaranteed.** ``workout_today`` only ever +goes false->true within a day, so in the ordinary case the budget starts at the +floor and rises when a workout lands. That is a property of the *input*, not +something enforced here: nothing persists a per-day high-water mark, so +anything that changes the resolved answer mid-day -- editing the config, or +deploying this coupling 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. + +That was a deliberate call (2026-08-29): the alternative is a persisted +``granted_budget_seconds`` on ``PlaytimeState`` taking ``max(resolved, +granted)``. If a mid-day drop ever bites, that is the fix -- not a nudge to +these numbers. + +**One seam.** :func:`resolve_budget_seconds` is called from ``rules_for``, 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. + +Uses ``http.client`` rather than ``urllib.request`` deliberately, matching +``screen_locker._http_workout_fetch``: it takes a host and a path rather than a +URL, so there is no scheme for a bad config value to turn into a ``file://`` +read, and no lint waiver is needed to say so. +""" + +from __future__ import annotations + +import http.client +import json +import logging +import time +from typing import TYPE_CHECKING, Final +from urllib.parse import urlsplit + +if TYPE_CHECKING: + from steam_backlog_enforcer.config import Config + +logger = logging.getLogger(__name__) + +# Loopback, so a short timeout is generous. The daemon ticks frequently and +# must never block on this. +_TIMEOUT_SECONDS: Final = 2.0 + +# A successful answer is reused for this long. The daemon would otherwise make +# one HTTP call per tick, and the answer changes at most a few times a day. The +# cost is that a workout logged now raises the budget within a minute rather +# than instantly -- invisible next to a two-hour swing. +_CACHE_TTL_SECONDS: Final = 60.0 + +_SECONDS_PER_HOUR: Final = 3600.0 + +# Only *successes* are cached. A failure is retried on the very next tick, so a +# restarted status server takes effect immediately instead of serving the floor +# for another minute. A dict rather than a module global so no function here +# needs a `global` statement to write it. +_cache: dict[str, tuple[float, bool]] = {} + + +def reset_cache() -> None: + """Drop the cached answer, forcing the next call to re-ask.""" + _cache.clear() + + +def _fetch_workout_today(url: str) -> bool: + """Read ``gaming.workout_today`` from the screen-locker status API. + + Args: + url: The status endpoint, e.g. ``http://127.0.0.1:8770/api/status``. + + Returns: + Whether a counted workout is logged for today. + + Raises: + OSError: The endpoint could not be reached. + ValueError: The response was not usable JSON, or not a 200. + KeyError: The payload lacked the expected fields. + """ + split = urlsplit(url) + conn = http.client.HTTPConnection( + split.hostname or "127.0.0.1", + split.port or 80, + timeout=_TIMEOUT_SECONDS, + ) + try: + conn.request("GET", split.path or "/api/status") + resp = conn.getresponse() + body = resp.read() + if resp.status != http.HTTPStatus.OK: + msg = f"status {resp.status} {resp.reason}" + raise ValueError(msg) + finally: + conn.close() + return bool(json.loads(body)["gaming"]["workout_today"]) + + +def workout_logged_today(config: Config) -> bool | None: + """Ask the screen locker whether today has a counted workout. + + Args: + config: Loaded user configuration, for the status URL. + + Returns: + True or False when the locker answered, ``None`` when it could not be + reached or its answer could not be understood. ``None`` is deliberately + distinct from ``False`` so the caller can log which one happened. + """ + url = config.workout_status_url + now = time.monotonic() + cached = _cache.get(url) + if cached is not None and now - cached[0] < _CACHE_TTL_SECONDS: + return cached[1] + + try: + answer = _fetch_workout_today(url) + except OSError as exc: + logger.warning( + "Could not reach the screen-locker status API at %s (%s) — " + "applying the unearned gaming budget. Check " + "`systemctl --user status screen-locker-web`.", + url, + exc, + ) + return None + except (ValueError, KeyError, TypeError) as exc: + logger.warning( + "Screen-locker status API at %s returned something unusable (%s) — " + "applying the unearned gaming budget.", + url, + exc, + ) + return None + + _cache[url] = (now, answer) + return answer + + +def resolve_budget_seconds(config: Config) -> float: + """Return today's gaming budget in seconds. + + Args: + config: Loaded user configuration. + + Returns: + ``daily_gaming_seconds`` when a counted workout is logged today, and + ``unearned_gaming_seconds`` otherwise -- including whenever the answer + could not be obtained at all. + """ + earned = float(config.daily_gaming_seconds) + unearned = float(config.unearned_gaming_seconds) + if unearned > earned: + # A floor above the ceiling would silently *reward* skipping. Refuse + # the config rather than enforce it backwards. + logger.error( + "unearned_gaming_seconds (%.0f) exceeds daily_gaming_seconds " + "(%.0f); clamping to the earned budget so skipping a workout can " + "never buy more time than doing one.", + unearned, + earned, + ) + unearned = earned + + answer = workout_logged_today(config) + if answer is None: + return unearned + if answer: + return earned + logger.info( + "No counted workout logged today — gaming budget is %.1fh not %.1fh.", + unearned / _SECONDS_PER_HOUR, + earned / _SECONDS_PER_HOUR, + ) + return unearned diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index 731fe2a..feae330 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -83,11 +83,15 @@ class Config: one allowed set. Raising this weakens enforcement proportionally. """ daily_gaming_seconds: int = 8 * 3600 - """Gaming seconds allowed per gaming day, which starts at 06:00 local. - - Spending this budget shuts Steam down and masks the launcher binaries until - the next 06:00. See :mod:`steam_backlog_enforcer._playtime`. - """ + """Seconds allowed per gaming day *once earned* by a workout; without one + the day gets ``unearned_gaming_seconds``. Spending whichever applies shuts + Steam down and masks the launchers until the next 06:00.""" + unearned_gaming_seconds: int = 6 * 3600 + """Seconds allowed on a day with no counted workout: a floor the day starts + at and rises from, never a later cut, and the fail-closed answer when the + locker is unreachable. Rationale in ``_workout_budget``.""" + workout_status_url: str = "http://127.0.0.1:8770/api/status" + """Where to ask whether today has a counted workout. Loopback only.""" count_launcher_processes: bool = True """Whether time spent in non-Steam launchers counts against the budget. diff --git a/steam_backlog_enforcer/tests/_no_workout_http.py b/steam_backlog_enforcer/tests/_no_workout_http.py new file mode 100644 index 0000000..b0fae8a --- /dev/null +++ b/steam_backlog_enforcer/tests/_no_workout_http.py @@ -0,0 +1,54 @@ +"""Autouse stub for the workout-to-budget lookup. + +``rules_for`` resolves the daily budget by asking the screen locker whether +today has a counted workout, over loopback HTTP. Left unstubbed, every test +that builds rules would make a real request: slow, and -- far worse -- +answered differently depending on whether ``screen-locker-web.service`` happens +to be running and whether the machine's owner has trained today. Tests would +pass in the morning and fail in the evening. + +The stub returns the *earned* budget, which is what every pre-coupling test was +written against. Tests that care about the coupling patch a level below this, +at ``_workout_budget._fetch_workout_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. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +if TYPE_CHECKING: + from collections.abc import Iterator + + from steam_backlog_enforcer.config import Config + + +def _earned_budget(config: Config) -> float: + """Return the earned budget, as a day with a logged workout would. + + Args: + config: Loaded user configuration. + + Returns: + ``daily_gaming_seconds`` as a float. + """ + return float(config.daily_gaming_seconds) + + +@pytest.fixture(autouse=True) +def _no_workout_http() -> Iterator[None]: + """Stop rules_for from making a real HTTP call to the screen locker. + + Yields: + None, with the budget resolver stubbed for the whole test. + """ + with patch( + "steam_backlog_enforcer._playtime_state.resolve_budget_seconds", + side_effect=_earned_budget, + ): + yield diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index 3344305..a0ed707 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -23,10 +23,11 @@ import pytest # Imported for its autouse side effect: naming it here registers it. from steam_backlog_enforcer.tests._isolate_playtime import _isolate_playtime from steam_backlog_enforcer.tests._no_subprocess import _block_real_subprocesses +from steam_backlog_enforcer.tests._no_workout_http import _no_workout_http # Re-exported so ruff --fix does not delete the imports above: pytest # registers autouse fixtures by name, so they look unused to the linter. -__all__ = ["_block_real_subprocesses", "_isolate_playtime"] +__all__ = ["_block_real_subprocesses", "_isolate_playtime", "_no_workout_http"] from steam_backlog_enforcer.tests._total_block_paths import ( BLOCK, diff --git a/steam_backlog_enforcer/tests/test_budget_view.py b/steam_backlog_enforcer/tests/test_budget_view.py index 2bcf763..a0b43a8 100644 --- a/steam_backlog_enforcer/tests/test_budget_view.py +++ b/steam_backlog_enforcer/tests/test_budget_view.py @@ -23,7 +23,15 @@ 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 -RULES = rules_for(Config(), demo=False) +# 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 +# call to the screen locker and pins RULES to whatever budget today happens to +# have earned. Stub it explicitly so the constant is the earned budget always. +with patch( + "steam_backlog_enforcer._playtime_state.resolve_budget_seconds", + side_effect=lambda config: float(config.daily_gaming_seconds), +): + RULES = rules_for(Config(), demo=False) class TestStateAccess: diff --git a/steam_backlog_enforcer/tests/test_playtime_policy.py b/steam_backlog_enforcer/tests/test_playtime_policy.py index bd376a0..67ac2be 100644 --- a/steam_backlog_enforcer/tests/test_playtime_policy.py +++ b/steam_backlog_enforcer/tests/test_playtime_policy.py @@ -119,3 +119,66 @@ class TestPolicyEnforcementDisabled: mock_rec.assert_called_once_with(should_block=False) mock_shutdown.assert_not_called() assert out.is_blocked() is False + + +class TestPolicyBudgetRoseMidDay: + """Logging a workout raises the budget past what has already been spent. + + Before the workout coupling the budget could not move within a day, so + ``blocked_at`` and ``warned_seconds`` were only ever reset by the 06:00 + rollover. Both now have to come down with the mounts. + """ + + def _blocked_at_six_hours(self) -> PlaytimeState: + """Return state that hit the unearned 6h budget and got cut off.""" + return PlaytimeState( + day_key=TODAY, + seconds=6 * 3600, + blocked_at=1.0, + warned_seconds=[3600, 1800, 600, 300], + ) + + def test_the_block_is_released(self) -> None: + """reconcile is asked to unmount, which is what re-enables gaming.""" + with ( + patch("steam_backlog_enforcer._playtime.reconcile") as mock_rec, + patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"), + ): + _policy(self._blocked_at_six_hours(), _rules(), now=NOW) + mock_rec.assert_called_once_with(should_block=False) + + def test_blocked_at_is_cleared(self) -> None: + """Otherwise the UI reports "blocked" while gaming works fine.""" + with ( + patch("steam_backlog_enforcer._playtime.reconcile"), + patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"), + ): + out = _policy(self._blocked_at_six_hours(), _rules(), now=NOW) + assert out.is_blocked() is False + + def test_warnings_are_re_armed(self) -> None: + """Thresholds key on seconds *remaining*, so a raise re-crosses them. + + Left alone, the extra two hours would arrive with no warning at all. + """ + with ( + patch("steam_backlog_enforcer._playtime.reconcile"), + patch( + "steam_backlog_enforcer._playtime_cutoff.notify_desktop_user" + ) as mock_notify, + ): + out = _policy(self._blocked_at_six_hours(), _rules(), now=NOW) + # Two hours remain, so nothing is due yet -- but the record is clear, + # so the 1h/30m/10m/5m warnings will fire again as it drains. + assert out.warned_seconds == [] + mock_notify.assert_not_called() + + def test_an_unblocked_day_is_left_alone(self) -> None: + """The release path must not fire on an ordinary under-budget tick.""" + state = PlaytimeState(day_key=TODAY, seconds=100.0, warned_seconds=[3600]) + with ( + patch("steam_backlog_enforcer._playtime.reconcile"), + patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"), + ): + out = _policy(state, _rules(), now=NOW) + assert out.warned_seconds == [3600] diff --git a/steam_backlog_enforcer/tests/test_workout_budget.py b/steam_backlog_enforcer/tests/test_workout_budget.py new file mode 100644 index 0000000..3bc67e2 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_workout_budget.py @@ -0,0 +1,245 @@ +"""Tests for _workout_budget: the workout-to-gaming-budget coupling. + +The property that matters most here is that every way of *not* getting an +answer yields the unearned floor. If any of them fell through to the earned +budget, the whole coupling could be defeated by stopping one user service, and +a silent 8h would be indistinguishable from an earned one. +""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from steam_backlog_enforcer import _workout_budget +from steam_backlog_enforcer._workout_budget import ( + reset_cache, + resolve_budget_seconds, + workout_logged_today, +) +from steam_backlog_enforcer.config import Config + +if TYPE_CHECKING: + from collections.abc import Iterator + +_PKG = "steam_backlog_enforcer._workout_budget" + +_EARNED = 8 * 3600 +_UNEARNED = 6 * 3600 + + +@pytest.fixture(autouse=True) +def _clear_cache() -> Iterator[None]: + """Keep the module memo from leaking answers between tests. + + Yields: + None, with the cache empty before and after. + """ + reset_cache() + yield + reset_cache() + + +def _config(**over: object) -> Config: + """Build a config with the budget fields under test. + + Args: + over: Field overrides. + + Returns: + A Config instance; never loaded from or saved to disk. + """ + config = Config() + config.daily_gaming_seconds = _EARNED + config.unearned_gaming_seconds = _UNEARNED + config.workout_status_url = "http://127.0.0.1:8770/api/status" + for key, value in over.items(): + setattr(config, key, value) + return config + + +class TestWorkoutLoggedToday: + """The locker's answer, or None when there was not one.""" + + def test_true_when_a_workout_is_logged(self) -> None: + """A truthy workout_today comes back as True.""" + with patch(f"{_PKG}._fetch_workout_today", return_value=True): + assert workout_logged_today(_config()) is True + + def test_false_when_none_is_logged(self) -> None: + """A falsy workout_today comes back as False, not None.""" + with patch(f"{_PKG}._fetch_workout_today", return_value=False): + assert workout_logged_today(_config()) is False + + def test_unreachable_server_is_none_not_false(self) -> None: + """ "Could not ask" must stay distinguishable from "asked, no".""" + with patch(f"{_PKG}._fetch_workout_today", side_effect=OSError("refused")): + assert workout_logged_today(_config()) is None + + def test_unreachable_server_says_why( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """The warning names the URL and how to check the unit.""" + with ( + caplog.at_level(logging.WARNING), + patch(f"{_PKG}._fetch_workout_today", side_effect=OSError("refused")), + ): + workout_logged_today(_config()) + assert "127.0.0.1:8770" in caplog.text + assert "screen-locker-web" in caplog.text + + @pytest.mark.parametrize( + "error", + [ValueError("not json"), KeyError("gaming"), TypeError("bad shape")], + ) + def test_an_unusable_answer_is_none(self, error: Exception) -> None: + """Malformed JSON, a missing key or a wrong shape all fail closed.""" + with patch(f"{_PKG}._fetch_workout_today", side_effect=error): + assert workout_logged_today(_config()) is None + + def test_a_successful_answer_is_cached(self) -> None: + """The daemon ticks often; one HTTP call per tick would be silly.""" + with patch(f"{_PKG}._fetch_workout_today", return_value=True) as fetch: + workout_logged_today(_config()) + workout_logged_today(_config()) + assert fetch.call_count == 1 + + def test_a_failure_is_not_cached(self) -> None: + """A restarted status server must take effect on the next tick.""" + with patch(f"{_PKG}._fetch_workout_today", side_effect=OSError("x")) as fetch: + workout_logged_today(_config()) + workout_logged_today(_config()) + assert fetch.call_count == 2 + + def test_the_cache_expires(self) -> None: + """A workout logged later in the day still raises the budget.""" + with ( + patch(f"{_PKG}._fetch_workout_today", return_value=False) as fetch, + patch(f"{_PKG}.time.monotonic", side_effect=[0.0, 999.0]), + ): + workout_logged_today(_config()) + workout_logged_today(_config()) + assert fetch.call_count == 2 + + def test_different_urls_do_not_share_a_cache_entry(self) -> None: + """Keyed by URL, so a test or a retarget cannot read a stale answer.""" + with patch(f"{_PKG}._fetch_workout_today", return_value=True) as fetch: + workout_logged_today(_config()) + workout_logged_today( + _config(workout_status_url="http://127.0.0.1:9999/api/status"), + ) + assert fetch.call_count == 2 + + +class TestResolveBudgetSeconds: + """The number the enforcer actually bills against.""" + + def test_a_workout_earns_the_full_budget(self) -> None: + """8h once a counted workout is logged today.""" + with patch(f"{_PKG}._fetch_workout_today", return_value=True): + assert resolve_budget_seconds(_config()) == float(_EARNED) + + def test_no_workout_gets_the_floor(self) -> None: + """6h on a day with nothing logged.""" + with patch(f"{_PKG}._fetch_workout_today", return_value=False): + assert resolve_budget_seconds(_config()) == float(_UNEARNED) + + def test_an_unreachable_locker_gets_the_floor(self) -> None: + """Fail closed: a dead locker costs gaming time, never grants it.""" + with patch(f"{_PKG}._fetch_workout_today", side_effect=OSError("refused")): + assert resolve_budget_seconds(_config()) == float(_UNEARNED) + + def test_the_cut_is_explained(self, caplog: pytest.LogCaptureFixture) -> None: + """A 2h swing should never be silent.""" + with ( + caplog.at_level(logging.INFO), + patch(f"{_PKG}._fetch_workout_today", return_value=False), + ): + resolve_budget_seconds(_config()) + assert "6.0h" in caplog.text + assert "8.0h" in caplog.text + + def test_a_floor_above_the_ceiling_is_refused( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Misconfigured, skipping would otherwise buy MORE time than working out.""" + config = _config(unearned_gaming_seconds=10 * 3600) + with ( + caplog.at_level(logging.ERROR), + patch(f"{_PKG}._fetch_workout_today", return_value=False), + ): + assert resolve_budget_seconds(config) == float(_EARNED) + assert "exceeds daily_gaming_seconds" in caplog.text + + +class TestFetchWorkoutToday: + """The HTTP read itself, over a real loopback server.""" + + def test_reads_the_flag(self, http_stub: str) -> None: + """A well-formed payload yields the boolean.""" + assert _workout_budget._fetch_workout_today(http_stub) is True + + def test_a_non_200_is_an_error(self, http_stub_500: str) -> None: + """An error status must not be parsed as "no workout".""" + with pytest.raises(ValueError, match="status 500"): + _workout_budget._fetch_workout_today(http_stub_500) + + +@pytest.fixture +def http_stub() -> Iterator[str]: + """Serve a status payload saying a workout was logged today. + + Yields: + The URL of the stub endpoint. + """ + yield from _serve(200, json.dumps({"gaming": {"workout_today": True}}).encode()) + + +@pytest.fixture +def http_stub_500() -> Iterator[str]: + """Serve a 500, as a broken payload builder would. + + Yields: + The URL of the stub endpoint. + """ + yield from _serve(500, b"status error") + + +def _serve(status: int, body: bytes) -> Iterator[str]: + """Run a one-route HTTP server on an ephemeral loopback port. + + Args: + status: Status code to return. + body: Body to return. + + Yields: + The URL of the endpoint. + """ + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + import threading + + class _Handler(BaseHTTPRequestHandler): + def log_message(self, *args: object) -> None: + """Stay quiet in test output.""" + + def do_GET(self) -> None: + """Serve the canned response.""" + self.send_response(status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}/api/status" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2)