mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-10 19:27:16 +01:00
Compare commits
2 Commits
0527bc680d
...
17e300af0a
| Author | SHA1 | Date | |
|---|---|---|---|
| 17e300af0a | |||
| afcc1a47c7 |
@ -199,6 +199,7 @@ their `max_gaming_minutes` + pure bonuses again (240/120 min = 2/min; full day =
|
||||
from the start day, cutting all of it, 180 with LeetCode's formerly free hour (0.20.0); maturity never delays it. Ledger `~/.local/share/focus_project/ledger.json`:
|
||||
signed `credit` rows, `entry_id` `<project>-<YYYYMMDD>-m<n>`, `detail.minutes` (positive int),
|
||||
`detail.ended_at` (else `day`). `focus_project` unusable = no project (logged).
|
||||
Unbacked timer time pays 0 from 2026-11-08 (focus_project 0.10.0).
|
||||
|
||||
## Adding a gate
|
||||
|
||||
|
||||
88
earned_time/earned_time/tests/test_focus_backing.py
Normal file
88
earned_time/earned_time/tests/test_focus_backing.py
Normal file
@ -0,0 +1,88 @@
|
||||
# Copyright (c) 2026 Krzysztof Rudnicki
|
||||
"""Regression: a timer-only focus day after the anti-cheat cutover pays 0.
|
||||
|
||||
Runs the REAL ``focus_project`` (the sibling package of this monorepo):
|
||||
timer rows -> harvester -> signed ledger -> ``credit_units``. earned_time
|
||||
itself does not change; this pins that the gate reaches pay.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime
|
||||
import importlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
from earned_time import credit_units, earners_for
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from types import ModuleType
|
||||
|
||||
DAY = date(2026, 11, 9) # after focus_project.BACKING_FROM
|
||||
_SIBLING = Path(__file__).resolve().parents[3] / "focus_project"
|
||||
|
||||
|
||||
def at(hour: int, minute: int) -> float:
|
||||
return datetime(2026, 11, 9, hour, minute).astimezone().timestamp()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def real_fp(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, key_file: Path
|
||||
) -> ModuleType:
|
||||
"""The real focus_project on redirected state (HOME, data dir, key)."""
|
||||
monkeypatch.setenv("HOME", str(tmp_path / "home"))
|
||||
monkeypatch.setenv("FOCUS_PROJECT_DATA", str(tmp_path / "data"))
|
||||
monkeypatch.setenv("FOCUS_PROJECT_KEY", str(key_file))
|
||||
monkeypatch.setenv("FOCUS_PROJECT_LEETCODE_LEDGER", str(tmp_path / "lc.json"))
|
||||
monkeypatch.delitem(sys.modules, "focus_project") # the conftest's fake
|
||||
monkeypatch.syspath_prepend(str(_SIBLING))
|
||||
fp = importlib.import_module("focus_project")
|
||||
fp.create_project(
|
||||
"hr",
|
||||
name="HR",
|
||||
start=DAY,
|
||||
end=DAY,
|
||||
replaces=["leetcode"],
|
||||
daily_minutes=120,
|
||||
unlock_minutes=30,
|
||||
)
|
||||
return fp
|
||||
|
||||
|
||||
def pay(fp: ModuleType, tmp_path: Path, key_file: Path, *rows: object) -> int | None:
|
||||
log = tmp_path / "data" / "activity.jsonl"
|
||||
log.write_text("".join(json.dumps(r) + "\n" for r in rows), encoding="utf-8")
|
||||
harvest = importlib.import_module("focus_project.harvest")
|
||||
assert harvest.harvest(now=at(23, 0)) == 0
|
||||
(item,) = (e for e in earners_for(DAY) if e.name == "focus")
|
||||
return credit_units(item, tmp_path / "data" / "ledger.json", key_file, DAY)
|
||||
|
||||
|
||||
def row(source: str, kind: str, ts: float) -> dict[str, object]:
|
||||
return {"ts": ts, "source": source, "kind": kind, "project": "hr"}
|
||||
|
||||
|
||||
TIMER = (
|
||||
row("timer", "start", at(9, 0)),
|
||||
row("timer", "checkin", at(9, 5)),
|
||||
row("timer", "stop", at(9, 10)),
|
||||
)
|
||||
|
||||
|
||||
def test_timer_only_day_pays_zero(
|
||||
real_fp: ModuleType, tmp_path: Path, key_file: Path
|
||||
) -> None:
|
||||
assert date(2026, 11, 8) == real_fp.BACKING_FROM
|
||||
assert pay(real_fp, tmp_path, key_file, *TIMER) == 0
|
||||
|
||||
|
||||
def test_backed_timer_day_still_pays(
|
||||
real_fp: ModuleType, tmp_path: Path, key_file: Path
|
||||
) -> None:
|
||||
beat = row("tutor", "heartbeat", at(9, 5))
|
||||
assert pay(real_fp, tmp_path, key_file, *TIMER, beat) == 10
|
||||
@ -36,6 +36,10 @@ uncovers today. A NEW project starting today may replace gates and set any targe
|
||||
new obligation, not a reduction (but see the floor).
|
||||
Floor (0.7.1, coordinator decision for kuhy): `daily_minutes >= 30 x replaced gates` (min 30; `book` = `reading`),
|
||||
enforced on create and update (`daily_floor(replaces)`), so a day cannot be made cheap.
|
||||
Backed timer (0.10.0, days from `BACKING_FROM` = 2026-11-08): a timer span earns only if a
|
||||
same-project `tutor`/`sql_drills`/`leetcode` beat (or accepted submission) is in it, 5 min
|
||||
before it, or bridges it (beats <= 20 min apart). `unbacked_minutes_for(day)` /
|
||||
`status()["unbacked_minutes"]` show the rest. A beat backs only the spans it touches (the tutor chain makes a graded card back the whole stretch). Known gap: any coach turn backs.
|
||||
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`).
|
||||
|
||||
|
||||
@ -31,10 +31,12 @@ 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.minutes import minutes_for, unbacked_minutes_for
|
||||
from focus_project.sources.activity import BACKING_FROM
|
||||
from focus_project.status import status
|
||||
|
||||
__all__ = [
|
||||
"BACKING_FROM",
|
||||
"Evidence",
|
||||
"Limits",
|
||||
"Milestone",
|
||||
@ -59,5 +61,6 @@ __all__ = [
|
||||
"run_checks",
|
||||
"stands_down",
|
||||
"status",
|
||||
"unbacked_minutes_for",
|
||||
"update_project",
|
||||
]
|
||||
|
||||
@ -8,6 +8,10 @@ minutes (floor). Time not yet lived is never credited: every interval is
|
||||
clipped at "now", so an event stamped ahead of the clock (a fast phone clock)
|
||||
earns only once the real clock reaches it. An unreadable registry reads as 0
|
||||
(logged): fail closed.
|
||||
|
||||
Unbacked minutes (0.10.0): timer time the anti-cheat gate refused
|
||||
(:mod:`focus_project.sources.activity`) and no other source covers, on the
|
||||
local day up to "now"; carry and the cap play no part. Shown, never paid.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -15,9 +19,10 @@ from __future__ import annotations
|
||||
from datetime import date # noqa: TC003
|
||||
import logging
|
||||
|
||||
from focus_project import _api, _clock, carry
|
||||
from focus_project import _api, _clock, carry, sources
|
||||
from focus_project._errors import RegistryError
|
||||
from focus_project._own import union_seconds
|
||||
from focus_project.sources import activity
|
||||
|
||||
_LOG = logging.getLogger(__name__)
|
||||
|
||||
@ -38,4 +43,26 @@ def minutes_for(day: date, now: float | None = None) -> int:
|
||||
return carry.balance(day, moment).counted(project.daily_on(day))
|
||||
|
||||
|
||||
__all__ = ["minutes_for", "union_seconds"]
|
||||
def unbacked_minutes_for(day: date, now: float | None = None) -> int:
|
||||
"""Whole minutes of timer time on ``day`` that the gate did not count.
|
||||
|
||||
0 with no project, before :data:`~focus_project.sources.activity.BACKING_FROM`,
|
||||
or with an unreadable registry (logged).
|
||||
"""
|
||||
try:
|
||||
project = _api.active(day)
|
||||
if project is None:
|
||||
return 0
|
||||
lo, hi = sources.day_bounds(day)
|
||||
bounds = (lo, min(hi, _clock.now() if now is None else now))
|
||||
unbacked = sources.clip(activity.split_timer(day, project)[1], bounds)
|
||||
counted = sources.clip(sources.collect(day, project), bounds)
|
||||
except (RegistryError, OSError) as exc:
|
||||
_LOG.warning("focus_project registry unreadable, 0 unbacked: %s", exc)
|
||||
return 0
|
||||
# |unbacked minus counted| = |unbacked union counted| - |counted|.
|
||||
gap = union_seconds(unbacked + counted) - union_seconds(counted)
|
||||
return int(gap // 60)
|
||||
|
||||
|
||||
__all__ = ["minutes_for", "unbacked_minutes_for", "union_seconds"]
|
||||
|
||||
@ -13,23 +13,36 @@ heartbeat covering the five minutes before it. Only events recorded for
|
||||
the open segment, like a ``checkin``: within :data:`MAX_GAP_S` of the last
|
||||
beat it extends the segment; later it ends the segment at that last beat. It
|
||||
never opens a segment. Other heartbeat sources never chain the timer.
|
||||
|
||||
0.10.0 anti-cheat (days from :data:`BACKING_FROM`): a timer span earns only
|
||||
if real activity of the same project backs it, whole span or nothing. A
|
||||
backing event is a ``tutor``/``sql_drills``/``leetcode`` heartbeat or an
|
||||
accepted LeetCode submission (leetcode projects only); ``browser`` never
|
||||
backs. A span ``(a, b)`` is backed if a backing event lies in
|
||||
``[a - 300, b]``, or between two consecutive backing events at most
|
||||
:data:`BRIDGE_S` apart. Check-ins alone earn 0. Earlier days keep the old
|
||||
rule (every span counts), so paid days never change.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
import importlib
|
||||
from itertools import pairwise
|
||||
import json
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from focus_project._paths import activity_path
|
||||
from focus_project.sources import Interval, heartbeat_interval
|
||||
from focus_project.sources import HEARTBEAT_WINDOW_S, Interval, heartbeat_interval
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from datetime import date
|
||||
|
||||
from focus_project._model import Project
|
||||
|
||||
MAX_GAP_S = 600.0
|
||||
BACKING_FROM: Final = date(2026, 11, 8)
|
||||
BACKING_SOURCES: Final = frozenset({"tutor", "sql_drills", "leetcode"})
|
||||
BRIDGE_S: Final = 1200.0
|
||||
_LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -102,12 +115,50 @@ def sessions(project_id: str) -> list[Interval]:
|
||||
return merged
|
||||
|
||||
|
||||
def intervals(day: date, project: Project) -> list[Interval]: # noqa: ARG001
|
||||
"""Timer segments plus heartbeats of ``project`` (the registry clips)."""
|
||||
def backing_times(events: list[dict[str, Any]], project: Project) -> list[float]:
|
||||
"""Sorted times of the events that may back a timer span.
|
||||
|
||||
The LeetCode ledger is read in its own guard: if it fails, only the
|
||||
submissions are lost, never the heartbeats (fail closed, narrow).
|
||||
"""
|
||||
times = [float(e["ts"]) for e in events if e.get("source") in BACKING_SOURCES]
|
||||
try:
|
||||
# Imported lazily: it pulls in earned_time, and the timer must keep
|
||||
# working (heartbeat backing only) where that is not installed.
|
||||
leetcode = importlib.import_module("focus_project.sources.leetcode")
|
||||
times += leetcode.submission_times(project)
|
||||
except Exception: # noqa: BLE001 (lose the submissions, keep the beats)
|
||||
_LOG.exception("leetcode backing unavailable, heartbeats only")
|
||||
return sorted(times)
|
||||
|
||||
|
||||
def is_backed(span: Interval, times: list[float]) -> bool:
|
||||
"""Whether a backing event lies in or just before ``span``, or bridges it."""
|
||||
lo, hi = span
|
||||
if any(lo - HEARTBEAT_WINDOW_S <= t <= hi for t in times):
|
||||
return True
|
||||
return any(a <= lo and hi <= b and b - a <= BRIDGE_S for a, b in pairwise(times))
|
||||
|
||||
|
||||
def split_timer(
|
||||
day: date, project: Project, events: list[dict[str, Any]] | None = None
|
||||
) -> tuple[list[Interval], list[Interval]]:
|
||||
"""``(backed, unbacked)`` timer spans of ``project``, judged for ``day``."""
|
||||
found = read_events(project.id) if events is None else events
|
||||
chain = [e for e in found if e.get("source") == "timer" or _is_beat(e)]
|
||||
spans = timer_intervals(chain)
|
||||
if day < BACKING_FROM:
|
||||
return spans, []
|
||||
times = backing_times(found, project)
|
||||
backed = [s for s in spans if is_backed(s, times)]
|
||||
return backed, [s for s in spans if s not in backed]
|
||||
|
||||
|
||||
def intervals(day: date, project: Project) -> list[Interval]:
|
||||
"""Backed timer segments plus heartbeats of ``project`` (the registry clips)."""
|
||||
events = read_events(project.id)
|
||||
timer = [e for e in events if e.get("source") == "timer" or _is_beat(e)]
|
||||
beats = [e for e in events if e.get("source") != "timer"]
|
||||
return [
|
||||
*timer_intervals(timer),
|
||||
*split_timer(day, project, events)[0],
|
||||
*(heartbeat_interval(float(e["ts"])) for e in beats),
|
||||
]
|
||||
|
||||
@ -55,8 +55,8 @@ def _submitted_at(row: object, key: bytes) -> float | None:
|
||||
return None
|
||||
|
||||
|
||||
def intervals(day: date, project: Project) -> list[Interval]: # noqa: ARG001
|
||||
"""One heartbeat per accepted submission (the registry clips to the day)."""
|
||||
def submission_times(project: Project) -> list[float]:
|
||||
"""Times of verified accepted submissions; [] for a non-leetcode project."""
|
||||
if not any("leetcode" in pattern.lower() for pattern in project.site_patterns):
|
||||
return []
|
||||
key = ledger.read_key()
|
||||
@ -64,4 +64,9 @@ def intervals(day: date, project: Project) -> list[Interval]: # noqa: ARG001
|
||||
_LOG.warning("leetcode source: no signing key, contributes 0")
|
||||
return []
|
||||
stamps = (_submitted_at(row, key) for row in ledger.read_rows(ledger_file()))
|
||||
return [heartbeat_interval(ts) for ts in stamps if ts is not None]
|
||||
return [ts for ts in stamps if ts is not None]
|
||||
|
||||
|
||||
def intervals(day: date, project: Project) -> list[Interval]: # noqa: ARG001
|
||||
"""One heartbeat per accepted submission (the registry clips to the day)."""
|
||||
return [heartbeat_interval(ts) for ts in submission_times(project)]
|
||||
|
||||
@ -6,7 +6,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from focus_project import _api, _milestones, carry, evidence
|
||||
from focus_project import _api, _milestones, carry, evidence, minutes
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from datetime import date
|
||||
@ -36,7 +36,8 @@ def status(day: date) -> dict[str, Any]:
|
||||
|
||||
``minutes`` is what was really worked plus what the day took from the
|
||||
bank, so it may exceed ``daily_minutes`` (121/120); ``counted_minutes``
|
||||
is the capped figure that pays.
|
||||
is the capped figure that pays. ``unbacked_minutes`` is timer time the
|
||||
anti-cheat gate refused (shown as "not counted", never paid).
|
||||
"""
|
||||
project = _api.active(day)
|
||||
if project is None:
|
||||
@ -52,6 +53,7 @@ def status(day: date) -> dict[str, Any]:
|
||||
"unlock_minutes": None,
|
||||
"ends": None,
|
||||
"replaces": [],
|
||||
"unbacked_minutes": 0,
|
||||
"milestones": [],
|
||||
"missed": False,
|
||||
}
|
||||
@ -72,6 +74,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)),
|
||||
"unbacked_minutes": minutes.unbacked_minutes_for(day),
|
||||
# Interview projects: [{id, due ISO, state}] and whether any is red.
|
||||
"milestones": rows,
|
||||
"missed": missed,
|
||||
|
||||
@ -137,6 +137,7 @@ def test_status_active_reports_the_worked_minutes() -> None:
|
||||
"unlock_minutes": 30,
|
||||
"ends": "2026-11-05",
|
||||
"replaces": ["leetcode"],
|
||||
"unbacked_minutes": 0,
|
||||
"milestones": [],
|
||||
"missed": False,
|
||||
}
|
||||
@ -160,6 +161,7 @@ def test_status_inactive() -> None:
|
||||
"unlock_minutes",
|
||||
"ends",
|
||||
"replaces",
|
||||
"unbacked_minutes",
|
||||
"milestones",
|
||||
"missed",
|
||||
}
|
||||
|
||||
181
focus_project/focus_project/tests/test_backing.py
Normal file
181
focus_project/focus_project/tests/test_backing.py
Normal file
@ -0,0 +1,181 @@
|
||||
# Copyright (c) 2026 Krzysztof Rudnicki
|
||||
"""0.10.0: a timer span earns only if a same-project beat backs it."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import focus_project as fp
|
||||
from focus_project import _api, minutes
|
||||
from focus_project.sources import activity
|
||||
from focus_project.tests import test_minutes as tm
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest
|
||||
|
||||
DAY = date(2026, 11, 8) # BACKING_FROM
|
||||
|
||||
|
||||
def at(hour: int, minute: int = 0) -> float:
|
||||
return tm.at(hour, minute, day=DAY)
|
||||
|
||||
|
||||
def proj() -> None:
|
||||
fp.create_project(
|
||||
"hr",
|
||||
name="HR prep",
|
||||
start=DAY,
|
||||
end=date(2026, 11, 12),
|
||||
replaces=["leetcode"],
|
||||
daily_minutes=120,
|
||||
unlock_minutes=30,
|
||||
site_patterns=["leetcode.com"],
|
||||
)
|
||||
|
||||
|
||||
def rows(*items: tuple[str, str, float]) -> None:
|
||||
tm.events(*items)
|
||||
|
||||
|
||||
def timer_only() -> None:
|
||||
rows(
|
||||
("timer", "start", at(9)),
|
||||
("timer", "checkin", at(9, 5)),
|
||||
("timer", "stop", at(9, 10)),
|
||||
)
|
||||
|
||||
|
||||
def test_timer_only_stretch_pays_zero_and_shows_unbacked() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
assert minutes.minutes_for(DAY) == 0
|
||||
assert minutes.unbacked_minutes_for(DAY) == 10
|
||||
assert fp.status(DAY)["unbacked_minutes"] == 10
|
||||
|
||||
|
||||
def test_a_beat_in_the_span_backs_all_of_it() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
rows(("sql_drills", "heartbeat", at(9, 5)))
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
assert minutes.minutes_for(DAY) == 10
|
||||
|
||||
|
||||
def test_a_beat_backs_only_the_spans_it_touches() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
rows(("sql_drills", "heartbeat", at(9, 7)))
|
||||
# 9:00-9:05 stays unbacked, but the beat itself pays 9:02-9:07.
|
||||
assert minutes.unbacked_minutes_for(DAY) == 2
|
||||
|
||||
|
||||
def test_a_beat_at_the_end_of_the_span_backs_it() -> None:
|
||||
proj()
|
||||
rows(
|
||||
("timer", "start", at(9)),
|
||||
("timer", "stop", at(9, 10)),
|
||||
("sql_drills", "heartbeat", at(9, 10)),
|
||||
)
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
assert minutes.minutes_for(DAY) == 10
|
||||
|
||||
|
||||
def test_a_beat_just_before_the_span_backs_it() -> None:
|
||||
proj()
|
||||
rows(
|
||||
("sql_drills", "heartbeat", at(8, 56)), # within 300 s before 9:00
|
||||
("timer", "start", at(9)),
|
||||
("timer", "stop", at(9, 5)),
|
||||
)
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
|
||||
|
||||
def test_a_late_beat_does_not_back_it() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
rows(("sql_drills", "heartbeat", at(9, 40)))
|
||||
assert minutes.unbacked_minutes_for(DAY) == 10
|
||||
|
||||
|
||||
def test_beats_up_to_twenty_minutes_apart_bridge_the_span() -> None:
|
||||
proj()
|
||||
rows(
|
||||
("tutor", "heartbeat", at(8, 50)),
|
||||
("timer", "start", at(9)),
|
||||
("timer", "checkin", at(9, 5)),
|
||||
("timer", "stop", at(9, 8)),
|
||||
("tutor", "heartbeat", at(9, 10)),
|
||||
)
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
|
||||
|
||||
def test_beats_over_twenty_minutes_apart_do_not_bridge() -> None:
|
||||
proj()
|
||||
rows(
|
||||
("tutor", "heartbeat", at(8, 30)),
|
||||
("timer", "start", at(9)),
|
||||
("timer", "stop", at(9, 8)),
|
||||
("tutor", "heartbeat", at(9, 40)),
|
||||
)
|
||||
assert minutes.unbacked_minutes_for(DAY) == 8
|
||||
|
||||
|
||||
def test_browser_never_backs() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
rows(("browser", "heartbeat", at(9, 7)))
|
||||
# The beat pays its own 9:02-9:07 but backs no timer span.
|
||||
assert minutes.unbacked_minutes_for(DAY) == 5
|
||||
assert minutes.minutes_for(DAY) == 5
|
||||
|
||||
|
||||
def test_days_before_backing_from_keep_the_old_rule() -> None:
|
||||
fp.create_project(
|
||||
"hr",
|
||||
name="Old",
|
||||
start=date(2026, 11, 2),
|
||||
end=date(2026, 11, 5),
|
||||
replaces=["leetcode"],
|
||||
daily_minutes=120,
|
||||
unlock_minutes=30,
|
||||
site_patterns=["leetcode.com"],
|
||||
)
|
||||
old = date(2026, 11, 2)
|
||||
rows(("timer", "start", tm.at(9)), ("timer", "stop", tm.at(9, 10)))
|
||||
assert minutes.minutes_for(old) == 10
|
||||
assert minutes.unbacked_minutes_for(old) == 0
|
||||
|
||||
|
||||
def test_no_project_is_zero_unbacked() -> None:
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
assert fp.status(DAY)["unbacked_minutes"] == 0
|
||||
|
||||
|
||||
def test_unbacked_is_clipped_at_now() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
assert minutes.unbacked_minutes_for(DAY, now=at(9, 4)) == 4
|
||||
|
||||
|
||||
def test_split_timer_returns_backed_and_unbacked() -> None:
|
||||
proj()
|
||||
timer_only()
|
||||
project = _api.active(DAY)
|
||||
assert project is not None
|
||||
backed, unbacked = activity.split_timer(DAY, project)
|
||||
assert backed == []
|
||||
assert unbacked == [(at(9), at(9, 5)), (at(9, 5), at(9, 10))]
|
||||
|
||||
|
||||
def test_backing_from_is_exported() -> None:
|
||||
assert date(2026, 11, 8) == fp.BACKING_FROM
|
||||
assert fp.unbacked_minutes_for is minutes.unbacked_minutes_for
|
||||
|
||||
|
||||
def test_registry_error_reads_zero(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def boom(_day: date) -> None:
|
||||
raise OSError
|
||||
|
||||
monkeypatch.setattr("focus_project._api.active", boom)
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
131
focus_project/focus_project/tests/test_backing_edges.py
Normal file
131
focus_project/focus_project/tests/test_backing_edges.py
Normal file
@ -0,0 +1,131 @@
|
||||
# Copyright (c) 2026 Krzysztof Rudnicki
|
||||
"""More 0.10.0 backing cases: sources, projects, the cutover midnight, chains.
|
||||
|
||||
The core rule lives in ``test_backing.py``; these pin the edges the spec
|
||||
named (C5-C7) with leetcode-ledger and other-project fixtures.
|
||||
"""
|
||||
|
||||
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 minutes
|
||||
from focus_project._paths import activity_path
|
||||
from focus_project.tests.test_leetcode_source import ac_row, write
|
||||
from focus_project.tests.test_minutes import at
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
DAY = date(2026, 11, 9) # after the cutover
|
||||
PRE = date(2026, 11, 7) # before it
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
"""Nothing may fall back to the real home directory."""
|
||||
monkeypatch.setenv("HOME", str(tmp_path / "home"))
|
||||
|
||||
|
||||
def make(pid: str = "hr", patterns: tuple[str, ...] = ("leetcode.com",)) -> None:
|
||||
fp.create_project(
|
||||
pid,
|
||||
name=pid,
|
||||
start=date(2026, 11, 6),
|
||||
end=date(2026, 11, 12),
|
||||
replaces=["leetcode"],
|
||||
daily_minutes=240,
|
||||
unlock_minutes=30,
|
||||
site_patterns=list(patterns),
|
||||
)
|
||||
|
||||
|
||||
def rows(*items: tuple[str, str, float] | tuple[str, str, float, str]) -> None:
|
||||
"""Append ``(source, kind, ts[, project])`` events (project default hr)."""
|
||||
activity_path().parent.mkdir(parents=True, exist_ok=True)
|
||||
with activity_path().open("a", encoding="utf-8") as handle:
|
||||
for item in items:
|
||||
source, kind, ts = item[:3]
|
||||
pid = item[3] if len(item) == 4 else "hr"
|
||||
row = {"ts": ts, "source": source, "kind": kind, "project": pid}
|
||||
handle.write(json.dumps(row) + "\n")
|
||||
|
||||
|
||||
def timer(day: date, *marks: tuple[int, int]) -> None:
|
||||
"""A segment: start at the first mark, check-ins, stop at the last."""
|
||||
kinds = ["start", *["checkin"] * (len(marks) - 2), "stop"]
|
||||
rows(
|
||||
*(
|
||||
("timer", k, at(h, m, day=day))
|
||||
for k, (h, m) in zip(kinds, marks, strict=True)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def beat(source: str, h: int, m: int, day: date = DAY, pid: str = "hr") -> None:
|
||||
rows((source, "heartbeat", at(h, m, day=day), pid))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", ["sql_drills", "leetcode"])
|
||||
def test_sql_drills_and_leetcode_heartbeats_back_too(source: str) -> None:
|
||||
make()
|
||||
timer(DAY, (9, 0), (9, 5), (9, 10))
|
||||
beat(source, 9, 5)
|
||||
assert minutes.minutes_for(DAY) == 10
|
||||
|
||||
|
||||
def test_browser_and_other_project_beats_do_not_back() -> None:
|
||||
make() # "other" needs no registry entry: rows are filtered by id
|
||||
timer(DAY, (9, 0), (9, 5), (9, 10))
|
||||
beat("browser", 9, 4)
|
||||
beat("tutor", 9, 4, pid="other")
|
||||
assert minutes.minutes_for(DAY) == 5 # the browser beat's own window only
|
||||
assert minutes.unbacked_minutes_for(DAY) == 6 # 9:04-9:10
|
||||
|
||||
|
||||
def test_accepted_submission_backs_a_leetcode_project(tmp_path: Path) -> None:
|
||||
make()
|
||||
timer(DAY, (9, 0), (9, 5), (9, 10))
|
||||
write(tmp_path, ac_row(at(9, 4, day=DAY)))
|
||||
assert minutes.minutes_for(DAY) == 11 # 8:59-9:10
|
||||
|
||||
|
||||
def test_submission_does_not_back_a_non_leetcode_project(tmp_path: Path) -> None:
|
||||
make("hr", ("sqlzoo.net",))
|
||||
timer(DAY, (9, 0), (9, 5), (9, 10))
|
||||
write(tmp_path, ac_row(at(9, 4, day=DAY)))
|
||||
assert minutes.minutes_for(DAY) == 0
|
||||
|
||||
|
||||
def test_span_across_the_cutover_midnight_follows_the_day() -> None:
|
||||
make()
|
||||
rows(
|
||||
("timer", "start", at(23, 58, day=date(2026, 11, 7))),
|
||||
("timer", "checkin", at(0, 3, day=date(2026, 11, 8))),
|
||||
("timer", "stop", at(0, 5, day=date(2026, 11, 8))),
|
||||
)
|
||||
assert minutes.minutes_for(date(2026, 11, 7)) == 2
|
||||
assert minutes.minutes_for(date(2026, 11, 8)) == 0
|
||||
assert minutes.unbacked_minutes_for(date(2026, 11, 8)) == 5
|
||||
|
||||
|
||||
def test_tutor_chained_span_is_backed_but_earlier_reading_is_not() -> None:
|
||||
"""A tutor beat chains the segment (0.8.0) and backs the span it ends.
|
||||
|
||||
start 9:00, checkin 9:08, tutor beat 9:16: the span 9:08-9:16 ends at a
|
||||
backing beat and earns; 9:00-9:08 has no beat in or 5 min before it and
|
||||
nothing bridges it, so it shows as unbacked.
|
||||
"""
|
||||
make()
|
||||
rows(
|
||||
("timer", "start", at(9, 0, day=DAY)),
|
||||
("timer", "checkin", at(9, 8, day=DAY)),
|
||||
)
|
||||
beat("tutor", 9, 16)
|
||||
assert minutes.minutes_for(DAY) == 8
|
||||
assert minutes.unbacked_minutes_for(DAY) == 8
|
||||
@ -14,6 +14,8 @@ import pytest
|
||||
import focus_project as fp
|
||||
from focus_project import minutes, sources
|
||||
from focus_project.sources import leetcode
|
||||
from focus_project.tests.test_backing import DAY, proj, rows
|
||||
from focus_project.tests.test_backing import at as day_at
|
||||
from focus_project.tests.test_minutes import D1, at, project
|
||||
|
||||
KEY = b"k" * 32
|
||||
@ -127,3 +129,29 @@ def test_read_only_replay_of_real_ledger(monkeypatch: pytest.MonkeyPatch) -> Non
|
||||
assert got
|
||||
assert all(b - a == sources.HEARTBEAT_WINDOW_S for a, b in got)
|
||||
assert REAL.stat().st_mtime_ns == before
|
||||
|
||||
|
||||
def test_accepted_submission_backs_a_timer_span(tmp_path: Path) -> None:
|
||||
"""0.10.0: an accepted submission backs timer time on a backed day."""
|
||||
proj()
|
||||
rows(("timer", "start", day_at(9)), ("timer", "stop", day_at(9, 10)))
|
||||
assert minutes.unbacked_minutes_for(DAY) == 10
|
||||
write(tmp_path, ac_row(day_at(9, 10)))
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
assert minutes.minutes_for(DAY) == 10
|
||||
|
||||
|
||||
def test_unreadable_ledger_leaves_heartbeat_backing(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
def boom(_project: object) -> list[float]:
|
||||
raise OSError
|
||||
|
||||
monkeypatch.setattr(leetcode, "submission_times", boom)
|
||||
proj()
|
||||
rows(
|
||||
("timer", "start", day_at(9)),
|
||||
("timer", "stop", day_at(9, 10)),
|
||||
("tutor", "heartbeat", day_at(9, 10)),
|
||||
)
|
||||
assert minutes.unbacked_minutes_for(DAY) == 0
|
||||
|
||||
@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "focus_project"
|
||||
version = "0.9.1"
|
||||
version = "0.10.0"
|
||||
description = "Time-boxed focus projects: replace chosen gates, pay a daily-minutes gate"
|
||||
requires-python = ">=3.14"
|
||||
# Stdlib only on purpose: guards import it from the system python.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user