tests/test_pdi.py

"""Unit tests for the IRCC PDI ingester (canlex/pdi.py). Offline only.

Covers the two ways this ingester can quietly lose guidance: a sub-page that
fails to fetch disappearing from a full rebuild, and an upstream page whose
markup changes shape parsing to nothing -- the 2026-07-27 D-memo failure mode,
which here is caught by the corpus-write guard rather than by preservation.
"""
import unittest

from canlex import pdi

PRCARD = ("https://www.canada.ca/en/immigration-refugees-citizenship/corporate/"
          "publications-manuals/operational-bulletins-manuals/"
          "permanent-residence/card/apply.html")
RENEW = PRCARD.replace("apply.html", "renew.html")


def _chunk(url, text, chunk_id="pdi-prcard-1-0"):
    return {"id": chunk_id, "source_url": url, "text": text}


class PreserveFailedTests(unittest.TestCase):
    STORED = [_chunk(PRCARD, "how to apply", "pdi-prcard-1-0"),
              _chunk(PRCARD, "supporting documents", "pdi-prcard-1-1"),
              _chunk(RENEW, "how to renew", "pdi-prcard-2-0")]

    def test_unfetchable_page_keeps_its_last_good_chunks(self):
        chunks, preserved = pdi.preserve_failed(
            [], [(PRCARD, "HTTPError: 404")], self.STORED)
        self.assertEqual(preserved, [PRCARD])
        self.assertEqual([c["text"] for c in chunks],
                         ["how to apply", "supporting documents"])

    def test_page_dropped_from_the_index_is_not_resurrected(self):
        # RENEW is gone from this run but never errored -- IRCC retired it, and
        # retired instructions must not come back as if they were current.
        fresh = [_chunk(PRCARD, "how to apply")]
        chunks, preserved = pdi.preserve_failed(fresh, [], self.STORED)
        self.assertEqual((chunks, preserved), (fresh, []))

    def test_freshly_scraped_page_wins_over_the_stored_copy(self):
        fresh = [_chunk(PRCARD, "how to apply (2026 wording)")]
        chunks, preserved = pdi.preserve_failed(
            fresh, [(PRCARD, "HTTPError: 404")], self.STORED)
        self.assertEqual((chunks, preserved), (fresh, []))

    def test_page_we_never_stored_cannot_be_preserved(self):
        new_url = PRCARD.replace("apply.html", "replace.html")
        chunks, preserved = pdi.preserve_failed(
            [], [(new_url, "TimeoutError: ")], self.STORED)
        self.assertEqual((chunks, preserved), ([], []))

    def test_failed_identity_is_the_page_url_not_the_error_text(self):
        # failures are (url, why) pairs; only the url identifies the page.
        chunks, preserved = pdi.preserve_failed(
            [], [(PRCARD, PRCARD)], self.STORED)
        self.assertEqual(preserved, [PRCARD])
        self.assertEqual(len(chunks), 2)

    def test_first_ever_run_has_nothing_to_preserve(self):
        fresh = [_chunk(PRCARD, "how to apply")]
        chunks, preserved = pdi.preserve_failed(
            fresh, [(RENEW, "HTTPError: 404")], [])
        self.assertEqual((chunks, preserved), (fresh, []))


class PreservedIdCollisionTests(unittest.TestCase):
    def test_preserved_ids_are_uniquified_against_this_run(self):
        # A chunk id embeds the page's position in the index listing, so when a
        # page drops out the ids shift and a preserved chunk can collide with a
        # live one. Duplicate ids orphan chunks in the embeddings loader.
        stored = [_chunk(RENEW, "how to renew", "pdi-prcard-2-0")]
        fresh = [_chunk(PRCARD, "how to apply", "pdi-prcard-2-0")]
        chunks, _preserved = pdi.preserve_failed(
            fresh, [(RENEW, "HTTPError: 404")], stored)
        pdi.uniquify_ids(chunks)
        self.assertEqual(len({c["id"] for c in chunks}), len(chunks))
        self.assertEqual(chunks[0]["id"], "pdi-prcard-2-0")


class PageChunkTests(unittest.TestCase):
    PAGE = ("<main><h1>Permanent resident card</h1>"
            "<time property='dateModified'>2026-05-04</time>"
            "<h2>Eligibility</h2><p>" + "An applicant must be a PR. " * 8 +
            "</p><h2>Fees</h2><p>" + "The fee is fifty dollars. " * 8 +
            "</p></main>")
    SRC = {"code": "pdi-prcard", "short": "PDI PR Card", "name": "PDI — PR card",
           "index": "https://x/card.html", "scope": "/card"}

    def test_headings_become_chunks_tagged_with_their_page(self):
        chunks = pdi._page_chunks(self.SRC, "https://x/card.html", self.PAGE, 0)
        self.assertEqual([c["marginal_note"] for c in chunks],
                         ["Eligibility", "Fees"])
        self.assertEqual({c["source_url"] for c in chunks},
                         {"https://x/card.html"})
        self.assertTrue(all(c["doc_type"] == "memorandum" for c in chunks))

    def test_reshaped_page_yields_nothing_to_write(self):
        # What canada.ca would serve if the instructions moved behind a script:
        # the parse succeeds and produces zero chunks, which is why the write
        # has to be guarded rather than trusted.
        reshaped = "<main><h1>Permanent resident card</h1><div id='app'></div></main>"
        self.assertEqual(
            pdi._page_chunks(self.SRC, "https://x/card.html", reshaped, 0), [])


if __name__ == "__main__":
    unittest.main()