tests/test_engine_sweep.py

"""Every curated pairing through the step-2 engine, over the input space.

The example-based tests passed while 29 held pairings crashed on any dated
offence (penalty_history stored as null; roadmap T0.2). This sweep runs every
pairing x {no date, 2002-06-01, 2010-01-01, 2020-01-01} x {foreign national,
permanent resident} x {completed, attempted} through _resolve_pairing,
_pairing_block and _equivalency_report, and asserts that nothing raises, the
branch is a known code, and a band (max_penalty_lt_10) is asserted only when
the branch states one.

It runs twice: over the curated JSON in the working tree, and over the
APPROVED baseline's curated JSON (data/curated/review_state.json, read with
`git show`) -- the next deploy ships main's code with that approved data
(RUNBOOK "composed release"), so the code must run cleanly on it too.

Pure functions over the curated files and a stub index: no corpus, no model.
"""
import json
import subprocess
import traceback
import unittest
from pathlib import Path
from unittest import mock

from canlex import screening as sc

ROOT = Path(__file__).resolve().parent.parent
CURATED = ROOT / "data" / "curated"

DATES = (None, "2002-06-01", "2010-01-01", "2020-01-01")
STATUSES = ("foreign-national", "permanent-resident")
# Every code branch_for can return, plus 'tie' (roadmap T1.4 adds it for an
# unresolved tie; listed now so that fix does not have to touch the sweep).
KNOWN_BRANCHES = set(sc._BRANCH_TEXT) | {"tie"}
NO_BAND = {"depends", "unknown", "tie", "tran-open", "limb-open"}


def approved_commit():
    try:
        state = json.loads((CURATED / "review_state.json").read_text(encoding="utf-8"))
        return state["approved"]["commit"]
    except (OSError, ValueError, KeyError):
        return None


def git_show_json(commit, path):
    """The JSON at commit:path, or None when git or the commit is unavailable
    (a shallow clone, an exported tarball)."""
    try:
        out = subprocess.run(["git", "-C", str(ROOT), "show", f"{commit}:{path}"],
                             capture_output=True, timeout=60)
    except (OSError, subprocess.SubprocessError):
        return None
    if out.returncode != 0:
        return None
    return json.loads(out.stdout.decode("utf-8"))


def _result(out):
    lines = [l for l in out.splitlines() if l.startswith("CANLEX_RESULT: ")]
    return lines, (json.loads(lines[-1][len("CANLEX_RESULT: "):]) if lines else None)


class _SweepMixin:
    """Subclasses set `data` (the curated equivalency dataset) in setUpClass."""
    data = None

    @classmethod
    def setUpClass(cls):
        from canlex import server
        cls.server = server
        cls._saved_index = server._INDEX
        server._INDEX = object.__new__(type("Stub", (), {"semantic": False}))

    @classmethod
    def tearDownClass(cls):
        cls.server._INDEX = cls._saved_index
        # restore the working tree's alias table (the sweep may have loaded
        # the baseline's)
        cls.server._load_equivalency()

    def _band_problems(self, where, f):
        out = []
        branch, lt10, mx = f.get("branch"), f.get("max_penalty_lt_10"), f.get("max_years")
        if branch not in KNOWN_BRANCHES:
            out.append(f"{where}: unknown branch {branch!r}")
        if branch in NO_BAND and lt10 is not None:
            out.append(f"{where}: branch {branch} but max_penalty_lt_10={lt10}")
        if branch == "serious-criminality" and not (mx is not None and mx >= 10 and lt10 is False):
            out.append(f"{where}: serious-criminality with max_years={mx}, lt10={lt10}")
        if branch == "criminality" and not (mx is not None and mx < 10 and lt10 is True):
            out.append(f"{where}: criminality with max_years={mx}, lt10={lt10}")
        return out

    def test_resolve_and_block_every_pairing(self):
        s = self.server
        problems = []
        for p in self.data["pairings"]:
            base = (p.get("us_terms") or [p["id"]])[0]
            values = (None, True, False) if p.get("depends_on") == "value" else (None,)
            for offense in (base, f"attempted {base}"):
                for date in DATES:
                    for status in STATUSES:
                        for value in values:
                            where = f"{p['id']} / {offense!r} / {date} / {status} / value={value}"
                            try:
                                f = s._resolve_pairing(p, offense, date, status, value)
                                text, fields = s._pairing_block(p, offense, date, status,
                                                                value_over=value)
                                json.dumps(fields)
                            except Exception:
                                problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                continue
                            problems += self._band_problems(where, f)
                            if fields.get("branch") != f.get("branch"):
                                problems.append(f"{where}: block branch {fields.get('branch')} "
                                                f"!= resolved {f.get('branch')}")
                            if not isinstance(text, str) or not text.strip():
                                problems.append(f"{where}: empty block")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")

    def test_equivalency_report_every_pairing(self):
        from unittest import mock
        s, data = self.server, self.data

        def load():
            sc.load_aliases(data.get("aliases"))
            return data

        problems = []
        with mock.patch.object(s, "_load_equivalency", side_effect=load):
            for p in data["pairings"]:
                base = (p.get("us_terms") or [p["id"]])[0]
                for offense in (base, f"attempted {base}"):
                    for date in DATES:
                        for status in STATUSES:
                            where = f"{p['id']} / {offense!r} / {date} / {status}"
                            try:
                                out = s._equivalency_report(offense, date, status)
                            except Exception:
                                problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                continue
                            lines, r = _result(out)
                            if len(lines) != 1:
                                problems.append(f"{where}: {len(lines)} CANLEX_RESULT lines")
                                continue
                            problems += self._band_problems(where, r)
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")

    def test_every_inchoate_form(self):
        """Roadmap T2.4 (track A step A10): every pairing's first term as the
        object of each inchoate kind -- an attempt, an accessory after the
        fact, a bare accessory, counselling (solicitation), a conspiracy,
        aiding and abetting -- through the step-2 engine: nothing raises, one
        result line, every band (and each open reading's) matches its
        maximum, and an open inchoate reading always asks its fact."""
        from unittest import mock
        s, data = self.server, self.data

        def load():
            sc.load_aliases(data.get("aliases"))
            return data

        forms = ("attempted {}", "accessory after the fact to {}", "accessory to {}",
                 "solicitation to commit {}", "conspiracy to commit {}",
                 "aiding and abetting {}")
        problems = []
        with mock.patch.object(s, "_load_equivalency", side_effect=load):
            for p in data["pairings"]:
                base = (p.get("us_terms") or [p["id"]])[0]
                for form in forms:
                    offense = form.format(base)
                    for date in (None, "2010-01-01"):
                        for status in STATUSES:
                            where = f"{p['id']} / {offense!r} / {date} / {status}"
                            try:
                                out = s._equivalency_report(offense, date, status)
                            except Exception:
                                problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                continue
                            lines, r = _result(out)
                            if len(lines) != 1:
                                problems.append(f"{where}: {len(lines)} CANLEX_RESULT lines")
                                continue
                            problems += self._band_problems(where, r)
                            for c in r.get("tran_candidates") or []:
                                problems += self._band_problems(
                                    f"{where} [{c.get('limb_label')}]", c)
                            inc = r.get("inchoate") or {}
                            if inc.get("open") and not any(
                                    n.get("fact") == sc.INCHOATE_FACT for n in r.get("needs") or []):
                                problems.append(f"{where}: an open inchoate reading asks nothing")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")

    def test_every_resolver_answer(self):
        """Pairings with curated element resolvers (roadmap T2.10, track A step
        A9): every answer, written in the facts, through the step-2 engine at
        every date and status, completed and attempted. A resolving answer
        gives a band that matches its maximum; one that resolves nothing (or
        whose element is not in force at the date) leaves 'depends'. The
        approved baseline has no resolvers, so this is a no-op there."""
        s = self.server
        problems = []
        for p in self.data["pairings"]:
            for r in p.get("resolvers") or []:
                base = (p.get("us_terms") or [p["id"]])[0]
                for o in r.get("outcomes") or []:
                    facts = f"{r['fact']}: {o['answer']}"
                    for offense in (base, f"attempted {base}"):
                        for date in DATES:
                            for status in STATUSES:
                                where = f"{p['id']} / {facts!r} / {offense!r} / {date} / {status}"
                                try:
                                    f = s._resolve_pairing(p, offense, date, status,
                                                           facts_text=facts)
                                    text, fields = s._pairing_block(p, offense, date, status,
                                                                    facts_text=facts)
                                    json.dumps(fields)
                                except Exception:
                                    problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                    continue
                                problems += self._band_problems(where, f)
                                if o.get("max_years") is None and f["branch"] != "depends":
                                    problems.append(f"{where}: resolved to {f['branch']} on an "
                                                    "answer that names no maximum")
                                if f.get("resolver") and f["resolver"]["answer"] != o["answer"]:
                                    problems.append(f"{where}: resolved on another answer")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")

    def test_every_limb_term(self):
        """Pairings with curated limbs (roadmap T1.13): each limb's own terms,
        with an amount on either side of the quantity line where one decides,
        through the step-2 engine -- and each reading of an open result keeps
        a band that matches its maximum. The approved baseline has no limbs,
        so this is a no-op there."""
        s = self.server
        problems = []
        for p in self.data["pairings"]:
            for l in p.get("limbs") or []:
                terms = list(l.get("terms") or [])[:2] or [(p.get("us_terms") or [p["id"]])[0]]
                if l.get("quantity"):
                    terms += [f"{terms[0]} 2 kg", f"{terms[0]} 5 kg"]
                for offense in terms + [f"attempted {terms[0]}"]:
                    for date in DATES + ("2026-08-01",):
                        for status in STATUSES:
                            where = f"{p['id']} limb {l['key']} / {offense!r} / {date} / {status}"
                            try:
                                f = s._resolve_pairing(p, offense, date, status)
                                text, fields = s._pairing_block(p, offense, date, status)
                                json.dumps(fields)
                            except Exception:
                                problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                continue
                            problems += self._band_problems(where, f)
                            for c in f.get("tran_candidates") or []:
                                problems += self._band_problems(f"{where} reading {c}", c)
                            if fields.get("branch") != f.get("branch"):
                                problems.append(f"{where}: block branch {fields.get('branch')} "
                                                f"!= resolved {f.get('branch')}")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")


    def test_s37_block_every_pairing(self):
        """The s. 37 block (roadmap T2.5) for every pairing's first term x
        organization fact x status: no exception, and a flag only where the
        pairing's also_engages (held; absent on the baseline) or the
        organization fact raises one."""
        s = self.server
        problems = []
        for p in self.data["pairings"]:
            base = (p.get("us_terms") or [p["id"]])[0]
            for org in (None, True, False):
                for status in STATUSES:
                    try:
                        lines, fields = s._s37_block([p], org, status, base)
                        json.dumps(fields)
                    except Exception:
                        problems.append(f"{p['id']} / {org} / {status}: "
                                        f"{traceback.format_exc(limit=3)}")
                        continue
                    if not p.get("also_engages") and not org and fields is not None:
                        problems.append(f"{p['id']} / {org}: flag without a source")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")


    def test_canada_entry_every_pairing(self):
        """A Canadian entry (roadmap T2.2) for every pairing's first term x status
        x sentence/custody through canlex_screen_record: nothing raises, one
        CANLEX_RESULT line, a known Canadian branch, a band only where the branch
        states one, and a removal path only for a reportable answer (a permanent
        resident's only through the Immigration Division)."""
        from canlex.rehab import CANADA_BRANCH_TEXT
        s, data = self.server, self.data

        def load():
            sc.load_aliases(data.get("aliases"))
            return data

        combos = ((None, None, None), ("custody", 8, "2010-01-01"),
                  ("conditional-sentence", 12, None))
        problems = []
        with mock.patch.object(s, "_load_equivalency", side_effect=load):
            for p in data["pairings"]:
                base = (p.get("us_terms") or [p["id"]])[0]
                for status in STATUSES:
                    for sentence, custody, date in combos:
                        where = f"{p['id']} / {base!r} / {status} / {sentence}"
                        try:
                            out = s.canlex_screen_record(
                                base, "convicted", country="Canada", status=status,
                                canada_sentence=sentence, custody_imposed_months=custody,
                                offence_date=date, today="2026-09-12")
                        except Exception:
                            problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                            continue
                        lines, r = _result(out)
                        if len(lines) != 1:
                            problems.append(f"{where}: {len(lines)} result lines")
                            continue
                        cb = r["canada_branch"]
                        if cb not in CANADA_BRANCH_TEXT:
                            problems.append(f"{where}: canada_branch {cb!r}")
                        if r["branch"] not in KNOWN_BRANCHES:
                            problems.append(f"{where}: branch {r['branch']!r}")
                        if cb == "36(1)(a)" and r["branch"] != "serious-criminality":
                            problems.append(f"{where}: 36(1)(a) as {r['branch']}")
                        if r["branch"] in NO_BAND | {"none"} and r["max_penalty_lt_10"] is not None:
                            problems.append(f"{where}: band with branch {r['branch']}")
                        if r["removal"] and cb not in ("36(1)(a)", "36(2)(a)",
                                                       "36(2)(a)-two-offences"):
                            problems.append(f"{where}: removal {r['removal']} for {cb}")
                        if status == "permanent-resident" and r["removal"] not in (None, "id-referral"):
                            problems.append(f"{where}: PR removal {r['removal']}")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")

    def test_youth_every_band(self):
        """The age at the offence (roadmap T2.6) for a sample of pairings x
        foreign / Canadian x under 12 / 16 / adult x a youth and a plain
        disposition: nothing raises, and the band follows the age."""
        s, data = self.server, self.data

        def load():
            sc.load_aliases(data.get("aliases"))
            return data

        problems = []
        with mock.patch.object(s, "_load_equivalency", side_effect=load):
            for p in data["pairings"][::7]:
                base = (p.get("us_terms") or [p["id"]])[0]
                for where_kw in ({"state": "TX"}, {"country": "Canada"}):
                    for age, band in ((10, "under-12"), (16, "youth"), (30, "adult")):
                        for disp in ("convicted", "juvenile adjudication"):
                            where = f"{p['id']} / {where_kw} / {age} / {disp}"
                            try:
                                out = s.canlex_screen_record(
                                    base, disp, offence_date="2015-01-01",
                                    age_at_offence=age, today="2026-09-12", **where_kw)
                            except Exception:
                                problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                continue
                            lines, r = _result(out)
                            if len(lines) != 1 or (r.get("youth_screen") or {}).get("band") != band:
                                problems.append(f"{where}: youth_screen {r.get('youth_screen')}")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")

    def test_record_limbs_every_pairing(self):
        """Whole records, limb by limb (roadmap T2.3): each pairing's first term,
        dated and undated, beside an old assault, a recent disorderly conduct or
        two Canadian summary-only convictions, through canlex_screen_history.
        Nothing raises; the limb results agree with the record-level answer (a
        deemed answer has a paragraph whose every limb holds; not-settled has
        two readings and no single table); s. 18.1 and the Canadian answers
        agree; the retired APPLY CONDITIONAL hedge never appears."""
        s, data = self.server, self.data

        def load():
            sc.load_aliases(data.get("aliases"))
            return data
        can = {"offense": "causing a disturbance", "disposition": "convicted",
               "jurisdiction": "canada", "canada_max_years": 2,
               "canada_offence_mode": "summary-only", "canada_sentence": "fine"}
        partners = (
            [{"offense": "simple assault", "disposition": "convicted", "state": "TX",
              "offence_date": "2004-01-01", "sentence_completed": "2005-01-01"}],
            [{"offense": "disorderly conduct", "disposition": "convicted", "state": "TX",
              "offence_date": "2019-01-01", "sentence_completed": "2019-02-01"}],
            [dict(can, offence_date="2015-01-01", sentence_completed="2015-06-01"),
             dict(can, offence_date="2016-01-01", sentence_completed="2016-06-01")])
        statuses = {None, "deemed-rehabilitated", "eligible-to-apply", "not-yet",
                    "undetermined", "not-settled", "not-inadmissible-pr"}
        problems = []
        with mock.patch.object(s, "_load_equivalency", side_effect=load):
            for p in data["pairings"]:
                base = (p.get("us_terms") or [p["id"]])[0]
                for dated in (True, False):
                    main = {"offense": base, "disposition": "convicted", "state": "TX"}
                    if dated:
                        main.update(offence_date="2010-01-01", sentence_completed="2011-01-01")
                    for k, extra in enumerate(partners):
                        status = "permanent-resident" if (k == 0 and not dated) else None
                        where = f"{p['id']} / dated={dated} / partner {k} / {status}"
                        try:
                            out = s.canlex_screen_history(
                                json.dumps([main] + extra), today="2026-09-12",
                                **({"status": status} if status else {}))
                        except Exception:
                            problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                            continue
                        lines, r = _result(out)
                        if len(lines) != 1:
                            problems.append(f"{where}: {len(lines)} result lines")
                            continue
                        st = r.get("rehab_status")
                        if st not in statuses or "APPLY CONDITIONAL" in out:
                            problems.append(f"{where}: rehab_status {st!r}")
                        for para in (r.get("s18") or []) + [x for rd in r.get("s18_readings") or []
                                                             for x in rd["paragraphs"]]:
                            for l in para["limbs"]:
                                if l["status"] not in ("pass", "fail", "unknown", "n/a") or \
                                        not l["chunk"].startswith("SOR-2002-227-s18"):
                                    problems.append(f"{where}: limb {l}")
                        if st == "deemed-rehabilitated" and r.get("s18") and not any(
                                pa["outcome"] == "deemed" and all(
                                    l["status"] in ("pass", "n/a") for l in pa["limbs"])
                                for pa in r["s18"]):
                            problems.append(f"{where}: deemed without every limb holding")
                        if st == "not-settled" and (r.get("s18") is not None or len(
                                {rd["status"] for rd in r.get("s18_readings") or []}) < 2):
                            problems.append(f"{where}: not-settled without two readings")
                        s181 = r.get("s18_1")
                        cbs = {e["canada_branch"] for e in r.get("canada_entries") or []}
                        if s181 and s181["outcome"] == "deemed" and cbs != {"s18.1-exempt"}:
                            problems.append(f"{where}: s. 18.1 exempt but {cbs}")
                        if s181 and s181["outcome"] == "not-deemed" and "s18.1-exempt" in cbs:
                            problems.append(f"{where}: s. 18.1 not met but {cbs}")
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")


class WorkingTreeSweepTests(_SweepMixin, unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.data = json.loads((CURATED / "us_equivalency.json").read_text(encoding="utf-8"))


class ApprovedBaselineSweepTests(_SweepMixin, unittest.TestCase):
    """main's code on the approved curated JSON -- what a composed release
    ships (data/curated/review_state.json names the commit)."""

    @classmethod
    def setUpClass(cls):
        commit = approved_commit()
        data = git_show_json(commit, "data/curated/us_equivalency.json") if commit else None
        if data is None:
            raise unittest.SkipTest("approved baseline not readable with git here")
        super().setUpClass()
        cls.data = data


class TransborderSweepTests(unittest.TestCase):
    """canlex_transborder (roadmap T2.1) over Acts x sections x modes x goods
    x status, with the curated (c)-(g) table and without it (the approved
    baseline has no table: what a composed release ships): nothing raises,
    one CANLEX_RESULT line, a known branch, and a removal path only for a
    prescribed offence and a foreign national."""
    BRANCHES = {"transborder-criminality", "not-prescribed", "not-reached-pr",
                "table-held", "not-settled"}
    CASES = (("Criminal Code", ("95", "320.15", "320.15(1)", "266", None)),
             ("IRPA", ("124", "124(1)(b)", "117")),
             ("Firearms Act", ("106", "110", "109")),
             ("Customs Act", ("159(1)", "159", "12", "156", "160", "11")),
             ("CDSA", ("4", "5", "7", "46", "46.3(1)")),
             ("Cannabis Act", ("8", "11", "12", "15")),
             ("Excise Act, 2001", ("32",)), (None, (None,)))

    def test_sweep(self):
        from canlex import server
        saved = server._INDEX
        server._INDEX = object.__new__(type("Stub", (), {"semantic": False}))
        problems = []
        try:
            for table in ("present", "absent"):
                patch = (mock.patch.object(server, "_load_transborder", return_value=None)
                         if table == "absent" else mock.patch.object(server, "_load_transborder",
                                                                     wraps=server._load_transborder))
                with patch:
                    for act, sections in self.CASES:
                        for sec in sections:
                            for mode in (None, "hybrid", "summary-only"):
                                for goods in (None, "cannabis", "firearm-or-weapon"):
                                    for status in (None, "permanent-resident"):
                                        where = f"{table} {act} {sec} {mode} {goods} {status}"
                                        try:
                                            out = server.canlex_transborder(
                                                act=act, section=sec, mode=mode, goods=goods,
                                                status=status,
                                                offence=None if act else "impaired driving")
                                        except Exception:
                                            problems.append(f"{where}: {traceback.format_exc(limit=3)}")
                                            continue
                                        lines, r = _result(out)
                                        if len(lines) != 1:
                                            problems.append(f"{where}: {len(lines)} result lines")
                                            continue
                                        if r["branch"] not in self.BRANCHES:
                                            problems.append(f"{where}: branch {r['branch']}")
                                        if r["removal"] and (r["prescribed"] is not True
                                                             or r["person"] != "foreign-national"):
                                            problems.append(f"{where}: removal {r['removal']}")
                                        if r["branch"] == "transborder-criminality" and not r["paragraph"]:
                                            problems.append(f"{where}: no paragraph")
        finally:
            server._INDEX = saved
        self.assertEqual(problems[:20], [], f"{len(problems)} problems")


class NullHistoryTests(unittest.TestCase):
    def test_null_penalty_history_is_no_history(self):
        p = {"max_years": 10, "penalty_history": None}
        self.assertEqual(sc.penalty_at(p, "2006-01-01")[:2], (10, ""))
        self.assertEqual(sc.penalty_at({"max_years": 5}, "2006-01-01")[:2], (5, ""))


if __name__ == "__main__":
    unittest.main()