"""Unit tests for C-32 (work-record registration: intake/decision kinds, CUST-WP-0061-T02) — detection and file write-back only. The live create-in-DB path is covered by the fix_repo integration path, proven manually against a real repo/API (see CUST-WP-0061 progress notes); these tests stay offline and use a synthetic registry fixture, matching the conventions in test_work_record_check.py. """ from __future__ import annotations import sys import textwrap from pathlib import Path import pytest import yaml sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) from consistency_check import ( # noqa: E402 ConsistencyReport, _check_work_record_registration, _classify_work_record_kind, _inject_yaml_block_field, _load_work_record_kind_registry, ) REGISTRY_YAML = textwrap.dedent( """ version: "0.1" kinds: - 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}$'] legacy_patterns: - pattern: '^DEC-[0-9]{4}-[0-9]{3}$' source: binky DecisionQueue - kind: engagement id_patterns: ['^[A-Z]+-ENG-[0-9]{4}-[0-9]{3}$'] legacy_patterns: - pattern: '^OH-[0-9]{4}-[0-9]{3}$' source: binky OfficeHourQueue - kind: workplan id_patterns: ['^[A-Z]+-WP-[0-9]{4}$'] """ ) @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 TestClassifyWorkRecordKind: def test_classifies_intake(self, registry_file): kinds = _load_work_record_kind_registry() assert _classify_work_record_kind("BINKY-IN-0001", kinds) == "intake" def test_classifies_legacy_awq(self, registry_file): kinds = _load_work_record_kind_registry() assert _classify_work_record_kind("AWQ-010", kinds) == "intake" def test_classifies_decision(self, registry_file): kinds = _load_work_record_kind_registry() assert _classify_work_record_kind("DEC-2026-004", kinds) == "decision" def test_unregistered_returns_none(self, registry_file): kinds = _load_work_record_kind_registry() assert _classify_work_record_kind("FOO-QX-001", kinds) is None class TestCheckWorkRecordRegistration: def _report(self, repo_dir: Path) -> ConsistencyReport: report = ConsistencyReport(repo_slug="test", repo_path=str(repo_dir)) _check_work_record_registration(repo_dir, report) return report def test_intake_without_hub_id_is_flagged_fixable(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text( "```yaml\nid: BINKY-IN-0001\ntitle: needs registration\n```\n", encoding="utf-8", ) report = self._report(tmp_path) assert len(report.issues) == 1 issue = report.issues[0] assert issue.check_id == "C-32" assert issue.fixable is True assert issue._fix_context["kind"] == "intake" assert issue._fix_context["rid"] == "BINKY-IN-0001" def test_intake_with_hub_id_is_skipped(self, tmp_path, registry_file): (tmp_path / "queue.md").write_text( '```yaml\nid: BINKY-IN-0001\ntitle: already linked\n' 'state_hub_intake_id: "019f8000-0000-7000-8000-000000000000"\n```\n', encoding="utf-8", ) report = self._report(tmp_path) assert report.issues == [] def test_decision_without_hub_id_is_flagged_fixable(self, tmp_path, registry_file): (tmp_path / "DecisionQueue.md").write_text( "```yaml\nid: DEC-2026-005\ntitle: a new decision\n```\n", encoding="utf-8", ) report = self._report(tmp_path) assert len(report.issues) == 1 assert report.issues[0]._fix_context["kind"] == "decision" def test_engagement_without_hub_id_is_info_not_fixable(self, tmp_path, registry_file): (tmp_path / "OfficeHourQueue.md").write_text( "```yaml\nid: OH-2026-001\ntitle: bank call\n```\n", encoding="utf-8", ) report = self._report(tmp_path) assert len(report.issues) == 1 assert report.issues[0].severity == "INFO" assert report.issues[0].fixable is False def test_workplan_kind_is_untouched_by_c32(self, tmp_path, registry_file): """Workplans keep their own C-06 registration path; C-32 must not double-register them.""" (tmp_path / "workplans" / "CUST-WP-0099-x.md").parent.mkdir(parents=True) (tmp_path / "workplans" / "CUST-WP-0099-x.md").write_text( "---\nid: CUST-WP-0099\n---\n```yaml\nid: CUST-WP-0099\ntitle: x\n```\n", encoding="utf-8", ) report = self._report(tmp_path) assert report.issues == [] def test_template_placeholder_ignored(self, tmp_path, registry_file): (tmp_path / "AGENTS.md").write_text( "```yaml\nid: BINKY-IN-NNNN\ntitle: template\n```\n", encoding="utf-8" ) report = self._report(tmp_path) assert report.issues == [] def test_no_op_when_registry_unavailable(self, tmp_path, monkeypatch): monkeypatch.setenv("WORK_RECORD_REGISTRY", str(tmp_path / "missing.yaml")) monkeypatch.setattr(Path, "home", lambda: tmp_path / "nohome") (tmp_path / "queue.md").write_text( "```yaml\nid: BINKY-IN-0001\ntitle: x\n```\n", encoding="utf-8" ) report = self._report(tmp_path) assert report.issues == [] class TestInjectYamlBlockField: def test_injects_field_into_matching_block(self, tmp_path): md = tmp_path / "queue.md" md.write_text( "```yaml\nid: BINKY-IN-0001\ntitle: needs registration\n```\n", encoding="utf-8", ) changed = _inject_yaml_block_field( md, "state_hub_intake_id", "019f8000-0000-7000-8000-000000000000", "BINKY-IN-0001" ) assert changed is True text = md.read_text() assert 'state_hub_intake_id: "019f8000-0000-7000-8000-000000000000"' in text # round-trips as valid yaml block = text.split("```yaml\n", 1)[1].rsplit("```", 1)[0] parsed = yaml.safe_load(block) assert parsed["state_hub_intake_id"] == "019f8000-0000-7000-8000-000000000000" def test_does_not_touch_other_blocks(self, tmp_path): md = tmp_path / "queue.md" md.write_text( "```yaml\nid: BINKY-IN-0001\ntitle: first\n```\n" "```yaml\nid: BINKY-IN-0002\ntitle: second\n```\n", encoding="utf-8", ) _inject_yaml_block_field(md, "state_hub_intake_id", "abc-123", "BINKY-IN-0001") text = md.read_text() assert "abc-123" in text second_block = text.split("BINKY-IN-0002")[1] assert "state_hub_intake_id" not in second_block def test_no_op_when_id_not_found(self, tmp_path): md = tmp_path / "queue.md" original = "```yaml\nid: BINKY-IN-0001\ntitle: x\n```\n" md.write_text(original, encoding="utf-8") changed = _inject_yaml_block_field(md, "state_hub_intake_id", "abc-123", "BINKY-IN-9999") assert changed is False assert md.read_text() == original def test_idempotent_when_field_already_set(self, tmp_path): md = tmp_path / "queue.md" md.write_text( '```yaml\nid: BINKY-IN-0001\nstate_hub_intake_id: "already-there"\n```\n', encoding="utf-8", ) changed = _inject_yaml_block_field(md, "state_hub_intake_id", "new-value", "BINKY-IN-0001") assert changed is False assert "already-there" in md.read_text() assert "new-value" not in md.read_text()