diff --git a/focus_project/README.md b/focus_project/README.md index 739facf..3ff7e12 100644 --- a/focus_project/README.md +++ b/focus_project/README.md @@ -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:`, `phone:`, `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/.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. diff --git a/focus_project/focus_project/__init__.py b/focus_project/focus_project/__init__.py index 3f02862..cde6cd3 100644 --- a/focus_project/focus_project/__init__.py +++ b/focus_project/focus_project/__init__.py @@ -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", diff --git a/focus_project/focus_project/_api.py b/focus_project/focus_project/_api.py index 1ffe5e0..8abf0c2 100644 --- a/focus_project/focus_project/_api.py +++ b/focus_project/focus_project/_api.py @@ -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(): diff --git a/focus_project/focus_project/_interview.py b/focus_project/focus_project/_interview.py new file mode 100644 index 0000000..76aad73 --- /dev/null +++ b/focus_project/focus_project/_interview.py @@ -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) diff --git a/focus_project/focus_project/_kind.py b/focus_project/focus_project/_kind.py new file mode 100644 index 0000000..4fca6ff --- /dev/null +++ b/focus_project/focus_project/_kind.py @@ -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 diff --git a/focus_project/focus_project/_milestones.py b/focus_project/focus_project/_milestones.py new file mode 100644 index 0000000..7c56a59 --- /dev/null +++ b/focus_project/focus_project/_milestones.py @@ -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 diff --git a/focus_project/focus_project/_model.py b/focus_project/focus_project/_model.py index 97c5ab9..a6b3ac8 100644 --- a/focus_project/focus_project/_model.py +++ b/focus_project/focus_project/_model.py @@ -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}" diff --git a/focus_project/focus_project/_mutate.py b/focus_project/focus_project/_mutate.py new file mode 100644 index 0000000..b76e001 --- /dev/null +++ b/focus_project/focus_project/_mutate.py @@ -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 + ) diff --git a/focus_project/focus_project/_paths.py b/focus_project/focus_project/_paths.py index 7377a9e..f617b53 100644 --- a/focus_project/focus_project/_paths.py +++ b/focus_project/focus_project/_paths.py @@ -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 diff --git a/focus_project/focus_project/_trips.py b/focus_project/focus_project/_trips.py new file mode 100644 index 0000000..ee9df1b --- /dev/null +++ b/focus_project/focus_project/_trips.py @@ -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)) diff --git a/focus_project/focus_project/checks.py b/focus_project/focus_project/checks.py new file mode 100644 index 0000000..6011ffb --- /dev/null +++ b/focus_project/focus_project/checks.py @@ -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) diff --git a/focus_project/focus_project/evidence.py b/focus_project/focus_project/evidence.py new file mode 100644 index 0000000..c031d77 --- /dev/null +++ b/focus_project/focus_project/evidence.py @@ -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) diff --git a/focus_project/focus_project/status.py b/focus_project/focus_project/status.py index c1cbc73..e6f6f19 100644 --- a/focus_project/focus_project/status.py +++ b/focus_project/focus_project/status.py @@ -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, } diff --git a/focus_project/focus_project/tests/conftest.py b/focus_project/focus_project/tests/conftest.py index 123e843..640a57e 100644 --- a/focus_project/focus_project/tests/conftest.py +++ b/focus_project/focus_project/tests/conftest.py @@ -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)) diff --git a/focus_project/focus_project/tests/helpers.py b/focus_project/focus_project/tests/helpers.py new file mode 100644 index 0000000..2848633 --- /dev/null +++ b/focus_project/focus_project/tests/helpers.py @@ -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)} diff --git a/focus_project/focus_project/tests/test_activity_status.py b/focus_project/focus_project/tests/test_activity_status.py index a84b398..88e1984 100644 --- a/focus_project/focus_project/tests/test_activity_status.py +++ b/focus_project/focus_project/tests/test_activity_status.py @@ -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", } diff --git a/focus_project/focus_project/tests/test_checks.py b/focus_project/focus_project/tests/test_checks.py new file mode 100644 index 0000000..07b8553 --- /dev/null +++ b/focus_project/focus_project/tests/test_checks.py @@ -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 diff --git a/focus_project/focus_project/tests/test_evidence.py b/focus_project/focus_project/tests/test_evidence.py new file mode 100644 index 0000000..58c476a --- /dev/null +++ b/focus_project/focus_project/tests/test_evidence.py @@ -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] diff --git a/focus_project/focus_project/tests/test_interview_model.py b/focus_project/focus_project/tests/test_interview_model.py new file mode 100644 index 0000000..0269fea --- /dev/null +++ b/focus_project/focus_project/tests/test_interview_model.py @@ -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 diff --git a/focus_project/focus_project/tests/test_milestones.py b/focus_project/focus_project/tests/test_milestones.py new file mode 100644 index 0000000..addeb39 --- /dev/null +++ b/focus_project/focus_project/tests/test_milestones.py @@ -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 diff --git a/focus_project/focus_project/tests/test_ratchet.py b/focus_project/focus_project/tests/test_ratchet.py new file mode 100644 index 0000000..bd4df71 --- /dev/null +++ b/focus_project/focus_project/tests/test_ratchet.py @@ -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 diff --git a/focus_project/focus_project/tests/test_status_interview.py b/focus_project/focus_project/tests/test_status_interview.py new file mode 100644 index 0000000..419e1c7 --- /dev/null +++ b/focus_project/focus_project/tests/test_status_interview.py @@ -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()