"""Unit tests for the curated-commentary pipeline (canlex/commentary.py and the canlex_us_disposition matcher in canlex/server.py). Offline: they run on a minimal in-memory dataset, not the curated file, and write only into a tempdir. python -m unittest discover -s tests """ import contextlib import io import json import sys import tempfile import unittest from pathlib import Path from canlex import commentary from canlex.commentary import BANNER, _entry_text from canlex.server import _match_dispositions def entry(**over): e = { "id": "test-disp", "names": ["withheld adjudication", "withholding of adjudication"], "is_conviction": "depends", "status": "no-authority", "analysis": "Body of the analysis.", "state_variations": [ {"state": "Florida", "note": "FL note", "is_conviction": "depends"}], "authorities": [ {"cite": "Case v Canada", "court": "FC", "pin": "para 1", "holding": "Held something."}], "guidance": [{"ref": "Guide", "note": "A note."}], "interpretation": "Reasoned view.", } e.update(over) return e class EntryTextTests(unittest.TestCase): def test_carries_banner_and_flags(self): text = _entry_text(entry()) self.assertTrue(text.startswith(BANNER)) # legacy 'depends' renders under the 5-point vocabulary self.assertIn("FACT-SPECIFIC", text) self.assertIn("NO AUTHORITY LOCATED", text) self.assertIn("INTERPRETATION (no direct authority", text) def test_five_point_verdicts_render(self): text = _entry_text(entry(is_conviction="likely-no", bottom_line="Check completion.")) self.assertIn("LIKELY NO", text) self.assertIn("BOTTOM LINE: Check completion.", text) def test_no_interpretation_block_when_empty(self): text = _entry_text(entry(interpretation="", status="settled")) self.assertNotIn("INTERPRETATION", text) class MatcherTests(unittest.TestCase): DATA = {"dispositions": [ entry(), entry(id="other", names=["state pardon"], state_variations=[]), ]} def test_matches_by_name_tokens(self): got = _match_dispositions(self.DATA, "court withheld adjudication", None) self.assertEqual(got[0]["id"], "test-disp") def test_state_boost_breaks_ties(self): got = _match_dispositions(self.DATA, "adjudication", "FL") self.assertTrue(got and got[0]["id"] == "test-disp") def test_no_match_returns_empty(self): self.assertEqual( _match_dispositions(self.DATA, "entirely unrelated words", None), []) def dataset(n): """A curated file holding n dispositions -> n + 2 chunks (one methodology entry, n disposition entries, one Florida state page).""" return {"reviewed": "2026-07-01", "methodology": [{"id": "m1", "title": "How this was built", "text": "Method body."}], "dispositions": [entry(id=f"d{i}", names=[f"disposition {i}"]) for i in range(n)]} class _TempCorpus: """Points the module's curated inputs and its output at a tempdir.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) root = Path(self.tmp.name) self.out = root / "commentary.json" self.curated = root / "us_dispositions.json" # No equivalency file: build() already treats it as optional, so the # chunk arithmetic below stays down to the dispositions dataset. for name, tmp_path in (("OUT", self.out), ("CURATED", self.curated), ("EQUIV", root / "us_equivalency.json")): self.addCleanup(setattr, commentary, name, getattr(commentary, name)) setattr(commentary, name, tmp_path) def curate(self, n): self.curated.write_text(json.dumps(dataset(n)), encoding="utf-8") def stored(self): return json.loads(self.out.read_text(encoding="utf-8")) def run_build(self, **kw): with contextlib.redirect_stdout(io.StringIO()) as out: ok = commentary.build(**kw) return ok, out.getvalue() class BuildWriteTests(_TempCorpus, unittest.TestCase): """The write path. build() re-renders the whole corpus from the curated files, so a truncated or half-edited dataset must not replace good chunks. """ def test_first_run_writes_and_reports_success(self): self.curate(3) ok, out = self.run_build() self.assertTrue(ok) self.assertEqual(len(self.stored()), 5) self.assertIn("commentary chunks", out) def test_collapsed_dataset_is_refused_and_leaves_the_corpus_alone(self): self.curate(20) self.run_build() self.curate(1) # dataset truncated to a stub ok, out = self.run_build() self.assertFalse(ok) self.assertIn("REFUSING", out) self.assertEqual(len(self.stored()), 22) def test_allow_shrink_lets_a_known_drop_through(self): self.curate(20) self.run_build() self.curate(1) ok, _out = self.run_build(allow_shrink=True) self.assertTrue(ok) self.assertEqual(len(self.stored()), 3) def test_normal_churn_still_writes(self): self.curate(20) self.run_build() self.curate(19) self.assertTrue(self.run_build()[0]) self.assertEqual(len(self.stored()), 21) def test_writes_one_space_indent(self): # The stored corpus is indent=1; any other value rewrites every line # of commentary.json and buries the real change in the diff. self.curate(1) self.run_build() self.assertTrue( self.out.read_text(encoding="utf-8").startswith('[\n {\n "id"')) def test_non_ascii_is_not_escaped(self): self.curate(1) self.run_build() self.assertIn("—", self.out.read_text(encoding="utf-8")) class MainExitTests(_TempCorpus, unittest.TestCase): """A refused write has to fail the process: refresh.py shells out to `py -m canlex.commentary` and only sees the exit code.""" def argv(self, *args): self.addCleanup(setattr, sys, "argv", sys.argv) sys.argv = ["canlex.commentary", *args] def run_main(self): with self.assertRaises(SystemExit) as caught: with contextlib.redirect_stdout(io.StringIO()): commentary.main() return caught.exception.code def test_exits_zero_on_a_good_run(self): self.curate(3) self.argv() self.assertEqual(self.run_main(), 0) def test_exits_non_zero_when_the_write_is_refused(self): self.curate(20) self.run_build() self.curate(1) self.argv() self.assertEqual(self.run_main(), 1) def test_allow_shrink_flag_is_parsed(self): self.curate(20) self.run_build() self.curate(1) self.argv("--allow-shrink") self.assertEqual(self.run_main(), 0) self.assertEqual(len(self.stored()), 3) if __name__ == "__main__": unittest.main()