diff --git a/state-hub/tests/test_consistency_check.py b/state-hub/tests/test_consistency_check.py new file mode 100644 index 0000000..a03c9b7 --- /dev/null +++ b/state-hub/tests/test_consistency_check.py @@ -0,0 +1,328 @@ +"""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, + get_tasks_from_workplan, + 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"