File size: 8,994 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 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | """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 = ("<main><h2>Notes</h2><p>1. This Chapter applies.</p>"
"<table><tr><th>Tariff Item</th></tr>"
"<tr><td>98.01</td><td></td><td>Conveyances</td></tr>"
"<tr><td>9801.10.10</td><td>10</td><td>Vehicles</td>"
"<td>-</td><td>Free</td><td>Free</td></tr>"
"</table></main>")
CH99_HTML = ("<main><table><tr><th>Tariff Item</th></tr>"
"<tr><td>99.01</td><td></td><td>Temporary importations</td></tr>"
"<tr><td>9901.00.00</td><td></td><td>Fishing equipment</td>"
"<td>-</td><td>Free</td><td></td></tr>"
"</table></main>")
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("<div><table></table></div>",
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('<main><table id="tariff"></table></main>',
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("<main><h2>Notes</h2><p>1. Applies.</p></main>",
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="<main></main>")))
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()
|