"""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()