Beemer
Deploy: the side-session batch (ebe9fff; approved_code 5a6b38d) -- guidance sections served whole with what is filed under them; IRB Chairperson's Guidelines under their own numbering and dated by their latest amendment; ENF 3/4/5 sections split correctly again; canlex_citing names a guidance section as get_section heads it and suggests a search that finds 1,088 of 1,252 (was 883); canlex_whats_new reads a date as the latest day it can mean. Eval 381 questions: Hit@1 260, Hit@5 364 as before.
91227f5 | """The authority registry (data/curated/authorities.json, roadmap T3.2) against | |
| the corpus and the curated datasets, and as the tools print it. | |
| Wave-3 verifier (2026-09-15): the registry's notes recorded corrections to the | |
| curated holdings (Nshogoza, Popic, Abid, Garcia, Lu, Barnett, Drake) that were | |
| never applied, the module was wired into no tool, and this file -- which the | |
| registry's own provenance names -- did not exist. Every decision key, pinned | |
| chunk and 'contains' phrase is checked here against data/processed; every | |
| correction a row records against the curated file it names; every cite the | |
| curated datasets use must resolve. | |
| The approved f5cf54d registry predates the render flag and the 'corrected' | |
| lists: on a composed release the rendering and correction tests skip, and the | |
| corpus checks read its phrases after quote normalisation (its Sittampalam | |
| phrases used straight quotes). | |
| """ | |
| import json | |
| import re | |
| import unittest | |
| from pathlib import Path | |
| from unittest import mock | |
| from canlex import authorities as A | |
| ROOT = Path(__file__).resolve().parent.parent | |
| CURATED = ROOT / "data" / "curated" | |
| PROCESSED = ROOT / "data" / "processed" | |
| FILES = ("caselaw.json", "enf.json", "enf-fr.json", "irb_papers.json", "irb_guidelines.json", | |
| "pdi.json", "policy.json", "commentary.json") | |
| CURATED_WITH_AUTHORITIES = ("us_equivalency.json", "us_dispositions.json", "transborder.json") | |
| def registry(): | |
| return json.loads((CURATED / "authorities.json").read_text(encoding="utf-8")) | |
| def curated(name): | |
| path = CURATED / name | |
| return json.loads(path.read_text(encoding="utf-8")) if path.exists() else None | |
| def cited(obj, out): | |
| """Every authority object ({cite, court|holding, ...}) in a curated file.""" | |
| if isinstance(obj, dict): | |
| if isinstance(obj.get("cite"), str) and ("holding" in obj or "court" in obj): | |
| out.append(obj) | |
| for v in obj.values(): | |
| cited(v, out) | |
| elif isinstance(obj, list): | |
| for v in obj: | |
| cited(v, out) | |
| return out | |
| _CHUNKS = {} | |
| def corpus(): | |
| """{chunk id: (file, chunk)} over the files the registry points into.""" | |
| if not _CHUNKS: | |
| for f in FILES: | |
| try: | |
| for c in json.loads((PROCESSED / f).read_text(encoding="utf-8")): | |
| _CHUNKS[c["id"]] = (f, c) | |
| except (OSError, ValueError): | |
| continue | |
| return _CHUNKS | |
| class RegistryCorpusTests(unittest.TestCase): | |
| """Every row against data/processed (both baselines).""" | |
| def setUpClass(cls): | |
| cls.reg = registry() | |
| cls.chunks = corpus() | |
| if not any(f == "caselaw.json" for f, _ in cls.chunks.values()): | |
| raise unittest.SkipTest("caselaw.json not readable here") | |
| cls.decisions = {(c.get("act_short"), c.get("act_code")) | |
| for f, c in cls.chunks.values() if f == "caselaw.json"} | |
| cls.strict = A.renderable(cls.reg) | |
| def test_decision_keys_resolve(self): | |
| missing = [] | |
| for key, row in self.reg["authorities"].items(): | |
| dk = row.get("decision_key") | |
| if row["status"] == "in-corpus": | |
| if not dk or (dk["act_short"], dk["act_code"]) not in self.decisions: | |
| missing.append(key) | |
| else: | |
| self.assertFalse(dk, key) | |
| self.assertFalse(row.get("pinned"), key) | |
| self.assertEqual(missing, []) | |
| def test_pins_belong_to_their_decision_and_contain_the_phrase(self): | |
| """Verbatim on the rendered registry (the phrase is what an officer is | |
| shown); after quote normalisation on the approved one.""" | |
| bad = [] | |
| for key, row in self.reg["authorities"].items(): | |
| dk = row.get("decision_key") or {} | |
| for p in row.get("pinned") or []: | |
| got = self.chunks.get(p["chunk"]) | |
| if got is None: | |
| bad.append((key, p["chunk"], "missing")) | |
| continue | |
| f, c = got | |
| if (c.get("act_short"), c.get("act_code")) != (dk.get("act_short"), dk.get("act_code")): | |
| bad.append((key, p["chunk"], "another decision")) | |
| text = c.get("text") or "" | |
| if self.strict: | |
| ok = p["contains"] in text | |
| else: | |
| ok = A._norm(p["contains"]).lower() in A._norm(text).lower() | |
| if not ok: | |
| bad.append((key, p["chunk"], p["contains"])) | |
| self.assertEqual(bad, []) | |
| def test_secondary_sources_exist_and_are_ordered(self): | |
| order = self.reg["secondary_status_order"] | |
| kinds = {"in-corpus decision": {"caselaw.json"}, | |
| "archived IRB paper": {"irb_papers.json"}, | |
| "withdrawn (unofficial ATIP copy)": {"enf.json", "enf-fr.json"}} | |
| bad = [] | |
| for key, row in self.reg["authorities"].items(): | |
| last = -1 | |
| for s in row.get("secondary") or []: | |
| got = self.chunks.get(s["chunk"]) | |
| if got is None: | |
| bad.append((key, s["chunk"], "missing")) | |
| continue | |
| if s["status"] not in order: | |
| bad.append((key, s["chunk"], s["status"])) | |
| continue | |
| i = order.index(s["status"]) | |
| if i < last: | |
| bad.append((key, s["chunk"], "out of order")) | |
| last = max(last, i) | |
| if s["status"] in kinds and got[0] not in kinds[s["status"]]: | |
| bad.append((key, s["chunk"], f"{s['status']} in {got[0]}")) | |
| self.assertEqual(bad, []) | |
| def test_aliases_are_bidirectional(self): | |
| rows = self.reg["authorities"] | |
| bad = [] | |
| for cite, key in self.reg["aliases"].items(): | |
| if key not in rows: | |
| bad.append((cite, key, "no row")) | |
| elif cite not in rows[key].get("cites", []) + rows[key].get("parallel_cites", []): | |
| bad.append((cite, key, "not in the row's cites")) | |
| for key, row in rows.items(): | |
| for cite in row.get("cites", []): | |
| if self.reg["aliases"].get(cite) != key: | |
| bad.append((key, cite, "cite not aliased")) | |
| self.assertEqual(bad, []) | |
| def test_counts_match(self): | |
| got = {} | |
| for row in self.reg["authorities"].values(): | |
| got[row["status"]] = got.get(row["status"], 0) + 1 | |
| self.assertEqual(got, self.reg["counts"]) | |
| def test_every_curated_cite_resolves(self): | |
| """0 unlinked: a cite a curated file uses that the registry does not | |
| hold would print 'unlinked' beside an officer's answer.""" | |
| loaded = A.load() | |
| unlinked = [] | |
| for name in CURATED_WITH_AUTHORITIES: | |
| data = curated(name) | |
| for a in cited(data, []) if data else []: | |
| if A.lookup(a["cite"], loaded)[0] is None: | |
| unlinked.append((name, a["cite"])) | |
| self.assertEqual(unlinked, []) | |
| def test_the_statute_the_corrected_holdings_name(self): | |
| """Popic: Criminal Code s. 393(3); Nshogoza: ss. 127(1) and 139(2).""" | |
| c46 = {} | |
| for c in json.loads((PROCESSED / "C-46.json").read_text(encoding="utf-8")): | |
| if c.get("lang", "en") == "en": | |
| c46.setdefault(str(c["section"]), []).append(c["text"]) | |
| self.assertTrue(any("unlawfully obtains transportation by land, water or air" in t | |
| for t in c46["393"])) | |
| self.assertTrue(any("to obstruct, pervert or defeat the course of justice" in t | |
| for t in c46["139"])) | |
| self.assertTrue(any("disobeys a lawful order made by a court of justice" in t | |
| for t in c46["127"])) | |
| class CorrectionsTests(unittest.TestCase): | |
| """The corrections the registry records are in the curated files (the | |
| A12 notes claimed them; they were never applied until 2026-09-15).""" | |
| def setUpClass(cls): | |
| cls.reg = registry() | |
| if not A.renderable(cls.reg): | |
| raise unittest.SkipTest("registry not rendered (the approved f5cf54d baseline)") | |
| cls.text = {n: (CURATED / n).read_text(encoding="utf-8") | |
| for n in CURATED_WITH_AUTHORITIES if (CURATED / n).exists()} | |
| def test_every_correction_is_applied(self): | |
| bad = [] | |
| for key, row in self.reg["authorities"].items(): | |
| for c in row.get("corrected") or []: | |
| blob = json.dumps(json.loads(self.text[c["file"]]), ensure_ascii=False) | |
| if c["old"] in blob: | |
| bad.append((key, c["file"], "old string still there", c["old"])) | |
| if c["new"] not in blob: | |
| bad.append((key, c["file"], "new string missing", c["new"])) | |
| self.assertEqual(bad, []) | |
| def test_a_note_claiming_a_correction_carries_the_correction(self): | |
| claim = re.compile(r"\b(corrected|completed|removed)\b[^.]*\b20\d\d-\d\d-\d\d", re.I) | |
| bare = [k for k, row in self.reg["authorities"].items() | |
| if any(claim.search(n) for n in row.get("notes") or []) and not row.get("corrected")] | |
| self.assertEqual(bare, []) | |
| def test_the_verifier_named_holdings(self): | |
| eq = json.dumps(curated("us_equivalency.json"), ensure_ascii=False) | |
| disp = json.dumps(curated("us_dispositions.json"), ensure_ascii=False) | |
| for want in ("The judicial review was DISMISSED", "s. 393(3)", "(paras 13-18;", | |
| "Garcia 2021 FC 141, paras 51-52", "presumption against retrospectivity"): | |
| self.assertIn(want, eq) | |
| for want in ("The Whatcom County Superior Court vacated", | |
| "Barnett v Canada (MCI) (1996), 33 Imm LR (2d) 1, 109 FTR 154 (FCTD)"): | |
| self.assertIn(want, disp) | |
| self.assertNotIn("Taiwan", disp) | |
| def test_pin_labels_are_officer_facing(self): | |
| bad = [(k, p["pin"]) for k, row in self.reg["authorities"].items() | |
| for p in row.get("pinned") or [] if re.search(r"curated|->|wrong", p["pin"])] | |
| self.assertEqual(bad, []) | |
| class RenderTests(unittest.TestCase): | |
| """The registry as the tools print it (roadmap T3.2).""" | |
| def setUpClass(cls): | |
| from canlex import server | |
| cls.server = server | |
| cls.reg = A.load() | |
| if not A.renderable(cls.reg): | |
| raise unittest.SkipTest("registry not rendered (the approved f5cf54d baseline)") | |
| cls.patch = mock.patch.object(server, "_index", side_effect=RuntimeError("no index")) | |
| cls.patch.start() | |
| server._load_equivalency() | |
| def tearDownClass(cls): | |
| cls.patch.stop() | |
| def test_no_pointer_names_canlex_case(self): | |
| for name in CURATED_WITH_AUTHORITIES: | |
| for a in cited(curated(name) or {}, []): | |
| res = A.resolve(a["cite"], self.reg) | |
| text = A.pointer(res, a.get("pin")) | |
| self.assertNotIn("canlex_case", text, a["cite"]) | |
| self.assertEqual("full text in CanLex" in text, | |
| res["status"] == "in-corpus" and res["present"] is not False, | |
| a["cite"]) | |
| def test_the_disposition_answer_carries_pointers_and_the_read_note(self): | |
| out = self.server.canlex_us_disposition("expungement") | |
| self.assertIn("full text in CanLex: canlex_search_legislation doc_type caselaw, " | |
| "act='Burgon'", out) | |
| self.assertIn("not in the CanLex corpus; quoted in Saini 2001 FCA 311, para 9", out) | |
| self.assertIn(A.READ_NOTE, out) | |
| out = self.server.canlex_us_disposition("deferred adjudication", state="TX") | |
| self.assertNotIn("canlex_case", out) | |
| def test_the_equivalency_block_carries_pointers(self): | |
| out = self.server.canlex_us_equivalency("perjury") | |
| self.assertIn("The judicial review was DISMISSED", out) | |
| self.assertIn("act='Nshogoza'; pinned para 55 (application for judicial review " | |
| "dismissed)", out) | |
| self.assertIn(A.READ_NOTE, out) | |
| def test_an_unrendered_registry_prints_bare_bullets(self): | |
| auths = [{"cite": "Nshogoza v Canada (MCI), 2015 FC 1211", "court": "FC", "holding": "h"}] | |
| plain = {k: v for k, v in self.reg.items() if k != "render"} | |
| self.assertEqual(A.authority_lines(auths, registry=plain), | |
| ["- Nshogoza v Canada (MCI), 2015 FC 1211 (FC): h"]) | |
| self.assertEqual(A.read_note(plain), A.NO_POINTER_NOTE) | |
| self.assertEqual(A.superseded_entries(plain), []) | |
| with mock.patch.object(A, "load", return_value=None): | |
| self.assertIn(A.HELD_NOTE, A.authority_lines(auths)[-1]) | |
| def test_review_notes_are_never_printed(self): | |
| res = A.resolve("Abid v Canada (MCI), 2011 FC 164", self.reg) | |
| text = A.pointer(res, None) | |
| self.assertIn("pinned paras 13-14 ('area of intersection')", text) | |
| self.assertNotIn("curated", text) | |
| self.assertIsNone(A._officer_pin({"pin": "para 13 (curated source) -> 'x'"})) | |
| def test_verified_text_quotes_the_general_life_clause(self): | |
| for section in ("344", "273"): | |
| r = A.statute_sentence("C-46", section, "imprisonment for life") | |
| self.assertIn("in any other case, to imprisonment for life", r["sentence"], section) | |
| self.assertNotIn("restricted firearm", r["sentence"], section) | |
| head, _, tail = r["sentence"].partition(" ... ") | |
| chunk = " ".join(A._norm(c["text"]) for c in A._act_sections("C-46")[section]) | |
| self.assertIn(head, chunk) | |
| self.assertIn(tail, chunk) | |
| def test_verified_text_prints_under_the_maximum(self): | |
| out = self.server.canlex_us_equivalency("robbery") | |
| i = out.index("Maximum penalty (verified against the corpus text)") | |
| self.assertIn("VERIFIED TEXT: C-46 s. 344 (current to ", out[i:]) | |
| def test_the_registry_supplies_the_saini_note(self): | |
| reg = A.superseded_entries() | |
| const = self.server.SUPERSEDED_CITES | |
| self.assertEqual(len(reg), len(const)) | |
| for k in ("note", "appellate_citation", "read_present", "read_absent", "read_unknown"): | |
| self.assertEqual(reg[0][k], const[0][k], k) | |
| self.assertEqual(reg[0]["pattern"].pattern, const[0]["pattern"].pattern) | |
| self.assertEqual(self.server._superseded_specs()[0]["id"], "saini-2000-fctd") | |
| def test_the_all_index_tags_every_row(self): | |
| out = self.server.canlex_us_disposition("all") | |
| rows = [l for l in out.splitlines() if l.startswith(" - ")] | |
| self.assertTrue(rows) | |
| tag = re.compile(r" \[(verified \d{4}-\d{2}-\d{2}|NOT VERIFIED|unverified)\]$") | |
| self.assertEqual([l for l in rows if not tag.search(l)], []) | |
| def test_commentary_prints_brief_pointers(self): | |
| from canlex.commentary import _entry_text | |
| e = next(e for e in curated("us_dispositions.json")["dispositions"] | |
| if e["id"] == "expungement") | |
| text = _entry_text(e) | |
| self.assertIn("full text in CanLex: canlex_search_legislation doc_type caselaw, " | |
| "act='Drake'; pinned", text) | |
| self.assertNotIn("\"", text.split("act='Drake'; pinned", 1)[1].split("\n", 1)[0]) | |
| class FileForTests(unittest.TestCase): | |
| """authorities._file_for names the processed file holding a chunk id; the | |
| manuals' ids (cp-14-1-1, op-22-fr-5-1) fell through to caselaw.json until | |
| 2026-10-23, and until 2026-10-24 the guidelines' ids (irbg-2-1-p1) did | |
| too while the IRB decisions in caselaw.json (irb-TB4-05778-36) were sent | |
| to irb_guidelines.json. An unreadable file (a git-LFS pointer) is skipped.""" | |
| def test_each_file_maps_its_own_ids_back_to_itself(self): | |
| for name in ("caselaw.json", "enf.json", "enf-fr.json", "irb_papers.json", | |
| "irb_guidelines.json", "pdi.json", "policy.json", "commentary.json", | |
| "manuals.json", "manuals-fr.json", "obs.json", "obs-fr.json"): | |
| try: | |
| chunks = json.loads((PROCESSED / name).read_text(encoding="utf-8")) | |
| except (OSError, ValueError): | |
| continue | |
| ids = [c["id"] for c in chunks] | |
| self.assertEqual([i for i in ids if A._file_for(i) != name], [], name) | |
| if __name__ == "__main__": | |
| unittest.main() | |