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

focus_project: interview milestones, evidence

Browse Source 
Project.kind (focus default; old rows load unchanged) plus interview
fields test_at, invite_url, trips, weekend_answers. Interview budget
30 min weekdays / 120 Sat-Sun, exempt from daily_floor, unlock clamped
after the 0.7.0 history ratchet. Milestones are generated (rules,
trip_ask, phone, mock, slot), never stored. Append-only evidence store
behind the _paths seam. Calendar dropped: slot = slot entry + slot_confirm.

Failure paths:
- unanswered trip question: fail closed (reads as a trip, phone due)
- late no-trip answer cannot erase an overdue phone milestone: closed
- phone source missing/stale/crashed: unknown, locks: fail closed
- trip_ask answer: evidence row first, then registry; a crash between
  leaves the question open: fail closed
- add_trip/answer_weekend loosening: ValueError, nothing written
- owed_lock with unreadable registry or evidence: fail open, logged
  (focus-guard's unreadable-registry rule); test day never locks
- status() milestones with unreadable evidence: empty list, logged
- torn or non-utf8 evidence lines: skipped and logged

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWZUTC9Zzdzsx7VW7dm
This commit is contained in:
Krzysztof Rudnicki 2026-11-10 15:38:51 +01:00
parent 5a87868763
commit 4c1ffd900b
22 changed files with 2022 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
focus_project/README.md
View File

@ -38,3 +38,12 @@ Floor (0.7.1, coordinator decision for kuhy): `daily_minutes >= 30 x replaced ga
enforced on create and update (`daily_floor(replaces)`), so a day cannot be made cheap.
Open the work: `python3 -m focus_project.open [--no-site]` (tutor focus screen, `work_url`).
Install: `./install.sh` (stdlib only, editable user install for `/usr/bin/python3`).
Interview prep (`kind="interview"`, 0.9): `create_project(..., kind="interview", test_at=, invite_url=, trips=, weekend_answers=)`;
`end` must be the test day. Budget: 30 min weekdays / 120 Sat-Sun (`daily_on`), exempt from `daily_floor`, unlock clamped.
Milestones are generated, never stored: `milestones(project, now)` -> `rules`, `trip_ask:<sat>`, `phone:<trip start>`, `mock`, `slot`
with states pending/done/missed/late_done/unknown. `owed_lock(now)` = earliest overdue uncompleted one (never on test day; wire it
after focus-guard's sick/free-day stand-down). `status(day)` gains `milestones` and `missed`.
Evidence is append-only (`evidence/<id>.jsonl` + copied files): `record_evidence(id, milestone, kind, value, file=, now=)`
(`file`, `tap`, `score`, `slot`, `slot_confirm`, tutor's `phone_check`). `add_trip` / `answer_weekend` only get stricter (ValueError).
`run_checks(id)` runs the `phone` source (reads the latest `phone_check` row) and latches changes; `register_source` swaps it.

15
focus_project/focus_project/__init__.py
View File

@ -23,25 +23,40 @@ from focus_project._api import (
stands_down,
)
from focus_project._errors import RegistryError
from focus_project._milestones import Milestone, milestones, owed_lock
from focus_project._model import Limits, Project
from focus_project._mutate import add_trip, answer_weekend, record_evidence
from focus_project._trips import Trip
from focus_project._update import Update, apply_update, update_project
from focus_project.activity import record_activity
from focus_project.checks import register_source, run_checks
from focus_project.evidence import Evidence
from focus_project.minutes import minutes_for
from focus_project.status import status
__all__ = [
"Evidence",
"Limits",
"Milestone",
"Project",
"RegistryError",
"Trip",
"Update",
"active",
"add_trip",
"answer_weekend",
"apply_update",
"create_project",
"daily_floor",
"end_project",
"list_projects",
"milestones",
"minutes_for",
"owed_lock",
"record_activity",
"record_evidence",
"register_source",
"run_checks",
"stands_down",
"status",
"update_project",

51
focus_project/focus_project/_api.py
View File

@ -4,19 +4,24 @@
from __future__ import annotations
from dataclasses import replace
from datetime import date # noqa: TC003
import logging
import re
import time
from urllib.parse import urlsplit
from typing import TYPE_CHECKING
from focus_project import _clock, _store
from focus_project import _clock, _interview, _store
from focus_project._errors import RegistryError
from focus_project._kind import FOCUS
from focus_project._model import GATES, Project, normalize_gate
if TYPE_CHECKING:
from datetime import date, datetime
from focus_project._trips import Trip
_LOG = logging.getLogger(__name__)
_ID = re.compile(r"[a-z0-9][a-z0-9-]*")
MAX_WORK_URL = 2000
MAX_WORK_URL = _interview.MAX_URL
# A project must ask real work: the focus earner pays per minute of ``daily``.
MIN_DAILY_PER_GATE = 30
@ -53,17 +58,7 @@ def stands_down(gate: str, day: date) -> bool:
def _validate_work_url(url: str) -> None:
"""Empty, or a plain http(s) URL with a host: it goes to ``xdg-open``."""
if not url:
return
parts = urlsplit(url) if url.isprintable() and " " not in url else None
if (
parts is None
or len(url) > MAX_WORK_URL
or parts.scheme not in {"http", "https"}
or not parts.netloc
):
msg = "work_url must be an http(s) URL without spaces (or empty)"
raise ValueError(msg)
_interview.check_http_url(url, "work_url")
def daily_floor(replaces: tuple[str, ...]) -> int:
@ -85,8 +80,9 @@ def _validate(project: Project) -> None:
if not project.replaces or not set(project.replaces) <= GATES:
msg = f"replaces must be a non-empty subset of gates {sorted(GATES)}"
raise ValueError(msg)
_interview.validate(project)
floor = daily_floor(project.replaces)
if project.daily_minutes < floor:
if project.kind == FOCUS and project.daily_minutes < floor:
msg = (
f"daily_minutes must be at least {floor} "
f"({MIN_DAILY_PER_GATE} per replaced gate)"
@ -109,8 +105,19 @@ def create_project( # noqa: PLR0913
subjects: list[str] | None = None,
site_patterns: list[str] | None = None,
work_url: str = "",
kind: str = FOCUS,
test_at: datetime | None = None,
invite_url: str = "",
trips: list[Trip] | None = None,
weekend_answers: dict[date, bool] | None = None,
) -> Project:
"""Add a project; ValueError on bad input, a duplicate id or an overlap."""
"""Add a project; ValueError on bad input, a duplicate id or an overlap.
``kind="interview"`` takes ``test_at`` (naive local; its date must be
``end``), ``invite_url``, ``trips`` and ``weekend_answers``; an interview
is exempt from :func:`daily_floor` and runs the 30/120 budget.
"""
created_ms = time.time_ns() // 1_000_000
project = Project(
id=project_id,
name=name,
@ -122,7 +129,15 @@ def create_project( # noqa: PLR0913
subjects=tuple(subjects or ()),
site_patterns=tuple(site_patterns or ()),
work_url=work_url,
created_ms=time.time_ns() // 1_000_000,
created_ms=created_ms,
kind=kind,
test_at=test_at,
invite_url=invite_url,
trips=tuple(
t if t.entered_ms else replace(t, entered_ms=created_ms)
for t in trips or ()
),
weekend_answers=tuple(sorted((weekend_answers or {}).items())),
)
_validate(project)
with _store.locked():

89
focus_project/focus_project/_interview.py Normal file
View File

@ -0,0 +1,89 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Validation of the interview-only fields."""
from __future__ import annotations
from datetime import date, timedelta
from typing import TYPE_CHECKING
from urllib.parse import urlsplit
from focus_project._kind import FOCUS, KINDS, WEEKDAY_MINUTES
if TYPE_CHECKING:
from focus_project._model import Project
MAX_URL = 2000
SATURDAY = 5
def check_http_url(url: str, label: str) -> None:
"""Empty, or a plain http(s) URL with a host: it goes to ``xdg-open``."""
if not url:
return
parts = urlsplit(url) if url.isprintable() and " " not in url else None
if (
parts is None
or len(url) > MAX_URL
or parts.scheme not in {"http", "https"}
or not parts.netloc
):
msg = f"{label} must be an http(s) URL without spaces (or empty)"
raise ValueError(msg)
def saturdays(project: Project) -> list[date]:
"""The Saturdays whose weekend lies inside D1..T-1."""
if project.test_at is None:
return []
last = project.test_at.date() - timedelta(days=1)
day = project.start + timedelta(days=(SATURDAY - project.start.weekday()) % 7)
found: list[date] = []
while day <= last:
found.append(day)
day += timedelta(days=7)
return found
def _has_interview_fields(project: Project) -> bool:
return bool(
project.test_at
or project.invite_url
or project.trips
or project.weekend_answers
)
def _validate_interview(project: Project) -> None:
if project.test_at is None or project.test_at.tzinfo is not None:
msg = "an interview needs test_at as a naive local datetime"
raise ValueError(msg)
test_day = project.test_at.date()
if project.end != test_day:
msg = f"end must be the test day {test_day}"
raise ValueError(msg)
if project.daily_minutes < WEEKDAY_MINUTES:
msg = f"daily_minutes must be at least {WEEKDAY_MINUTES}"
raise ValueError(msg)
for trip in project.trips:
if trip.end < trip.start or trip.start > test_day:
msg = f"trip {trip.start}..{trip.end} must start by the test day"
raise ValueError(msg)
weekends = saturdays(project)
for day, _ in project.weekend_answers:
if day not in weekends:
msg = f"{day} is not a weekend inside the project before the test day"
raise ValueError(msg)
def validate(project: Project) -> None:
"""ValueError unless the kind and its interview fields are coherent."""
if project.kind not in KINDS:
msg = f"kind must be one of {sorted(KINDS)}"
raise ValueError(msg)
check_http_url(project.invite_url, "invite_url")
if project.kind == FOCUS:
if _has_interview_fields(project):
msg = "test_at, invite_url, trips, weekend_answers need kind='interview'"
raise ValueError(msg)
return
_validate_interview(project)

23
focus_project/focus_project/_kind.py Normal file
View File

@ -0,0 +1,23 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Project kinds and the interview budget (the one place the numbers live)."""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from datetime import date
FOCUS = "focus"
INTERVIEW = "interview"
KINDS = frozenset({FOCUS, INTERVIEW})
# kuhy, 2026-11-10: near-zero weekday energy, weekends are for learning.
WEEKDAY_MINUTES = 30
WEEKEND_MINUTES = 120
_FIRST_WEEKEND_DAY = 5 # Saturday
def interview_minutes(day: date) -> int:
"""The daily target of an interview project on ``day`` (Sat/Sun = 120)."""
return WEEKEND_MINUTES if day.weekday() >= _FIRST_WEEKEND_DAY else WEEKDAY_MINUTES

218
focus_project/focus_project/_milestones.py Normal file
View File

@ -0,0 +1,218 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Interview milestones: generated from the project, judged from evidence.
Nothing here is stored or editable. ``due`` and the state are recomputed from
the project (kind, test_at, trips, weekend answers) and the append-only
evidence rows, so a milestone cannot be moved or ticked off by editing a row.
Failure modes (R8): an unanswered trip question reads as a trip (fail
closed); a check that could not run is ``unknown`` and counts as not done
(fail closed); an unreadable registry or evidence file arms no lock (fail
open, logged), matching focus-guard's "unreadable registry" rule.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime, time, timedelta
import logging
from typing import TYPE_CHECKING, Any, Literal
from focus_project import _api, _interview, evidence
from focus_project._errors import RegistryError
from focus_project._kind import INTERVIEW
if TYPE_CHECKING:
from collections.abc import Callable
from focus_project._model import Project
_LOG = logging.getLogger(__name__)
State = Literal["pending", "done", "missed", "late_done", "unknown"]
PHONE_LEAD = timedelta(hours=12)
_RED = frozenset({"missed", "unknown"})
@dataclass(frozen=True)
class Milestone:
"""One generated obligation of an interview project."""
id: str
due: datetime
state: State
completed: bool
@property
def red(self) -> bool:
"""Past due and not done (also: the check could not run)."""
return self.state in _RED
def to_dict(self) -> dict[str, Any]:
"""The status() row: id, due ISO, state."""
return {"id": self.id, "due": self.due.isoformat(), "state": self.state}
def _at(day: date, hour: int, minute: int = 0) -> datetime:
return datetime.combine(day, time(hour, minute))
def _trip_due(trip_start: date, entered_ms: int) -> datetime:
"""Start - 12 h, or the moment it was entered if that left < 12 h."""
nominal = _at(trip_start, 0) - PHONE_LEAD
if entered_ms > 0:
entered = evidence.from_ms(entered_ms)
if nominal - entered < PHONE_LEAD:
return entered
return nominal
def _is_trip(
*,
answered: bool | None,
answer_ms: int | None,
ask_due: datetime,
phone_due: datetime,
now: datetime,
) -> bool:
"""Whether the weekend counts as a trip (fail closed).
Unanswered once the question is due = a trip. A no-trip answer given
only after the phone prep was already due cannot erase that overdue
milestone: the answer row's own timestamp decides, not a stored flag.
"""
if answered is None:
return now >= ask_due
if answered:
return True
return answer_ms is not None and answer_ms >= evidence.to_ms(phone_due)
def plan(
project: Project, test_day: date, now: datetime, rows: list[evidence.Evidence]
) -> dict[str, datetime]:
"""Milestone id -> due; ``project`` must be a validated interview."""
dues = {
"rules": _at(project.start, 23, 59),
"mock": _at(test_day - timedelta(days=2), 23, 59),
"slot": _at(test_day - timedelta(days=1), 12),
}
answers = dict(project.weekend_answers)
for sat in _interview.saturdays(project):
ask_due = _at(sat - timedelta(days=2), 23, 59)
phone_due = _at(sat, 0) - PHONE_LEAD
dues[f"trip_ask:{sat}"] = ask_due
covered = any(t.covers(sat) for t in project.trips)
row = evidence.first(rows, f"trip_ask:{sat}", "answer")
answer_ms = None if row is None else row.ms
if not covered and _is_trip(
answered=answers.get(sat),
answer_ms=answer_ms,
ask_due=ask_due,
phone_due=phone_due,
now=now,
):
dues[f"phone:{sat}"] = phone_due
for trip_row in project.trips:
dues[f"phone:{trip_row.start}"] = _trip_due(trip_row.start, trip_row.entered_ms)
return dues
def _when(row: evidence.Evidence | None) -> datetime | None:
return None if row is None else row.at
Rows = list[evidence.Evidence]
Completion = tuple[datetime | None, bool]
def _by_evidence(kind: str) -> Callable[[str, Rows, Project], Completion]:
"""Done at the first evidence row of ``kind`` (rules file, mock score)."""
def judge(mid: str, rows: Rows, _project: Project) -> Completion:
return _when(evidence.first(rows, mid, kind)), False
return judge
def _trip_ask(mid: str, rows: Rows, project: Project) -> Completion:
"""Done at the answer row; an answer with no row was given at creation."""
if date.fromisoformat(mid.partition(":")[2]) not in dict(project.weekend_answers):
return None, False
return _when(evidence.first(rows, mid, "answer")) or evidence.from_ms(0), False
def _slot(mid: str, rows: Rows, _project: Project) -> Completion:
"""Done once the latest slot entry has a confirm tap at or after it."""
slot = evidence.last(rows, mid, "slot")
if slot is None:
return None, False
confirm = evidence.first(rows, mid, "slot_confirm", since_ms=slot.ms)
return (None if confirm is None else max(slot.at, confirm.at)), False
def _phone(mid: str, rows: Rows, _project: Project) -> Completion:
"""Done when the offline tap and a passing auto check both exist.
A tap with no verdict at all (or a dark source) is ``unknown``.
"""
tap = evidence.first(rows, mid, "tap")
passed, latest, _ = evidence.check_summary(rows, mid)
if tap is not None and passed is not None:
return max(tap.at, evidence.from_ms(passed)), False
return None, tap is not None and latest is None
_JUDGES = {
"rules": _by_evidence("file"),
"mock": _by_evidence("score"),
"trip_ask": _trip_ask,
"slot": _slot,
"phone": _phone,
}
def _judge(
mid: str, due: datetime, rows: Rows, project: Project, now: datetime
) -> Milestone:
done_at, unknown = _JUDGES[mid.partition(":")[0]](mid, rows, project)
if done_at is not None:
state: State = "done" if done_at <= due else "late_done"
return Milestone(mid, due, state, completed=True)
if now < due:
return Milestone(mid, due, "pending", completed=False)
return Milestone(mid, due, "unknown" if unknown else "missed", completed=False)
def milestones(project: Project, now: datetime) -> list[Milestone]:
"""The project's milestones at ``now`` (naive local), earliest due first.
Empty for a focus project. May raise OSError reading the evidence file.
"""
if project.kind != INTERVIEW or project.test_at is None:
return []
rows = evidence.read(project.id)
dues = plan(project, project.test_at.date(), now, rows)
found = [_judge(mid, due, rows, project, now) for mid, due in dues.items()]
return sorted(found, key=lambda m: (m.due, m.id))
def owed_lock(now: datetime) -> Milestone | None:
"""The earliest overdue, uncompleted milestone that locks, or None.
Locks (missed or unknown) from the moment it is due, every day of the
project except the test day itself, where nothing locks. It does not
consult sickdays/freedays: focus-guard's ``_off(today)`` precondition
already stands the lock down on those days, so wire this after it.
Unreadable state arms nothing (fail open, logged).
"""
try:
project = _api.active(now.date())
if project is None or project.test_at is None:
return None
if now.date() >= project.test_at.date():
return None
owed = [m for m in milestones(project, now) if not m.completed and now >= m.due]
except (RegistryError, OSError) as exc:
_LOG.warning("interview state unreadable, no milestone lock: %s", exc)
return None
return owed[0] if owed else None

47
focus_project/focus_project/_model.py
View File

@ -4,10 +4,12 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from datetime import date, datetime
from typing import Any
from focus_project._errors import RegistryError
from focus_project._kind import FOCUS, INTERVIEW, interview_minutes
from focus_project._trips import Trip
GATES = frozenset({"leetcode", "reading", "tutor"})
_ALIASES = {"book": "reading"}
@ -18,6 +20,17 @@ def normalize_gate(gate: str) -> str:
return _ALIASES.get(gate, gate)
def _answers(raw: dict[str, Any] | None) -> tuple[tuple[date, bool], ...]:
"""Parse the saturday -> bool map; a non-bool answer is a bad row."""
pairs = []
for key, value in (raw or {}).items():
if not isinstance(value, bool):
msg = f"weekend answer for {key} is not true/false"
raise TypeError(msg)
pairs.append((date.fromisoformat(key), value))
return tuple(sorted(pairs))
@dataclass(frozen=True)
class Limits:
"""The minute targets that held up to and including ``until``."""
@ -70,19 +83,30 @@ class Project:
work_url: str = ""
# Earlier targets, oldest first: an edit applies from its day, never back.
history: tuple[Limits, ...] = ()
# "focus" (every pre-interview row) or "interview"; the rest is interview-only.
kind: str = FOCUS
# Local naive datetime of the test; its date is the project's last day.
test_at: datetime | None = None
invite_url: str = ""
trips: tuple[Trip, ...] = ()
# Saturday -> True (trip that weekend) / False (no trip); sorted pairs.
weekend_answers: tuple[tuple[date, bool], ...] = ()
def _limits_on(self, day: date) -> Limits | None:
return next((h for h in self.history if day <= h.until), None)
def daily_on(self, day: date) -> int:
"""The daily target that held on ``day``."""
if self.kind == INTERVIEW:
return interview_minutes(day)
past = self._limits_on(day)
return self.daily_minutes if past is None else past.daily_minutes
def unlock_on(self, day: date) -> int:
"""The unlock target that held on ``day``."""
"""The unlock target that held on ``day`` (clamped for an interview)."""
past = self._limits_on(day)
return self.unlock_minutes if past is None else past.unlock_minutes
held = self.unlock_minutes if past is None else past.unlock_minutes
return min(held, interview_minutes(day)) if self.kind == INTERVIEW else held
def replaces_on(self, day: date) -> tuple[str, ...]:
"""The gates this project switched off on ``day``."""
@ -104,7 +128,7 @@ class Project:
def to_dict(self) -> dict[str, Any]:
"""The projects.json row."""
return {
row: dict[str, Any] = {
"id": self.id,
"name": self.name,
"start": self.start.isoformat(),
@ -118,13 +142,23 @@ class Project:
"created_ms": self.created_ms,
"work_url": self.work_url,
"history": [h.to_dict() for h in self.history],
"kind": self.kind,
}
if self.kind == INTERVIEW:
row["test_at"] = self.test_at.isoformat() if self.test_at else None
row["invite_url"] = self.invite_url
row["trips"] = [t.to_dict() for t in self.trips]
row["weekend_answers"] = {
day.isoformat(): trip for day, trip in self.weekend_answers
}
return row
@classmethod
def from_dict(cls, row: Any) -> Project: # noqa: ANN401
"""Parse a registry row, or raise :class:`RegistryError`."""
try:
ended = row.get("ended_on")
test_at = row.get("test_at")
return cls(
id=str(row["id"]),
name=str(row["name"]),
@ -139,6 +173,11 @@ class Project:
created_ms=int(row.get("created_ms", 0)),
work_url=str(row.get("work_url") or ""),
history=tuple(Limits.from_dict(h) for h in row.get("history", ())),
kind=str(row.get("kind") or FOCUS),
test_at=datetime.fromisoformat(test_at) if test_at else None,
invite_url=str(row.get("invite_url") or ""),
trips=tuple(Trip.from_dict(t) for t in row.get("trips", ())),
weekend_answers=_answers(row.get("weekend_answers")),
)
except (KeyError, TypeError, ValueError, AttributeError) as exc:
msg = f"bad project row {row!r}: {exc}"

178
focus_project/focus_project/_mutate.py Normal file
View File

@ -0,0 +1,178 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Interview mutations: trips, weekend answers, evidence. Stricter only."""
from __future__ import annotations
from dataclasses import replace
from datetime import date, datetime
import math
from typing import TYPE_CHECKING
from focus_project import _interview, _milestones, _store, evidence
from focus_project._kind import INTERVIEW
from focus_project._trips import Trip, add
if TYPE_CHECKING:
from pathlib import Path
from focus_project._model import Project
# Which evidence kinds each milestone family accepts from a caller.
# ``check`` and ``answer`` are internal: written by run_checks/answer_weekend.
_ACCEPTS = {
"rules": {"file"},
"mock": {"score"},
"slot": {"slot", "slot_confirm"},
"phone": {"tap"},
}
# tutor's worker reports phone readiness under the bare id "phone".
_PHONE_REPORT = "phone"
def local_now(now: datetime | None) -> datetime:
"""``now`` or the clock's local time."""
return evidence.clock_now() if now is None else now
def find(existing: list[Project], project_id: str) -> tuple[int, Project]:
"""Index and interview project; KeyError unknown, ValueError focus kind."""
for i, project in enumerate(existing):
if project.id == project_id:
if project.kind != INTERVIEW:
msg = f"{project_id!r} is not an interview project (no milestones)"
raise ValueError(msg)
return i, project
raise KeyError(project_id)
def add_trip(
project_id: str, start: date, end: date, *, now: datetime | None = None
) -> Project:
"""Announce a trip; it may only grow (see :func:`focus_project._trips.add`)."""
stamp = local_now(now)
with _store.locked():
existing = _store.read_all()
i, project = find(existing, project_id)
trips = add(project.trips, Trip(start, end, evidence.to_ms(stamp)))
updated = replace(project, trips=trips)
_interview.validate(updated)
existing[i] = updated
_store.write_all(existing)
return updated
def answer_weekend(
project_id: str, saturday: date, *, trip: bool, now: datetime | None = None
) -> Project:
"""Answer the trip question of a weekend; True never flips back to False."""
stamp = local_now(now)
with _store.locked():
existing = _store.read_all()
i, project = find(existing, project_id)
if saturday not in _interview.saturdays(project):
msg = f"{saturday} is not a weekend of {project_id!r} before its test day"
raise ValueError(msg)
answers = dict(project.weekend_answers)
current = answers.get(saturday)
if current is trip:
return project
if current is True:
msg = "a trip answer may only get stricter: it stays True"
raise ValueError(msg)
if not trip and any(t.covers(saturday) for t in project.trips):
msg = f"an announced trip covers {saturday}: it cannot be a no-trip weekend"
raise ValueError(msg)
updated = replace(
project, weekend_answers=tuple(sorted({**answers, saturday: trip}.items()))
)
# Evidence first: a crash between the two leaves the question open
# (fail closed), never an answer with no time.
evidence.append(project_id, f"trip_ask:{saturday}", "answer", trip, stamp)
existing[i] = updated
_store.write_all(existing)
return updated
def _number(value: object) -> float:
if (
isinstance(value, bool)
or not isinstance(value, int | float)
or not math.isfinite(value)
or value < 0
):
msg = "a score must be a finite, non-negative number"
raise ValueError(msg)
return value
def _count(value: object, label: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
msg = f"{label} must be a non-negative integer"
raise ValueError(msg)
return value
def _clean(
project_id: str, kind: str, value: object, file: Path | None, stamp: datetime
) -> object:
"""The value to store for ``kind``; ValueError if it is not acceptable."""
if kind == "file":
if file is None or not file.is_file() or not file.stat().st_size:
msg = "a non-empty file is required"
raise ValueError(msg)
return evidence.store_file(project_id, file, stamp)
if kind == "score":
return _number(value)
if kind == "slot":
moment = datetime.fromisoformat(str(value)) if isinstance(value, str) else None
if moment is None or moment.tzinfo is not None:
msg = "slot must be a naive local ISO datetime"
raise ValueError(msg)
return moment.isoformat()
if kind == "phone_check":
if not isinstance(value, dict) or not isinstance(value.get("plan"), bool):
msg = "phone_check needs {plan: bool, due: int, seen_ms: int}"
raise ValueError(msg)
return {
"plan": value["plan"],
"due": _count(value.get("due"), "due"),
"seen_ms": _count(value.get("seen_ms"), "seen_ms"),
}
if kind == "slot_confirm" and value not in {True, None}:
msg = "slot_confirm is a tap: value True"
raise ValueError(msg)
return True # tap, slot_confirm
def record_evidence( # noqa: PLR0913
project_id: str,
milestone: str,
kind: str,
value: object = None,
*,
file: Path | None = None,
now: datetime | None = None,
) -> evidence.Evidence:
"""Append evidence for a milestone; ValueError if it does not fit.
Kinds: ``file`` (rules, copied from ``file``), ``tap`` (phone:*),
``score`` (mock), ``slot`` (ISO datetime) and ``slot_confirm`` (slot),
``phone_check`` ({plan, due, seen_ms}, milestone "phone", from tutor).
There is no way to delete a row.
"""
stamp = local_now(now)
_, project = find(_store.read_all(), project_id)
if milestone == _PHONE_REPORT:
accepted = {"phone_check"}
else:
known = {m.id for m in _milestones.milestones(project, stamp)}
accepted = _ACCEPTS.get(milestone.partition(":")[0], set())
if milestone not in known:
msg = f"{milestone!r} is not a milestone of {project_id!r}"
raise ValueError(msg)
if kind not in accepted:
msg = f"{milestone!r} takes evidence kinds {sorted(accepted)}, not {kind!r}"
raise ValueError(msg)
return evidence.append(
project_id, milestone, kind, _clean(project_id, kind, value, file, stamp), stamp
)

15
focus_project/focus_project/_paths.py
View File

@ -30,3 +30,18 @@ def activity_path() -> Path:
def lock_path() -> Path:
"""The registry's writer lock."""
return data_dir() / "projects.lock"
def evidence_dir() -> Path:
"""Interview-prep evidence: one jsonl per project plus copied files."""
return data_dir() / "evidence"
def evidence_path(project_id: str) -> Path:
"""The append-only evidence rows of one project."""
return evidence_dir() / f"{project_id}.jsonl"
def evidence_files_dir(project_id: str) -> Path:
"""Where evidence files (screenshots) of one project are copied."""
return evidence_dir() / project_id

61
focus_project/focus_project/_trips.py Normal file
View File

@ -0,0 +1,61 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Trips: days away from the PC. They can only grow (ratchet)."""
from __future__ import annotations
from dataclasses import dataclass, replace
from datetime import date
from typing import Any
@dataclass(frozen=True)
class Trip:
"""An inclusive away range; ``entered_ms`` is when it was announced."""
start: date
end: date
entered_ms: int = 0
def covers(self, day: date) -> bool:
"""Whether ``day`` is inside the trip."""
return self.start <= day <= self.end
def to_dict(self) -> dict[str, Any]:
"""The projects.json row."""
return {
"start": self.start.isoformat(),
"end": self.end.isoformat(),
"entered_ms": self.entered_ms,
}
@classmethod
def from_dict(cls, row: dict[str, Any]) -> Trip:
"""Parse a trip row (errors surface as RegistryError upstream)."""
return cls(
start=date.fromisoformat(row["start"]),
end=date.fromisoformat(row["end"]),
entered_ms=int(row.get("entered_ms", 0)),
)
def add(trips: tuple[Trip, ...], new: Trip) -> tuple[Trip, ...]:
"""``trips`` plus ``new``; ValueError if that would loosen an old trip.
A new range touching an old trip must cover it: it replaces the old one
(keeping its entry time when the start did not move earlier). Anything
that starts later or ends earlier than what is on file is refused.
"""
kept: list[Trip] = []
for old in trips:
if old.start > new.end or new.start > old.end:
kept.append(old)
elif new.start <= old.start and new.end >= old.end:
if new.start == old.start:
new = replace(new, entered_ms=old.entered_ms)
else:
msg = (
f"a trip may only get stricter: pass a range that covers "
f"{old.start}..{old.end}"
)
raise ValueError(msg)
return tuple(sorted([*kept, new], key=lambda t: t.start))

106
focus_project/focus_project/checks.py Normal file
View File

@ -0,0 +1,106 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Pluggable check sources for the auto-checked milestones, and latching.
A source is ``fn(project, now) -> True | False | None``; ``None`` means "could
not check" (logged) and counts as not done. Only ``phone`` is auto-checked
(the calendar slot check was dropped, 2026-11-10: ``slot`` is two taps).
:func:`run_checks` is the only place sources run: status()/milestones() stay
cheap and read the latched ``check`` rows it appends (only on a change).
"""
from __future__ import annotations
from collections.abc import Callable
from datetime import datetime, timedelta
import logging
from focus_project import _milestones, _mutate, _store, evidence
from focus_project._model import Project
_LOG = logging.getLogger(__name__)
CheckSource = Callable[[Project, datetime], bool | None]
DAY_MS = 24 * 3600 * 1000
# A phone check starts 24 h before the milestone is due.
WINDOW = timedelta(hours=24)
def _report_ok(value: object, report_ms: int) -> bool:
"""A plan exists, cards are due, the phone was seen <= 24 h before the report."""
if not isinstance(value, dict):
return False
try:
return bool(
value["plan"] is True
and int(value["due"]) > 0
and report_ms - int(value["seen_ms"]) <= DAY_MS
)
except KeyError, TypeError, ValueError:
return False
def phone_source(project: Project, now: datetime) -> bool | None:
"""Read tutor's latest ``phone_check`` report (<= 24 h old).
No report, a stale one or one stamped ahead of the clock = ``None``
(the worker is not reporting). A fresh one passes only if the plan
exists, cards are due and the phone was seen within 24 h of the report.
"""
report = evidence.last(evidence.read(project.id), "phone", "phone_check")
if report is None:
_LOG.warning("phone check: no tutor report for %s", project.id)
return None
if not 0 <= evidence.to_ms(now) - report.ms <= DAY_MS:
_LOG.warning("phone check: tutor report for %s is stale or ahead", project.id)
return None
return _report_ok(report.value, report.ms)
SOURCES: dict[str, CheckSource] = {"phone": phone_source}
def register_source(name: str, source: CheckSource) -> None:
"""Replace the callable behind an auto check (``phone``)."""
if name not in SOURCES:
msg = f"unknown check source {name!r}; known: {sorted(SOURCES)}"
raise ValueError(msg)
SOURCES[name] = source
def _run(project: Project, now: datetime) -> bool | None:
try:
return SOURCES["phone"](project, now)
except Exception as exc: # noqa: BLE001 (pluggable code: any failure = unknown)
_LOG.warning("phone check source failed: %s", exc)
return None
def run_checks(
project_id: str, *, now: datetime | None = None
) -> dict[str, bool | None]:
"""Run the auto checks of open phone milestones; latch what changed.
Returns milestone id -> verdict for the ones checked (inside their
24 h window, not yet done). KeyError for an unknown project; a focus
project has nothing to check.
"""
stamp = _mutate.local_now(now)
project = next((p for p in _store.read_all() if p.id == project_id), None)
if project is None:
raise KeyError(project_id)
open_ids = [
m.id
for m in _milestones.milestones(project, stamp)
if m.id.startswith("phone:")
and m.state not in {"done", "late_done"}
and stamp >= m.due - WINDOW
]
if not open_ids:
return {}
verdict = _run(project, stamp)
rows = evidence.read(project_id)
for mid in open_ids:
_, latest, seen = evidence.check_summary(rows, mid)
if verdict != latest and not (verdict is None and not seen):
evidence.append(project_id, mid, "check", verdict, stamp)
return dict.fromkeys(open_ids, verdict)

159
focus_project/focus_project/evidence.py Normal file
View File

@ -0,0 +1,159 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The interview evidence store: append-only rows plus copied files.
One jsonl per project (``{ms, milestone, kind, value}``) under the
:mod:`focus_project._paths` seam. There is deliberately no delete or rewrite
function: a milestone once evidenced stays evidenced.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
import itertools
import json
import logging
import os
import re
import shutil
from typing import TYPE_CHECKING
from focus_project import _clock, _paths
if TYPE_CHECKING:
from pathlib import Path
_LOG = logging.getLogger(__name__)
@dataclass(frozen=True)
class Evidence:
"""One evidence row."""
ms: int
milestone: str
kind: str
value: object
@property
def at(self) -> datetime:
"""When it was recorded, as a naive local datetime."""
return from_ms(self.ms)
def to_ms(moment: datetime) -> int:
"""Unix milliseconds of a naive local datetime."""
return int(moment.timestamp() * 1000)
def from_ms(ms: int) -> datetime:
"""Naive local datetime of unix milliseconds."""
return datetime.fromtimestamp(ms / 1000, tz=UTC).astimezone().replace(tzinfo=None)
def clock_now() -> datetime:
"""The clock seam (:mod:`focus_project._clock`) as a naive local datetime."""
return from_ms(int(_clock.now() * 1000))
def append(
project_id: str, milestone: str, kind: str, value: object, now: datetime
) -> Evidence:
"""Append one row (fsynced) and return it."""
row = Evidence(to_ms(now), milestone, kind, value)
path = _paths.evidence_path(project_id)
path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(
{"ms": row.ms, "milestone": milestone, "kind": kind, "value": value}
)
with path.open("a", encoding="utf-8") as handle:
handle.write(line + "\n")
handle.flush()
os.fsync(handle.fileno())
return row
def store_file(project_id: str, source: Path, now: datetime) -> str:
"""Copy ``source`` beside the rows under a name that never collides."""
target = _paths.evidence_files_dir(project_id)
target.mkdir(parents=True, exist_ok=True)
safe = re.sub(r"[^A-Za-z0-9._-]", "_", source.name)[:80]
stamp = to_ms(now)
for attempt in itertools.count():
name = f"{stamp}-{safe}" if not attempt else f"{stamp}-{attempt}-{safe}"
try:
with (target / name).open("xb") as out, source.open("rb") as src:
shutil.copyfileobj(src, out)
out.flush()
os.fsync(out.fileno())
except FileExistsError:
continue
return name
raise AssertionError # pragma: no cover (itertools.count never ends)
def _parse(line: str) -> Evidence | None:
try:
row = json.loads(line)
if not isinstance(row, dict) or isinstance(row["ms"], bool):
return None
return Evidence(
int(row["ms"]), str(row["milestone"]), str(row["kind"]), row["value"]
)
except ValueError, KeyError, TypeError:
return None
def read(project_id: str) -> list[Evidence]:
"""Every row in file order; a missing file is empty, bad lines are skipped.
A torn last line (crash mid-append) must not wipe out the whole history,
so malformed lines are logged and ignored; an unreadable file raises.
"""
try:
path = _paths.evidence_path(project_id)
text = path.read_text(encoding="utf-8", errors="replace")
except FileNotFoundError:
return []
rows: list[Evidence] = []
for number, line in enumerate(text.splitlines(), start=1):
parsed = _parse(line)
if parsed is None:
_LOG.warning(
"evidence %s line %d: skipping malformed row", project_id, number
)
else:
rows.append(parsed)
return rows
def first(
rows: list[Evidence], milestone: str, kind: str, since_ms: int = 0
) -> Evidence | None:
"""The earliest matching row at or after ``since_ms``."""
return next(
(
r
for r in rows
if (r.milestone, r.kind) == (milestone, kind) and r.ms >= since_ms
),
None,
)
def last(rows: list[Evidence], milestone: str, kind: str) -> Evidence | None:
"""The latest matching row."""
return next(
(r for r in reversed(rows) if (r.milestone, r.kind) == (milestone, kind)),
None,
)
def check_summary(
rows: list[Evidence], milestone: str
) -> tuple[int | None, bool | None, bool]:
"""``(first passing ms, latest verdict, any check row)`` for a milestone."""
checks = [r for r in rows if (r.milestone, r.kind) == (milestone, "check")]
passed = next((r.ms for r in checks if r.value is True), None)
latest = checks[-1].value if checks else None
return passed, (latest if isinstance(latest, bool) else None), bool(checks)

34
focus_project/focus_project/status.py
View File

@ -3,10 +3,32 @@
from __future__ import annotations
from datetime import date # noqa: TC003
from typing import Any
import logging
from typing import TYPE_CHECKING, Any
from focus_project import _api, carry
from focus_project import _api, _milestones, carry, evidence
if TYPE_CHECKING:
from datetime import date
from focus_project._model import Project
_LOG = logging.getLogger(__name__)
def _milestone_block(project: Project) -> tuple[list[dict[str, Any]], bool]:
"""Milestone rows and whether any is red; unreadable evidence shows none.
Display path, so it fails open (logged): the lock path, ``owed_lock``,
has its own handling.
"""
now = evidence.clock_now()
try:
items = _milestones.milestones(project, now)
except OSError as exc:
_LOG.warning("interview evidence unreadable, no milestones shown: %s", exc)
return [], False
return [m.to_dict() for m in items], any(m.red for m in items)
def status(day: date) -> dict[str, Any]:
@ -30,9 +52,12 @@ def status(day: date) -> dict[str, Any]:
"unlock_minutes": None,
"ends": None,
"replaces": [],
"milestones": [],
"missed": False,
}
daily = project.daily_on(day)
bank = carry.balance(day)
rows, missed = _milestone_block(project)
return {
"project": project.id,
"name": project.name,
@ -47,4 +72,7 @@ def status(day: date) -> dict[str, Any]:
"unlock_minutes": project.unlock_on(day),
"ends": project.last_day.isoformat(),
"replaces": list(project.replaces_on(day)),
# Interview projects: [{id, due ISO, state}] and whether any is red.
"milestones": rows,
"missed": missed,
}

15
focus_project/focus_project/tests/conftest.py
View File

@ -11,7 +11,7 @@ from typing import TYPE_CHECKING
import pytest
from focus_project import _clock
from focus_project import _clock, _paths, checks
if TYPE_CHECKING:
from pathlib import Path
@ -40,3 +40,16 @@ def _far_future_clock(monkeypatch: pytest.MonkeyPatch) -> None:
def data(tmp_path: Path) -> Path:
"""The redirected data directory (may not exist yet)."""
return tmp_path / "data"
@pytest.fixture(autouse=True)
def _interview_isolation(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, _redirect_state: None
) -> None:
"""Evidence lives under the redirected data dir; sources are a private copy.
Fails closed (R3): if the path seam ever resolves outside ``tmp_path`` the
test errors before it can write a production evidence row.
"""
assert _paths.evidence_dir().is_relative_to(tmp_path)
monkeypatch.setattr(checks, "SOURCES", dict(checks.SOURCES))

52
focus_project/focus_project/tests/helpers.py Normal file
View File

@ -0,0 +1,52 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Shared fixtures-as-functions for the interview-prep tests."""
from __future__ import annotations
from datetime import date, datetime, time
from typing import TYPE_CHECKING
import focus_project as fp
if TYPE_CHECKING:
from pathlib import Path
START = date(2026, 11, 10) # Tuesday: D1
TEST_DAY = date(2026, 11, 27) # Friday: T
TEST_AT = datetime.combine(TEST_DAY, time(10, 0))
SAT1 = date(2026, 11, 14)
SAT2 = date(2026, 11, 21)
def at(day: int, hour: int = 0, minute: int = 0) -> datetime:
"""A November 2026 local datetime."""
return datetime.combine(date(2026, 11, day), time(hour, minute))
def make(pid: str = "iv", /, **kw: object) -> fp.Project:
"""Create an interview project D1 = Nov 10, T = Nov 27."""
args: dict[str, object] = {
"name": "Interview",
"start": START,
"end": TEST_DAY,
"replaces": ["leetcode", "reading", "tutor"],
"daily_minutes": 120,
"unlock_minutes": 30,
"kind": "interview",
"test_at": TEST_AT,
"invite_url": "https://app.codility.com/c/run/abc",
}
args.update(kw)
return fp.create_project(pid, **args) # type: ignore[arg-type]
def shot(tmp_path: Path, name: str = "rules.png") -> Path:
"""A non-empty screenshot stand-in."""
path = tmp_path / name
path.write_bytes(b"\x89PNG rules")
return path
def states(project: fp.Project, now: datetime) -> dict[str, str]:
"""Milestone id -> state."""
return {m.id: m.state for m in fp.milestones(project, now)}

4
focus_project/focus_project/tests/test_activity_status.py
View File

@ -137,6 +137,8 @@ def test_status_active_reports_the_worked_minutes() -> None:
"unlock_minutes": 30,
"ends": "2026-11-05",
"replaces": ["leetcode"],
"milestones": [],
"missed": False,
}
@ -158,6 +160,8 @@ def test_status_inactive() -> None:
"unlock_minutes",
"ends",
"replaces",
"milestones",
"missed",
}

193
focus_project/focus_project/tests/test_checks.py Normal file
View File

@ -0,0 +1,193 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Pluggable check sources, the phone source, and latching."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import pytest
import focus_project as fp
from focus_project import _paths, checks, evidence
from focus_project.tests.helpers import SAT1, at, make
if TYPE_CHECKING:
from collections.abc import Callable
from datetime import datetime
MID = "phone:2026-11-14"
def _ms(moment: datetime) -> int:
return int(moment.timestamp() * 1000)
def _report(
moment: datetime,
*,
plan: bool = True,
due: int = 5,
seen: datetime | None = None,
) -> None:
value = {"plan": plan, "due": due, "seen_ms": _ms(seen or moment)}
fp.record_evidence("iv", "phone", "phone_check", value, now=moment)
def _phone(now: datetime) -> bool | None:
return checks.phone_source(fp.list_projects()[0], now)
def test_phone_source_without_a_report_is_unknown(
caplog: pytest.LogCaptureFixture,
) -> None:
make()
with caplog.at_level("WARNING"):
assert _phone(at(13, 9)) is None
assert "phone" in caplog.text
def test_phone_source_passes_on_a_fresh_good_report() -> None:
make()
_report(at(13, 8), seen=at(12, 20))
assert _phone(at(13, 9)) is True
@pytest.mark.parametrize(
"bad",
[{"plan": False}, {"due": 0}, {"seen": at(11, 7)}],
)
def test_phone_source_fails_on_any_bad_part(bad: dict[str, object]) -> None:
make()
_report(at(13, 8), **bad) # type: ignore[arg-type]
assert _phone(at(13, 9)) is False
def test_phone_source_stale_or_future_report_is_unknown() -> None:
make()
_report(at(13, 8))
assert _phone(at(14, 8, 1)) is None # the worker stopped reporting
assert _phone(at(13, 7)) is None # stamped ahead of the clock
def test_phone_source_uses_the_latest_report() -> None:
make()
_report(at(13, 8), plan=False)
_report(at(13, 9))
assert _phone(at(13, 10)) is True
def test_phone_source_hand_edited_row_is_a_failure() -> None:
make()
_report(at(13, 8))
path = _paths.evidence_path("iv")
row = {"ms": _ms(at(13, 9)), "milestone": "phone", "kind": "phone_check"}
path.write_text(path.read_text() + json.dumps({**row, "value": "ok"}) + "\n")
assert _phone(at(13, 10)) is False
def _patch(
monkeypatch: pytest.MonkeyPatch, fn: Callable[[fp.Project, datetime], bool | None]
) -> None:
monkeypatch.setitem(checks.SOURCES, "phone", fn)
def _trip_project() -> None:
make()
fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 9))
def test_run_checks_latches_a_pass(monkeypatch: pytest.MonkeyPatch) -> None:
_trip_project()
_patch(monkeypatch, lambda p, n: True)
assert fp.run_checks("iv", now=at(13, 9)) == {MID: True}
assert fp.run_checks("iv", now=at(13, 10)) == {MID: True}
rows = [r for r in evidence.read("iv") if r.kind == "check"]
assert [(r.milestone, r.value) for r in rows] == [(MID, True)]
def test_run_checks_only_inside_the_window(monkeypatch: pytest.MonkeyPatch) -> None:
_trip_project()
_patch(monkeypatch, lambda p, n: True)
assert fp.run_checks("iv", now=at(12, 11, 59)) == {} # >24 h before due
assert fp.run_checks("iv", now=at(12, 12)) == {MID: True}
def test_run_checks_records_changes_not_every_tick(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_trip_project()
answers: list[bool | None] = [None, False, False, None, None]
_patch(monkeypatch, lambda p, n: answers.pop(0))
for minute in range(5):
fp.run_checks("iv", now=at(13, 9, minute))
rows = [r.value for r in evidence.read("iv") if r.kind == "check"]
assert rows == [False, None] # leading None writes nothing
def test_run_checks_source_crash_is_unknown_and_loud(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
_trip_project()
def boom(p: fp.Project, n: datetime) -> bool:
msg = "tutor down"
raise OSError(msg)
_patch(monkeypatch, boom)
with caplog.at_level("WARNING"):
assert fp.run_checks("iv", now=at(13, 9)) == {MID: None}
assert "tutor down" in caplog.text
def test_run_checks_skips_finished_milestones(monkeypatch: pytest.MonkeyPatch) -> None:
_trip_project()
fp.record_evidence("iv", MID, "tap", value=True, now=at(12, 9))
evidence.append("iv", MID, "check", True, at(12, 10))
_patch(monkeypatch, lambda p, n: pytest.fail("already done"))
assert fp.run_checks("iv", now=at(13, 9)) == {}
def test_run_checks_ignores_focus_projects() -> None:
fp.create_project(
"f",
name="F",
start=at(10).date(),
end=at(12).date(),
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
assert fp.run_checks("f", now=at(11, 9)) == {}
with pytest.raises(KeyError):
fp.run_checks("ghost", now=at(11, 9))
def test_register_source_swaps_the_callable(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(checks, "SOURCES", dict(checks.SOURCES))
def fn(p: fp.Project, n: datetime) -> bool:
return True
fp.register_source("phone", fn)
assert checks.SOURCES["phone"] is fn
with pytest.raises(ValueError, match="source"):
fp.register_source("calendar", fn)
def test_end_to_end_phone_milestone_goes_done() -> None:
_trip_project()
fp.record_evidence("iv", MID, "tap", value=True, now=at(12, 9))
_report(at(13, 9))
assert fp.run_checks("iv", now=at(13, 9, 30)) == {MID: True}
state = {m.id: m.state for m in fp.milestones(fp.list_projects()[0], at(13, 10))}
assert state[MID] == "done"
def test_report_missing_a_field_fails() -> None:
make()
path = _paths.evidence_path("iv")
path.parent.mkdir(parents=True, exist_ok=True)
row = {"ms": _ms(at(13, 8)), "milestone": "phone", "kind": "phone_check"}
path.write_text(json.dumps({**row, "value": {"plan": True}}) + "\n")
assert _phone(at(13, 9)) is False

126
focus_project/focus_project/tests/test_evidence.py Normal file
View File

@ -0,0 +1,126 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The append-only evidence store and record_evidence validation."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import pytest
import focus_project as fp
from focus_project import _paths, evidence
from focus_project.tests.helpers import at, make, shot
if TYPE_CHECKING:
from pathlib import Path
def test_paths_are_isolated(tmp_path: Path) -> None:
assert _paths.evidence_dir().is_relative_to(tmp_path)
assert _paths.evidence_path("iv").name == "iv.jsonl"
assert _paths.evidence_files_dir("iv").name == "iv"
def test_file_is_copied_and_row_appended(tmp_path: Path) -> None:
make()
src = shot(tmp_path)
row = fp.record_evidence("iv", "rules", "file", file=src, now=at(10, 9))
assert row.kind == "file"
stored = _paths.evidence_files_dir("iv") / str(row.value)
assert stored.read_bytes() == src.read_bytes()
line = json.loads(_paths.evidence_path("iv").read_text().splitlines()[0])
assert set(line) == {"ms", "milestone", "kind", "value"}
assert line["milestone"] == "rules"
def test_rows_append_never_rewrite() -> None:
make()
fp.record_evidence("iv", "mock", "score", 71, now=at(24, 9))
fp.record_evidence("iv", "mock", "score", 80.5, now=at(24, 10))
assert [r.value for r in evidence.read("iv")] == [71, 80.5]
assert not hasattr(fp, "delete_evidence")
assert not hasattr(evidence, "delete")
def test_same_name_files_do_not_overwrite(tmp_path: Path) -> None:
make()
one = fp.record_evidence("iv", "rules", "file", file=shot(tmp_path), now=at(10, 9))
two = fp.record_evidence("iv", "rules", "file", file=shot(tmp_path), now=at(10, 9))
assert one.value != two.value
@pytest.mark.parametrize(
("milestone", "kind", "value"),
[
("rules", "tap", True), # wrong kind for the milestone
("rules", "check", True), # internal kind
("trip_ask:2026-11-14", "answer", True), # use answer_weekend
("nope", "tap", True), # not a milestone of the project
("mock", "score", "high"),
("mock", "score", True),
("mock", "score", float("nan")),
("mock", "score", -1),
("slot", "slot", "tomorrow"),
("slot", "slot", "2026-11-26T10:00:00+02:00"), # aware
("slot", "slot_confirm", False),
("phone", "phone_check", {"plan": True}),
("phone", "phone_check", {"plan": "yes", "due": 1, "seen_ms": 1}),
("phone", "phone_check", "ok"),
],
)
def test_bad_evidence_refused(milestone: str, kind: str, value: object) -> None:
make()
with pytest.raises(ValueError): # noqa: PT011
fp.record_evidence("iv", milestone, kind, value, now=at(12, 9))
def test_file_kind_needs_a_real_nonempty_file(tmp_path: Path) -> None:
make()
empty = tmp_path / "e.png"
empty.write_bytes(b"")
for bad in (None, tmp_path / "missing.png", empty, tmp_path):
with pytest.raises(ValueError, match="file"):
fp.record_evidence("iv", "rules", "file", file=bad, now=at(10, 9))
def test_unknown_project_is_keyerror() -> None:
with pytest.raises(KeyError):
fp.record_evidence("ghost", "mock", "score", 1, now=at(24, 9))
def test_focus_project_has_no_milestones() -> None:
fp.create_project(
"f",
name="F",
start=at(10).date(),
end=at(12).date(),
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
with pytest.raises(ValueError, match="milestone"):
fp.record_evidence("f", "mock", "score", 1, now=at(10, 9))
def test_torn_and_foreign_lines_are_skipped(caplog: pytest.LogCaptureFixture) -> None:
make()
fp.record_evidence("iv", "mock", "score", 70, now=at(24, 9))
path = _paths.evidence_path("iv")
path.write_text(path.read_text() + '{"ms": 1, "milest\n[1]\n{"ms":"x"}\n')
with caplog.at_level("WARNING"):
rows = evidence.read("iv")
assert [r.value for r in rows] == [70]
assert "skipping" in caplog.text
def test_missing_file_reads_empty() -> None:
assert evidence.read("nobody") == []
def test_garbage_bytes_are_skipped_not_fatal() -> None:
make()
fp.record_evidence("iv", "mock", "score", 70, now=at(24, 9))
path = _paths.evidence_path("iv")
path.write_bytes(path.read_bytes() + b"\xff\xfe\x80 not utf-8\n")
assert [r.value for r in evidence.read("iv")] == [70]

175
focus_project/focus_project/tests/test_interview_model.py Normal file
View File

@ -0,0 +1,175 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Project.kind, interview fields, validation, budget."""
from __future__ import annotations
from datetime import date
import json
from typing import TYPE_CHECKING
import pytest
import focus_project as fp
from focus_project import _interview, _milestones
from focus_project._errors import RegistryError
from focus_project.tests.helpers import SAT1, START, TEST_AT, TEST_DAY, at, make
if TYPE_CHECKING:
from pathlib import Path
LIVE_ROW = {
"id": "hackerrank-prep",
"name": "HackerRank prep",
"start": "2026-11-02",
"end": "2026-11-05",
"replaces": ["leetcode", "reading", "tutor"],
"daily_minutes": 120,
"unlock_minutes": 30,
"subjects": ["a"],
"site_patterns": ["leetcode.com"],
"ended_on": None,
"created_ms": 1793603609345,
"work_url": "https://www.hackerrank.com/dashboard",
"history": [],
}
def test_live_projects_json_shape_loads(data: Path) -> None:
data.mkdir()
(data / "projects.json").write_text(json.dumps({"projects": [LIVE_ROW]}))
[p] = fp.list_projects()
assert p.kind == "focus"
assert p.test_at is None
assert p.trips == ()
assert p.weekend_answers == ()
assert p.daily_on(date(2026, 11, 3)) == 120
snap = fp.status(date(2026, 11, 3))
assert snap["milestones"] == []
assert snap["missed"] is False
def test_focus_row_stays_free_of_interview_keys(data: Path) -> None:
fp.create_project(
"f",
name="F",
start=START,
end=TEST_DAY,
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
row = json.loads((data / "projects.json").read_text())["projects"][0]
assert row["kind"] == "focus"
assert not {"test_at", "invite_url", "trips", "weekend_answers"} & set(row)
def test_interview_roundtrip_and_file_shape(data: Path) -> None:
p = make(
trips=[fp.Trip(date(2026, 11, 20), date(2026, 11, 22))],
weekend_answers={SAT1: False},
)
[back] = fp.list_projects()
assert back == p
assert back.kind == "interview"
assert back.test_at == TEST_AT
assert back.trips[0].entered_ms > 0
row = json.loads((data / "projects.json").read_text())["projects"][0]
assert row["test_at"] == "2026-11-27T10:00:00"
assert row["trips"][0]["start"] == "2026-11-20"
assert row["weekend_answers"] == {"2026-11-14": False}
assert row["invite_url"].startswith("https://")
@pytest.mark.parametrize(
"bad",
[
{"test_at": None},
{"test_at": at(26, 10)}, # end != test day
{"test_at": at(9, 10), "end": date(2026, 11, 9)},
{"invite_url": "ftp://x.example/t"},
{"invite_url": "javascript:alert(1)"},
{"invite_url": "https://a b.example"},
{"kind": "bogus"},
{"trips": [fp.Trip(date(2026, 11, 28), date(2026, 11, 29))]},
{"trips": [fp.Trip(date(2026, 11, 20), date(2026, 11, 19))]},
{"weekend_answers": {date(2026, 11, 15): True}}, # a Sunday
{"weekend_answers": {date(2026, 11, 28): True}}, # after T
{"daily_minutes": 29},
],
)
def test_interview_validation(bad: dict[str, object]) -> None:
with pytest.raises(ValueError): # noqa: PT011
make(**bad)
def test_focus_kind_rejects_interview_fields() -> None:
with pytest.raises(ValueError, match="interview"):
make(kind="focus")
def test_empty_invite_url_is_allowed() -> None:
assert make(invite_url="").invite_url == ""
def test_budget_30_weekdays_120_weekend() -> None:
p = make(unlock_minutes=60)
assert p.daily_on(date(2026, 11, 10)) == 30 # Tue
assert p.daily_on(date(2026, 11, 13)) == 30 # Fri
assert p.daily_on(SAT1) == 120
assert p.daily_on(date(2026, 11, 15)) == 120 # Sun
assert p.unlock_on(date(2026, 11, 10)) == 30 # clamped to the day's target
assert p.unlock_on(SAT1) == 60
def test_budget_flows_through_status() -> None:
make()
assert fp.status(date(2026, 11, 10))["daily_minutes"] == 30
assert fp.status(SAT1)["daily_minutes"] == 120
def test_interview_exempt_from_floor_but_focus_is_not() -> None:
make(daily_minutes=30, unlock_minutes=30) # 3 gates, floor 90
with pytest.raises(ValueError, match="at least"):
fp.create_project(
"f",
name="F",
start=date(2026, 12, 1),
end=date(2026, 12, 2),
replaces=["leetcode", "reading"],
daily_minutes=30,
unlock_minutes=10,
)
def test_non_interview_has_no_weekends() -> None:
focus = fp.create_project(
"f",
name="F",
start=START,
end=TEST_DAY,
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
assert _interview.saturdays(focus) == []
def test_trip_without_entry_time_uses_the_nominal_due() -> None:
assert _milestones._trip_due(date(2026, 11, 13), 0) == at(12, 12)
def test_non_boolean_weekend_answer_is_a_bad_row(data: Path) -> None:
make()
path = data / "projects.json"
rows = json.loads(path.read_text())
rows["projects"][0]["weekend_answers"] = {"2026-11-14": "false"}
path.write_text(json.dumps(rows))
with pytest.raises(RegistryError):
fp.list_projects()
def test_interview_unlock_edit_keeps_the_ratchet() -> None:
make(unlock_minutes=60)
fp.update_project("iv", today=SAT1, unlock_minutes=10)
assert fp.list_projects()[0].unlock_on(SAT1) == 60 # lower holds from tomorrow
assert fp.list_projects()[0].unlock_on(date(2026, 11, 15)) == 10

234
focus_project/focus_project/tests/test_milestones.py Normal file
View File

@ -0,0 +1,234 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Milestone generation, states and the lock that falls out of them."""
from __future__ import annotations
from datetime import date
from typing import TYPE_CHECKING
import focus_project as fp
from focus_project import evidence
from focus_project.tests.helpers import SAT1, SAT2, at, make, shot, states
if TYPE_CHECKING:
from pathlib import Path
import pytest
def _due(p: fp.Project, now_day: int = 10) -> dict[str, str]:
return {m.id: m.due.isoformat() for m in fp.milestones(p, at(now_day, 8))}
def test_table_of_dues() -> None:
assert _due(make()) == {
"rules": "2026-11-10T23:59:00",
"trip_ask:2026-11-14": "2026-11-12T23:59:00",
"trip_ask:2026-11-21": "2026-11-19T23:59:00",
"mock": "2026-11-25T23:59:00",
"slot": "2026-11-26T12:00:00",
}
def test_sorted_by_due_and_all_pending_at_start() -> None:
project = make()
items = fp.milestones(project, at(10, 8))
assert [m.due for m in items] == sorted(m.due for m in items)
assert {m.state for m in items} == {"pending"}
assert items[0].to_dict() == {
"id": "rules",
"due": "2026-11-10T23:59:00",
"state": "pending",
}
def test_focus_project_has_none() -> None:
p = fp.create_project(
"f",
name="F",
start=date(2026, 11, 10),
end=date(2026, 11, 12),
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
assert fp.milestones(p, at(11, 8)) == []
def test_unanswered_weekend_is_a_trip_once_ask_is_due() -> None:
p = make()
assert "phone:2026-11-14" not in states(p, at(12, 23, 58))
got = states(p, at(13, 8))
assert got["trip_ask:2026-11-14"] == "missed"
assert got["phone:2026-11-14"] == "pending" # due Fri 12:00
assert states(p, at(13, 12))["phone:2026-11-14"] == "missed"
assert _due(p, 13)["phone:2026-11-14"] == "2026-11-13T12:00:00"
def test_answers_decide_the_phone_milestone() -> None:
make()
no = fp.answer_weekend("iv", SAT1, trip=False, now=at(11, 9))
assert "phone:2026-11-14" not in states(no, at(13, 13))
assert states(no, at(13, 13))["trip_ask:2026-11-14"] == "done"
yes = fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 10))
assert states(yes, at(11, 11))["phone:2026-11-14"] == "pending"
def test_explicit_trip_covers_its_saturday() -> None:
make()
p = fp.add_trip("iv", date(2026, 11, 13), date(2026, 11, 15), now=at(10, 9))
got = states(p, at(13, 8))
assert "phone:2026-11-13" in got
assert "phone:2026-11-14" not in got
assert _due(p)["phone:2026-11-13"] == "2026-11-12T12:00:00"
def test_late_trip_is_due_the_moment_it_is_entered() -> None:
make()
p = fp.add_trip("iv", date(2026, 11, 13), date(2026, 11, 14), now=at(12, 8))
assert _due(p, 12)["phone:2026-11-13"] == "2026-11-12T08:00:00"
assert states(p, at(12, 8))["phone:2026-11-13"] == "missed"
far = fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 22), now=at(12, 9))
assert _due(far, 12)["phone:2026-11-20"] == "2026-11-19T12:00:00"
def _clean(tmp_path: Path) -> None:
"""Everything due before the slot is done on time; weekend 2 is no trip."""
fp.record_evidence("iv", "rules", "file", file=shot(tmp_path), now=at(10, 9))
fp.answer_weekend("iv", SAT1, trip=False, now=at(11, 9))
fp.answer_weekend("iv", SAT2, trip=False, now=at(11, 9))
fp.record_evidence("iv", "mock", "score", 70, now=at(24, 9))
def test_slot_is_due_t_minus_1_at_noon_and_locks_that_evening(tmp_path: Path) -> None:
make()
_clean(tmp_path)
assert fp.owed_lock(at(26, 11, 59)) is None
owed = fp.owed_lock(at(26, 12))
assert owed is not None
assert owed.id == "slot"
assert owed.state == "missed"
assert fp.owed_lock(at(26, 20)) is not None
def test_test_day_never_locks_but_shows_red(tmp_path: Path) -> None:
make()
assert fp.owed_lock(at(27, 9)) is None
assert fp.status(date(2026, 11, 27))["milestones"] is not None
assert {m.state for m in fp.milestones(fp.list_projects()[0], at(27, 9))} >= {
"missed"
}
def test_owed_lock_is_the_earliest_missed() -> None:
make()
owed = fp.owed_lock(at(13, 8))
assert owed is not None
assert owed.id == "rules"
def test_late_completion_unlocks(tmp_path: Path) -> None:
make()
fp.answer_weekend("iv", SAT1, trip=False, now=at(11, 9))
owed = fp.owed_lock(at(11, 10))
assert owed is not None
assert owed.id == "rules"
fp.record_evidence("iv", "rules", "file", file=shot(tmp_path), now=at(11, 10))
p = fp.list_projects()[0]
assert states(p, at(11, 10))["rules"] == "late_done"
assert fp.owed_lock(at(11, 10, 1)) is None
def test_on_time_is_done_at_the_boundary(tmp_path: Path) -> None:
make()
fp.record_evidence("iv", "rules", "file", file=shot(tmp_path), now=at(10, 23, 58))
p = fp.list_projects()[0]
assert states(p, at(11, 8))["rules"] == "done"
assert states(p, at(10, 23, 59))["rules"] == "done"
def test_mock_needs_a_score(tmp_path: Path) -> None:
p = make()
assert states(p, at(26, 8))["mock"] == "missed"
fp.record_evidence("iv", "mock", "score", 64, now=at(26, 8))
assert states(fp.list_projects()[0], at(26, 9))["mock"] == "late_done"
def test_slot_needs_entry_and_confirm_tap() -> None:
p = make()
fp.record_evidence("iv", "slot", "slot", "2026-11-26T20:00:00", now=at(25, 9))
assert states(p, at(25, 10))["slot"] == "pending"
assert states(p, at(26, 12))["slot"] == "missed" # entered, never confirmed
fp.record_evidence("iv", "slot", "slot_confirm", value=True, now=at(25, 11))
assert states(p, at(25, 12))["slot"] == "done"
fp.record_evidence("iv", "slot", "slot", "2026-11-26T21:00:00", now=at(25, 13))
assert states(p, at(25, 14))["slot"] == "pending" # moved: confirm again
fp.record_evidence("iv", "slot", "slot_confirm", value=True, now=at(26, 13))
assert states(p, at(26, 14))["slot"] == "late_done"
assert not [s for s in states(p, at(26, 14)).values() if s == "unknown"]
def _phone_setup() -> None:
make()
fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 9))
def test_phone_milestone_states() -> None:
_phone_setup()
p = fp.list_projects()[0]
mid = "phone:2026-11-14"
assert states(p, at(13, 13))[mid] == "missed" # no tap, no check
fp.record_evidence("iv", mid, "tap", value=True, now=at(12, 9))
assert states(p, at(13, 13))[mid] == "unknown" # tap but nothing checked
evidence.append("iv", mid, "check", value=False, now=at(13, 11))
assert states(p, at(13, 13))[mid] == "missed" # checked and failing
evidence.append("iv", mid, "check", value=None, now=at(13, 11, 30))
assert states(p, at(13, 13))[mid] == "unknown" # source went dark
evidence.append("iv", mid, "check", value=True, now=at(13, 11, 40))
assert states(p, at(13, 13))[mid] == "done"
def test_phone_passed_after_due_is_late_done() -> None:
_phone_setup()
mid = "phone:2026-11-14"
fp.record_evidence("iv", mid, "tap", value=True, now=at(12, 9))
evidence.append("iv", mid, "check", value=True, now=at(13, 14))
assert states(fp.list_projects()[0], at(13, 15))[mid] == "late_done"
def test_unknown_locks_like_missed(tmp_path: Path) -> None:
_phone_setup()
mid = "phone:2026-11-14"
fp.record_evidence("iv", "rules", "file", file=shot(tmp_path), now=at(10, 9))
fp.answer_weekend("iv", SAT2, trip=False, now=at(11, 9))
fp.record_evidence("iv", mid, "tap", value=True, now=at(12, 9))
owed = fp.owed_lock(at(13, 13))
assert owed is not None
assert (owed.id, owed.state) == (mid, "unknown")
def test_unreadable_registry_fails_open_and_loud(
data: Path, caplog: pytest.LogCaptureFixture
) -> None:
data.mkdir()
(data / "projects.json").write_text("{nope")
with caplog.at_level("WARNING"):
assert fp.owed_lock(at(13, 8)) is None
assert "unreadable" in caplog.text
def test_no_project_no_lock() -> None:
assert fp.owed_lock(at(13, 8)) is None
def test_focus_project_never_locks_on_milestones() -> None:
fp.create_project(
"f",
name="F",
start=date(2026, 11, 10),
end=date(2026, 11, 20),
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
assert fp.owed_lock(at(13, 8)) is None

172
focus_project/focus_project/tests/test_ratchet.py Normal file
View File

@ -0,0 +1,172 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""add_trip / answer_weekend only ever get stricter."""
from __future__ import annotations
from datetime import date
import pytest
import focus_project as fp
from focus_project import evidence
from focus_project.tests.helpers import SAT1, at, make, states
def _ms(day: int, hour: int) -> int:
return int(at(day, hour).timestamp() * 1000)
def test_add_trip_records_when_it_was_entered() -> None:
make()
p = fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 22), now=at(10, 9))
assert p.trips == (fp.Trip(date(2026, 11, 20), date(2026, 11, 22), _ms(10, 9)),)
assert fp.list_projects() == [p]
def test_extending_keeps_entry_time_moving_earlier_resets_it() -> None:
make()
fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 21), now=at(10, 9))
later = fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 23), now=at(11, 9))
assert later.trips == (fp.Trip(date(2026, 11, 20), date(2026, 11, 23), _ms(10, 9)),)
earlier = fp.add_trip("iv", date(2026, 11, 19), date(2026, 11, 23), now=at(12, 9))
assert earlier.trips == (
fp.Trip(date(2026, 11, 19), date(2026, 11, 23), _ms(12, 9)),
)
def test_second_disjoint_trip_and_idempotent_repeat() -> None:
make()
fp.add_trip("iv", date(2026, 11, 13), date(2026, 11, 14), now=at(10, 9))
fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 21), now=at(10, 10))
again = fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 21), now=at(11, 9))
assert len(again.trips) == 2
assert again.trips[1].entered_ms == _ms(10, 10)
@pytest.mark.parametrize(
("start", "end"),
[
(date(2026, 11, 21), date(2026, 11, 22)), # starts later: loosens
(date(2026, 11, 20), date(2026, 11, 21)), # ends earlier: shortens
(date(2026, 11, 21), date(2026, 11, 21)), # strictly inside
(date(2026, 11, 22), date(2026, 11, 25)), # partial overlap, later start
],
)
def test_trip_may_not_shrink_or_start_later(start: date, end: date) -> None:
make()
before = fp.add_trip("iv", date(2026, 11, 20), date(2026, 11, 22), now=at(10, 9))
with pytest.raises(ValueError, match=r"stricter|cover"):
fp.add_trip("iv", start, end, now=at(11, 9))
assert fp.list_projects() == [before]
@pytest.mark.parametrize(
("start", "end"),
[
(date(2026, 11, 20), date(2026, 11, 19)),
(date(2026, 11, 28), date(2026, 11, 29)),
],
)
def test_bad_trip_ranges(start: date, end: date) -> None:
make()
with pytest.raises(ValueError): # noqa: PT011
fp.add_trip("iv", start, end, now=at(10, 9))
def test_trip_on_wrong_project() -> None:
with pytest.raises(KeyError):
fp.add_trip("ghost", date(2026, 11, 20), date(2026, 11, 21), now=at(10, 9))
fp.create_project(
"f",
name="F",
start=date(2026, 12, 1),
end=date(2026, 12, 3),
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
with pytest.raises(ValueError, match="interview"):
fp.add_trip("f", date(2026, 12, 2), date(2026, 12, 3), now=at(10, 9))
def test_answer_false_to_true_ok_true_to_false_refused() -> None:
make()
fp.answer_weekend("iv", SAT1, trip=False, now=at(11, 9))
yes = fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 10))
assert dict(yes.weekend_answers) == {SAT1: True}
with pytest.raises(ValueError, match="stricter"):
fp.answer_weekend("iv", SAT1, trip=False, now=at(11, 11))
assert dict(fp.list_projects()[0].weekend_answers) == {SAT1: True}
def test_repeating_an_answer_writes_no_second_row() -> None:
make()
fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 9))
fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 10))
rows = [r for r in evidence.read("iv") if r.milestone == "trip_ask:2026-11-14"]
assert [(r.kind, r.value) for r in rows] == [("answer", True)]
@pytest.mark.parametrize(
"day", [date(2026, 11, 15), date(2026, 11, 7), date(2026, 11, 28)]
)
def test_answer_only_for_a_weekend_of_the_project(day: date) -> None:
make()
with pytest.raises(ValueError, match="weekend"):
fp.answer_weekend("iv", day, trip=True, now=at(11, 9))
def test_no_trip_answer_contradicting_an_explicit_trip_is_refused() -> None:
make()
fp.add_trip("iv", date(2026, 11, 13), date(2026, 11, 15), now=at(10, 9))
with pytest.raises(ValueError, match="trip covers"):
fp.answer_weekend("iv", SAT1, trip=False, now=at(11, 9))
def test_late_answer_unlocks_but_is_late_done() -> None:
p = make()
assert states(p, at(13, 8))["trip_ask:2026-11-14"] == "missed"
p = fp.answer_weekend("iv", SAT1, trip=False, now=at(13, 8))
assert states(p, at(13, 9))["trip_ask:2026-11-14"] == "late_done"
def test_answer_on_wrong_project() -> None:
with pytest.raises(KeyError):
fp.answer_weekend("ghost", SAT1, trip=True, now=at(11, 9))
def test_find_skips_other_projects() -> None:
fp.create_project(
"f",
name="F",
start=date(2026, 10, 1),
end=date(2026, 10, 3),
replaces=["leetcode"],
daily_minutes=30,
unlock_minutes=10,
)
make()
p = fp.answer_weekend("iv", SAT1, trip=True, now=at(11, 9))
assert dict(p.weekend_answers) == {SAT1: True}
def test_late_no_trip_answer_cannot_erase_an_overdue_phone_milestone() -> None:
make()
assert states(fp.list_projects()[0], at(13, 13))["phone:2026-11-14"] == "missed"
p = fp.answer_weekend("iv", SAT1, trip=False, now=at(13, 13))
got = states(p, at(13, 14))
assert got["trip_ask:2026-11-14"] == "late_done"
assert got["phone:2026-11-14"] == "missed"
def test_no_trip_answer_before_the_phone_is_due_drops_it() -> None:
make()
p = fp.answer_weekend("iv", SAT1, trip=False, now=at(13, 11, 59))
assert "phone:2026-11-14" not in states(p, at(13, 14))
def test_no_trip_answer_given_at_creation_has_no_phone() -> None:
p = make(weekend_answers={SAT1: False})
got = states(p, at(13, 14))
assert got["trip_ask:2026-11-14"] == "done"
assert "phone:2026-11-14" not in got

72
focus_project/focus_project/tests/test_status_interview.py Normal file
View File

@ -0,0 +1,72 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""status(day) carries the milestones for i3bar / daily-limits / tutor."""
from __future__ import annotations
from datetime import date
from typing import TYPE_CHECKING
import focus_project as fp
from focus_project import _clock, evidence
from focus_project.tests.helpers import at, make
if TYPE_CHECKING:
from pathlib import Path
import pytest
def _now(monkeypatch: pytest.MonkeyPatch, day: int, hour: int) -> None:
stamp = at(day, hour).timestamp()
monkeypatch.setattr(_clock, "now", lambda: stamp)
def test_idle_status_has_the_keys() -> None:
snap = fp.status(date(2026, 11, 10))
assert snap["milestones"] == []
assert snap["missed"] is False
def test_status_lists_milestones_and_goes_red(monkeypatch: pytest.MonkeyPatch) -> None:
make()
_now(monkeypatch, 10, 9)
snap = fp.status(date(2026, 11, 10))
assert snap["missed"] is False
assert snap["milestones"][0]["id"] == "rules"
assert set(snap["milestones"][0]) == {"id", "due", "state"}
_now(monkeypatch, 11, 9)
snap = fp.status(date(2026, 11, 11))
assert snap["missed"] is True
assert snap["milestones"][0]["state"] == "missed"
assert snap["milestones"][0]["due"] == "2026-11-10T23:59:00"
def test_status_stays_red_on_test_day(monkeypatch: pytest.MonkeyPatch) -> None:
make()
_now(monkeypatch, 27, 9)
assert fp.status(date(2026, 11, 27))["missed"] is True
assert fp.owed_lock(at(27, 9)) is None
def test_status_survives_unreadable_evidence(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
make()
_now(monkeypatch, 11, 9)
def boom(project_id: str) -> list[evidence.Evidence]:
msg = "disk"
raise OSError(msg)
monkeypatch.setattr(evidence, "read", boom)
with caplog.at_level("WARNING"):
snap = fp.status(date(2026, 11, 11))
assert snap["milestones"] == []
assert snap["missed"] is False
assert "evidence unreadable" in caplog.text
assert fp.owed_lock(at(11, 9)) is None
def test_isolation_canary(tmp_path: Path) -> None:
make()
assert not (tmp_path / "data" / "evidence").exists()