File size: 7,997 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 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 | """Unit tests for the AMPS Master Penalty Document ingester (canlex/amps.py).
Offline only. The ingester is a full rebuild, so a run that loses its network
half way through -- or an index CBSA rebuilds into something the link regex no
longer matches -- used to write the shortfall straight over the stored corpus.
These cover the module's own half of the fix: which contraventions count as
"failed this run", and that the identity used to preserve them is the same one
parse_page stamps on a chunk.
"""
import contextlib
import io
import json
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from canlex import amps
class PreserveFailedTests(unittest.TestCase):
STORED = [{"section": "C005", "text": "records not kept"},
{"section": "C348", "text": "no advance data"}]
def test_unfetchable_contravention_keeps_its_last_good_chunk(self):
chunks, preserved = amps.preserve_failed(
[], [("c005", "ValueError: empty response")], self.STORED)
self.assertEqual(preserved, ["C005"])
self.assertEqual([c["text"] for c in chunks], ["records not kept"])
def test_index_codes_are_lower_case_but_sections_are_not(self):
# The whole preserve step turns on this: build() collects the index's
# lower-case c-number, a chunk carries it upper-cased.
_chunks, preserved = amps.preserve_failed([], [("c348", "404")],
self.STORED)
self.assertEqual(preserved, ["C348"])
def test_contravention_dropped_upstream_is_not_resurrected(self):
# C348 vanished from the index rather than failing; it stays gone.
chunks, preserved = amps.preserve_failed([], [("c005", "404")],
self.STORED)
self.assertEqual(preserved, ["C005"])
self.assertNotIn("C348", [c["section"] for c in chunks])
def test_freshly_scraped_chunk_wins_over_the_stored_copy(self):
fresh = [{"section": "C005", "text": "new penalty amounts"}]
chunks, preserved = amps.preserve_failed(fresh, [("c005", "404")],
self.STORED)
self.assertEqual((chunks, preserved), (fresh, []))
def test_contravention_we_never_had_cannot_be_preserved(self):
chunks, preserved = amps.preserve_failed([], [("c999", "404")],
self.STORED)
self.assertEqual((chunks, preserved), ([], []))
def test_a_clean_run_preserves_nothing(self):
fresh = [{"section": "C005", "text": "x"}]
chunks, preserved = amps.preserve_failed(fresh, [], self.STORED)
self.assertEqual((chunks, preserved), (fresh, []))
def test_total_scrape_failure_restores_the_whole_corpus(self):
# Every page erroring is the shape of the outage the guard is for: the
# stored corpus survives intact instead of being written away.
failed = [("c005", "timeout"), ("c348", "timeout")]
chunks, preserved = amps.preserve_failed([], failed, self.STORED)
self.assertEqual(preserved, ["C005", "C348"])
self.assertEqual(len(chunks), len(self.STORED))
class ChunkIdentityTests(unittest.TestCase):
"""preserve_failed keys on `section`; parse_page must keep filling it."""
HTML = ("<main><h1>Administrative Monetary Penalty C005</h1>"
"<p>Person failed to keep the prescribed records for the "
"prescribed period.</p>"
"<table><tr><td>First</td><td>$500</td></tr></table></main>"
'<time property="dateModified">2026-05-01</time>')
def test_parsed_section_is_the_preserve_key(self):
chunk = amps.parse_page(self.HTML, "c005", "https://x/c005-eng.html")
self.assertEqual(chunk["section"], "C005")
_chunks, preserved = amps.preserve_failed([], [("c005", "404")],
[chunk])
self.assertEqual(preserved, ["C005"])
def test_one_chunk_per_contravention_so_the_key_is_unique(self):
chunk = amps.parse_page(self.HTML, "c005", "https://x/c005-eng.html")
self.assertEqual(chunk["id"], "amps-c005")
class BuildWiringTests(unittest.TestCase):
"""build() with the network stubbed out: is the guard actually reached?"""
INDEX = ('<a href="/trade-commerce/amps/contraventions-infractions/'
'c005-eng.html">C005</a>')
PAGE = ("<main><h1>Administrative Monetary Penalty C005</h1>"
"<p>Person failed to keep the prescribed records for the "
"prescribed period.</p></main>")
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.dir = Path(tmp.name)
self.out = self.dir / "amps.json"
def _build(self, index_html, **kwargs):
def stub(url, _dest):
return (index_html if url == amps.INDEX_URL
else self.PAGE).encode("utf-8")
with mock.patch.object(amps, "PROCESSED_DIR", self.dir), \
mock.patch.object(amps, "_fetch", side_effect=stub), \
contextlib.redirect_stdout(io.StringIO()) as out:
return amps.build(**kwargs), out.getvalue()
def test_an_index_that_scrapes_to_nothing_cannot_empty_the_corpus(self):
# CBSA rebuilding the MPD index into something _LINK no longer matches.
self.out.write_text(json.dumps([{"section": f"C{i:03d}"}
for i in range(50)]), encoding="utf-8")
ok, log = self._build("<table id='mpd'></table>")
self.assertFalse(ok)
self.assertIn("REFUSING", log)
self.assertEqual(
len(json.loads(self.out.read_text(encoding="utf-8"))), 50)
def test_allow_shrink_lets_an_understood_drop_through(self):
self.out.write_text(json.dumps([{"section": f"C{i:03d}"}
for i in range(50)]), encoding="utf-8")
ok, _log = self._build("<table id='mpd'></table>", allow_shrink=True)
self.assertTrue(ok)
self.assertEqual(json.loads(self.out.read_text(encoding="utf-8")), [])
def test_a_healthy_run_writes_two_space_indent(self):
# The stored file is indent=2; writing any other width would rewrite
# every line of a 440KB data file as a spurious diff.
ok, log = self._build(self.INDEX)
self.assertTrue(ok)
self.assertTrue(self.out.read_text(encoding="utf-8").startswith('[\n {'))
self.assertIn("1 contraventions -> amps.json", log)
class MainExitCodeTests(unittest.TestCase):
def test_refused_write_exits_non_zero(self):
with mock.patch.object(amps, "build", return_value=False), \
mock.patch.object(amps.sys, "argv", ["amps"]):
with self.assertRaises(SystemExit) as caught:
amps.main()
self.assertEqual(caught.exception.code, 1)
def test_successful_run_exits_zero(self):
with mock.patch.object(amps, "build", return_value=True), \
mock.patch.object(amps.sys, "argv", ["amps"]):
with self.assertRaises(SystemExit) as caught:
amps.main()
self.assertEqual(caught.exception.code, 0)
def test_allow_shrink_flag_reaches_build(self):
with mock.patch.object(amps, "build", return_value=True) as build, \
mock.patch.object(amps.sys, "argv", ["amps", "--allow-shrink"]):
with self.assertRaises(SystemExit):
amps.main()
self.assertTrue(build.call_args.kwargs["allow_shrink"])
def test_allow_shrink_is_off_by_default(self):
with mock.patch.object(amps, "build", return_value=True) as build, \
mock.patch.object(amps.sys, "argv", ["amps"]):
with self.assertRaises(SystemExit):
amps.main()
self.assertFalse(build.call_args.kwargs["allow_shrink"])
if __name__ == "__main__":
unittest.main()
|