"""Unit tests for the stateless / pure-function layer of consistency_check.py. Covers: - parse_frontmatter — YAML frontmatter splitting - parse_task_blocks — ```task``` block extraction - get_tasks_from_workplan — block vs. frontmatter fallback - ConsistencyReport — issue accumulation and severity filtering - render_text — text rendering smoke tests - report_to_dict — serialisation shape No network calls, no DB, no live API — these tests run fully offline. """ from __future__ import annotations import sys from pathlib import Path import pytest # Make scripts/ importable without installing sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) from consistency_check import ( ConsistencyReport, Issue, FILE_TO_DB_WORKSTREAM_STATUS, get_tasks_from_workplan, normalise_workstream_status, parse_frontmatter, parse_task_blocks, render_text, report_to_dict, ) # --------------------------------------------------------------------------- # parse_frontmatter # --------------------------------------------------------------------------- class TestParseFrontmatter: def test_valid_frontmatter(self): text = "---\nid: WP-001\ntitle: Test\nstatus: active\n---\n\n# Body" meta, body = parse_frontmatter(text) assert meta["id"] == "WP-001" assert meta["title"] == "Test" assert meta["status"] == "active" assert "# Body" in body def test_no_frontmatter_returns_empty_meta(self): text = "# Just a heading\n\nSome body text." meta, body = parse_frontmatter(text) assert meta == {} assert body == text def test_empty_frontmatter(self): text = "---\n---\n\nBody only." meta, body = parse_frontmatter(text) assert isinstance(meta, dict) assert "Body only." in body def test_frontmatter_with_quoted_strings(self): text = '---\nid: "WP-007"\ntitle: "My Workplan"\n---\nBody' meta, _ = parse_frontmatter(text) assert meta["id"] == "WP-007" assert meta["title"] == "My Workplan" def test_incomplete_frontmatter_delimiter(self): """Only one --- means no valid frontmatter block.""" text = "---\nid: foo\nNo closing delimiter" meta, body = parse_frontmatter(text) assert meta == {} def test_body_preserved_fully(self): text = "---\nid: X\n---\nLine1\nLine2\nLine3" _, body = parse_frontmatter(text) assert "Line1" in body assert "Line2" in body assert "Line3" in body # --------------------------------------------------------------------------- # parse_task_blocks # --------------------------------------------------------------------------- class TestParseTaskBlocks: def test_single_task_block(self): body = "## T01\n\n```task\nid: WP-T01\nstatus: done\npriority: high\n```\n" tasks = parse_task_blocks(body) assert len(tasks) == 1 assert tasks[0]["id"] == "WP-T01" assert tasks[0]["status"] == "done" def test_multiple_task_blocks(self): body = ( "## T01\n```task\nid: T01\nstatus: done\n```\n\n" "## T02\n```task\nid: T02\nstatus: todo\n```\n" ) tasks = parse_task_blocks(body) assert len(tasks) == 2 assert tasks[0]["id"] == "T01" assert tasks[1]["id"] == "T02" def test_no_task_blocks(self): body = "# Workplan\n\nNo task blocks here.\n" tasks = parse_task_blocks(body) assert tasks == [] def test_task_block_with_all_fields(self): body = ( "```task\n" "id: CUST-WP-0008-T01\n" "status: done\n" "priority: critical\n" "assignee: custodian\n" 'state_hub_task_id: "abc-123"\n' "```\n" ) tasks = parse_task_blocks(body) assert len(tasks) == 1 t = tasks[0] assert t["id"] == "CUST-WP-0008-T01" assert t["status"] == "done" assert t["priority"] == "critical" assert t["state_hub_task_id"] == "abc-123" def test_ignores_non_task_fenced_blocks(self): body = ( "```python\nprint('hello')\n```\n\n" "```task\nid: T01\nstatus: todo\n```\n" "```bash\necho hi\n```\n" ) tasks = parse_task_blocks(body) assert len(tasks) == 1 assert tasks[0]["id"] == "T01" # --------------------------------------------------------------------------- # get_tasks_from_workplan # --------------------------------------------------------------------------- class TestGetTasksFromWorkplan: def test_prefers_task_blocks_over_frontmatter(self): meta = {"tasks": [{"id": "from-frontmatter", "status": "todo"}]} body = "```task\nid: from-block\nstatus: done\n```\n" tasks = get_tasks_from_workplan(meta, body) assert len(tasks) == 1 assert tasks[0]["id"] == "from-block" def test_falls_back_to_frontmatter_tasks(self): meta = {"tasks": [{"id": "FM-01", "status": "todo"}, {"id": "FM-02", "status": "done"}]} body = "# No task blocks here\n" tasks = get_tasks_from_workplan(meta, body) assert len(tasks) == 2 assert tasks[0]["id"] == "FM-01" assert tasks[1]["id"] == "FM-02" def test_returns_empty_when_no_tasks(self): meta = {} body = "# Just prose\n" tasks = get_tasks_from_workplan(meta, body) assert tasks == [] def test_ignores_non_list_frontmatter_tasks(self): meta = {"tasks": "not-a-list"} body = "# No blocks\n" tasks = get_tasks_from_workplan(meta, body) assert tasks == [] # --------------------------------------------------------------------------- # ConsistencyReport — issue accumulation and severity filtering # --------------------------------------------------------------------------- class TestConsistencyReport: def _report(self): return ConsistencyReport(repo_slug="test-repo", repo_path="/tmp/test-repo") def test_add_fail_issue(self): r = self._report() r.add(severity="FAIL", check_id="C-01", message="No workplans/ dir") assert len(r.issues) == 1 assert len(r.failures) == 1 assert len(r.warnings) == 0 assert len(r.infos) == 0 def test_add_warn_issue(self): r = self._report() r.add(severity="WARN", check_id="C-04", message="Status drift", fixable=True) assert len(r.warnings) == 1 assert r.warnings[0].fixable is True def test_add_info_issue(self): r = self._report() r.add(severity="INFO", check_id="C-08", message="Completed orphan") assert len(r.infos) == 1 def test_mixed_severities(self): r = self._report() r.add(severity="FAIL", check_id="C-01", message="F1") r.add(severity="FAIL", check_id="C-03", message="F2") r.add(severity="WARN", check_id="C-04", message="W1") r.add(severity="INFO", check_id="C-08", message="I1") assert len(r.failures) == 2 assert len(r.warnings) == 1 assert len(r.infos) == 1 assert len(r.issues) == 4 def test_add_returns_issue(self): r = self._report() issue = r.add(severity="FAIL", check_id="C-01", message="msg") assert isinstance(issue, Issue) assert issue.severity == "FAIL" def test_empty_report(self): r = self._report() assert r.failures == [] assert r.warnings == [] assert r.infos == [] assert r.issues == [] # --------------------------------------------------------------------------- # render_text # --------------------------------------------------------------------------- class TestRenderText: def _clean_report(self): return ConsistencyReport(repo_slug="my-repo", repo_path="/path/to/repo") def test_pass_result_shown(self): r = self._clean_report() text = render_text(r) assert "PASS" in text assert "my-repo" in text def test_fail_result_shown(self): r = self._clean_report() r.add(severity="FAIL", check_id="C-01", message="No workplans/ directory") text = render_text(r) assert "FAIL" in text assert "C-01" in text assert "No workplans/ directory" in text def test_warn_result_shown(self): r = self._clean_report() r.add(severity="WARN", check_id="C-04", message="Status drift", fixable=True) text = render_text(r) assert "WARN" in text assert "[fixable]" in text def test_fix_applied_shown(self): r = self._clean_report() r.fixes_applied.append("C-04: updated status active→done for WP-001") text = render_text(r) assert "FIXES APPLIED" in text assert "C-04" in text def test_summary_counts_correct(self): r = self._clean_report() r.add(severity="FAIL", check_id="C-01", message="f") r.add(severity="WARN", check_id="C-04", message="w") r.add(severity="INFO", check_id="C-08", message="i") text = render_text(r) assert "1 fail" in text assert "1 warn" in text assert "1 info" in text def test_info_hidden_when_show_info_false(self): r = self._clean_report() r.add(severity="INFO", check_id="C-08", message="Completed orphan") text = render_text(r, show_info=False) assert "C-08" not in text # --------------------------------------------------------------------------- # report_to_dict # --------------------------------------------------------------------------- class TestReportToDict: def test_clean_report_result_is_pass(self): r = ConsistencyReport(repo_slug="r", repo_path="/p") d = report_to_dict(r) assert d["result"] == "pass" assert d["summary"] == {"fail": 0, "warn": 0, "info": 0} assert d["issues"] == [] def test_fail_result(self): r = ConsistencyReport(repo_slug="r", repo_path="/p") r.add(severity="FAIL", check_id="C-01", message="missing") d = report_to_dict(r) assert d["result"] == "fail" assert d["summary"]["fail"] == 1 def test_warn_only_result(self): r = ConsistencyReport(repo_slug="r", repo_path="/p") r.add(severity="WARN", check_id="C-04", message="drift") d = report_to_dict(r) assert d["result"] == "warn" assert d["summary"]["warn"] == 1 assert d["summary"]["fail"] == 0 def test_issue_fields_serialised(self): r = ConsistencyReport(repo_slug="r", repo_path="/p") r.add( severity="FAIL", check_id="C-03", message="stale ref", file_path="workplans/WP.md", db_id="abc-123", file_value="abc-123", db_value="", ) d = report_to_dict(r) issue = d["issues"][0] assert issue["severity"] == "FAIL" assert issue["check_id"] == "C-03" assert issue["file_path"] == "workplans/WP.md" assert issue["db_id"] == "abc-123" def test_fixes_applied_in_dict(self): r = ConsistencyReport(repo_slug="r", repo_path="/p") r.fixes_applied.append("C-04: status fixed") d = report_to_dict(r) assert "C-04: status fixed" in d["fixes_applied"] def test_repo_slug_and_path_preserved(self): r = ConsistencyReport(repo_slug="the-custodian", repo_path="/home/worsch/the-custodian") d = report_to_dict(r) assert d["repo_slug"] == "the-custodian" assert d["repo_path"] == "/home/worsch/the-custodian" # --------------------------------------------------------------------------- # Status vocabulary normalisation # --------------------------------------------------------------------------- class TestNormaliseWorkstreamStatus: """FILE_TO_DB_WORKSTREAM_STATUS maps workplan file vocabulary to DB vocabulary. Workplan files use task-style "done"; the DB workstream API uses "completed". The C-04 check and fix code must normalise before comparing or PATCHing. """ def test_done_maps_to_completed(self): assert normalise_workstream_status("done") == "completed" def test_completed_is_identity(self): assert normalise_workstream_status("completed") == "completed" def test_active_is_identity(self): assert normalise_workstream_status("active") == "active" def test_blocked_is_identity(self): assert normalise_workstream_status("blocked") == "blocked" def test_archived_is_identity(self): assert normalise_workstream_status("archived") == "archived" def test_unknown_value_returned_as_is(self): # Don't crash on unexpected values — return them unchanged assert normalise_workstream_status("foobar") == "foobar" def test_map_constant_covers_done(self): assert "done" in FILE_TO_DB_WORKSTREAM_STATUS assert FILE_TO_DB_WORKSTREAM_STATUS["done"] == "completed" def test_c04_no_spurious_drift_when_done_vs_completed(self): """done (file) vs completed (DB) must NOT be reported as C-04 drift.""" assert normalise_workstream_status("done") == normalise_workstream_status("completed") def test_c04_real_drift_still_detected(self): """done (file) vs active (DB) IS real drift and must be detected.""" assert normalise_workstream_status("done") != normalise_workstream_status("active")