File size: 3,861 Bytes
43ab322 | 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 | """Unit tests for the D-memo ingester (canlex/dmemo.py). Offline only.
Covers the two failure modes that cost the corpus on 2026-07-27: CBSA replaced
the static index with a JSON-fed DataTable (the link-scrape found zero memos),
and the ingester wrote that empty result straight over 1,870 good chunks.
"""
import json
import unittest
from canlex import dmemo
class DatasourceUrlTests(unittest.TestCase):
RAW = json.dumps({"data": [
{"NUMBER": '<a href="/publications/dm-md/d1/d1-2-1-eng.html">D1-2-1</a>',
"STATUS": "Active"},
{"NUMBER": '<a href="/publications/dm-md/d8/d8-2-2-eng.html">D8-2-2</a>',
"STATUS": "Active"},
{"NUMBER": "D8-11-1", "STATUS": "Cancelled"},
{"NUMBER": '<a href="/publications/dm-md/d10/d10-14-48-eng.html">'
'D10-14-48</a>', "STATUS": "Cancelled"},
]}).encode("utf-8")
def test_extracts_absolute_urls(self):
self.assertEqual(
dmemo.datasource_urls(self.RAW),
["https://www.cbsa-asfc.gc.ca/publications/dm-md/d1/d1-2-1-eng.html",
"https://www.cbsa-asfc.gc.ca/publications/dm-md/d8/d8-2-2-eng.html"])
def test_cancelled_memos_are_skipped_even_when_still_linked(self):
# Retired guidance must not surface as if it were current.
self.assertNotIn("d10-14-48", " ".join(dmemo.datasource_urls(self.RAW)))
def test_deduplicates(self):
row = {"NUMBER": '<a href="/publications/dm-md/d1/d1-2-1-eng.html">x</a>',
"STATUS": "Active"}
raw = json.dumps({"data": [row, row]}).encode("utf-8")
self.assertEqual(len(dmemo.datasource_urls(raw)), 1)
def test_empty_datasource_yields_nothing(self):
self.assertEqual(dmemo.datasource_urls(b'{"data": []}'), [])
class IndexFallbackTests(unittest.TestCase):
def test_scrapes_links_off_the_legacy_index(self):
html = ('<a href="/publications/dm-md/d1/d1-2-1-eng.html">D1-2-1</a>'
'<a href="/publications/dm-md/menu-eng.html">menu</a>')
self.assertEqual(
dmemo.index_urls(html),
["https://www.cbsa-asfc.gc.ca/publications/dm-md/d1/d1-2-1-eng.html"])
def test_rebuilt_index_has_no_links_to_find(self):
# What CBSA now serves: an empty table, memo rows loaded by JS.
self.assertEqual(dmemo.index_urls('<table id="dmemo-tbl"></table>'), [])
class PreserveFailedTests(unittest.TestCase):
STORED = [{"section": "D10-14-3", "text": "seasonal duties"},
{"section": "D10-14-3", "text": "part two"},
{"section": "D1-2-1", "text": "special services"}]
def test_unfetchable_memo_keeps_its_last_good_chunks(self):
failures = [("https://x/publications/dm-md/d10/d10-14-3-eng.html", "404")]
chunks, preserved = dmemo.preserve_failed([], failures, self.STORED)
self.assertEqual(preserved, ["D10-14-3"])
self.assertEqual(len(chunks), 2)
def test_freshly_parsed_chunks_win_over_stored(self):
fresh = [{"section": "D10-14-3", "text": "new text"}]
failures = [("https://x/publications/dm-md/d10/d10-14-3-eng.html", "404")]
chunks, preserved = dmemo.preserve_failed(fresh, failures, self.STORED)
self.assertEqual(preserved, [])
self.assertEqual(chunks, fresh)
def test_memo_we_never_had_cannot_be_preserved(self):
failures = [("https://x/publications/dm-md/d8/d8-2-2-eng.html", "404")]
chunks, preserved = dmemo.preserve_failed([], failures, self.STORED)
self.assertEqual((chunks, preserved), ([], []))
def test_nothing_happens_without_failures(self):
fresh = [{"section": "D1-2-1", "text": "x"}]
chunks, preserved = dmemo.preserve_failed(fresh, [], self.STORED)
self.assertEqual((chunks, preserved), (fresh, []))
if __name__ == "__main__":
unittest.main()
|