File size: 4,188 Bytes
6ce7899 | 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 | """Unit tests for the delegation-instrument ingester (canlex/delegation.py).
Offline only. The ingester rebuilds delegation.json from scratch on every run,
so an instrument whose page moves or changes shape drops out of the corpus
silently -- the failure mode that emptied dmemos.json on 2026-07-27. These cover
the identity the run's failures are tracked by, not the shared write guard
(tests/test_common.py owns that).
"""
import unittest
from canlex import delegation
SRC = {
"code": "cbsa-2023-05",
"act_code": "CBSA-IRPA-DELEG-2023-05",
"act_short": "CBSA Deleg 2023-05-08",
"act_name": "Delegation of Authority ...",
"url": "https://example.invalid/irpa-lipr-2023-05-08-eng.html",
"effective": "2023-05-08",
}
class ChunkIdentityTests(unittest.TestCase):
"""preserve_failed keys on act_code, so the parsers must stamp it."""
def test_schedule_items_carry_the_instrument_act_code(self):
html = ("<main><table class='table-bordered'>"
"<tr><td>1.</td><td>A55(1)</td><td>Arrest with warrant</td>"
"<td><p class='h4'>CBSA</p><ul><li>Officer</li></ul></td></tr>"
"</table></main>")
chunks = delegation.parse_cbsa(html, SRC)
self.assertTrue(chunks)
self.assertEqual({c["act_code"] for c in chunks}, {SRC["act_code"]})
def test_narrative_instruments_carry_it_too(self):
html = "<main><p>The following positions are authorized.</p></main>"
chunks = delegation.parse_cbsa_narrative(html, SRC)
self.assertEqual([c["act_code"] for c in chunks], [SRC["act_code"]])
class PreserveFailedTests(unittest.TestCase):
STORED = [{"act_code": "CBSA-IRPA-DELEG-2023-05", "text": "item 1"},
{"act_code": "CBSA-IRPA-DELEG-2023-05", "text": "item 2"},
{"act_code": "IRCC-IL3-DELEG", "text": "IL3 item"}]
def test_unfetchable_instrument_keeps_its_last_good_chunks(self):
failures = [("CBSA-IRPA-DELEG-2023-05", "HTTPError: 404")]
chunks, preserved = delegation.preserve_failed([], failures, self.STORED)
self.assertEqual(preserved, ["CBSA-IRPA-DELEG-2023-05"])
self.assertEqual([c["text"] for c in chunks], ["item 1", "item 2"])
def test_an_instrument_that_parses_to_nothing_is_a_failure_too(self):
# A reachable page whose table markup changed -- the CBSA D-memo trap.
failures = [("IRCC-IL3-DELEG", "no content parsed")]
_chunks, preserved = delegation.preserve_failed([], failures,
self.STORED)
self.assertEqual(preserved, ["IRCC-IL3-DELEG"])
def test_freshly_parsed_chunks_win_over_stored(self):
fresh = [{"act_code": "CBSA-IRPA-DELEG-2023-05", "text": "new text"}]
failures = [("CBSA-IRPA-DELEG-2023-05", "no content parsed")]
chunks, preserved = delegation.preserve_failed(fresh, failures,
self.STORED)
self.assertEqual((chunks, preserved), (fresh, []))
def test_instrument_retired_from_sources_is_not_resurrected(self):
# Removing an entry from SOURCES is deliberate; it never fails, so it
# must stay out of the corpus.
failures = [("CBSA-IRPA-DELEG-2023-05", "HTTPError: 404")]
chunks, preserved = delegation.preserve_failed([], failures,
self.STORED)
self.assertNotIn("IRCC-IL3-DELEG", preserved)
self.assertNotIn("IL3 item", [c["text"] for c in chunks])
def test_instrument_we_never_stored_cannot_be_preserved(self):
failures = [("CBSA-IRPA-PEACEOFF-2022-08", "HTTPError: 404")]
chunks, preserved = delegation.preserve_failed([], failures,
self.STORED)
self.assertEqual((chunks, preserved), ([], []))
def test_a_clean_run_preserves_nothing(self):
fresh = [{"act_code": "IRCC-IL3-DELEG", "text": "x"}]
chunks, preserved = delegation.preserve_failed(fresh, [], self.STORED)
self.assertEqual((chunks, preserved), (fresh, []))
if __name__ == "__main__":
unittest.main()
|