cc-os/plugins/os-doc-hygiene/tests/test_validate_report.py

"""
Tests for scripts/validate_report.py

Wires the machine-report schema validator into pytest (Group 5 correctness
gate). Two layers:

  1. End-to-end against the GOLDEN fixtures (Layer-2 reversion protection):
     examples/golden/valid_report.json → PASS (exit 0),
     examples/golden/invalid_report.json → FAIL (exit 1).
     The golden files are human-gated; they are NOT mutated here.

  2. Branch coverage via crafted dicts fed straight into ReportValidator —
     the raw_tokens requirement, the op_type↔exact_edit biconditional
     (invariant #11), and the safety_tier derivation rule (invariant #10).

sys.path is patched here (no shared conftest) so the test file is
self-contained.
"""

from __future__ import annotations

import copy
import json
import subprocess
import sys
from pathlib import Path

import pytest

_SCRIPTS_DIR = Path(__file__).parent.parent / "scripts"
if str(_SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS_DIR))

from validate_report import ReportValidator, derive_safety_tier  # noqa: E402

_GOLDEN_DIR = Path(__file__).parent.parent / "examples" / "golden"
_VALIDATOR = _SCRIPTS_DIR / "validate_report.py"


def _validate(report: dict) -> list:
    return ReportValidator(report).validate()


def _minimal_valid_report() -> dict:
    """A schema-valid single-entry report used as a base for mutation tests."""
    return {
        "schema_version": "1.0",
        "tool_version": "0.1.0",
        "generated_at": "2026-06-24T00:00:00Z",
        "scan": {
            "project_root": "/p",
            "scope_globs": ["**/*.md"],
            "excluded_dirs": ["build"],
            "files_scanned": 1,
        },
        "shortlist": ["a.md"],
        "entries": [
            {
                "path": "a.md",
                "category": {"class": "bloat", "subtype": "distill"},
                "signals": [],
                "op": "Condense.",
                "op_type": "generative",
                "is_destructive": False,
                "is_reversible": True,
                "safety_tier": "confirm",
                "token_estimate": {"raw_tokens": 100},
            }
        ],
        "promotion_candidates": [],
    }


# ===========================================================================
# Layer 1 — golden fixtures end-to-end via the CLI (exit codes)
# ===========================================================================

class TestGoldenFixturesCLI:

    def _run(self, fixture: Path):
        return subprocess.run(
            [sys.executable, str(_VALIDATOR), str(fixture)],
            capture_output=True,
            text=True,
        )

    def test_valid_golden_passes(self):
        result = self._run(_GOLDEN_DIR / "valid_report.json")
        assert result.returncode == 0, result.stdout
        payload = json.loads(result.stdout)
        assert payload["valid"] is True
        assert payload["violations"] == []

    def test_invalid_golden_fails(self):
        result = self._run(_GOLDEN_DIR / "invalid_report.json")
        assert result.returncode == 1
        payload = json.loads(result.stdout)
        assert payload["valid"] is False
        assert len(payload["violations"]) >= 1

    def test_invalid_golden_flags_safety_tier_mismatch(self):
        """The golden invalid report has an auto tier on a destructive op."""
        result = self._run(_GOLDEN_DIR / "invalid_report.json")
        payload = json.loads(result.stdout)
        fields = {v["field"] for v in payload["violations"]}
        assert any("safety_tier" in f for f in fields)

    def test_usage_error_exit_code(self):
        result = subprocess.run(
            [sys.executable, str(_VALIDATOR)],
            capture_output=True,
            text=True,
        )
        assert result.returncode == 2


# ===========================================================================
# Layer 2 — crafted-dict branch coverage
# ===========================================================================

class TestBaseFixtureIsValid:

    def test_minimal_report_has_no_violations(self):
        assert _validate(_minimal_valid_report()) == []


class TestRawTokensRequirement:

    def test_missing_raw_tokens_is_rejected(self):
        r = _minimal_valid_report()
        r["entries"][0]["token_estimate"] = {"injection_frequency": None}
        violations = _validate(r)
        assert any(
            v["field"].endswith("token_estimate.raw_tokens") for v in violations
        )

    def test_non_numeric_raw_tokens_is_rejected(self):
        r = _minimal_valid_report()
        r["entries"][0]["token_estimate"] = {"raw_tokens": "lots"}
        violations = _validate(r)
        assert any(
            v["field"].endswith("token_estimate.raw_tokens") for v in violations
        )


class TestOpTypeExactEditBiconditional:
    """Invariant #11: exact_edit present IFF op_type == deterministic."""

    def test_generative_with_exact_edit_is_rejected(self):
        r = _minimal_valid_report()
        # base entry is generative; bolt on a forbidden exact_edit
        r["entries"][0]["exact_edit"] = {
            "kind": "insert-frontmatter",
            "key": "hygiene",
            "value": "frozen",
        }
        violations = _validate(r)
        assert any(v["field"].endswith("exact_edit") for v in violations)

    def test_deterministic_without_exact_edit_is_rejected(self):
        r = _minimal_valid_report()
        e = r["entries"][0]
        e["op_type"] = "deterministic"
        e["is_destructive"] = False
        e["is_reversible"] = True
        e["safety_tier"] = "auto"  # derived for deterministic+nondestr+reversible
        # no exact_edit present
        violations = _validate(r)
        assert any(
            v["field"].endswith("exact_edit")
            and "MUST carry" in v["message"]
            for v in violations
        )

    def test_deterministic_with_valid_exact_edit_passes(self):
        r = _minimal_valid_report()
        e = r["entries"][0]
        e["op_type"] = "deterministic"
        e["is_destructive"] = False
        e["is_reversible"] = True
        e["safety_tier"] = "auto"
        e["exact_edit"] = {
            "kind": "insert-frontmatter",
            "key": "hygiene",
            "value": "frozen",
        }
        assert _validate(r) == []


class TestSafetyTierDerivation:
    """Invariant #10: recorded safety_tier must equal the derivation."""

    def test_truth_table(self):
        # generative ⇒ confirm regardless of booleans
        assert derive_safety_tier("generative", False, True) == "confirm"
        assert derive_safety_tier("generative", True, False) == "confirm"
        # destructive ⇒ confirm
        assert derive_safety_tier("deterministic", True, True) == "confirm"
        # irreversible ⇒ confirm
        assert derive_safety_tier("deterministic", False, False) == "confirm"
        # deterministic + non-destructive + reversible ⇒ auto (only auto case)
        assert derive_safety_tier("deterministic", False, True) == "auto"

    def test_recorded_tier_mismatch_is_rejected(self):
        """generative op recorded as 'auto' must be rejected."""
        r = _minimal_valid_report()
        r["entries"][0]["safety_tier"] = "auto"  # wrong: generative ⇒ confirm
        violations = _validate(r)
        assert any(
            v["field"].endswith("safety_tier") and "mismatch" in v["message"]
            for v in violations
        )

    def test_correct_tier_accepted(self):
        r = _minimal_valid_report()
        # base is generative with confirm — already correct
        assert not any(
            v["field"].endswith("safety_tier") for v in _validate(r)
        )


class TestCategoryEnum:

    def test_bad_subtype_for_class_rejected(self):
        r = _minimal_valid_report()
        r["entries"][0]["category"] = {"class": "stale", "subtype": "distill"}
        violations = _validate(r)
        assert any(v["field"].endswith("category.subtype") for v in violations)


class TestLifecycleKinds:
    """Group 3 (task 3.1): delete/extract-then-delete kinds + the lifecycle
    object + the ADR-0039 tier branch in derive_safety_tier."""

    def _lifecycle_delete_entry(self, **lifecycle_overrides) -> dict:
        lifecycle = {
            "rule_ref": "autoresearch/*/**",
            "lifetime": "temporary",
            "git_state": {"tracked": True, "clean": True},
        }
        lifecycle.update(lifecycle_overrides)
        return {
            "path": "autoresearch/run-1/notes.md",
            "category": {"class": "stale", "subtype": "orphaned"},
            "signals": [],
            "op": "Delete expired temporary run.",
            "op_type": "deterministic",
            "is_destructive": True,
            "is_reversible": True,
            "safety_tier": "auto",
            "exact_edit": {
                "kind": "delete",
                "anchor": {"start_line": 1, "end_line": 10},
                "expected_sha256": "a" * 64,
                "lifecycle": lifecycle,
            },
            "token_estimate": {"raw_tokens": 10},
        }

    def _base_report(self, entry: dict) -> dict:
        r = _minimal_valid_report()
        r["shortlist"] = [entry["path"]]
        r["entries"] = [entry]
        return r

    # -- kind enum + required sub-fields -----------------------------------

    def test_delete_kind_is_accepted(self):
        r = self._base_report(self._lifecycle_delete_entry())
        assert _validate(r) == []

    def test_extract_then_delete_requires_extraction_target(self):
        entry = self._lifecycle_delete_entry()
        entry["exact_edit"]["kind"] = "extract-then-delete"
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("extraction_target") for v in violations)

    def test_extract_then_delete_repo_durable_requires_dest(self):
        entry = self._lifecycle_delete_entry()
        entry["exact_edit"]["kind"] = "extract-then-delete"
        entry["exact_edit"]["extraction_target"] = "repo-durable"
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("extraction_dest") for v in violations)

    def test_extract_then_delete_repo_durable_with_dest_passes(self):
        entry = self._lifecycle_delete_entry()
        entry["exact_edit"]["kind"] = "extract-then-delete"
        entry["exact_edit"]["extraction_target"] = "repo-durable"
        entry["exact_edit"]["extraction_dest"] = "docs/adr/0099-extracted.md"
        r = self._base_report(entry)
        assert _validate(r) == []

    def test_extract_then_delete_cross_repo_needs_no_dest(self):
        entry = self._lifecycle_delete_entry()
        entry["exact_edit"]["kind"] = "extract-then-delete"
        entry["exact_edit"]["extraction_target"] = "cross-repo"
        r = self._base_report(entry)
        assert _validate(r) == []

    def test_missing_lifecycle_object_is_rejected(self):
        entry = self._lifecycle_delete_entry()
        del entry["exact_edit"]["lifecycle"]
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("exact_edit.lifecycle") for v in violations)

    def test_both_served_fields_is_rejected(self):
        entry = self._lifecycle_delete_entry(
            lifetime="delete-once-served",
            served_when_path="docs/plans/archive/{name}",
            served_when="the plan shipped",
        )
        entry["exact_edit"]["safety_tier_note"] = None  # no-op, keep shape
        r = self._base_report(entry)
        violations = _validate(r)
        assert any("exactly one" in v["message"] for v in violations)

    def test_neither_served_field_on_delete_once_served_is_rejected(self):
        entry = self._lifecycle_delete_entry(lifetime="delete-once-served")
        r = self._base_report(entry)
        violations = _validate(r)
        assert any("exactly one" in v["message"] for v in violations)

    def test_temporary_with_served_field_is_rejected(self):
        entry = self._lifecycle_delete_entry(served_when_path="x/archive/{name}")
        r = self._base_report(entry)
        violations = _validate(r)
        assert any("not carry served_when_path" in v["message"] for v in violations)

    def test_missing_git_state_is_rejected(self):
        entry = self._lifecycle_delete_entry()
        del entry["exact_edit"]["lifecycle"]["git_state"]
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("git_state") for v in violations)

    # -- tier derivation branch ----------------------------------------------

    def test_scanner_proven_tracked_clean_derives_auto(self):
        entry = self._lifecycle_delete_entry()  # temporary + tracked+clean
        r = self._base_report(entry)
        assert _validate(r) == []
        assert entry["safety_tier"] == "auto"

    def test_scanner_proven_dirty_forces_confirm(self):
        entry = self._lifecycle_delete_entry(git_state={"tracked": True, "clean": False})
        entry["safety_tier"] = "auto"  # wrong: dirty forces confirm
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("safety_tier") and "mismatch" in v["message"] for v in violations)

    def test_scanner_proven_untracked_forces_confirm(self):
        entry = self._lifecycle_delete_entry(git_state={"tracked": False, "clean": False})
        entry["safety_tier"] = "auto"
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("safety_tier") and "mismatch" in v["message"] for v in violations)

    def test_classifier_judged_served_when_always_confirm_even_tracked_clean(self):
        entry = self._lifecycle_delete_entry(
            lifetime="delete-once-served",
            served_when="the effort this plan describes has shipped",
            git_state={"tracked": True, "clean": True},
        )
        entry["safety_tier"] = "auto"  # wrong: classifier-judged always confirm
        r = self._base_report(entry)
        violations = _validate(r)
        assert any(v["field"].endswith("safety_tier") and "mismatch" in v["message"] for v in violations)

    def test_classifier_judged_served_when_confirm_is_accepted(self):
        entry = self._lifecycle_delete_entry(
            lifetime="delete-once-served",
            served_when="the effort this plan describes has shipped",
            git_state={"tracked": True, "clean": True},
        )
        entry["safety_tier"] = "confirm"
        r = self._base_report(entry)
        assert _validate(r) == []

    def test_served_when_path_tracked_clean_derives_auto(self):
        entry = self._lifecycle_delete_entry(
            lifetime="delete-once-served",
            served_when_path="openspec/changes/archive/{id}/",
            git_state={"tracked": True, "clean": True},
        )
        r = self._base_report(entry)
        assert _validate(r) == []
        assert entry["safety_tier"] == "auto"

    def test_non_lifecycle_branches_unchanged(self):
        """The pre-existing non-lifecycle truth table is untouched."""
        assert derive_safety_tier("generative", False, True) == "confirm"
        assert derive_safety_tier("deterministic", True, True) == "confirm"
        assert derive_safety_tier("deterministic", False, False) == "confirm"
        assert derive_safety_tier("deterministic", False, True) == "auto"

    def test_derive_safety_tier_signature_accepts_lifecycle_kw(self):
        assert derive_safety_tier(
            "deterministic", True, True,
            lifecycle={"lifetime": "temporary", "git_state": {"tracked": True, "clean": True}},
        ) == "auto"


class TestPromotionCandidatesShape:

    def test_envelope_requires_promotion_candidates(self):
        r = _minimal_valid_report()
        del r["promotion_candidates"]
        violations = _validate(r)
        assert any(v["field"] == "promotion_candidates" for v in violations)

    def test_empty_promotion_candidates_is_valid(self):
        r = _minimal_valid_report()
        r["promotion_candidates"] = []
        assert _validate(r) == []

    def test_well_formed_candidate_is_valid(self):
        r = _minimal_valid_report()
        r["promotion_candidates"] = [
            {"path": "docs/plans/x.md", "name": "archive-bucket", "pitch": "Move it to archive/ once."}
        ]
        assert _validate(r) == []

    def test_candidate_missing_field_is_rejected(self):
        r = _minimal_valid_report()
        r["promotion_candidates"] = [{"path": "docs/plans/x.md", "name": "archive-bucket"}]
        violations = _validate(r)
        assert any(v["field"].endswith(".pitch") for v in violations)

    def test_promotion_candidates_must_be_a_list(self):
        r = _minimal_valid_report()
        r["promotion_candidates"] = {"not": "a list"}
        violations = _validate(r)
        assert any(v["field"] == "promotion_candidates" for v in violations)


class TestEnvelope:

    def test_missing_top_level_field_rejected(self):
        r = _minimal_valid_report()
        del r["generated_at"]
        violations = _validate(r)
        assert any(v["field"] == "generated_at" for v in violations)

    def test_entry_path_not_in_shortlist_rejected(self):
        r = _minimal_valid_report()
        r["entries"][0]["path"] = "ghost.md"  # not in shortlist
        violations = _validate(r)
        assert any(
            v["field"].endswith(".path") and "shortlist" in v["message"]
            for v in violations
        )