"""The live tracker suite (Gitea #156) and the proof that its safety rules hold. The `live_tracker` tests need a tracker and skip without one; the rest of this file never does. Which tracker, how to start one, and what to do with a red run: `instructions/dev/tracker-testing.md`. """ from __future__ import annotations import contextlib import http.server import json import threading from datetime import date from pathlib import Path import pytest from typer.testing import CliRunner from chemenu import config_shapes, settings from chemenu.cli import app from chemenu.review import CHECK_WAITING_OVERDUE, run_review from chemenu.frontmatter_io import write_page from chemenu.tasks.protocol import ( OpenItems, ProjectSummary, ReadSource, SomedayItem, ) from chemenu.tests import sp_headless from chemenu.tests import tracker_live as live from chemenu.tests.tracker_live import LiveAbort, LiveTarget runner = CliRunner() # --- the live scenario --------------------------------------------------------------------- @pytest.fixture(scope="session") def live_pool(): pool: dict[str, "LiveTarget | None"] = {} yield pool for target in pool.values(): if target is not None: target.cleanup() @pytest.fixture(params=live.KINDS) def target(request, live_pool) -> LiveTarget: kind = request.param if kind not in live_pool: try: live_pool[kind] = live.build_target(kind) except LiveAbort as exc: pytest.fail(f"live tracker '{kind}' refused to start: {exc}") found = live_pool[kind] if found is None: if kind in live.required_kinds(): pytest.fail(f"CHEMENU_LIVE_REQUIRE names '{kind}' but no such tracker is configured") pytest.skip(f"no live '{kind}' tracker configured (instructions/dev/tracker-testing.md)") return found def _project_page(root: Path, name: str) -> None: write_page( root / "kb" / "gtd" / "technik" / f"{name}.md", { "type": "types/project.md", "state": "active", "responsibility": "technik", "created": "2026-01-01", "modified": "2026-01-01", "provenance": "general", "summary": f"Live-test project {name}.", }, f"\n# {name}\n\n## Ziel\n\nLive test.\n", ) def _provisioner(target: LiveTarget): """Only a tracker the suite owns end to end gets its marker created for it.""" if target.kind != "caldav": return None def provision() -> None: reader = live.reader_for(target.config) from chemenu.tasks import caldav as cd cfg = cd.CalDAVConfig.from_section( settings.Section.from_data(config_shapes.CALDAV, target.config["caldav"])) writer = cd.CalDAVWriter(cfg, reader) for name in (live.MARKER_PROJECT, cfg.inbox_list, cfg.someday_list): if name not in {p.name for p in reader.projects()}: writer.create_project(name) return provision def _open_titles(reader) -> dict[str, str]: return {item.title: item.id for item in reader.open_items(live.MARKER_PROJECT).items} @pytest.mark.live_tracker def test_the_documented_workflow_runs_against_the_real_tracker( target, kb_dir, tmp_path, monkeypatch, record_property, capsys, ): root = kb_dir.parent profile = tmp_path / "live-profile.json" profile.write_text(json.dumps(target.config), encoding="utf-8") assert "plain" not in profile.read_text(encoding="utf-8") monkeypatch.setenv("WIKITOOL_TASKS_CONFIG", str(profile)) live.export_secrets(target, monkeypatch.setenv) with capsys.disabled(): print(f"\n[tracker-live] {target.label} - version {target.version} - run {live.RUN_ID}") record_property("tracker_kind", target.kind) record_property("tracker_version", target.version) reader = live.reader_for(target.config) live.guard(reader, provision=_provisioner(target)) _project_page(root, live.MARKER_PROJECT) open_title, waiting_title = live.run_id_titles() overdue = live.overdue_day(date.today()) try: created = runner.invoke(app, ["task", "new", "--project", live.MARKER_PROJECT, "--title", open_title]) assert created.exit_code == 0, created.output created = runner.invoke(app, ["task", "new", "--project", live.MARKER_PROJECT, "--title", waiting_title, "--waiting", "--follow-up-at", overdue.isoformat()]) assert created.exit_code == 0, created.output listing = runner.invoke(app, ["task", "list", "--project", live.MARKER_PROJECT]) assert listing.exit_code == 0, listing.output assert open_title in listing.output and waiting_title in listing.output titles = _open_titles(reader) assert open_title in titles and waiting_title in titles waiting = {w.title: w for w in reader.open_items(live.MARKER_PROJECT).waiting} assert waiting_title in waiting and open_title not in waiting assert waiting[waiting_title].follow_up_at == overdue report = run_review(root, today=date.today()) assert any( f.check == CHECK_WAITING_OVERDUE and f.item_id == titles[waiting_title] for f in report.findings ), [f for f in report.findings] for title in (open_title, waiting_title): closed = runner.invoke(app, ["task", "close", "--id", titles[title]]) assert closed.exit_code == 0, closed.output assert not {open_title, waiting_title} & set(_open_titles(reader)) finally: for title, item_id in _open_titles(reader).items(): if title.startswith(live.ITEM_PREFIX): runner.invoke(app, ["task", "close", "--id", item_id]) # --- the safety rules, against fakes ------------------------------------------------------- class _RecordingReader: """A tracker that knows the listed projects and counts every call that could write.""" def __init__(self, names: list[str]): self._names = names self.writes = 0 def projects(self): return [ProjectSummary(name=n, created=None) for n in self._names] def open_items(self, project_name): return OpenItems(count=0, waiting=(), items=()) def someday_items(self) -> list[SomedayItem]: return [] def source(self): return ReadSource(kind="fake", detail="fake") def test_guard_aborts_when_the_marker_project_is_missing_and_nothing_was_written(): reader = _RecordingReader(["Inbox", "Somebody's real project"]) with pytest.raises(LiveAbort, match="Chemenu Live-Test"): live.guard(reader) assert reader.writes == 0 def test_guard_accepts_the_marker_in_any_case_and_spacing(): live.guard(_RecordingReader([" chemenu LIVE-test "])) def test_guard_provisions_only_when_a_provisioner_is_given(): names: list[str] = [] reader = _RecordingReader(names) def provision(): names.append(live.MARKER_PROJECT) live.guard(reader, provision=provision) assert names == [live.MARKER_PROJECT] with pytest.raises(LiveAbort): live.guard(_RecordingReader([])) @contextlib.contextmanager def _answering_health_server(): class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): # noqa: N802 body = json.dumps({"ok": True, "data": {"server": "up", "rendererReady": True}}).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *args): # noqa: D401 pass server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: yield server.server_address[1] finally: server.shutdown() thread.join(timeout=5) def test_start_refuses_when_the_api_port_already_answers(tmp_path): executable = tmp_path / "superproductivity" executable.write_text("#!/bin/sh\nexit 0\n") executable.chmod(0o755) with _answering_health_server() as port: with pytest.raises(LiveAbort, match="already answers"): sp_headless.start(executable, tmp_path / "profile", marker_project="x", port=port) assert not (tmp_path / "profile").exists() def test_a_profile_with_a_home_relative_path_is_refused(tmp_path, monkeypatch): monkeypatch.setattr(live, "REPO_ROOT", tmp_path) (tmp_path / live.PROFILE_DIR).mkdir() (tmp_path / live.PROFILE_DIR / "mine.json").write_text(json.dumps({ "schema": 1, "tasks": {"provider": "superproductivity", **live.THRESHOLDS}, "superproductivity": {"access": "snapshot", "backups_dir": "~/.config/sp/backups"}, })) with pytest.raises(LiveAbort, match="absolute"): live.load_profile("mine") @pytest.mark.parametrize("name", ["", "../escape", "a/b", ".hidden"]) def test_a_profile_name_cannot_leave_the_profile_directory(name): with pytest.raises(LiveAbort, match="bare profile name"): live.profile_path(name) def test_a_profile_is_read_exactly_as_written(tmp_path, monkeypatch): monkeypatch.setattr(live, "REPO_ROOT", tmp_path) (tmp_path / live.PROFILE_DIR).mkdir() (tmp_path / live.PROFILE_DIR / "mine.json").write_text(json.dumps({ "schema": 1, "tasks": {"provider": "superproductivity", **live.THRESHOLDS}, "superproductivity": {"access": "snapshot", "backups_dir": str(tmp_path)}, "live_test_version": "19.1.0", })) target = live.load_profile("mine") assert target.version == "19.1.0" assert "live_test_version" not in target.config def test_required_kinds_turn_a_missing_tracker_into_a_failure(): assert live.required_kinds({"CHEMENU_LIVE_REQUIRE": "caldav, sp"}) == {"caldav", "sp"} assert live.required_kinds({}) == set() assert live.build_target("caldav", {}) is None assert live.build_target("sp", {}) is None assert live.build_target("profile", {}) is None def test_caldav_target_needs_both_credentials(): with pytest.raises(LiveAbort, match="USER"): live.caldav_target_from_env({"CHEMENU_LIVE_CALDAV_URL": "http://127.0.0.1:5232/u/"}) target = live.caldav_target_from_env({ "CHEMENU_LIVE_CALDAV_URL": "http://127.0.0.1:5232/u/", "CHEMENU_LIVE_CALDAV_USER": "u", "CHEMENU_LIVE_CALDAV_PASSWORD": "p", }) assert target is not None and target.config["tasks"]["provider"] == "caldav" def test_item_titles_carry_the_run_prefix(): for title in live.run_id_titles(): assert title.startswith(live.ITEM_PREFIX) assert live.ITEM_PREFIX.startswith("[live-test ")