File size: 7,240 Bytes
9fce526 6ce7899 9fce526 6ce7899 9fce526 6ce7899 9fce526 6ce7899 9fce526 0b22e9a 9fce526 0b22e9a 9fce526 6ce7899 9fce526 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | """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()
|