the-custodian/tests/test_validate_work_records.py

"""Unit + integration tests for tools/validate_work_records.py
(CUST-WP-0060-T04, work-record CI validation gate).

Covers:
  - load_registry / classify — id pattern matching against the live canon
    registry, incl. grandfathered legacy schemes (AWQ-, DEC-, OH-, task
    single-digit / -LEGACY- variants, and the exact MASON-0001 bootstrap)
  - load_validators — jsonschema-backed per-kind checks (intake, decision,
    engagement): required-field gates on non-terminal records, historical
    grace on terminal ones
  - iter_blocks — multi-document yaml blocks, malformed-yaml handling that
    only errors when a registered id is actually at stake
  - main() end-to-end via subprocess: exit codes, template-placeholder
    skipping, --strict sidetrack escalation

Uses the live canon (this repo) as the default registry/schema source, so
these tests double as a regression check on the shipped artifacts — not
just the tool's logic in isolation.
"""
from __future__ import annotations

import subprocess
import sys
import textwrap
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT))

from tools.validate_work_records import (  # noqa: E402
    classify,
    iter_blocks,
    load_registry,
    load_validators,
)


@pytest.fixture(scope="module")
def kinds():
    return load_registry(REPO_ROOT)


@pytest.fixture(scope="module")
def validators():
    return load_validators(REPO_ROOT)


class TestClassify:
    def test_workplan_and_task_ids(self, kinds):
        assert classify("CUST-WP-0060", kinds) == "workplan"
        assert classify("CUST-WP-0060-T01", kinds) == "task"

    def test_repository_derived_multi_segment_workplan_prefix(self, kinds):
        assert classify("RAPP-OPENBAO-WP-0002", kinds) == "workplan"
        assert classify("RAPP-OPENBAO-WP-0002-T01", kinds) == "task"
        assert classify("RAIL-K8S-WP-0001", kinds) == "workplan"
        assert classify("RAPP--OPENBAO-WP-0002", kinds) is None
        assert classify("1RAPP-OPENBAO-WP-0002", kinds) is None

    def test_ad_hoc_daily_container_and_task_ids(self, kinds):
        assert classify("ACTIVITY-WP-ADHOC-2026-08-23", kinds) == "workplan"
        assert classify("ACTIVITY-WP-ADHOC-2026-08-23-T01", kinds) == "task"
        assert classify("RAPP-OPENBAO-WP-ADHOC-2026-08-23", kinds) == "workplan"

    def test_unqualified_ad_hoc_ids_are_grandfathered(self, kinds):
        assert classify("ADHOC-2026-08-23", kinds) == "workplan"
        assert classify("ADHOC-2026-08-23-T01", kinds) == "task"

    def test_new_kind_ids(self, kinds):
        assert classify("BINKY-IN-0001", kinds) == "intake"
        assert classify("BINKY-DEC-2026-001", kinds) == "decision"
        assert classify("BINKY-ENG-2026-001", kinds) == "engagement"

    @pytest.mark.parametrize(
        "legacy_id,expected_kind",
        [
            ("AWQ-010", "intake"),
            ("DEC-2026-004", "decision"),
            ("OH-2026-003", "engagement"),
            ("CUST-WP-0060-T1", "task"),
            ("MASON-0001", "workplan"),
            ("MASON-0001-T01", "task"),
        ],
    )
    def test_grandfathered_legacy_ids(self, kinds, legacy_id, expected_kind):
        assert classify(legacy_id, kinds) == expected_kind

    def test_unregistered_id_returns_none(self, kinds):
        assert classify("FOO-QX-001", kinds) is None

    def test_risk_register_species(self, kinds):
        assert classify("RISK-F-0008", kinds) == "register-entry"
        assert classify("RISK-REG-0001", kinds) == "register-entry"
        assert classify("RISK-N-0003", kinds) is None


class TestValidators:
    def test_intake_open_without_lane_errors(self, validators):
        errs = validators["intake"]({"id": "X-IN-0001", "title": "t", "status": "open"})
        assert any("lane" in e for e in errs)

    def test_intake_open_with_lane_passes(self, validators):
        errs = validators["intake"](
            {"id": "X-IN-0001", "title": "clean intake item", "status": "open", "lane": "green"}
        )
        assert errs == []

    def test_intake_closed_without_outcome_errors(self, validators):
        errs = validators["intake"]({"id": "X-IN-0001", "title": "t", "status": "closed"})
        assert any("outcome" in e for e in errs)

    def test_intake_promoted_without_promoted_to_errors(self, validators):
        errs = validators["intake"](
            {"id": "X-IN-0001", "title": "t", "status": "closed", "outcome": "promoted"}
        )
        assert any("promoted_to" in e for e in errs)

    def test_intake_promoted_with_promoted_to_passes(self, validators):
        errs = validators["intake"](
            {
                "id": "X-IN-0001",
                "title": "promoted intake item",
                "status": "closed",
                "outcome": "promoted",
                "promoted_to": "CUST-WP-0061",
            }
        )
        assert errs == []

    def test_decision_prepared_requires_recommendation_and_fallback(self, validators):
        errs = validators["decision"](
            {"id": "X-DEC-2026-001", "title": "t", "status": "prepared", "lane": "yellow"}
        )
        assert any("agent_recommendation" in e or "fallback_if_no_response" in e
                    for e in errs)

    def test_decision_resolved_requires_outcome_and_resolver(self, validators):
        errs = validators["decision"](
            {"id": "X-DEC-2026-001", "title": "t", "status": "resolved", "lane": "yellow"}
        )
        assert any("outcome" in e or "resolved_at" in e or "resolved_by" in e
                    for e in errs)

    def test_decision_full_prepared_package_passes(self, validators):
        errs = validators["decision"](
            {
                "id": "X-DEC-2026-001",
                "title": "full prepared decision package",
                "status": "prepared",
                "lane": "yellow",
                "agent_recommendation": "approve",
                "fallback_if_no_response": "no degradation",
            }
        )
        assert errs == []

    def test_engagement_prepared_requires_material_and_counterparty(self, validators):
        errs = validators["engagement"](
            {"id": "X-ENG-2026-001", "title": "t", "status": "prepared"}
        )
        assert any("prepared_material" in e or "counterparty" in e for e in errs)

    def test_engagement_queued_bare_passes(self, validators):
        errs = validators["engagement"](
            {"id": "X-ENG-2026-001", "title": "a queued engagement", "status": "queued"}
        )
        assert errs == []


class TestLoadValidatorsFallback:
    """The jsonschema-unavailable path — this is the exact failure mode
    that broke the first Forgejo CI run (node:20-bookworm runner had no
    python3-jsonschema until the apt install was added). Must degrade to
    minimal checks, not crash."""

    def test_fallback_used_when_jsonschema_unimportable(self, monkeypatch):
        monkeypatch.setitem(sys.modules, "jsonschema", None)
        # load_validators does `import jsonschema` internally; force reload
        # of the module under test isn't needed since the import happens at
        # call time, not at module import time.
        import importlib
        import tools.validate_work_records as vwr
        importlib.reload(vwr)
        try:
            validators = vwr.load_validators(REPO_ROOT)
            assert set(validators) == {"intake", "decision", "engagement"}
            errs = validators["intake"]({"id": "X-IN-0001", "status": "open"})
            assert any("missing title" in e for e in errs)
            assert any("missing lane" in e for e in errs)
            errs_ok = validators["intake"](
                {"id": "X-IN-0001", "status": "open", "title": "has a title",
                 "lane": "green"}
            )
            assert errs_ok == []
        finally:
            monkeypatch.delitem(sys.modules, "jsonschema", raising=False)
            importlib.reload(vwr)


class TestIterBlocks:
    def test_multi_document_yaml_block_yields_both_docs(self, tmp_path, kinds):
        md = tmp_path / "doc.md"
        md.write_text(
            "```yaml\nid: CUST-WP-0060\n---\nid: CUST-WP-0060-T01\n```\n",
            encoding="utf-8",
        )
        blocks = list(iter_blocks(md, kinds))
        ids = [b[1]["id"] for b in blocks if b[1]]
        assert ids == ["CUST-WP-0060", "CUST-WP-0060-T01"]

    def test_malformed_yaml_without_registered_id_is_silently_skipped(self, tmp_path, kinds):
        md = tmp_path / "doc.md"
        md.write_text("```yaml\nid: [unterminated\n```\n", encoding="utf-8")
        results = list(iter_blocks(md, kinds))
        assert all(err is None for _, _, err in results)

    def test_malformed_yaml_with_registered_id_errors(self, tmp_path, kinds):
        md = tmp_path / "doc.md"
        md.write_text(
            "```yaml\nid: CUST-WP-0060\nbad: [unterminated\n```\n", encoding="utf-8"
        )
        results = list(iter_blocks(md, kinds))
        assert any(err is not None for _, _, err in results)

    def test_ignores_non_mapping_documents(self, tmp_path, kinds):
        md = tmp_path / "doc.md"
        md.write_text("```yaml\n- just\n- a\n- list\n```\n", encoding="utf-8")
        assert list(iter_blocks(md, kinds)) == []


class TestMainEndToEnd:
    """Invokes the real CLI via subprocess against synthetic repo fixtures."""

    def _run(self, repo: Path, *extra_args: str) -> subprocess.CompletedProcess:
        return subprocess.run(
            [sys.executable, str(REPO_ROOT / "tools/validate_work_records.py"),
             "--repo", str(repo), "--canon", str(REPO_ROOT), *extra_args],
            capture_output=True, text=True,
        )

    def test_clean_repo_exits_zero(self, tmp_path):
        (tmp_path / "workplan.md").write_text(
            textwrap.dedent(
                """
                ```yaml
                id: X-IN-0001
                title: "a clean intake item"
                status: open
                lane: green
                ```
                """
            ),
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 0, result.stdout
        assert "0 errors" in result.stdout

    def test_invalid_record_exits_one_with_message(self, tmp_path):
        (tmp_path / "bad.md").write_text(
            textwrap.dedent(
                """
                ```yaml
                id: X-DEC-2026-001
                title: "missing required prepared fields"
                status: prepared
                lane: yellow
                ```
                """
            ),
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 1
        assert "X-DEC-2026-001" in result.stdout

    def test_template_placeholder_ids_are_not_records(self, tmp_path):
        (tmp_path / "AGENTS.md").write_text(
            "```yaml\nid: BINKY-WP-NNNN\n```\n```yaml\nid: X-WP-0001-TNN\n```\n",
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 0
        assert "0 checked" in result.stdout

    def test_unregistered_id_warns_but_does_not_fail_by_default(self, tmp_path):
        (tmp_path / "notes.md").write_text(
            "```yaml\nid: FOO-QX-001\ntitle: rogue\n```\n", encoding="utf-8"
        )
        result = self._run(tmp_path)
        assert result.returncode == 0
        assert "WARN" in result.stdout
        assert "FOO-QX-001" in result.stdout

    def test_strict_mode_escalates_sidetrack_warning_to_failure(self, tmp_path):
        (tmp_path / "notes.md").write_text(
            "```yaml\nid: FOO-QX-001\ntitle: rogue\n```\n", encoding="utf-8"
        )
        result = self._run(tmp_path, "--strict")
        assert result.returncode == 1

    def test_grandfathered_legacy_ids_pass_unmodified(self, tmp_path):
        (tmp_path / "queue.md").write_text(
            textwrap.dedent(
                """
                ```yaml
                id: AWQ-010
                title: "legacy intake, unrenamed"
                lane: green
                status: open
                ```
                ```yaml
                id: DEC-2026-004
                title: "legacy decision, unrenamed"
                status: resolved
                lane: red
                outcome: approved
                resolved_at: "2026-07-20"
                resolved_by: Bernd
                ```
                """
            ),
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 0, result.stdout
        assert "2 checked" in result.stdout

    def test_mason_bootstrap_ids_pass_unmodified(self, tmp_path):
        (tmp_path / "legacy-mason.md").write_text(
            textwrap.dedent(
                """
                ```yaml
                id: MASON-0001
                title: "legacy Mason bootstrap"
                status: finished
                ```
                ```task
                id: MASON-0001-T01
                status: done
                priority: high
                ```
                """
            ),
            encoding="utf-8",
        )
        result = self._run(tmp_path, "--strict")
        assert result.returncode == 0, result.stdout
        assert "2 checked" in result.stdout

    def test_task_fence_rejects_non_task_registered_id(self, tmp_path):
        (tmp_path / "wrong-fence.md").write_text(
            "```task\nid: CUST-WP-0063\nstatus: active\n```\n",
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 1
        assert "registered as workplan, not task" in result.stdout

    def test_terminal_record_grace_does_not_mask_structural_errors(self, tmp_path):
        """Historical grace waives missing-field ('required') errors on
        terminal records, but a genuine value violation (bad lane enum)
        must still fail — grace is not a blanket skip for closed work."""
        (tmp_path / "archived.md").write_text(
            textwrap.dedent(
                """
                ```yaml
                id: X-DEC-2026-003
                title: "resolved decision with an invalid lane value"
                status: resolved
                lane: not-a-real-lane
                outcome: approved
                resolved_at: "2026-01-01"
                resolved_by: Bernd
                ```
                """
            ),
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 1, result.stdout
        assert "X-DEC-2026-003" in result.stdout

    def test_terminal_record_gets_historical_grace(self, tmp_path):
        """A resolved/closed record missing spine fields (e.g. an old
        archived decision without `lane`) is not penalised for
        completeness — only for structural errors."""
        (tmp_path / "archived.md").write_text(
            textwrap.dedent(
                """
                ```yaml
                id: X-DEC-2026-002
                title: "old resolved decision, minimal fields"
                status: resolved
                outcome: approved
                resolved_at: "2026-01-01"
                resolved_by: Bernd
                ```
                """
            ),
            encoding="utf-8",
        )
        result = self._run(tmp_path)
        assert result.returncode == 0, result.stdout