core/tests/test_refusal_taxonomy_lane.py

"""ADR-0163 Phase A — refusal-taxonomy lane tests."""

from __future__ import annotations

import ast
import hashlib
import json
import re
from pathlib import Path

import pytest

from evals.framework import discover_lanes, get_lane, load_cases, run_lane
from evals.refusal_taxonomy.runner import build_report, categorize_cases
from evals.refusal_taxonomy.shape_categories import (
    SHAPE_CATEGORY_ORDER,
    ShapeCategory,
    categorize,
)
from scripts.build_refusal_taxonomy_cases import (
    build_cases,
    extract_statement,
)

_REPO_ROOT = Path(__file__).resolve().parent.parent
_LANE_ROOT = _REPO_ROOT / "evals" / "refusal_taxonomy"
_CASES_PATH = _LANE_ROOT / "public" / "v1" / "cases.jsonl"
_REPORT_PATH = _LANE_ROOT / "v1" / "report.json"
_GSM8K_REPORT = (
    _REPO_ROOT / "evals" / "gsm8k_math" / "train_sample" / "v1" / "report.json"
)
_SHAPE_CATEGORIES_PATH = _LANE_ROOT / "shape_categories.py"


# ---------------------------------------------------------------------------
# Case-set integrity
# ---------------------------------------------------------------------------


def test_cases_file_exists_and_nonempty():
    assert _CASES_PATH.exists(), f"expected case set at {_CASES_PATH}"
    cases = load_cases(_CASES_PATH)
    assert len(cases) == 50, "v1 sample should mirror the 50-case GSM8K train sample"


def test_case_schema_valid():
    cases = load_cases(_CASES_PATH)
    for case in cases:
        assert set(case.keys()) >= {"case_id", "statement", "refusal_reason"}
        assert isinstance(case["case_id"], str) and case["case_id"].strip()
        assert isinstance(case["statement"], str) and case["statement"].strip()
        assert isinstance(case["refusal_reason"], str) and case["refusal_reason"].strip()


def test_cases_sorted_by_id():
    cases = load_cases(_CASES_PATH)
    ids = [c["case_id"] for c in cases]
    assert ids == sorted(ids), "cases.jsonl must be deterministically sorted"


# ---------------------------------------------------------------------------
# Lane discovery
# ---------------------------------------------------------------------------


def test_lane_auto_discoverable():
    names = [lane.name for lane in discover_lanes()]
    assert "refusal_taxonomy" in names


def test_lane_run_via_framework():
    lane = get_lane("refusal_taxonomy")
    result = run_lane(lane, version="v1", split="public")
    assert result.metrics["total"] == 50
    assert result.metrics["categorized_rate"] == pytest.approx(0.72)


# ---------------------------------------------------------------------------
# Enum coverage + exhaustiveness
# ---------------------------------------------------------------------------


def test_shape_category_order_covers_enum():
    assert set(SHAPE_CATEGORY_ORDER) == set(ShapeCategory)
    assert len(SHAPE_CATEGORY_ORDER) == len(ShapeCategory)


def test_every_category_value_reachable_by_a_rule():
    # For each non-UNCATEGORIZED category, provide a tiny synthetic
    # statement that the categorizer routes to it.  If a future change
    # eliminates a rule, this test fails loudly.
    probes: dict[ShapeCategory, str] = {
        ShapeCategory.NESTED_QUESTION_TARGET:
            "If Jen has 150 ducks, how many total birds does she have?",
        ShapeCategory.UNIT_PARTITION:
            "She splits it up into 25-foot sections.",
        ShapeCategory.RATE_WITH_CURRENCY: "Tina makes $18.00 an hour.",
        ShapeCategory.COMPARATIVE_WITH_UNIT:
            "Her grandfather is 7 times her age.",
        ShapeCategory.FRACTIONAL_RATE_OF_CHANGE:
            "His fish ate half of them.",
        ShapeCategory.INDEFINITE_QUANTITY: "There are some kids in camp.",
        ShapeCategory.TEMPORAL_AGGREGATION:
            "Mark does a gig every other day for 2 weeks.",
        ShapeCategory.CONDITIONAL_QUANTITY:
            "If she had two more, she would have plenty.",
        ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY:
            "Marnie makes bead bracelets.",
        ShapeCategory.UNCATEGORIZED:
            "Nicole collected 400 Pokemon cards.",
    }
    for category, probe in probes.items():
        assert categorize(probe) is category, (
            f"probe for {category.value!r} routed to {categorize(probe).value!r}"
        )


def test_added_category_cites_three_examples():
    """Enforce ADR-0163 §Risks: every category cites ≥ 3 refused statements
    OR is documented in its docstring as a reserve slot with rationale.

    This test parses the shape_categories.py source and verifies that each
    rule predicate docstring contains either three "case " citations or the
    word "reserve".
    """

    source = _SHAPE_CATEGORIES_PATH.read_text()
    tree = ast.parse(source)
    rule_funcs = [
        node for node in ast.walk(tree)
        if isinstance(node, ast.FunctionDef)
        and node.name.startswith("_is_")
    ]
    assert rule_funcs, "expected rule predicates named _is_*"
    case_cite_re = re.compile(r"case\s+\d{4}", re.IGNORECASE)
    for func in rule_funcs:
        doc = ast.get_docstring(func) or ""
        cites = len(case_cite_re.findall(doc))
        if cites < 3 and "reserve" not in doc.lower():
            pytest.fail(
                f"{func.name}: needs ≥ 3 case citations or a 'reserve' note "
                f"(found {cites} citations)"
            )


# ---------------------------------------------------------------------------
# Categorizer determinism + purity
# ---------------------------------------------------------------------------


def test_categorizer_is_deterministic():
    cases = load_cases(_CASES_PATH)
    runs = [
        [categorize(c["statement"]).value for c in cases]
        for _ in range(5)
    ]
    assert all(run == runs[0] for run in runs)


def test_categorizer_is_pure_no_io(tmp_path, monkeypatch):
    # Sentinel — fail if the categorizer touches the filesystem or os.environ.
    def fail_open(*_a, **_kw):
        raise AssertionError("categorize() must not perform I/O")
    monkeypatch.setattr("builtins.open", fail_open)
    monkeypatch.setattr("os.environ", {})
    # Drive a handful of probes; any I/O attempt explodes the test.
    for statement in (
        "Tina makes $18.00 an hour.",
        "There are some kids in camp.",
        "She splits it up into 25-foot sections.",
        "If Jen has 150 ducks, how many total birds does she have?",
    ):
        categorize(statement)


def test_categorizer_rejects_non_string():
    with pytest.raises(TypeError):
        categorize(None)  # type: ignore[arg-type]


def test_categorizer_no_llm_or_ml_imports():
    """Per ADR-0163 §Constraint #4: no LLM call, no embedding, no learned model."""

    source = _SHAPE_CATEGORIES_PATH.read_text()
    banned = (
        "openai", "anthropic", "huggingface", "transformers", "torch",
        "tensorflow", "sklearn", "numpy", "sentence_transformers",
        "requests", "httpx", "urllib",
    )
    for token in banned:
        assert token not in source.lower(), (
            f"shape_categories.py must not reference {token!r} — "
            "rules-only doctrine"
        )


# ---------------------------------------------------------------------------
# Histogram correctness
# ---------------------------------------------------------------------------


def test_histogram_synthetic_fixture():
    cases = [
        {"case_id": "x1", "statement": "There are some kids in camp.",
         "refusal_reason": "r"},
        {"case_id": "x2", "statement": "Marnie makes bead bracelets.",
         "refusal_reason": "r"},
        {"case_id": "x3", "statement": "Tina makes $18.00 an hour.",
         "refusal_reason": "r"},
    ]
    report = build_report(cases)
    assert report.metrics["total"] == 3
    assert report.metrics["by_category"]["indefinite_quantity"] == 1
    assert report.metrics["by_category"]["descriptive_setup_no_quantity"] == 1
    assert report.metrics["by_category"]["rate_with_currency"] == 1
    assert report.metrics["uncategorized"] == 0
    assert report.metrics["categorized_rate"] == pytest.approx(1.0)


def test_histogram_includes_all_categories_even_when_zero():
    cases = [
        {"case_id": "x1", "statement": "Marnie makes bead bracelets.",
         "refusal_reason": "r"},
    ]
    report = build_report(cases)
    keys = set(report.metrics["by_category"].keys())
    assert keys == {c.value for c in SHAPE_CATEGORY_ORDER}


def test_v1_report_uncategorized_under_fifty_percent():
    payload = json.loads(_REPORT_PATH.read_text())
    rate = payload["metrics"]["categorized_rate"]
    assert rate >= 0.5, (
        f"<50% categorized signals taxonomy/data mismatch; got {rate:.2%}"
    )


# ---------------------------------------------------------------------------
# Replay determinism
# ---------------------------------------------------------------------------


def test_report_replays_byte_identical():
    cases = load_cases(_CASES_PATH)
    r1 = build_report(cases)
    r2 = build_report(cases)
    assert r1.metrics == r2.metrics
    assert r1.case_details == r2.case_details


def test_committed_report_matches_categorizer():
    cases = load_cases(_CASES_PATH)
    fresh = build_report(cases)
    committed = json.loads(_REPORT_PATH.read_text())
    assert fresh.metrics["case_digest"] == committed["metrics"]["case_digest"]
    assert fresh.metrics["by_category"] == committed["metrics"]["by_category"]


# ---------------------------------------------------------------------------
# Helper script
# ---------------------------------------------------------------------------


def test_helper_extracts_statement_from_reason():
    reason = (
        "candidate_graph: no admissible candidate for statement: "
        "'Tina makes $18.00 an hour.'"
    )
    assert extract_statement(reason) == "Tina makes $18.00 an hour."


def test_helper_handles_question_envelope():
    reason = (
        "candidate_graph: no admissible candidate for question: "
        "'If Jen has 150 ducks, how many total birds does she have?'"
    )
    assert extract_statement(reason) == (
        "If Jen has 150 ducks, how many total birds does she have?"
    )


def test_helper_rebuilds_cases_matching_committed():
    rebuilt = build_cases(_GSM8K_REPORT)
    committed = load_cases(_CASES_PATH)
    assert rebuilt == committed


# ---------------------------------------------------------------------------
# Read-only invariant
# ---------------------------------------------------------------------------


def _tree_digest(root: Path) -> str:
    h = hashlib.sha256()
    if not root.exists():
        return "absent"
    for path in sorted(p for p in root.rglob("*") if p.is_file()):
        h.update(str(path.relative_to(root)).encode("utf-8"))
        h.update(b"\0")
        h.update(path.read_bytes())
        h.update(b"\0")
    return h.hexdigest()


def test_lane_run_does_not_mutate_protected_trees():
    teaching = _REPO_ROOT / "teaching"
    packs = _REPO_ROOT / "packs"
    lp_data = _REPO_ROOT / "language_packs" / "data"

    before = (
        _tree_digest(teaching),
        _tree_digest(packs),
        _tree_digest(lp_data),
    )

    cases = load_cases(_CASES_PATH)
    build_report(cases)

    after = (
        _tree_digest(teaching),
        _tree_digest(packs),
        _tree_digest(lp_data),
    )
    assert before == after, (
        "refusal-taxonomy lane must be read-only over teaching/, packs/, "
        "and language_packs/data/"
    )