tests/test_eval.py

"""The retrieval gate's own scorer (canlex/eval.py) and the knob sweep
(canlex/sweep.py), offline (the code audit of 2026-09-21, missing item 2:
eval.py was 0 % covered while RUNBOOK names `py -m canlex.eval` as the
release gate, so a regression in _matches / must_match / score_item /
summarize would have moved every reported number silently).

Synthetic results only: no index, no corpus, no subprocess -- the sweep's
eval run is a mocked subprocess.

    python -m unittest tests.test_eval
"""
import contextlib
import io
import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest import mock

from canlex import eval as ev
from canlex import sweep


def res(act_short="IRPA", act_code="I-2.5", section="36", **extra):
    r = {"id": f"{act_code}-s{section or 'x'}", "act_short": act_short, "act_code": act_code,
         "section": section, "marginal_note": "", "doc_type": "legislation"}
    r.update(extra)
    return r


class MatchesTests(unittest.TestCase):
    def test_an_answer_names_the_act_by_short_name_or_code_and_the_section(self):
        r = res()
        self.assertTrue(ev._matches(r, [("IRPA", "36")]))
        self.assertTrue(ev._matches(r, [("I-2.5", "36")]))
        self.assertTrue(ev._matches(r, [("irpa", "36")]))          # case-folded act
        self.assertFalse(ev._matches(r, [("IRPA", "35")]))
        self.assertFalse(ev._matches(r, [("IRPR", "36")]))
        self.assertFalse(ev._matches(r, []))

    def test_an_empty_section_matches_any_chunk_of_the_act(self):
        # case-law answers: the chunks carry no section number
        self.assertTrue(ev._matches(res(section="36"), [("IRPA", "")]))
        self.assertTrue(ev._matches(res("Baker", "fca-1", ""), [("Baker", "")]))
        self.assertTrue(ev._matches(res("Baker", "fca-1", "17"), [("Baker", "")]))
        self.assertFalse(ev._matches(res("Baker", "fca-1", ""), [("Tran", "")]))
        # the second answer of a list still counts
        self.assertTrue(ev._matches(res(), [("IRPR", "18"), ("IRPA", "")]))


class MustMatchTests(unittest.TestCase):
    def test_a_marginal_note_key_compares_case_and_whitespace_folded(self):
        r = res("ENF 5", "enf-5", "3", marginal_note="Deemed   rehabilitation")
        self.assertTrue(ev.must_match(r, {"act_short": "ENF 5", "marginal_note": "deemed rehabilitation"}))
        self.assertTrue(ev.must_match(r, {"act_short": "enf-5", "marginal_note": "Deemed Rehabilitation"}))
        self.assertFalse(ev.must_match(r, {"act_short": "ENF 5", "marginal_note": "Rehabilitation"}))
        self.assertFalse(ev.must_match(r, {"act_short": "ENF 6", "marginal_note": "deemed rehabilitation"}))

    def test_a_section_key_and_a_bare_act_key(self):
        r = res()
        self.assertTrue(ev.must_match(r, {"act_short": "IRPA", "section": "36"}))
        self.assertFalse(ev.must_match(r, {"act_short": "IRPA", "section": "35"}))
        self.assertTrue(ev.must_match(r, {"act_short": "IRPA", "section": ""}))
        self.assertTrue(ev.must_match(r, {"act_short": "I-2.5"}))          # any chunk of the act
        self.assertFalse(ev.must_match(r, {"act_short": "Baker"}))
        # a marginal-note key wins over a section key on the same item
        self.assertFalse(ev.must_match(r, {"act_short": "IRPA", "section": "36", "marginal_note": "x"}))


class GuidanceActsTests(unittest.TestCase):
    def test_only_guidance_chunks_count_and_the_empty_name_is_dropped(self):
        chunks = [res("ENF 5", "enf-5", "", doc_type="memorandum"),
                  res("", "pdi-1", "", doc_type="memorandum"),
                  res("IRPA", "I-2.5", "36"),
                  res("Baker", "fca-1", "", doc_type="caselaw")]
        self.assertEqual(ev.guidance_acts(chunks), frozenset({"enf 5", "enf-5", "pdi-1"}))
        self.assertEqual(ev.guidance_acts([]), frozenset())


class ScoreItemTests(unittest.TestCase):
    ITEM = {"query": "q", "answers": [["IRPA", "36"]]}

    def ranked(self, hit_at):
        rows = [res("IRPR", "SOR-2002-227", str(i)) for i in range(1, 8)]
        if hit_at:
            rows[hit_at - 1] = res()
        return rows

    def test_rank_is_the_first_matching_result_or_zero(self):
        self.assertEqual(ev.score_item(self.ITEM, self.ranked(1))["rank"], 1)
        self.assertEqual(ev.score_item(self.ITEM, self.ranked(3))["rank"], 3)
        self.assertEqual(ev.score_item(self.ITEM, self.ranked(0))["rank"], 0)
        self.assertEqual(ev.score_item(self.ITEM, [])["rank"], 0)
        self.assertEqual(ev.score_item({"query": "q"}, self.ranked(1))["rank"], 0)   # no answers

    def test_a_cosurfaced_result_is_never_ranked(self):
        rows = [dict(res(), cosurfaced=True)] + self.ranked(2)
        self.assertEqual(ev.score_item(self.ITEM, rows)["rank"], 2)
        rows = [dict(res(), cosurfaced=True)] + self.ranked(0)
        self.assertEqual(ev.score_item(self.ITEM, rows)["rank"], 0)

    def test_coverage_and_guidance_are_none_when_the_item_asks_for_neither(self):
        s = ev.score_item(self.ITEM, self.ranked(1), guidance=frozenset({"enf 5"}))
        self.assertEqual((s["cov6"], s["cov6_rn"], s["g6"], s["g6_rn"]), (None, None, None, None))

    def test_coverage_reads_the_top_view_k_and_read_next_adds_to_it(self):
        item = dict(self.ITEM, must_see=[{"act_short": "IRPA", "section": "36"},
                                         {"act_short": "ENF 5", "marginal_note": "Deemed rehabilitation"}])
        enf = res("ENF 5", "enf-5", "3", marginal_note="Deemed rehabilitation", doc_type="memorandum")
        rows = self.ranked(1)
        s = ev.score_item(item, rows)
        self.assertEqual((s["cov6"], s["cov6_rn"]), (0.5, 0.5))
        s = ev.score_item(item, rows, read_next=[enf])
        self.assertEqual((s["cov6"], s["cov6_rn"]), (0.5, 1.0))
        # a key satisfied past VIEW_K is not in view
        rows = self.ranked(1)[:ev.VIEW_K] + [enf]
        self.assertEqual(ev.score_item(item, rows)["cov6"], 0.5)
        rows = self.ranked(1)[:ev.VIEW_K - 1] + [enf]
        self.assertEqual(ev.score_item(item, rows)["cov6"], 1.0)

    def test_guided_needs_a_guidance_result_that_is_an_answer_or_a_key(self):
        guidance = frozenset({"enf 5", "enf-5"})
        enf = res("ENF 5", "enf-5", "3", marginal_note="Deemed rehabilitation", doc_type="memorandum")
        item = {"query": "q", "answers": [["ENF 5", ""]]}
        self.assertTrue(ev.score_item(item, [enf], guidance=guidance)["g6"])
        self.assertFalse(ev.score_item(item, self.ranked(0), guidance=guidance)["g6"])
        # the guidance result must be one the item names: another memo does not count
        other = res("ENF 6", "enf-6", "1", doc_type="memorandum")
        s = ev.score_item(item, [other], read_next=[enf], guidance=guidance)
        self.assertEqual((s["g6"], s["g6_rn"]), (False, True))
        # named through a must_see key rather than an answer
        keyed = {"query": "q", "answers": [["IRPA", "36"]],
                 "must_see": [{"act_short": "ENF 5", "marginal_note": "Deemed rehabilitation"}]}
        self.assertTrue(ev.score_item(keyed, [enf], guidance=guidance)["g6"])
        self.assertIsNone(ev.score_item(keyed, [enf], guidance=frozenset())["g6"])


class MetricsTests(unittest.TestCase):
    def test_metrics_on_a_fixed_rank_list(self):
        h1, h3, h5, h10, mrr = ev._metrics([1, 3, 0, 6])
        self.assertEqual((h1, h3, h5, h10), (0.25, 0.5, 0.5, 0.75))
        self.assertAlmostEqual(mrr, (1 + 1 / 3 + 1 / 6) / 4)
        self.assertEqual(ev._metrics([]), (0.0, 0.0, 0.0, 0.0, 0.0))
        self.assertEqual(ev._metrics([1, 1]), (1.0, 1.0, 1.0, 1.0, 1.0))

    def test_summarize_averages_the_optional_measures_over_the_items_that_have_them(self):
        scores = [{"rank": 1, "cov6": 1.0, "cov6_rn": 1.0, "g6": True, "g6_rn": True},
                  {"rank": 3, "cov6": 0.0, "cov6_rn": 0.5, "g6": None, "g6_rn": None},
                  {"rank": 0, "cov6": None, "cov6_rn": None, "g6": False, "g6_rn": True}]
        m = ev.summarize(scores)
        self.assertEqual((m["n"], m["hit1"], m["hit3"], m["hit10"]), (3, 1 / 3, 2 / 3, 2 / 3))
        self.assertAlmostEqual(m["mrr"], (1 + 1 / 3) / 3)
        self.assertEqual((m["cov6"], m["cov6_rn"], m["n_cov6"]), (0.5, 0.75, 2))
        self.assertEqual((m["g6"], m["g6_rn"], m["n_g6"]), (0.5, 1.0, 2))
        empty = ev.summarize([])
        self.assertEqual((empty["n"], empty["cov6"], empty["n_cov6"]), (0, None, 0))

    def test_the_report_line_prints_the_optional_measures_only_when_present(self):
        with contextlib.redirect_stdout(io.StringIO()) as out:
            ev._report_line("ALL", [{"rank": 1, "cov6": None, "cov6_rn": None, "g6": None, "g6_rn": None}])
            ev._report_line("x", [{"rank": 2, "cov6": 0.5, "cov6_rn": 1.0, "g6": True, "g6_rn": True}])
        lines = out.getvalue().splitlines()
        self.assertIn("ALL", lines[0])
        self.assertIn("Hit@1 1.00", lines[0])
        self.assertNotIn("cov@6", lines[0])
        self.assertIn("cov@6 0.50/1.00 (n=1)", lines[1])
        self.assertIn("guid@6 1.00/1.00 (n=1)", lines[1])


class QuestionSetTests(unittest.TestCase):
    def _with_questions(self, items):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        path = Path(tmp.name) / "questions.json"
        path.write_text(json.dumps(items), encoding="utf-8")
        return mock.patch.object(ev, "QUESTIONS", path)

    def test_defaults_are_applied_and_the_sweep_slice_drops_holdouts_and_no_answer(self):
        items = [{"query": "a", "answers": [["IRPA", "36"]]},
                 {"query": "b", "answers": [["IRPA", "36"]], "holdout": True},
                 {"query": "c", "style": "no-answer"},
                 {"query": "d", "answers": [["IRPA", "36"]], "style": "typo", "added": "2026-09-09"}]
        with self._with_questions(items):
            loaded = ev.load_questions()
            self.assertEqual([(i["style"], i["holdout"]) for i in loaded],
                             [("standard", False), ("standard", True), ("no-answer", False),
                              ("typo", False)])
            self.assertEqual([i["query"] for i in ev.sweep_questions()], ["a", "d"])
        self.assertTrue(ev.is_legacy(loaded[0]))
        self.assertTrue(ev.is_legacy(loaded[1]))
        self.assertFalse(ev.is_legacy(loaded[3]))      # dated: outside the frozen slice
        self.assertFalse(ev.is_legacy({"style": "typo"}))

    def test_a_missing_question_file_is_none_and_an_empty_sweep(self):
        with mock.patch.object(ev, "QUESTIONS", Path(tempfile.gettempdir()) / "no-such-canlex-questions.json"):
            self.assertIsNone(ev.load_questions())
            self.assertEqual(ev.sweep_questions(), [])


class _StubIndex:
    """A LegislationIndex stand-in for eval.run(): chunks, a canned search
    and the co-surfaced guidance the index would add to a view."""
    see_also_source = "stub"

    def __init__(self, chunks, hits):
        self.chunks, self.hits = chunks, hits

    def search(self, query, top_k=6):
        return list(self.hits.get(query, []))[:top_k]

    def _cosurface_guidance(self, top, keep=None):
        return list(top) + [len(self.chunks) - 1]         # the last chunk is guidance


class ReadNextTests(unittest.TestCase):
    def test_read_next_is_what_the_index_cosurfaces_past_the_top_view(self):
        chunks = [res(section=str(i)) for i in range(1, 9)] + [
            res("ENF 5", "enf-5", "3", doc_type="memorandum")]
        index = _StubIndex(chunks, {})
        positions = {c["id"]: i for i, c in enumerate(chunks)}
        results = [dict(c) for c in chunks[:8]]
        results[0]["cosurfaced"] = True                  # never part of the view
        results.append({"id": "unknown"})                # not in the index
        rn = ev.read_next_for(index, results, positions)
        self.assertEqual([c["act_short"] for c in rn], ["ENF 5"])


class RunTests(unittest.TestCase):
    """eval.run() end to end on a stub index, in both output modes."""

    def _run(self, items, hits, env=None):
        chunks = [res(section="36", marginal_note="Serious criminality"), res(section="35"),
                  res("ENF 5", "enf-5", "3", marginal_note="Deemed rehabilitation", doc_type="memorandum")]
        with mock.patch.object(ev, "load_questions", return_value=items), \
                mock.patch.object(ev, "LegislationIndex", lambda: _StubIndex(chunks, hits)), \
                mock.patch.dict(os.environ, env or {"CANLEX_EVAL_SWEEP": ""}), \
                contextlib.redirect_stdout(io.StringIO()) as out:
            ev.run()
        return out.getvalue()

    def _items(self):
        for item in [{"query": "hit", "answers": [["IRPA", "36"]], "style": "standard"},
                     {"query": "miss", "answers": [["IRPA", "35"]], "style": "standard", "holdout": True},
                     {"query": "typo", "answers": [["IRPA", "36"]], "style": "typo", "added": "2026-09-09",
                      "must_see": [{"act_short": "ENF 5", "marginal_note": "Deemed rehabilitation"}]},
                     {"query": "nothing", "style": "no-answer"}]:
            item.setdefault("holdout", False)
            yield item

    def _hits(self):
        hit = [res(section="36", marginal_note="Serious criminality", confidence=0.9)]
        return {"hit": hit, "typo": hit, "miss": [res(section="36", confidence=0.5)],
                "nothing": [res(section="36", confidence=0.8)]}

    def test_the_full_report_names_the_slices_the_misses_and_the_calibration_table(self):
        out = self._run(list(self._items()), self._hits())
        self.assertIn("3 scored questions", out)
        for label in ("ALL", "legacy (comparable)", "style: typo", "holdout (sweep-blind)",
                      "holdout (undated)", "sweep-visible"):
            self.assertIn(label, out)
        self.assertIn("1 miss(es)", out)
        self.assertIn("[absent (searched 20)] (standard) miss", out)
        self.assertIn("gold: IRPA s.35   |   top result: IRPA s.36", out)
        self.assertIn("OVER-CONFIDENT", out)        # 0.8 on the no-answer item
        self.assertIn("read-next source: stub", out)

    def test_sweep_mode_prints_the_vertical_block_over_the_sweep_slice_only(self):
        out = self._run(list(self._items()), self._hits(), env={"CANLEX_EVAL_SWEEP": "1"})
        self.assertIn("2 questions (sweep slice)", out)      # holdout and no-answer excluded
        self.assertIn("Hit@1:   1.00", out)
        self.assertIn("MRR:     1.00", out)
        self.assertNotIn("miss(es)", out)
        self.assertIsNotNone(sweep._METRIC_RE.search(out))    # what sweep.py parses

    def test_no_question_set_says_so_on_stderr(self):
        with mock.patch.object(ev, "load_questions", return_value=None), \
                contextlib.redirect_stderr(io.StringIO()) as err:
            ev.run()
        self.assertIn("No question set", err.getvalue())


class SweepRunEvalTests(unittest.TestCase):
    BLOCK = ("CanLex retrieval evaluation -- 2 questions (sweep slice)\n\n"
             "  Hit@1:   0.50\n  Hit@3:   0.75\n  Hit@5:   0.80\n  Hit@10:  0.90\n  MRR:     0.66\n")

    def _proc(self, rc=0, stdout="", stderr=""):
        return subprocess.CompletedProcess(args=[], returncode=rc, stdout=stdout, stderr=stderr)

    def test_the_metrics_are_parsed_from_the_subprocess_run_with_the_overrides_in_its_env(self):
        with mock.patch.object(sweep.subprocess, "run", return_value=self._proc(0, self.BLOCK)) as run:
            m = sweep._run_eval({"CANLEX_MN_WEIGHT": 0.005, "CANLEX_MN_CAP": 0.024})
        self.assertEqual({k: m[k] for k in ("hit1", "hit3", "hit5", "hit10", "mrr")},
                         {"hit1": 0.5, "hit3": 0.75, "hit5": 0.8, "hit10": 0.9, "mrr": 0.66})
        self.assertEqual(m["stdout"], self.BLOCK)
        cmd, kw = run.call_args.args[0], run.call_args.kwargs
        self.assertEqual(cmd[-2:], ["-m", "canlex.eval"])
        self.assertEqual(kw["env"]["CANLEX_EVAL_SWEEP"], "1")          # holdouts stay blind
        self.assertEqual((kw["env"]["CANLEX_MN_WEIGHT"], kw["env"]["CANLEX_MN_CAP"]), ("0.005", "0.024"))
        self.assertEqual(Path(kw["cwd"]), sweep.ROOT)

    def test_a_failed_or_unparsable_eval_raises(self):
        with mock.patch.object(sweep.subprocess, "run", return_value=self._proc(1, "", "boom")):
            with self.assertRaises(RuntimeError) as cm:
                sweep._run_eval({})
        self.assertIn("exit 1", str(cm.exception))
        self.assertIn("boom", str(cm.exception))
        with mock.patch.object(sweep.subprocess, "run", return_value=self._proc(0, "no metrics here")):
            with self.assertRaises(RuntimeError) as cm:
                sweep._run_eval({})
        self.assertIn("could not parse", str(cm.exception))

    def test_the_order_is_hit5_then_mrr(self):
        self.assertGreater(sweep._score({"hit5": 0.8, "mrr": 0.1}), sweep._score({"hit5": 0.7, "mrr": 0.9}))
        self.assertGreater(sweep._score({"hit5": 0.8, "mrr": 0.5}), sweep._score({"hit5": 0.8, "mrr": 0.4}))


class SweepMainTests(unittest.TestCase):
    def test_main_keeps_the_best_value_per_knob_and_writes_the_log_and_summary(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        log, summary = Path(tmp.name) / "sweep.log", Path(tmp.name) / "sweep.json"

        def fake_eval(values):
            # knob A: 2 beats the default 1 on Hit@5; knob B: nothing beats the default
            hit5 = 0.7 + (0.1 if values["A"] == 2 else 0.0) - (0.2 if values["B"] == 9 else 0.0)
            return {"hit1": 0.5, "hit3": 0.6, "hit5": hit5, "hit10": 0.9, "mrr": hit5 / 2, "stdout": ""}

        with mock.patch.object(sweep, "KNOBS", [("A", [1, 2], 1), ("B", [3, 9], 3)]), \
                mock.patch.object(sweep, "LOG", log), mock.patch.object(sweep, "SUMMARY", summary), \
                mock.patch.object(sweep, "_run_eval", side_effect=fake_eval) as run, \
                contextlib.redirect_stdout(io.StringIO()) as out:
            sweep.main()
        doc = json.loads(summary.read_text(encoding="utf-8"))
        self.assertEqual(doc["best"], {"A": 2, "B": 3})
        self.assertAlmostEqual(doc["best_metrics"]["hit5"], 0.8)
        # baseline + one run per non-default value (the default reuses the baseline)
        self.assertEqual(run.call_count, 3)
        self.assertEqual(len(doc["runs"]), 5)
        self.assertIn("! A: 1 -> 2", out.getvalue())
        self.assertNotIn("! B:", out.getvalue())
        text = log.read_text(encoding="utf-8")
        self.assertIn("Sweeping A in [1, 2]", text)
        self.assertIn("Best: {'A': 2, 'B': 3}", text)


if __name__ == "__main__":
    unittest.main()