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