| """The pure chain, the decisive facts, the confidence block, the result card |
| and completion IF SERVED AS IMPOSED (roadmap T4.1, T4.2, T4.7, T4.5; track A |
| step A14, 2026-09-15). |
| |
| The chain is a refactor with no verdict change of its own (proved against |
| the tree before the step: every screening case and a 60-record set gave the |
| same CANLEX_RESULT fields and the same memo text once the new blocks are |
| removed -- scratchpad a14_diff.py). These tests hold what the step adds: |
| - the decisive facts list only facts whose values change the branch, the |
| allegation paragraph or the rehabilitation status, and fold into ASSUMED; |
| - every screening tool's CANLEX_RESULT carries 'confidence'; |
| - detail 'card' gives about ten lines and the same result line; |
| - the IF SERVED EXACTLY AS IMPOSED footing never gives a status, and a fine |
| blocks it; its Guide 5312 rules are verbatim in the corpus; |
| - conviction_date bounds the Tran maximum; |
| - all of it runs on the APPROVED curated JSON (a composed release). |
| Offline over the real curated files and a stub index.""" |
| import datetime as dt |
| import json |
| import subprocess |
| import unittest |
| from pathlib import Path |
| from unittest import mock |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| PROCESSED = ROOT / "data" / "processed" |
| T = "2026-09-15" |
|
|
|
|
| def result_of(out): |
| lines = [l for l in out.splitlines() if l.startswith("CANLEX_RESULT: ")] |
| return json.loads(lines[-1][len("CANLEX_RESULT: "):]) |
|
|
|
|
| def _chunks(name): |
| path = PROCESSED / name |
| if not path.exists(): |
| return None |
| data = json.loads(path.read_text(encoding="utf-8")) |
| return data["chunks"] if isinstance(data, dict) else data |
|
|
|
|
| class _Base(unittest.TestCase): |
| @classmethod |
| def setUpClass(cls): |
| from canlex import server |
| cls.s = server |
| cls._saved = server._INDEX |
| server._INDEX = object.__new__(type("Stub", (), {"semantic": False})) |
|
|
| @classmethod |
| def tearDownClass(cls): |
| cls.s._INDEX = cls._saved |
|
|
| def record(self, **kw): |
| kw.setdefault("today", T) |
| return self.s.canlex_screen_record(**kw) |
|
|
| def history(self, entries, **kw): |
| kw.setdefault("today", T) |
| return self.s.canlex_screen_history(json.dumps(entries), **kw) |
|
|
| @staticmethod |
| def decisive(r): |
| return {d["fact"]: d for d in r["decisive_missing"]} |
|
|
|
|
| class ChainTests(_Base): |
| def test_the_chain_without_text_gives_the_tools_answer(self): |
| """render=False (the decisive re-runs) and the memo agree.""" |
| from canlex.screening import parse_date |
| cases = [dict(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2016-01-01"), |
| dict(offense="theft", disposition="convicted", state="TX", |
| offence_date="2018-01-01", sentence_completed="2018-06-01"), |
| dict(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01"), |
| dict(offense="assault", disposition="withheld adjudication", state="FL", |
| offence_date="2016-01-01"), |
| dict(offense="attempted burglary", disposition="convicted", state="CA", |
| offence_date="2005-01-01", sentence_completed="2007-01-01")] |
| for c in cases: |
| r = result_of(self.record(**c)) |
| j = self.s._jurisdiction(c.get("state"), None) |
| ch = self.s._chain_entry(c["offense"], c["disposition"], j, |
| parse_date(c.get("offence_date")), |
| parse_date(c.get("sentence_completed")), |
| "foreign-national", dt.date.fromisoformat(T)) |
| eq = ch["eq"] or {} |
| self.assertEqual((eq.get("pairing"), eq.get("branch"), eq.get("max_years")), |
| (r["pairing"], r["branch"], r["max_years"]), c) |
| self.assertEqual((ch["rehab"] or {}).get("status"), r.get("rehab_status"), c) |
|
|
| def test_allegation_paragraphs_are_the_ones_drafted(self): |
| p = {"canadian_offence": "X"} |
| for branch in ("serious-criminality", "criminality", "criminality-two-plus-only"): |
| for conv, unknown in ((True, False), (False, False), (False, True)): |
| for count in ("one", "two-plus-summary-only"): |
| lines = " ".join(self.s._allegation_lines({"branch": branch}, p, conv, |
| unknown, count)) |
| paras = [x for x, _ in self.s._allegation_paras(branch, conv, unknown, count)] |
| for para in paras: |
| self.assertIn(f"IRPA paragraph {para} ", lines, (branch, conv, count)) |
| self.assertEqual(lines.count("Draft allegation paragraph"), len(paras), |
| (branch, conv, unknown, count)) |
|
|
|
|
| class DecisiveFactsTests(_Base): |
| def test_a_straddled_maximum_lists_the_offence_date(self): |
| r = result_of(self.record(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01")) |
| d = self.decisive(r)["offence_date"] |
| self.assertEqual(sorted(v["branch"] for v in d["values"]), |
| ["criminality", "serious-criminality"]) |
| self.assertEqual({tuple(v["paragraphs"]) for v in d["values"]}, |
| {("36(2)(b)",), ("36(1)(b)",)}) |
|
|
| def test_a_tie_lists_its_deciding_fact(self): |
| r = result_of(self.record(offense="theft", disposition="convicted", state="TX", |
| offence_date="2018-01-01", sentence_completed="2018-06-01")) |
| d = self.decisive(r)["value_over_5000"] |
| self.assertEqual(sorted(v["branch"] for v in d["values"]), |
| ["criminality", "serious-criminality"]) |
| |
| r2 = result_of(self.record(offense="theft", disposition="convicted", state="TX", |
| offence_date="2018-01-01", sentence_completed="2018-06-01", |
| value_over_5000=False)) |
| self.assertNotIn("value_over_5000", self.decisive(r2)) |
|
|
| def test_status_other_record_and_relief_fold_into_assumed(self): |
| out = self.record(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2016-01-01") |
| r = result_of(out) |
| dec = self.decisive(r) |
| self.assertEqual({v["value"]: v["branch"] for v in dec["status"]["values"]}, |
| {"foreign-national": "criminality", |
| "permanent-resident": "not-inadmissible-pr"}) |
| self.assertEqual(dec["status"]["assumed"], "status-foreign-national") |
| self.assertEqual({v["value"]: v["rehab_status"] for v in dec["other_record"]["values"]}, |
| {False: ["deemed-rehabilitated"], True: ["apply-conditional"]}) |
| self.assertIn("relief", dec) |
| |
| top = out.split("## Step 1")[0] |
| self.assertIn(self.s.DECISIVE_HEAD, top) |
| self.assertLess(top.index(self.s.DECISIVE_HEAD), top.index("ASSUMED (not given")) |
| for code in ("status-foreign-national", "clean-record", "no-foreign-relief"): |
| self.assertIn(self.s._ASSUMED_TEXT[code] + self.s.DECISIVE_SUFFIX, top) |
| |
| self.assertEqual(r["assumed"], ["status-foreign-national", "application-unknown", |
| "adult-at-offence", "no-foreign-relief", |
| "clean-record"]) |
|
|
| def test_a_fact_given_is_not_re_run(self): |
| r = result_of(self.record(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2016-01-01", |
| status="permanent-resident", other_record=True)) |
| self.assertNotIn("status", r["decisive_checked"]) |
| self.assertNotIn("other_record", r["decisive_checked"]) |
|
|
| def test_the_age_is_re_run_and_never_decisive_for_a_foreign_record(self): |
| r = result_of(self.record(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2016-01-01")) |
| self.assertIn("age", r["decisive_checked"]) |
| self.assertNotIn("age", self.decisive(r)) |
|
|
| def test_no_decisive_fact_is_said(self): |
| out = self.record(offense="trespass", disposition="convicted", state="TX", |
| offence_date="2018-01-01", sentence_completed="2018-02-01", |
| status="foreign-national", other_record=True, |
| date_of_birth="1980-01-01") |
| r = result_of(out) |
| if not r["decisive_missing"]: |
| self.assertIn("THE ANSWER CHANGES WITH THESE FACTS: none found", out) |
|
|
| def test_only_changing_facts_are_kept(self): |
| for kw in (dict(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01"), |
| dict(offense="assault", disposition="convicted", state="NY", |
| offence_date="2016-01-01", sentence_completed="2017-01-01")): |
| r = result_of(self.record(**kw)) |
| for d in r["decisive_missing"]: |
| answers = {(v["branch"], tuple(v["paragraphs"]), tuple(v["rehab_status"])) |
| for v in d["values"]} |
| self.assertGreater(len(answers), 1, d) |
|
|
| def test_a_statute_codes_specification_is_re_run_with_its_answer(self): |
| """The step memo keys a match on the record context too: the former |
| UCMJ Art. 90's limbs are read from the specification answer, so a |
| match made without it must never be reused for a reading that has it |
| (found by the before/after diff, 2026-09-15). The Art. 90 element |
| split is held for review (85802df): skipped where it is absent, as a |
| composed release on the approved baseline is.""" |
| alias = ((self.s._load_equivalency() or {}).get("aliases") or {}).get("article 90") or {} |
| if not any(isinstance(r, dict) and r.get("elements") |
| for r in [alias] + list(alias.get("dated") or [])): |
| self.skipTest("held content absent: the UCMJ Art. 90 element split") |
| out = self.record(offense="Article 90", disposition="court-martial conviction", |
| offence_date="2016-01-01", sentence_completed="2017-01-01") |
| r = result_of(out) |
| self.assertEqual({e["answer"]: e["pairing"] for e in r["element_readings"]}, |
| {"assault": "assault-simple", "disobedience": "military-offences"}) |
| dec = self.decisive(r)["ucmj-specification"] |
| self.assertEqual({v["value"]: v["branch"] for v in dec["values"]}, |
| {"ucmj-specification: assault": "tie", |
| "ucmj-specification: disobedience": "none"}) |
|
|
| def test_a_canadian_record_re_runs_status(self): |
| r = result_of(self.record(offense="impaired driving", disposition="convicted", |
| country="Canada", offence_date="2019-06-01", |
| sentence_completed="2020-01-01", canada_sentence="fine")) |
| self.assertIn("status", r["decisive_checked"]) |
| self.assertIn("confidence", r) |
|
|
| def test_history_lists_an_entrys_offence_date_and_the_status(self): |
| r = result_of(self.history([ |
| dict(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01"), |
| dict(offense="theft", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2015-06-01", |
| value_over_5000=False)])) |
| dec = self.decisive(r) |
| self.assertIn("#1 offence_date", dec) |
| self.assertEqual({v["count"] for v in dec["#1 offence_date"]["values"]}, |
| {"multiple"}) |
| self.assertIn("#1 relief", r["decisive_checked"]) |
| self.assertIn("status", r["decisive_checked"]) |
| for d in r["decisive_missing"]: |
| self.assertGreater(len({json.dumps([v["entries"], v["count"], v["rehab_status"]]) |
| for v in d["values"]}), 1, d) |
|
|
|
|
| class ConfidenceTests(_Base): |
| KEYS = {"step1", "step2", "step3"} |
|
|
| def test_every_screening_tool_carries_it(self): |
| outs = { |
| "canlex_us_disposition": self.s.canlex_us_disposition(disposition="deferred adjudication", |
| state="TX"), |
| "canlex_us_equivalency": self.s.canlex_us_equivalency(offense="DUI", |
| offence_date="2015-01-01"), |
| "canlex_rehabilitation": self.s.canlex_rehabilitation( |
| basis="foreign-conviction", max_penalty_lt_10=True, count="one", |
| clean_record=True, sentence_completed="2016-01-01", today=T), |
| "canlex_screen_record": self.record(offense="DUI", disposition="convicted", |
| state="TX", offence_date="2015-01-01", |
| sentence_completed="2016-01-01"), |
| "canlex_screen_history": self.history([dict(offense="DUI", disposition="convicted", |
| state="TX", offence_date="2015-01-01", |
| sentence_completed="2016-01-01")]), |
| "canlex_transborder": self.s.canlex_transborder(act="Criminal Code", section="95", |
| offence="loaded handgun"), |
| } |
| for tool, out in outs.items(): |
| r = result_of(out) |
| self.assertEqual(r["tool"], tool) |
| self.assertEqual(set(r["confidence"]), self.KEYS, tool) |
|
|
| def test_the_values(self): |
| c = result_of(self.record(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", |
| sentence_completed="2016-01-01"))["confidence"] |
| self.assertEqual(c["step1"][0]["entry"], "conviction") |
| self.assertIn(c["step1"][0]["entry_status"], |
| ("settled", "judicially-considered", "guidance-only", "no-authority")) |
| s2 = c["step2"][0] |
| self.assertEqual((s2["pairing"], s2["maximum_source"]), ("dui", "history")) |
| self.assertRegex(s2["authorities_in_corpus"] or "0/0", r"^\d+/\d+$") |
| self.assertEqual(c["step3"], {"limbs_tested": False, "limbs_unknown": None}) |
| c2 = result_of(self.record(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2020-06-01", |
| sentence_completed="2021-06-01"))["confidence"] |
| self.assertEqual(c2["step2"][0]["maximum_source"], "current") |
| tie = result_of(self.record(offense="theft", disposition="convicted", state="TX", |
| offence_date="2018-01-01"))["confidence"] |
| self.assertEqual(sorted(x["pairing"] for x in tie["step2"]), ["theft-over", "theft-under"]) |
| h = result_of(self.history([dict(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", |
| sentence_completed="2016-01-01")]))["confidence"] |
| self.assertEqual(h["step1"][0]["n"], 1) |
| self.assertTrue(h["step3"]["limbs_tested"]) |
| self.assertIsInstance(h["step3"]["limbs_unknown"], int) |
|
|
| def test_a_verified_state_row_says_so(self): |
| from canlex.screening import step1_confidence |
| self.assertTrue(step1_confidence({"id": "x"}, {"verified": "2026-09-01"})["row_verified"]) |
| self.assertFalse(step1_confidence({"id": "x"}, {"unverifiable": "why"})["row_verified"]) |
| self.assertIsNone(step1_confidence({"id": "x"}, None)["row_verified"]) |
|
|
|
|
| class CardTests(_Base): |
| def body(self, out): |
| return [l for l in out.splitlines() if not l.startswith("CANLEX_RESULT: ")] |
|
|
| def test_the_card_is_short_and_carries_the_same_result(self): |
| kw = dict(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2016-01-01") |
| card = self.record(detail="card", **kw) |
| full = self.record(**kw) |
| body = self.body(card) |
| self.assertLessEqual(len(body), 13) |
| for head in ("# Result card", "Result:", "Step 1:", "Dates:", "Action:"): |
| self.assertTrue(any(l.startswith(head) for l in body), head) |
| self.assertEqual(body[-1], self.s.CARD_FULL) |
| rc, rf = result_of(card), result_of(full) |
| self.assertEqual(rc.pop("detail"), "card") |
| self.assertEqual(rc, rf) |
| self.assertNotIn("detail", rf) |
|
|
| def test_not_settled_names_the_deciding_fact(self): |
| card = self.record(detail="card", offense="theft", disposition="convicted", state="TX", |
| offence_date="2018-01-01", sentence_completed="2018-06-01") |
| result = next(l for l in self.body(card) if l.startswith("Result:")) |
| self.assertIn("NOT SETTLED", result) |
| self.assertIn("CAD 5,000", result) |
| self.assertIn(self.s.CARD_NOT_SETTLED_ACTION, card) |
|
|
| def test_an_open_result_says_the_assumed_facts_were_not_compared_per_reading(self): |
| """Wave-4 verifier (2026-09-15): on a tie or an open limb the status and |
| relief re-runs compare the whole answer, not each reading -- the memo, |
| the card and CANLEX_RESULT say so rather than implying they cannot |
| matter.""" |
| kw = dict(offense="assault causing bodily injury", disposition="convicted", |
| state="TX", offence_date="2016-01-01", sentence_completed="2016-06-01") |
| full, card = self.record(**kw), self.record(detail="card", **kw) |
| r = result_of(full) |
| self.assertEqual(r["branch"], "tie") |
| self.assertIn("status", r["decisive_whole_answer_only"]) |
| self.assertIn("relief", r["decisive_whole_answer_only"]) |
| self.assertIn("NOTE -- NOT SETTLED at step 2: the re-runs of status", full) |
| self.assertIn("compared on the whole answer only, not reading by reading", card) |
| settled = result_of(self.record(offense="robbery", disposition="convicted", state="IL", |
| offence_date="2001-01-01", |
| sentence_completed="2004-01-01")) |
| self.assertNotIn("decisive_whole_answer_only", settled) |
|
|
| def test_the_actions(self): |
| fn = self.record(detail="card", offense="DUI", disposition="convicted", state="TX", |
| offence_date="2020-06-01", sentence_completed="2021-06-01", |
| status="foreign-national") |
| self.assertIn("s. 44(1) report on IRPA s. 36(1)(b)", fn) |
| self.assertIn("IRPR s. 42(1)", fn) |
| |
| |
| self.assertIn("the officer 'shall allow' it -- mandatory unless s. 42(2) applies", fn) |
| pr = self.record(detail="card", offense="robbery", disposition="convicted", state="IL", |
| offence_date="2001-01-01", sentence_completed="2004-01-01", |
| status="permanent-resident") |
| self.assertIn("the Minister may refer it to the Immigration Division (IRPA s. 44(2))", |
| pr) |
| self.assertNotIn("is referred to the Immigration Division", pr) |
| none = self.record(detail="card", offense="trespass", disposition="convicted", |
| state="TX", offence_date="2018-01-01", |
| sentence_completed="2018-02-01") |
| self.assertIn(self.s.CARD_NO_GROUND_ACTION, none) |
| deemed = self.record(detail="card", offense="DUI", disposition="convicted", state="TX", |
| offence_date="2015-01-01", sentence_completed="2016-01-01") |
| self.assertIn(self.s.CARD_DEEMED_ACTION, deemed) |
|
|
| def test_history_card(self): |
| ents = [dict(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2016-03-01", sentence_completed="2017-03-01"), |
| dict(offense="disorderly conduct", disposition="pled guilty", state="TX", |
| offence_date="2019-05-05", sentence_completed="2019-06-01")] |
| card = self.history(ents, detail="card") |
| body = self.body(card) |
| self.assertLessEqual(len(body), 12) |
| self.assertTrue(body[0].startswith("# Result card -- a record of 2 entries")) |
| self.assertTrue(any(l.startswith("Entries: #1") for l in body)) |
| rc, rf = result_of(card), result_of(self.history(ents)) |
| rc.pop("detail") |
| self.assertEqual(rc, rf) |
|
|
| def test_a_bad_detail_is_refused(self): |
| self.assertTrue(self.record(offense="DUI", disposition="convicted", |
| detail="short").startswith("Error: detail")) |
|
|
|
|
| class SentenceTests(unittest.TestCase): |
| def test_the_common_forms(self): |
| from canlex.screening import parse_sentence |
| got = {t: [(p["kind"], p["n"], p["unit"], p["suspended"]) |
| for p in parse_sentence(t)["parts"]] |
| for t in ("3 YRS PROB", "30 DAYS JAIL SUSP", "$500 FINE", "12 MOS DL SUSP")} |
| self.assertEqual(got, {"3 YRS PROB": [("probation", 3.0, "years", False)], |
| "30 DAYS JAIL SUSP": [("imprisonment", 30.0, "days", True)], |
| "$500 FINE": [("fine", None, None, False)], |
| "12 MOS DL SUSP": [("driving", 12.0, "months", False)]}) |
| self.assertEqual(parse_sentence("$500 FINE")["parts"][0]["amount"], 500.0) |
| self.assertEqual(parse_sentence("40 HRS COMMUNITY SERVICE")["unread"], |
| ["40 HRS COMMUNITY SERVICE"]) |
|
|
| def test_a_thousands_comma_in_a_fine_is_one_part(self): |
| """The code audit of 2026-09-21 (HELD): '$1,000 fine' split at the comma into |
| a $1 fine and an unread '500 fine'. Before/after on '2 yrs prob, $1,500 fine': |
| blocked ('cannot read: 500 fine') -> blocked (the fine rule, amount $1,500). |
| The readings the audit proposed for '6 mos jail, 3 mos susp', '1 yr 6 mos |
| jail' and '5 years tdcj' are NOT added here: each is a new reading of a |
| record's words for the maintainer to approve one by one, so they stay |
| unread (pinned as such).""" |
| from canlex.screening import parse_sentence, projected_completion |
| p = parse_sentence("$1,000 fine") |
| self.assertEqual(([x["kind"] for x in p["parts"]], p["parts"][0]["amount"], p["unread"]), |
| (["fine"], 1000.0, [])) |
| p = parse_sentence("2 yrs prob, $1,500 fine") |
| self.assertEqual([(x["kind"], x["amount"]) for x in p["parts"]], |
| [("probation", None), ("fine", 1500.0)]) |
| self.assertEqual(p["unread"], []) |
| self.assertEqual(parse_sentence("$1,000,000 fine")["parts"][0]["amount"], 1000000.0) |
| |
| self.assertEqual([x["kind"] for x in parse_sentence("2 yrs prob,180 days jail")["parts"]], |
| ["probation", "imprisonment"]) |
| blocked = projected_completion("2 yrs prob, $1,500 fine", dt.date(2020, 1, 1))["blocked"] |
| self.assertIn("$1,500 fine", blocked) |
| self.assertIn("Guide 5312", blocked) |
| self.assertNotIn("cannot read", blocked) |
| |
| for s, unread in (("6 mos jail, 3 mos susp", ["3 mos susp"]), |
| ("1 yr 6 mos jail", ["1 yr 6 mos jail"]), |
| ("5 years tdcj", ["5 years tdcj"])): |
| self.assertEqual(parse_sentence(s)["unread"], unread, s) |
|
|
| def test_the_projection(self): |
| from canlex.screening import projected_completion |
| d = dt.date(2019, 3, 1) |
| p = projected_completion("3 YRS PROB, 30 DAYS JAIL SUSP", d) |
| self.assertEqual((p["date"], p["rule"], p["blocked"]), (dt.date(2022, 3, 1), "probation", None)) |
| self.assertEqual(projected_completion("30 DAYS JAIL SUSP", d)["date"], d) |
| self.assertEqual(projected_completion("180 DAYS JAIL, 170 SUSP", d)["date"], |
| dt.date(2019, 3, 11)) |
| self.assertEqual(projected_completion("3 MONTHS JAIL", dt.date(2002, 12, 13))["date"], |
| dt.date(2003, 3, 13)) |
| dl = projected_completion("3 MOS JAIL; 12 MOS DL SUSP", d) |
| self.assertEqual((dl["date"], dl["if_administrative"]), |
| (dt.date(2020, 3, 1), dt.date(2019, 6, 1))) |
| self.assertIn("fine", projected_completion("2 years probation, $500 fine", d)["blocked"]) |
| self.assertIn("cannot read", projected_completion("40 HRS COMMUNITY SERVICE", d)["blocked"]) |
|
|
| def test_months_end_on_the_last_day(self): |
| from canlex.screening import add_months |
| self.assertEqual(add_months(dt.date(2019, 1, 31), 1), dt.date(2019, 2, 28)) |
| self.assertEqual(add_months(dt.date(2019, 11, 30), 3), dt.date(2020, 2, 29)) |
|
|
| def test_guide_5312_rules_are_verbatim(self): |
| from canlex.screening import GUIDE_5312_CHUNK, GUIDE_5312_COURT, GUIDE_5312_RULES |
| chunks = _chunks("policy.json") |
| if chunks is None: |
| self.skipTest("corpus not ingested") |
| text = next(c["text"] for c in chunks if c["id"] == GUIDE_5312_CHUNK) |
| for rule in list(GUIDE_5312_RULES.values()) + [GUIDE_5312_COURT]: |
| self.assertIn(rule, text) |
|
|
| def test_the_statute_phrases_the_card_and_questions_rely_on(self): |
| irpr, irpa = _chunks("SOR-2002-227.json"), _chunks("I-2.5.json") |
| if irpr is None or irpa is None: |
| self.skipTest("corpus not ingested") |
| s42 = next(c["text"] for c in irpr if c["id"] == "SOR-2002-227-s42") |
| self.assertIn("shall allow the foreign national to withdraw their application and " |
| "leave Canada", s42) |
| self.assertIn("If a report is being prepared or has been prepared under subsection " |
| "44(1) of the Act", s42) |
| s44 = next(c["text"] for c in irpa if c["id"] == "I-2.5-s44") |
| self.assertIn("the Minister may refer the report to the Immigration Division", s44) |
| irpa_text = "\n".join(c["text"] for c in irpa) |
| self.assertIn("A foreign national is inadmissible on grounds of criminality for", |
| irpa_text) |
|
|
|
|
| class ProjectionMemoTests(_Base): |
| KW = dict(offense="DUI", disposition="convicted", state="TX", offence_date="2019-06-01") |
|
|
| def test_no_verdict_on_the_footing(self): |
| out = self.record(sentence_imposed="3 YRS PROB, 30 DAYS JAIL SUSP", |
| sentence_date="2020-01-10", **self.KW) |
| r = result_of(out) |
| self.assertEqual(r["completion_basis"], "if-served-as-imposed") |
| self.assertIsNone(r.get("rehab_status")) |
| p = r["if_served_as_imposed"] |
| self.assertEqual((p["completion"], p["earliest_apply"], p["source"]), |
| ("2023-01-10", "2028-01-10", "policy-imm-5312-12")) |
| from canlex.screening import IF_SERVED_LABEL |
| self.assertIn(IF_SERVED_LABEL, out) |
| self.assertIn("No verdict is given on this footing", out) |
|
|
| def test_a_fine_blocks_it(self): |
| r = result_of(self.record(sentence_imposed="2 years probation, $500 fine", |
| sentence_date="2020-01-10", **self.KW)) |
| self.assertEqual(r["completion_basis"], "not-projected") |
| self.assertIn("fine", r["if_served_as_imposed"]["blocked"]) |
|
|
| def test_a_given_completion_wins(self): |
| r = result_of(self.record(sentence_imposed="3 YRS PROB", sentence_date="2020-01-10", |
| sentence_completed="2023-01-10", **self.KW)) |
| self.assertEqual(r["completion_basis"], "sentence_completed") |
| self.assertNotIn("if_served_as_imposed", r) |
|
|
| def test_the_margin(self): |
| r = result_of(self.record(sentence_imposed="2 YRS PROB", sentence_date="2019-12-01", |
| offense="DUI", disposition="convicted", state="TX", |
| offence_date="2019-06-01")) |
| |
| self.assertTrue(r["if_served_as_imposed"]["margin"]) |
|
|
|
|
| class ConvictionDateTests(_Base): |
| def test_it_bounds_the_tran_maximum(self): |
| base = dict(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01") |
| self.assertEqual(result_of(self.record(**base))["branch"], "tran-open") |
| out = self.record(conviction_date="2018-06-01", **base) |
| r = result_of(out) |
| self.assertEqual((r["branch"], r["max_years"], r["assessed_basis"]), |
| ("criminality", 5, "inferred-from-conviction")) |
| self.assertIn("inferred from the conviction date", out) |
| |
| self.assertEqual(result_of(self.record(conviction_date="2019-01-10", **base))["branch"], |
| "tran-open") |
|
|
| def test_out_of_order_dates_are_said(self): |
| out = self.record(offense="DUI", disposition="convicted", state="TX", |
| offence_date="2018-01-01", conviction_date="2017-06-01", |
| sentence_date="2017-01-01") |
| self.assertIn("WARNING: the conviction date 2017-06-01 precedes the offence date", out) |
| self.assertIn("WARNING: the sentence date 2017-01-01 precedes", out) |
|
|
| def test_history_reads_it_too(self): |
| out = self.history([dict(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01", |
| conviction_date="2018-06-01")]) |
| self.assertIn("maximum inferred from the conviction date (2018-06-01)", out) |
|
|
|
|
| def _git_json(commit, path): |
| try: |
| out = subprocess.run(["git", "-C", str(ROOT), "show", f"{commit}:{path}"], |
| capture_output=True, timeout=60) |
| except (OSError, subprocess.SubprocessError): |
| return None |
| return json.loads(out.stdout.decode("utf-8")) if out.returncode == 0 else None |
|
|
|
|
| class ApprovedBaselineTests(_Base): |
| """main's code on the approved curated JSON (what a composed release |
| ships): every new block runs, and says nothing it cannot.""" |
|
|
| @classmethod |
| def setUpClass(cls): |
| super().setUpClass() |
| try: |
| state = json.loads((ROOT / "data" / "curated" / "review_state.json") |
| .read_text(encoding="utf-8")) |
| commit = state["approved"]["commit"] |
| except (OSError, ValueError, KeyError): |
| raise unittest.SkipTest("no review state") |
| cls.eq = _git_json(commit, "data/curated/us_equivalency.json") |
| cls.disp = _git_json(commit, "data/curated/us_dispositions.json") |
| if cls.eq is None or cls.disp is None: |
| raise unittest.SkipTest("approved baseline not readable with git here") |
|
|
| @classmethod |
| def tearDownClass(cls): |
| cls.s._load_equivalency() |
| super().tearDownClass() |
|
|
| def test_the_new_blocks_run_on_the_approved_data(self): |
| from canlex.screening import load_aliases |
|
|
| def eq(): |
| load_aliases(self.eq.get("aliases")) |
| return self.eq |
| with mock.patch.object(self.s, "_load_equivalency", eq), \ |
| mock.patch.object(self.s, "_load_dispositions", lambda: self.disp): |
| for detail in ("full", "card"): |
| for kw in (dict(offense="DUI", disposition="convicted", state="TX", |
| sentence_completed="2019-06-01", conviction_date="2018-06-01"), |
| dict(offense="theft", disposition="convicted", state="TX", |
| offence_date="2018-01-01", sentence_imposed="3 YRS PROB", |
| sentence_date="2018-03-01"), |
| dict(offense="assault", disposition="deferred adjudication", |
| state="TX", offence_date="2014-01-01"), |
| dict(offense="Article 90", disposition="court-martial conviction"), |
| dict(offense="impaired driving", disposition="convicted", |
| country="Canada", offence_date="2019-06-01")): |
| r = result_of(self.record(detail=detail, **kw)) |
| self.assertIn("confidence", r, kw) |
| self.assertIn("decisive_missing", r, kw) |
| r = result_of(self.history([dict(offense="DUI", disposition="convicted", |
| state="TX", sentence_completed="2019-06-01"), |
| dict(offense="theft", disposition="convicted", |
| state="TX", offence_date="2016-01-01", |
| sentence_completed="2016-02-01")], |
| detail=detail)) |
| self.assertIn("confidence", r) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
