test(CUST-WP-0008): add unit tests for consistency_check.py pure layer
33 offline tests covering: parse_frontmatter, parse_task_blocks, get_tasks_from_workplan, ConsistencyReport severity filtering, render_text output, and report_to_dict serialisation. Closes the DoD automated-tests gap for the Consistency Engine workstream. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
328
state-hub/tests/test_consistency_check.py
Normal file
328
state-hub/tests/test_consistency_check.py
Normal file
@@ -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"
|
||||
Reference in New Issue
Block a user