"""Unit tests for the Customs Tariff Schedule ingester. Offline only. The corpus is two chapters (98 and 99), so a single chapter that fetches badly or parses to nothing is a 30-70% collapse -- the same shape of failure that emptied dmemos.json on 2026-07-27. These cover how the module decides a chapter failed, how it hangs on to the last-good copy of one that did, and that a refused write reaches the shell as a non-zero exit. """ import json import sys import tempfile import unittest from pathlib import Path from unittest import mock from canlex import tariff_schedule as ts CH98_HTML = ("

Notes

1. This Chapter applies.

" "" "" "" "" "
Tariff Item
98.01Conveyances
9801.10.1010Vehicles-FreeFree
") CH99_HTML = ("
" "" "" "" "
Tariff Item
99.01Temporary importations
9901.00.00Fishing equipment-Free
") class ChapterIdentityTests(unittest.TestCase): def test_every_chunk_of_a_chapter_shares_one_identity(self): # Nothing else on a chunk names its chapter, so preserve_failed keys on # `part`; the Notes chunk and the heading chunks must agree on it. chunks = ts.parse_chapter(CH98_HTML, ts.SOURCES["ch98"]) self.assertEqual({c["part"] for c in chunks}, {ts._chapter_part("98")}) self.assertEqual(len(chunks), 2) # Notes + heading 98.01 def test_part_matches_what_the_stored_corpus_carries(self): self.assertEqual(ts._chapter_part("99"), "Schedule, Chapter 99") class ParseCollapseTests(unittest.TestCase): """A chapter that parses to nothing is the reshaped-page signature.""" def test_page_without_main_yields_nothing(self): self.assertEqual(ts.parse_chapter("
", ts.SOURCES["ch98"]), []) def test_js_rendered_table_yields_nothing(self): # An empty table whose rows arrive by script -- what CBSA did to the # D-memo index; the ingester must not read it as a valid chapter. self.assertEqual( ts.parse_chapter('
', ts.SOURCES["ch99"]), []) def test_notes_survive_a_missing_table(self): # A partial parse is not a failure -- the shrink guard, not # preserve_failed, is what catches those. chunks = ts.parse_chapter("

Notes

1. Applies.

", ts.SOURCES["ch99"]) self.assertEqual([c["section"] for c in chunks], ["Sch-Ch99-Notes"]) class PreserveFailedTests(unittest.TestCase): STORED = [{"part": "Schedule, Chapter 98", "section": "Sch-Ch98-Notes"}, {"part": "Schedule, Chapter 98", "section": "Sch-98.01"}, {"part": "Schedule, Chapter 99", "section": "Sch-99.01"}] def test_failed_chapter_keeps_its_last_good_chunks(self): chunks, preserved = ts.preserve_failed( [{"part": "Schedule, Chapter 99", "section": "Sch-99.01"}], [("98", "HTTPError: 404")], self.STORED) self.assertEqual(preserved, ["Schedule, Chapter 98"]) self.assertEqual(len(chunks), 3) def test_empty_parse_is_a_failure_and_is_preserved(self): # The reason string is never read; only the chapter identity is. chunks, preserved = ts.preserve_failed([], [("99", "no chunks parsed")], self.STORED) self.assertEqual(preserved, ["Schedule, Chapter 99"]) self.assertEqual([c["section"] for c in chunks], ["Sch-99.01"]) def test_chapter_that_did_not_fail_is_not_resurrected(self): # Chapter 98 simply is not in this run's output and did not error, so # whatever it used to hold stays gone. chunks, preserved = ts.preserve_failed( [{"part": "Schedule, Chapter 99", "section": "Sch-99.01"}], [], self.STORED) self.assertEqual(preserved, []) self.assertEqual(len(chunks), 1) def test_fresh_chunks_win_over_the_stored_copy(self): fresh = [{"part": "Schedule, Chapter 98", "section": "Sch-98.02"}] chunks, preserved = ts.preserve_failed(fresh, [("98", "boom")], self.STORED) self.assertEqual((chunks, preserved), (fresh, [])) def test_first_ever_run_has_nothing_to_preserve(self): chunks, preserved = ts.preserve_failed([], [("98", "boom")], []) self.assertEqual((chunks, preserved), ([], [])) class BuildTests(unittest.TestCase): """build() end to end with the network stubbed out.""" def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.out = Path(tmp.name) / "tariff_schedule.json" patch = mock.patch.object(ts, "OUT", self.out) patch.start() self.addCleanup(patch.stop) def _fetch_returning(self, pages): """Stub _fetch: url -> canned HTML, or an exception to raise.""" def fetch(url, dest): page = pages[url] if isinstance(page, Exception): raise page return page return fetch def _pages(self, ch98=CH98_HTML, ch99=CH99_HTML): return {ts.SOURCES["ch98"]["url"]: ch98, ts.SOURCES["ch99"]["url"]: ch99} def _build(self, pages, **kwargs): with mock.patch.object(ts, "_fetch", self._fetch_returning(pages)), \ mock.patch("builtins.print"): return ts.build(**kwargs) def _stored(self): return json.loads(self.out.read_text(encoding="utf-8")) def test_first_run_writes_with_the_existing_indent(self): self.assertTrue(self._build(self._pages())) text = self.out.read_text(encoding="utf-8") # indent=1: changing it would rewrite every line of the data file. self.assertTrue(text.startswith("[\n {\n"), text[:20]) self.assertEqual(len(self._stored()), 3) # ch98 Notes + 98.01 + 99.01 def test_unreachable_chapter_falls_back_to_the_stored_copy(self): self.assertTrue(self._build(self._pages())) ok = self._build(self._pages(ch98=OSError("connection reset"))) self.assertTrue(ok) self.assertEqual(len(self._stored()), 3) self.assertIn("Schedule, Chapter 98", {c["part"] for c in self._stored()}) def test_reshaped_chapter_falls_back_to_the_stored_copy(self): self.assertTrue(self._build(self._pages())) self.assertTrue(self._build(self._pages(ch99="
"))) self.assertEqual(len(self._stored()), 3) def test_collapse_that_cannot_be_preserved_is_refused(self): # A chapter that still parses, just to far fewer chunks: no failure to # preserve, so the shrink guard is the only thing standing in the way. self.out.write_text(json.dumps( [{"part": "Schedule, Chapter 98", "id": f"old-{i}"} for i in range(20)] + [{"part": "Schedule, Chapter 99", "id": f"old9-{i}"} for i in range(20)]), encoding="utf-8") self.assertFalse(self._build(self._pages())) self.assertEqual(len(self._stored()), 40) # untouched def test_allow_shrink_lets_a_reviewed_collapse_through(self): self.out.write_text(json.dumps([{"part": "Schedule, Chapter 98", "id": f"old-{i}"} for i in range(40)]), encoding="utf-8") self.assertTrue(self._build(self._pages(), allow_shrink=True)) self.assertEqual(len(self._stored()), 3) class MainTests(unittest.TestCase): def _main_with(self, argv, result): calls = [] def stub(allow_shrink=False): calls.append(allow_shrink) return result saved_argv, sys.argv = sys.argv, argv with mock.patch.object(ts, "build", stub): try: with self.assertRaises(SystemExit) as exit_info: ts.main() finally: sys.argv = saved_argv return calls, exit_info.exception.code def test_guard_is_on_unless_asked_otherwise(self): calls, code = self._main_with(["prog"], True) self.assertEqual((calls, code), ([False], 0)) def test_allow_shrink_flag_reaches_build(self): calls, code = self._main_with(["prog", "--allow-shrink"], True) self.assertEqual((calls, code), ([True], 0)) def test_refused_write_exits_non_zero(self): _calls, code = self._main_with(["prog"], False) self.assertEqual(code, 1) if __name__ == "__main__": unittest.main()