screen-locker/screen_locker/tests/test_mcp.py
Krzysztof kuhy Rudnicki 79879263ba Support multiple workouts per day; fix the dead sync; ban silent failures
The workout log stored ONE workout per calendar day, so a second same-day
workout was silently dropped and could never earn credit.

Multiple workouts per day:
- workout_log.json: {date: entry} -> {date: [entry, ...]}. Legacy entries are
  wrapped on read (_log_io.load_workout_log) so their HMACs stay valid — no
  re-signing; every write normalizes the file, migrating it progressively.
- New write chokepoint (_log_mixin.write_signed_entry) APPENDS and dedups by a
  stable workout_id, returning RecordResult so callers can tell a new workout
  from a duplicate. Re-verifying the same workout is a no-op.
- count_weekly_workouts is anti-gaming: each VERIFIED workout counts
  individually; self-reported manual entries count at most once per day. Four
  real workouts in one day satisfy the week; four manual entries count as one.
- Reward: the day's first counted workout gets the base +2h; an additional
  same-day VERIFIED workout gets +1h; an additional manual gets nothing. The
  old "+1h when weekly_count > 4" gate is replaced — weekly volume is the
  banked bonus's job and shouldn't be paid twice.
- Fixed: the unlock screen labelled the +1h "#{weekly_count - 4}", which can
  render negative under the new rule. Removed the dead runnerup_manual type.
- Flutter: the manual budget counts distinct DAYS, matching the PC.

PC->phone sync (it had never worked once):
- push_pc_manuals pushed from a side-store written at log-time, so anything
  logged before that feature existed could never publish. _manual_push now
  derives the crdt log directly from workout_log.json — the single source of
  truth — covering manual AND verified runs. The record id is the workout's own
  workout_id (derived for legacy entries to match the phone's format) and the
  HLC comes from the entry's own timestamp, so a re-push is byte-identical and
  the repo never churns.
- The phone reads every kind (readMergedWorkoutPayloads) and lists the PC's
  workouts under "SYNCED FROM PC". The manual budget still sees manual-only, so
  a verified run can't consume a self-report slot.
- Settings claimed "Connected." whenever a token STRING existed. It now
  verifies against the API on open, so a revoked token reports 401 instead of
  showing a green lie.

Never fail silently:
- 30 swallowed exceptions across 19 files now log WARNING with the concrete
  cause AND consequence — plus the token guards in _workout_sync that returned
  [] in total silence, the exact bug that hid the dead sync for weeks.
- New pre-commit hook (scripts/check_silent_failures.py) fails any except that
  neither re-raises nor logs loudly. ruff's S110 only catches `except: pass`;
  the real pattern here was `except: log.info(); return []`. Documented in
  CLAUDE.md.

Verified on real data: the Jul 13 run (pulled from its actual TCX over ADB) now
counts alongside that day's manual squash, and all three workouts are published
to kuhyx/syncs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Fv8ULVt3Q67GwCFbM7doHB
2026-07-17 09:22:31 +02:00

149 lines
5.0 KiB
Python

"""Tests for the read-only MCP server tools in ``_mcp``.
Helpers are patched at the ``_mcp`` module namespace (where they are imported),
keeping each tool's own logic under test while isolating the already-tested
leaf functions. ``get_status`` / ``explain_lock`` are exercised against a real
:class:`StatusSnapshot` so the ``dataclasses.asdict`` conversion (and its
JSON-friendliness) is covered for real, not mocked away.
"""
from __future__ import annotations
from dataclasses import asdict
import json
from unittest.mock import patch
from screen_locker import _mcp
from screen_locker._compliance_state import (
AutoUpgradeOpportunity,
LockExplanation,
PredicateResult,
)
from screen_locker._status_data import (
DayStatus,
ManualWorkoutBudgetStatus,
ShutdownProjection,
ShutdownProjectionDay,
SickBudgetStatus,
StatusSnapshot,
WeeklySummary,
)
def _lock_explanation() -> LockExplanation:
return LockExplanation(
fired=True,
stage="full_lock_pending_heat_check",
reason="No skip condition applies.",
trace=(PredicateResult("scheduled_skip", fired=False, reason="not a skip"),),
auto_upgrade=AutoUpgradeOpportunity(
would_attempt=False, via="none", reason="No pending opportunity."
),
heat_skip_evaluated=False,
)
def _snapshot() -> StatusSnapshot:
day = DayStatus(
date="2026-07-10",
label="Fri Jul 10",
entry_types=(),
source="",
counted=False,
day_count=0,
is_sick_day=False,
)
return StatusSnapshot(
today=day,
week=WeeklySummary(
days=(day,), counted_count=2, minimum=4, remaining=2, extra=0
),
bonus_hours_this_week=0,
streak=1,
early_bird_extended=False,
shutdown=ShutdownProjection(
tonight=(22, 22, 5),
rest_of_week=(
ShutdownProjectionDay(label="Mon", hour=21, speculative=False),
),
next_week_preview=(
ShutdownProjectionDay(label="Mon", hour=21, speculative=True),
),
explanation="explanation text",
),
lock_explanation=_lock_explanation(),
sick_budget=SickBudgetStatus(
used_7d=0,
budget_7d=1,
used_30d=0,
budget_30d=3,
used_90d=0,
budget_90d=10,
debt=0,
exhausted=False,
),
manual_workout_budget=ManualWorkoutBudgetStatus(
used_7d=0, budget_7d=2, used_30d=0, budget_30d=5, exhausted=False
),
generated_at="2026-07-10T12:00:00+00:00",
)
class TestReadTools:
def test_get_status_returns_json_friendly_asdict(self) -> None:
snap = _snapshot()
with patch.object(_mcp, "gather_status", return_value=snap) as gs:
out = _mcp.get_status()
gs.assert_called_once_with()
assert out == asdict(snap)
# asdict output must be pure JSON (no datetime/dataclass leaks).
json.dumps(out)
def test_get_summary_combines_line_and_state(self) -> None:
with (
patch.object(_mcp, "gather_status") as gs,
patch.object(_mcp, "format_summary_line", return_value="LINE") as fsl,
patch.object(_mcp, "_compliance_state_word", return_value="ok") as csw,
):
out = _mcp.get_summary()
assert out == {"summary_line": "LINE", "compliance_state": "ok"}
fsl.assert_called_once_with(gs.return_value)
csw.assert_called_once_with(gs.return_value)
def test_explain_lock_returns_lock_explanation_asdict(self) -> None:
snap = _snapshot()
with patch.object(_mcp, "gather_status", return_value=snap):
out = _mcp.explain_lock()
assert out == asdict(snap.lock_explanation)
assert out["fired"] is True
json.dumps(out)
class TestGetFlags:
def test_flags_dict_and_predicate_wiring(self) -> None:
with (
patch.object(_mcp, "load_history", return_value="HIST") as lh,
patch.object(_mcp, "has_logged_today", return_value=True) as hlt,
patch.object(_mcp, "is_scheduled_skip_today", return_value=False) as iss,
patch.object(_mcp, "is_early_bird_pending", return_value=True) as iep,
patch.object(_mcp, "is_sick_day_today", return_value=False) as isd,
):
out = _mcp.get_flags()
assert out == {
"has_logged_today": True,
"is_scheduled_skip_today": False,
"is_early_bird_pending": True,
"is_sick_day_today": False,
}
lh.assert_called_once_with()
hlt.assert_called_once_with(_mcp._LOG_FILE)
iss.assert_called_once_with(_mcp.SCHEDULED_SKIPS_FILE)
iep.assert_called_once_with(_mcp.EARLY_BIRD_PENDING_FILE)
isd.assert_called_once_with("HIST")
def test_main_runs_stdio_server() -> None:
with patch.object(_mcp.mcp, "run") as run:
_mcp.main()
run.assert_called_once_with()