"""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 = ("

Administrative Monetary Penalty C005

" "

Person failed to keep the prescribed records for the " "prescribed period.

" "
First$500
" '') 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 = ('C005') PAGE = ("

Administrative Monetary Penalty C005

" "

Person failed to keep the prescribed records for the " "prescribed period.

") 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("
") 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("
", 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()