"""Tests for C-33 (generated per-repo work-record index, CUST-WP-0061-T04): _generate_work_record_index() and _check_work_record_index_freshness(). Offline, synthetic registry fixture — same conventions as test_work_record_check.py / test_work_record_registration.py. """ from __future__ import annotations import sys import textwrap from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) from consistency_check import ( # noqa: E402 ConsistencyReport, _check_work_record_index_freshness, _generate_work_record_index, _work_record_index_document, ) REGISTRY_YAML = textwrap.dedent( """ version: "0.1" kinds: - kind: workplan id_patterns: ['^[A-Z]+-WP-[0-9]{4}$'] - kind: task id_patterns: ['^[A-Z]+-WP-[0-9]{4}-T[0-9]{2,3}$'] - kind: intake id_patterns: ['^[A-Z]+-IN-[0-9]{4}$'] legacy_patterns: - pattern: '^AWQ-[0-9]{3}$' source: binky AutopilotWorkQueue - kind: decision id_patterns: ['^[A-Z]+-DEC-[0-9]{4}-[0-9]{3}$'] - kind: engagement id_patterns: ['^[A-Z]+-ENG-[0-9]{4}-[0-9]{3}$'] """ ) @pytest.fixture def registry_file(tmp_path, monkeypatch): reg = tmp_path / "work-record-types.yaml" reg.write_text(REGISTRY_YAML, encoding="utf-8") monkeypatch.setenv("WORK_RECORD_REGISTRY", str(reg)) return reg class TestGenerateWorkRecordIndex: def test_empty_repo_returns_none(self, tmp_path, registry_file): assert _generate_work_record_index(tmp_path, "testrepo") is None def test_includes_workplan_and_tasks(self, tmp_path, registry_file): wp_dir = tmp_path / "workplans" wp_dir.mkdir() (wp_dir / "CUST-WP-0001-demo.md").write_text( textwrap.dedent( """\ --- id: CUST-WP-0001 status: active --- ```task id: CUST-WP-0001-T01 status: done priority: high ``` """ ), encoding="utf-8", ) content = _generate_work_record_index(tmp_path, "testrepo") assert content is not None assert "| workplan | CUST-WP-0001 | active | — | workplans/CUST-WP-0001-demo.md |" in content assert "| task | CUST-WP-0001-T01 | done | — |" in content def test_excludes_archived_workplans(self, tmp_path, registry_file): wp_dir = tmp_path / "workplans" / "archived" wp_dir.mkdir(parents=True) (wp_dir / "260101-CUST-WP-0000-old.md").write_text( "---\nid: CUST-WP-0000\nstatus: finished\n---\n", encoding="utf-8" ) assert _generate_work_record_index(tmp_path, "testrepo") is None def test_includes_intake_decision_engagement(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text( textwrap.dedent( """ ```yaml id: AWQ-010 title: qonto mcp status: open lane: green ``` ```yaml id: BINKY-DEC-2026-004 title: qonto approval status: resolved lane: red ``` ```yaml id: BINKY-ENG-2026-001 title: bank call status: queued ``` """ ), encoding="utf-8", ) content = _generate_work_record_index(tmp_path, "testrepo") assert "| intake | AWQ-010 | open | green | queue.md |" in content assert "| decision | BINKY-DEC-2026-004 | resolved | red | queue.md |" in content assert "| engagement | BINKY-ENG-2026-001 | queued | — | queue.md |" in content def test_ignores_unregistered_ids(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text( "```yaml\nid: FOO-QX-001\nstatus: open\n```\n", encoding="utf-8" ) assert _generate_work_record_index(tmp_path, "testrepo") is None def test_ignores_template_placeholders(self, tmp_path, registry_file): (tmp_path / "AGENTS.md").write_text( "```yaml\nid: BINKY-IN-NNNN\nstatus: open\n```\n", encoding="utf-8" ) assert _generate_work_record_index(tmp_path, "testrepo") is None def test_does_not_scan_its_own_output_file(self, tmp_path, registry_file): (tmp_path / "WORK-RECORDS.md").write_text( "| intake | AWQ-999 | open | green | ghost.md |\n", encoding="utf-8" ) assert _generate_work_record_index(tmp_path, "testrepo") is None def test_header_and_sort_order(self, tmp_path, registry_file): (tmp_path / "b.md").write_text("```yaml\nid: AWQ-002\nstatus: open\n```\n", encoding="utf-8") (tmp_path / "a.md").write_text("```yaml\nid: AWQ-001\nstatus: open\n```\n", encoding="utf-8") content = _generate_work_record_index(tmp_path, "testrepo") assert content.startswith("# Work Records — testrepo") assert "[auto]" in content # sorted by source path first (a.md before b.md) assert content.index("AWQ-001") < content.index("AWQ-002") class TestCheckWorkRecordIndexFreshness: def _report(self, repo_dir: Path, repo_slug: str = "testrepo") -> ConsistencyReport: report = ConsistencyReport(repo_slug=repo_slug, repo_path=str(repo_dir)) _check_work_record_index_freshness(repo_dir, repo_slug, report) return report def test_no_issue_when_no_records(self, tmp_path, registry_file): report = self._report(tmp_path) assert report.issues == [] def test_flags_missing_index(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text("```yaml\nid: AWQ-001\nstatus: open\n```\n", encoding="utf-8") report = self._report(tmp_path) assert len(report.issues) == 1 assert report.issues[0].check_id == "C-33" assert report.issues[0].fixable is True def test_no_issue_when_index_already_current(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text("```yaml\nid: AWQ-001\nstatus: open\n```\n", encoding="utf-8") content = _generate_work_record_index(tmp_path, "testrepo") (tmp_path / "WORK-RECORDS.md").write_text(content, encoding="utf-8") report = self._report(tmp_path) assert report.issues == [] def test_flags_stale_index_after_new_record_added(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text("```yaml\nid: AWQ-001\nstatus: open\n```\n", encoding="utf-8") content = _generate_work_record_index(tmp_path, "testrepo") (tmp_path / "WORK-RECORDS.md").write_text(content, encoding="utf-8") # a new record appears (tmp_path / "queue.md").write_text( "```yaml\nid: AWQ-001\nstatus: open\n```\n```yaml\nid: AWQ-002\nstatus: open\n```\n", encoding="utf-8", ) report = self._report(tmp_path) assert len(report.issues) == 1 assert report.issues[0].check_id == "C-33" def test_fix_context_carries_correct_content(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text("```yaml\nid: AWQ-001\nstatus: open\n```\n", encoding="utf-8") report = self._report(tmp_path) ctx = report.issues[0]._fix_context assert ctx["index_file"] == tmp_path / "WORK-RECORDS.md" assert "AWQ-001" in ctx["content"] def test_flags_stale_index_when_only_archived_workplans_remain( self, tmp_path, registry_file ): wp_dir = tmp_path / "workplans" / "archived" wp_dir.mkdir(parents=True) (wp_dir / "260101-CUST-WP-0000-old.md").write_text( "---\nid: CUST-WP-0000\nstatus: finished\n---\n", encoding="utf-8" ) (tmp_path / "WORK-RECORDS.md").write_text( "| workplan | CUST-WP-0000 | active | — | workplans/CUST-WP-0000-old.md |\n", encoding="utf-8", ) report = self._report(tmp_path) assert len(report.issues) == 1 assert report.issues[0].check_id == "C-33" content = report.issues[0]._fix_context["content"] assert "CUST-WP-0000" not in content assert "| Kind | ID | Status | Lane | Source |" in content def test_no_issue_when_empty_index_matches_archived_only_repo( self, tmp_path, registry_file ): wp_dir = tmp_path / "workplans" / "archived" wp_dir.mkdir(parents=True) (wp_dir / "260101-CUST-WP-0000-old.md").write_text( "---\nid: CUST-WP-0000\nstatus: finished\n---\n", encoding="utf-8" ) (tmp_path / "WORK-RECORDS.md").write_text( _work_record_index_document("testrepo", []), encoding="utf-8" ) report = self._report(tmp_path) assert report.issues == []