"""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) # The lane covers the *refused* subset of the 50-case GSM8K train sample. # As the reader improved, 6 cases now admit, leaving 44 refusals to # categorize (build_cases filters to verdict == "refused"). assert len(cases) == 44, "v1 sample should mirror the refused GSM8K train cases" 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"] == 44 # NOTE: a hard ``categorized_rate >= 0.95`` floor was removed here. It is # both redundant and perverse: the exact histogram (and therefore the rate) # is pinned by ``test_committed_report_matches_categorizer``, and as the # reader graduates *categorized* refusals into correct admissions, the # residual refusal set skews toward the uncategorized tail — so the rate # drifts DOWN precisely as the reader improves. We assert categorization is # still meaningful (the bulk of refusals carry a shape) without a target # that fights reader progress; the precise distribution is the committed # report's job. assert result.metrics["categorized_rate"] > 0.5 # --------------------------------------------------------------------------- # 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.CURRENCY_AMOUNT: "It cost $100,000 to open initially.", ShapeCategory.MULTIPLICATIVE_AGGREGATION: "Each survey has 10 questions.", ShapeCategory.DISCRETE_COUNT_STATEMENT: "Nicole collected 400 Pokemon cards.", ShapeCategory.UNCATEGORIZED: "John invests in a bank and gets 10% simple interest.", } 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/" )