stack: wikitool review - der Wochenrueckblick als Join zur Lesezeit (#125)
Files changed: - CHANGES.md - VERSION - tools/CONTRACT.md - tools/chemenu/cli.py - tools/chemenu/commands/review_cmd.py - tools/chemenu/commands/run_budget.py - tools/chemenu/review.py - tools/chemenu/tests/test_review.py
This commit is contained in:
1 parent
1875449b31
commit
80b57e0d01
8 files changed
+751
-5
No files matched your search
@@ -29,6 +29,7 @@ try:
|
||||
page_ops,
|
||||
provenance_cmd,
|
||||
raw_cmd,
|
||||
review_cmd,
|
||||
run_budget,
|
||||
search as search_module,
|
||||
touch as touch_module,
|
||||
@@ -158,6 +159,7 @@ app.command("rm")(page_ops.rm_command)
|
||||
app.command("move")(page_ops.move_command)
|
||||
app.command("lint")(lint_module.lint_command)
|
||||
app.command("search")(search_module.search_command)
|
||||
app.command("review")(review_cmd.review_command)
|
||||
app.command("publish")(git_publish.publish_command)
|
||||
app.command("sync")(git_publish.sync_command)
|
||||
app.command("doctor")(doctor.doctor_command)
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""`wikitool review` - the terminal adapter over `chemenu.review` (Gitea #125).
|
||||
|
||||
The checks, the join and the partial-report rule live in `chemenu.review`,
|
||||
which imports no CLI machinery. This module owns only what a terminal needs:
|
||||
the `--json` flag, the two render forms, and the exit code.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import typer
|
||||
|
||||
from chemenu import config
|
||||
from chemenu.commands._util import fail
|
||||
from chemenu.errors import ValidationError
|
||||
from chemenu.review import ALL_CHECKS, ReviewReport, run_review
|
||||
|
||||
__all__ = ["render_report", "report_to_dict", "review_command"]
|
||||
|
||||
|
||||
def render_report(report: ReviewReport) -> str:
|
||||
"""The `--json`-free rendering. One line per finding, `[check] project:
|
||||
message`, so a hit can be told apart from the summary line without a
|
||||
schema - the same shape `search`'s table takes for the same reason."""
|
||||
lines: list[str] = []
|
||||
if report.checks_skipped:
|
||||
lines.append("INCOMPLETE - the following check(s) did not run:")
|
||||
for check, reason in report.checks_skipped:
|
||||
lines.append(f" - {check}: {reason}")
|
||||
lines.append(
|
||||
f"Partial result: {report.kb_project_count} kb/ project page(s) found; no "
|
||||
"tracker cross-check for the check(s) above."
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if not report.findings:
|
||||
lines.append("No findings.")
|
||||
else:
|
||||
for finding in report.findings:
|
||||
lines.append(f"[{finding.check}] {finding.project}: {finding.message}")
|
||||
|
||||
lines.append("")
|
||||
lines.append(f"{len(report.findings)} finding(s), {len(report.checks_run)}/{len(ALL_CHECKS)} check(s) ran.")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def report_to_dict(report: ReviewReport) -> dict:
|
||||
"""The `--json` form. Carries the same three things the text form does -
|
||||
findings, which checks ran, which were skipped and why - so a caller never
|
||||
has to parse prose to tell a partial report from a complete one."""
|
||||
return {
|
||||
"findings": [
|
||||
{"check": finding.check, "project": finding.project, "message": finding.message}
|
||||
for finding in report.findings
|
||||
],
|
||||
"checks_run": list(report.checks_run),
|
||||
"checks_skipped": [
|
||||
{"check": check, "reason": reason} for check, reason in report.checks_skipped
|
||||
],
|
||||
"kb_project_count": report.kb_project_count,
|
||||
"complete": report.complete,
|
||||
}
|
||||
|
||||
|
||||
def review_command(
|
||||
json_out: bool = typer.Option(False, "--json", help="Print the findings as JSON."),
|
||||
):
|
||||
"""Run the weekly GTD review: join the task tracker against kb/gtd/ pages
|
||||
over the project name and report the five staleness/mismatch checks
|
||||
(#119 D10/D26). Read-only - stores nothing, not even a reports/ file
|
||||
(#119 D3), and is exempt from the Iteration Budget Gate like `search`."""
|
||||
try:
|
||||
report = run_review(config.ROOT)
|
||||
except ValidationError as exc:
|
||||
fail(str(exc))
|
||||
return
|
||||
|
||||
if json_out:
|
||||
typer.echo(json.dumps(report_to_dict(report), indent=2))
|
||||
else:
|
||||
typer.echo(render_report(report))
|
||||
|
||||
if not report.complete:
|
||||
# Printed above already - this is deliberately not fail(), which
|
||||
# would swallow the report just rendered behind a single ERROR line.
|
||||
# See chemenu.review.ReviewReport.complete: an incomplete report must
|
||||
# never exit 0 the way a quiet week does.
|
||||
raise typer.Exit(code=1)
|
||||
@@ -115,9 +115,11 @@ SKIP_COMMAND_PATHS = {
|
||||
# reading, not iterating: the budget exists to stop an agent looping over the
|
||||
# wiki's *state*, and charging for a search would penalise the one habit that
|
||||
# lowers cost - looking before reading. `doctor` is here for the same reason:
|
||||
# it only reads and reports, never mutates anything. Every command that
|
||||
# mutates anything stays counted.
|
||||
SKIP_COMMANDS = {"search", "doctor"}
|
||||
# it only reads and reports, never mutates anything. `review` (#125) joins the
|
||||
# task tracker against kb/gtd/ pages and stores nothing either (#119 D3) - the
|
||||
# same read-only argument as `search`, just over a different pair of sources.
|
||||
# Every command that mutates anything stays counted.
|
||||
SKIP_COMMANDS = {"search", "doctor", "review"}
|
||||
|
||||
|
||||
def is_exempt(command: str, args: list[str]) -> bool:
|
||||
|
||||
@@ -0,0 +1,262 @@
|
||||
"""`wikitool review` core (Gitea #125; design in #119 D3/D8/D10/D26).
|
||||
|
||||
The weekly review joins the task tracker (`chemenu.tasks`, #124) against the
|
||||
`kb/gtd/` project pages over the case-normalized project name
|
||||
(`chemenu.tasks.protocol.normalize_project_name`, #119 D8) - a join done at
|
||||
*read time* and never stored (#119 D3, the `reports/` posture: never
|
||||
re-derive, always compile, but nothing here is a compiled artifact). This
|
||||
module owns the five checks and their data flow; `chemenu.commands.review_cmd`
|
||||
owns the CLI adapter, flags and rendering.
|
||||
|
||||
Every provider read is wrapped individually so a single unreachable call
|
||||
degrades the affected checks rather than the whole report: `projects()`
|
||||
failing skips checks 1/2/3/4 (they all need the project list), `someday_items()`
|
||||
failing skips only check 5, and a single project's `open_items()` failing
|
||||
skips that one project everywhere without aborting the others. A `ReviewReport`
|
||||
with anything in `checks_skipped` is, by construction, never mistaken for a
|
||||
quiet week - see `ReviewReport.complete` and `commands.review_cmd`'s exit code.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from chemenu import config, kb_scan
|
||||
from chemenu.errors import ValidationError
|
||||
from chemenu.tasks import config as tasks_config
|
||||
from chemenu.tasks.protocol import OpenItems, TaskReader, normalize_project_name
|
||||
|
||||
CHECK_STALLED = "stalled"
|
||||
CHECK_WAITING_OVERDUE = "waiting_overdue"
|
||||
CHECK_UNPAGED_PROJECT = "unpaged_project"
|
||||
CHECK_NO_OPEN_LOOP = "no_open_loop"
|
||||
CHECK_SOMEDAY_STALE = "someday_stale"
|
||||
|
||||
# The checks that need the full tracker project list (#119 D26 checks 1/2/3/4)
|
||||
# - `projects()` failing skips all four together, since none of them can be
|
||||
# answered from `someday_items()` alone.
|
||||
_PROJECT_LIST_CHECKS = (CHECK_STALLED, CHECK_WAITING_OVERDUE, CHECK_UNPAGED_PROJECT, CHECK_NO_OPEN_LOOP)
|
||||
|
||||
ALL_CHECKS = (*_PROJECT_LIST_CHECKS, CHECK_SOMEDAY_STALE)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Finding:
|
||||
"""One reported mismatch. `project` is the display name - the kb/ page's
|
||||
title when a check is anchored on the kb/ side (checks 1 and 4, where the
|
||||
kb/ page is what the finding is about), the tracker's own project name
|
||||
otherwise (checks 2 and 3, where no kb/ page need exist)."""
|
||||
|
||||
check: str
|
||||
project: str
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReviewReport:
|
||||
findings: tuple[Finding, ...]
|
||||
checks_run: tuple[str, ...]
|
||||
checks_skipped: tuple[tuple[str, str], ...] # (check, reason)
|
||||
kb_project_count: int
|
||||
|
||||
@property
|
||||
def complete(self) -> bool:
|
||||
"""Whether every one of the five checks actually ran. `False` is the
|
||||
signal `commands.review_cmd` exits 1 on - a partial report must never
|
||||
read like a quiet week (see this module's docstring)."""
|
||||
return not self.checks_skipped
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _KbProject:
|
||||
title: str
|
||||
state: Optional[str]
|
||||
|
||||
|
||||
def _load_kb_projects(kb_dir: Path) -> dict[str, _KbProject]:
|
||||
"""Every `kb/gtd/` project page, keyed by case-normalized title (#119 D8).
|
||||
`page.kind` resolves through the type-spec's own `name:` field
|
||||
(`Page.kind`), so this finds a `project` page regardless of which
|
||||
`responsibility:` area it lives under."""
|
||||
pages = kb_scan.load_kb_pages(kb_dir)
|
||||
result: dict[str, _KbProject] = {}
|
||||
for page in pages.values():
|
||||
if page.kind != "project":
|
||||
continue
|
||||
result[normalize_project_name(page.title)] = _KbProject(
|
||||
title=page.title, state=page.frontmatter.get("state")
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _build_reader(cfg: tasks_config.TasksConfig) -> TaskReader:
|
||||
"""Dispatch on `cfg.provider` - the same inline shape
|
||||
`doctor.check_tasks_provider` uses, kept in sync with
|
||||
`tasks_config.KNOWN_PROVIDERS`."""
|
||||
if cfg.provider == "superproductivity":
|
||||
from chemenu.tasks import superproductivity as sp
|
||||
|
||||
sp_cfg = sp.SuperProductivityConfig.from_dict(cfg.provider_config)
|
||||
return sp.SuperProductivityReader(sp_cfg)
|
||||
raise ValidationError(f"No reader is wired up for task provider {cfg.provider!r}.")
|
||||
|
||||
|
||||
def _weeks_between(start: date, end: date) -> float:
|
||||
return (end - start).days / 7
|
||||
|
||||
|
||||
def _months_between(start: date, end: date) -> int:
|
||||
"""Whole calendar months between two dates, floored - `age_months` for
|
||||
check 5. Deliberately calendar-based rather than `days / 30`: an
|
||||
approximation would drift a fixed-days threshold away from what "N months"
|
||||
actually means at the boundary."""
|
||||
months = (end.year - start.year) * 12 + (end.month - start.month)
|
||||
if end.day < start.day:
|
||||
months -= 1
|
||||
return months
|
||||
|
||||
|
||||
def run_review(root: Path, *, today: Optional[date] = None) -> ReviewReport:
|
||||
"""Run the five weekly-review checks (#119 D10/D26) against the instance
|
||||
rooted at `root` and return their findings. Stores nothing (#119 D3): every
|
||||
provider read is a fresh call, and no file under `root` is touched.
|
||||
|
||||
Raises `ValidationError` for anything that is not "the provider is
|
||||
unreachable right now" - no `.wikitool-tasks.json` at all, or a
|
||||
malformed one. Those are configuration problems a retry cannot fix, so
|
||||
`commands.review_cmd` reports them as an ordinary exit-1 error rather than
|
||||
a partial report.
|
||||
"""
|
||||
today = today if today is not None else date.today()
|
||||
cfg = tasks_config.read_config(root)
|
||||
if cfg is None:
|
||||
raise ValidationError(
|
||||
f"No {config.TASKS_CONFIG_FILENAME} - no task tracker is configured, so there is "
|
||||
"nothing to join the weekly review against. Configure one first."
|
||||
)
|
||||
reader = _build_reader(cfg)
|
||||
kb_projects = _load_kb_projects(Path(root) / "kb")
|
||||
|
||||
findings: list[Finding] = []
|
||||
checks_run: list[str] = []
|
||||
checks_skipped: list[tuple[str, str]] = []
|
||||
|
||||
try:
|
||||
tracker_projects = reader.projects()
|
||||
except ValidationError as exc:
|
||||
reason = str(exc)
|
||||
checks_skipped.extend((check, reason) for check in _PROJECT_LIST_CHECKS)
|
||||
tracker_projects = None
|
||||
|
||||
if tracker_projects is not None:
|
||||
open_items_by_key: dict[str, tuple[str, OpenItems]] = {}
|
||||
broken_keys: set[str] = set()
|
||||
for project in tracker_projects:
|
||||
key = normalize_project_name(project.name)
|
||||
try:
|
||||
open_items_by_key[key] = (project.name, reader.open_items(project.name))
|
||||
except ValidationError:
|
||||
broken_keys.add(key)
|
||||
|
||||
# Check 1 - stalled: a tracker project with zero open items whose kb/
|
||||
# page is `state: active` (#119 D27's anti-noise core: dormant,
|
||||
# completed and abandoned never reach here).
|
||||
for project in tracker_projects:
|
||||
key = normalize_project_name(project.name)
|
||||
if key in broken_keys:
|
||||
continue
|
||||
_, items = open_items_by_key[key]
|
||||
if items.count != 0:
|
||||
continue
|
||||
kb = kb_projects.get(key)
|
||||
if kb is not None and kb.state == "active":
|
||||
findings.append(Finding(
|
||||
CHECK_STALLED, kb.title,
|
||||
f"Tracker project '{project.name}' has zero open items and the kb/ page "
|
||||
"is state: active.",
|
||||
))
|
||||
checks_run.append(CHECK_STALLED)
|
||||
|
||||
# Check 2 - waiting-for overdue: any WAITING item whose follow_up_at
|
||||
# is older than the threshold (#119 D9/D30 - never the due date).
|
||||
for _key, (name, items) in open_items_by_key.items():
|
||||
for item in items.waiting:
|
||||
if item.follow_up_at is None:
|
||||
continue
|
||||
age_days = (today - item.follow_up_at).days
|
||||
if age_days > cfg.thresholds.stalled_waiting_days:
|
||||
findings.append(Finding(
|
||||
CHECK_WAITING_OVERDUE, name,
|
||||
f"'{item.title}' is {age_days} day(s) past its follow_up_at "
|
||||
f"({item.follow_up_at.isoformat()}).",
|
||||
))
|
||||
checks_run.append(CHECK_WAITING_OVERDUE)
|
||||
|
||||
# Check 3 - tracker project with no kb/ page, older than the
|
||||
# threshold (#119 D26's noise brake: an age threshold, not a marker).
|
||||
for project in tracker_projects:
|
||||
key = normalize_project_name(project.name)
|
||||
if key in kb_projects or project.created is None:
|
||||
continue
|
||||
age_weeks = _weeks_between(project.created, today)
|
||||
if age_weeks > cfg.thresholds.unpaged_project_weeks:
|
||||
findings.append(Finding(
|
||||
CHECK_UNPAGED_PROJECT, project.name,
|
||||
f"No kb/ page for this tracker project after {age_weeks:.1f} week(s) "
|
||||
f"(created {project.created.isoformat()}).",
|
||||
))
|
||||
checks_run.append(CHECK_UNPAGED_PROJECT)
|
||||
|
||||
# Check 4 - kb/ page with no open loop: `state: active` but either no
|
||||
# tracker project of this name exists, or it has zero open items. This
|
||||
# is the reverse direction of check 3's join (#119 D8's beidseitig
|
||||
# unmatched report - a rename on either side must surface somewhere).
|
||||
tracker_by_key = {normalize_project_name(p.name): p for p in tracker_projects}
|
||||
for key, kb in kb_projects.items():
|
||||
if kb.state != "active":
|
||||
continue
|
||||
project = tracker_by_key.get(key)
|
||||
if project is None:
|
||||
findings.append(Finding(
|
||||
CHECK_NO_OPEN_LOOP, kb.title,
|
||||
"state: active, but no tracker project of this name exists.",
|
||||
))
|
||||
continue
|
||||
if key in broken_keys:
|
||||
continue
|
||||
_, items = open_items_by_key[key]
|
||||
if items.count == 0:
|
||||
findings.append(Finding(
|
||||
CHECK_NO_OPEN_LOOP, kb.title,
|
||||
f"state: active, but tracker project '{project.name}' has zero open items.",
|
||||
))
|
||||
checks_run.append(CHECK_NO_OPEN_LOOP)
|
||||
|
||||
# Check 5 - someday/maybe items untouched for longer than the threshold.
|
||||
# Independent of the tracker project list, so it still runs when
|
||||
# `projects()` above failed but `someday_items()` does not.
|
||||
try:
|
||||
someday = reader.someday_items()
|
||||
except ValidationError as exc:
|
||||
checks_skipped.append((CHECK_SOMEDAY_STALE, str(exc)))
|
||||
else:
|
||||
for item in someday:
|
||||
if item.modified is None:
|
||||
continue
|
||||
age_months = _months_between(item.modified, today)
|
||||
if age_months > cfg.thresholds.someday_stale_months:
|
||||
findings.append(Finding(
|
||||
CHECK_SOMEDAY_STALE, item.title,
|
||||
f"Untouched for {age_months} month(s) (last modified "
|
||||
f"{item.modified.isoformat()}).",
|
||||
))
|
||||
checks_run.append(CHECK_SOMEDAY_STALE)
|
||||
|
||||
return ReviewReport(
|
||||
findings=tuple(findings),
|
||||
checks_run=tuple(checks_run),
|
||||
checks_skipped=tuple(checks_skipped),
|
||||
kb_project_count=len(kb_projects),
|
||||
)
|
||||
@@ -0,0 +1,357 @@
|
||||
"""Tests for `chemenu.review` and `wikitool review` (Gitea #125).
|
||||
|
||||
`run_review` takes its root as an explicit argument and never touches
|
||||
`config.ROOT`, so these tests build a tree under `tmp_path` without needing
|
||||
`kb_dir`/`use_shipped_type_specs` - the real shipped `types/project.md` is
|
||||
what `config.ROOT` already falls back to under the hermetic fixture (see
|
||||
`instructions/dev/testing-conventions.md`), and that is the schema this
|
||||
feature is meant to be exercised against, not a synthetic stand-in.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
from datetime import date, datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from chemenu.cli import app
|
||||
from chemenu.commands.run_budget import is_exempt
|
||||
from chemenu.errors import ValidationError
|
||||
from chemenu.frontmatter_io import write_page
|
||||
from chemenu.tests.conftest import use_shipped_type_specs
|
||||
from chemenu.review import (
|
||||
CHECK_NO_OPEN_LOOP,
|
||||
CHECK_SOMEDAY_STALE,
|
||||
CHECK_STALLED,
|
||||
CHECK_UNPAGED_PROJECT,
|
||||
CHECK_WAITING_OVERDUE,
|
||||
run_review,
|
||||
)
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
TODAY = date(2026, 9, 19)
|
||||
|
||||
THRESHOLDS = {
|
||||
"stalled_waiting_days": 14,
|
||||
"unpaged_project_weeks": 3,
|
||||
"someday_stale_months": 5,
|
||||
}
|
||||
|
||||
|
||||
def _ms(year: int, month: int, day: int) -> int:
|
||||
return int(datetime(year, month, day, tzinfo=timezone.utc).timestamp() * 1000)
|
||||
|
||||
|
||||
def _entity_state(records: dict[str, dict]) -> dict:
|
||||
return {"ids": list(records.keys()), "entities": records}
|
||||
|
||||
|
||||
def _write_snapshot(root: Path, projects: dict, tasks: dict, tags: dict) -> Path:
|
||||
db_path = root / "db.json"
|
||||
db_path.write_text(
|
||||
json.dumps({
|
||||
"project": _entity_state(projects),
|
||||
"task": _entity_state(tasks),
|
||||
"tag": _entity_state(tags),
|
||||
}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
return db_path
|
||||
|
||||
|
||||
def _write_tasks_config(root: Path, db_path: Path, thresholds: dict | None = None) -> None:
|
||||
(root / ".wikitool-tasks.json").write_text(
|
||||
json.dumps({
|
||||
"schema": 1,
|
||||
"provider": "superproductivity",
|
||||
"thresholds": thresholds or THRESHOLDS,
|
||||
"superproductivity": {"db_path": str(db_path)},
|
||||
}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
|
||||
def _project_page(root: Path, name: str, state: str, *, area: str = "haus") -> None:
|
||||
write_page(
|
||||
root / "kb" / "gtd" / area / f"{name}.md",
|
||||
{
|
||||
"type": "types/project.md",
|
||||
"state": state,
|
||||
"responsibility": area,
|
||||
"created": "2026-01-01",
|
||||
"modified": "2026-01-01",
|
||||
"provenance": "general",
|
||||
"summary": f"Fixture project {name}.",
|
||||
},
|
||||
f"\n# {name}\n\n## Ziel\n\nFixture.\n",
|
||||
)
|
||||
|
||||
|
||||
def _tree(root: Path) -> dict[str, bytes]:
|
||||
return {
|
||||
str(p.relative_to(root)): p.read_bytes()
|
||||
for p in root.rglob("*") if p.is_file()
|
||||
}
|
||||
|
||||
|
||||
# --- check 1: stalled --------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("state,should_fire", [
|
||||
("active", True),
|
||||
("dormant", False),
|
||||
("completed", False),
|
||||
("abandoned", False),
|
||||
])
|
||||
def test_check1_stalled_only_fires_for_active(tmp_path, state, should_fire):
|
||||
projects = {"p1": {"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
|
||||
"taskIds": [], "backlogTaskIds": []}}
|
||||
db_path = _write_snapshot(tmp_path, projects, {}, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Ship Chemenu 7.0", state)
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
fired = any(f.check == CHECK_STALLED for f in report.findings)
|
||||
assert fired == should_fire
|
||||
|
||||
|
||||
# --- check 2: waiting overdue ------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("remind_day,should_fire", [
|
||||
(1, True), # 2026-01-01 -> far more than 14 days before TODAY
|
||||
(10, False), # 2026-09-10 -> 9 days before TODAY, under the threshold
|
||||
])
|
||||
def test_check2_waiting_overdue_threshold(tmp_path, remind_day, should_fire):
|
||||
month = 1 if remind_day == 1 else 9
|
||||
projects = {"p1": {"id": "p1", "title": "Kueche renovieren", "created": _ms(2026, 1, 1),
|
||||
"taskIds": ["t1"], "backlogTaskIds": []}}
|
||||
tasks = {"t1": {"id": "t1", "title": "Warte auf Angebot - Tobias", "isDone": False,
|
||||
"tagIds": ["tag-wait"], "remindAt": _ms(2026, month, remind_day)}}
|
||||
tags = {"tag-wait": {"id": "tag-wait", "title": "waiting"}}
|
||||
db_path = _write_snapshot(tmp_path, projects, tasks, tags)
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Kueche renovieren", "active")
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
fired = any(f.check == CHECK_WAITING_OVERDUE for f in report.findings)
|
||||
assert fired == should_fire
|
||||
|
||||
|
||||
# --- check 3: unpaged tracker project ----------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("created_year_month_day,should_fire", [
|
||||
((2026, 9, 15), False), # 4 days old, well under 3 weeks
|
||||
((2026, 8, 1), True), # ~7 weeks old
|
||||
])
|
||||
def test_check3_unpaged_project_age_threshold(tmp_path, created_year_month_day, should_fire):
|
||||
y, m, d = created_year_month_day
|
||||
projects = {"p1": {"id": "p1", "title": "No Page Yet", "created": _ms(y, m, d),
|
||||
"taskIds": [], "backlogTaskIds": []}}
|
||||
db_path = _write_snapshot(tmp_path, projects, {}, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
(tmp_path / "kb" / "gtd").mkdir(parents=True)
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
fired = any(f.check == CHECK_UNPAGED_PROJECT for f in report.findings)
|
||||
assert fired == should_fire
|
||||
|
||||
|
||||
def test_check3_and_check2_are_case_normalized_and_report_no_mismatch(tmp_path):
|
||||
"""'Kueche renovieren' and 'kueche renovieren' are the same project (#119
|
||||
D8) - no unpaged/no-open-loop finding from the case difference alone."""
|
||||
projects = {"p1": {"id": "p1", "title": "kueche renovieren", "created": _ms(2020, 1, 1),
|
||||
"taskIds": ["t1"], "backlogTaskIds": []}}
|
||||
tasks = {"t1": {"id": "t1", "title": "Irgendwas tun", "isDone": False, "tagIds": []}}
|
||||
db_path = _write_snapshot(tmp_path, projects, tasks, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Kueche renovieren", "active")
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
assert not any(f.check == CHECK_UNPAGED_PROJECT for f in report.findings)
|
||||
assert not any(f.check == CHECK_NO_OPEN_LOOP for f in report.findings)
|
||||
|
||||
|
||||
# --- check 4: kb/ page with no open loop -------------------------------------
|
||||
|
||||
|
||||
def test_check4_fires_when_no_tracker_project_exists(tmp_path):
|
||||
db_path = _write_snapshot(tmp_path, {}, {}, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Ghost Project", "active")
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
findings = [f for f in report.findings if f.check == CHECK_NO_OPEN_LOOP]
|
||||
assert len(findings) == 1
|
||||
assert findings[0].project == "Ghost Project"
|
||||
|
||||
|
||||
# --- run isolation and file mutation -----------------------------------------
|
||||
|
||||
|
||||
def test_a_run_touches_no_file(tmp_path):
|
||||
projects = {"p1": {"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
|
||||
"taskIds": [], "backlogTaskIds": ["t3"]}}
|
||||
tasks = {"t3": {"id": "t3", "title": "Someday item", "isDone": False, "tagIds": [],
|
||||
"updated": _ms(2020, 1, 1)}}
|
||||
db_path = _write_snapshot(tmp_path, projects, tasks, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Ship Chemenu 7.0", "active")
|
||||
|
||||
before = _tree(tmp_path)
|
||||
run_review(tmp_path, today=TODAY)
|
||||
assert _tree(tmp_path) == before
|
||||
|
||||
|
||||
# --- check 5: someday stale ---------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("updated_ymd,should_fire", [
|
||||
((2026, 8, 1), False), # ~1.5 months old
|
||||
((2025, 1, 1), True), # well over 5 months old
|
||||
])
|
||||
def test_check5_someday_stale_threshold(tmp_path, updated_ymd, should_fire):
|
||||
y, m, _d = updated_ymd
|
||||
projects = {"p1": {"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
|
||||
"taskIds": [], "backlogTaskIds": ["t3"]}}
|
||||
tasks = {"t3": {"id": "t3", "title": "Irgendwann Keller aufraeumen", "isDone": False,
|
||||
"tagIds": [], "updated": _ms(y, m, 1)}}
|
||||
db_path = _write_snapshot(tmp_path, projects, tasks, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
fired = any(f.check == CHECK_SOMEDAY_STALE for f in report.findings)
|
||||
assert fired == should_fire
|
||||
|
||||
|
||||
# --- provider/configuration errors -------------------------------------------
|
||||
|
||||
|
||||
def test_no_tasks_config_is_a_clear_validation_error(tmp_path):
|
||||
(tmp_path / "kb" / "gtd").mkdir(parents=True)
|
||||
with pytest.raises(ValidationError, match="no task tracker is configured"):
|
||||
run_review(tmp_path, today=TODAY)
|
||||
|
||||
|
||||
def test_unreachable_provider_skips_the_project_list_checks_but_not_someday(tmp_path):
|
||||
"""A db_path that does not exist is the unreachable-provider case: checks
|
||||
1/2/3/4 cannot run at all, but check 5 uses the same failing read and is
|
||||
skipped too - the report must say so, never look like a quiet week."""
|
||||
_write_tasks_config(tmp_path, tmp_path / "does-not-exist.json")
|
||||
_project_page(tmp_path, "Ship Chemenu 7.0", "active")
|
||||
|
||||
report = run_review(tmp_path, today=TODAY)
|
||||
assert not report.complete
|
||||
skipped_checks = {check for check, _reason in report.checks_skipped}
|
||||
assert skipped_checks == {
|
||||
CHECK_STALLED, CHECK_WAITING_OVERDUE, CHECK_UNPAGED_PROJECT,
|
||||
CHECK_NO_OPEN_LOOP, CHECK_SOMEDAY_STALE,
|
||||
}
|
||||
assert report.kb_project_count == 1
|
||||
assert report.findings == ()
|
||||
|
||||
|
||||
# --- CLI: exit codes, --json, budget exemption -------------------------------
|
||||
|
||||
|
||||
def test_cli_no_provider_exits_1_with_a_clear_message(tmp_path, monkeypatch):
|
||||
from chemenu import config
|
||||
|
||||
monkeypatch.setattr(config, "ROOT", tmp_path)
|
||||
use_shipped_type_specs(monkeypatch)
|
||||
(tmp_path / "kb" / "gtd").mkdir(parents=True)
|
||||
result = runner.invoke(app, ["review"])
|
||||
assert result.exit_code == 1
|
||||
assert "no task tracker is configured" in result.output
|
||||
|
||||
|
||||
def test_cli_incomplete_report_exits_1_and_still_prints(tmp_path, monkeypatch):
|
||||
from chemenu import config
|
||||
|
||||
monkeypatch.setattr(config, "ROOT", tmp_path)
|
||||
use_shipped_type_specs(monkeypatch)
|
||||
_write_tasks_config(tmp_path, tmp_path / "does-not-exist.json")
|
||||
(tmp_path / "kb" / "gtd").mkdir(parents=True)
|
||||
|
||||
result = runner.invoke(app, ["review"])
|
||||
assert result.exit_code == 1
|
||||
assert "INCOMPLETE" in result.output
|
||||
|
||||
|
||||
def test_cli_json_and_text_agree_on_findings(tmp_path, monkeypatch):
|
||||
"""The golden check (#125's own AC): the `--json` findings and the
|
||||
text-rendered findings must name exactly the same (check, project) pairs -
|
||||
the same posture `test_mcp_server.py` holds the MCP wire format to
|
||||
against the CLI's own `--json`."""
|
||||
from chemenu import config
|
||||
|
||||
monkeypatch.setattr(config, "ROOT", tmp_path)
|
||||
use_shipped_type_specs(monkeypatch)
|
||||
projects = {"p1": {"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
|
||||
"taskIds": [], "backlogTaskIds": []}}
|
||||
db_path = _write_snapshot(tmp_path, projects, {}, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Ship Chemenu 7.0", "active")
|
||||
|
||||
text_result = runner.invoke(app, ["review"])
|
||||
json_result = runner.invoke(app, ["review", "--json"])
|
||||
assert text_result.exit_code == 0
|
||||
assert json_result.exit_code == 0
|
||||
|
||||
payload = json.loads(json_result.output)
|
||||
json_pairs = {(f["check"], f["project"]) for f in payload["findings"]}
|
||||
|
||||
import re
|
||||
|
||||
text_pairs = {
|
||||
(m.group(1), m.group(2))
|
||||
for m in re.finditer(r"^\[(\S+)\] ([^:]+):", text_result.output, re.MULTILINE)
|
||||
}
|
||||
# A tracker project with zero open items and an active kb/ page satisfies
|
||||
# both check 1's and check 4's condition (#125's table: check 4's "no
|
||||
# open items" branch is not exclusive of check 1) - both fire.
|
||||
assert json_pairs == text_pairs == {
|
||||
(CHECK_STALLED, "Ship Chemenu 7.0"),
|
||||
(CHECK_NO_OPEN_LOOP, "Ship Chemenu 7.0"),
|
||||
}
|
||||
|
||||
|
||||
def test_review_is_exempt_from_the_iteration_budget():
|
||||
assert is_exempt("review", []) is True
|
||||
assert is_exempt("review", ["--json"]) is True
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
subprocess.run(["git", "--version"], capture_output=True).returncode != 0,
|
||||
reason="git not available",
|
||||
)
|
||||
def test_a_run_leaves_the_git_tree_untouched(tmp_path, monkeypatch):
|
||||
from chemenu import config
|
||||
|
||||
subprocess.run(["git", "init", "-q", "-b", "main"], cwd=tmp_path, check=True)
|
||||
subprocess.run(["git", "config", "user.name", "Fixture Author"], cwd=tmp_path, check=True)
|
||||
subprocess.run(["git", "config", "user.email", "fixture@example.com"], cwd=tmp_path, check=True)
|
||||
|
||||
projects = {"p1": {"id": "p1", "title": "Ship Chemenu 7.0", "created": _ms(2026, 1, 1),
|
||||
"taskIds": [], "backlogTaskIds": []}}
|
||||
db_path = _write_snapshot(tmp_path, projects, {}, {})
|
||||
_write_tasks_config(tmp_path, db_path)
|
||||
_project_page(tmp_path, "Ship Chemenu 7.0", "active")
|
||||
subprocess.run(["git", "add", "-A"], cwd=tmp_path, check=True)
|
||||
subprocess.run(["git", "commit", "-q", "-m", "fixture"], cwd=tmp_path, check=True)
|
||||
|
||||
monkeypatch.setattr(config, "ROOT", tmp_path)
|
||||
use_shipped_type_specs(monkeypatch)
|
||||
before = subprocess.run(
|
||||
["git", "status", "--porcelain"], cwd=tmp_path, capture_output=True, text=True, check=True
|
||||
).stdout
|
||||
runner.invoke(app, ["review", "--json"])
|
||||
after = subprocess.run(
|
||||
["git", "status", "--porcelain"], cwd=tmp_path, capture_output=True, text=True, check=True
|
||||
).stdout
|
||||
assert before == after == ""
|
||||
Reference in new issue
Block a user