tests/test_amps.py

"""Unit tests for the AMPS Master Penalty Document ingester (canlex/amps.py).

Offline only. The ingester is a full rebuild, so a run that loses its network
half way through -- or an index CBSA rebuilds into something the link regex no
longer matches -- used to write the shortfall straight over the stored corpus.
These cover the module's own half of the fix: which contraventions count as
"failed this run", and that the identity used to preserve them is the same one
parse_page stamps on a chunk.
"""
import contextlib
import io
import json
import tempfile
import unittest
from pathlib import Path
from unittest import mock

from canlex import amps


class PreserveFailedTests(unittest.TestCase):
    STORED = [{"section": "C005", "text": "records not kept"},
              {"section": "C348", "text": "no advance data"}]

    def test_unfetchable_contravention_keeps_its_last_good_chunk(self):
        chunks, preserved = amps.preserve_failed(
            [], [("c005", "ValueError: empty response")], self.STORED)
        self.assertEqual(preserved, ["C005"])
        self.assertEqual([c["text"] for c in chunks], ["records not kept"])

    def test_index_codes_are_lower_case_but_sections_are_not(self):
        # The whole preserve step turns on this: build() collects the index's
        # lower-case c-number, a chunk carries it upper-cased.
        _chunks, preserved = amps.preserve_failed([], [("c348", "404")],
                                                  self.STORED)
        self.assertEqual(preserved, ["C348"])

    def test_contravention_dropped_upstream_is_not_resurrected(self):
        # C348 vanished from the index rather than failing; it stays gone.
        chunks, preserved = amps.preserve_failed([], [("c005", "404")],
                                                 self.STORED)
        self.assertEqual(preserved, ["C005"])
        self.assertNotIn("C348", [c["section"] for c in chunks])

    def test_freshly_scraped_chunk_wins_over_the_stored_copy(self):
        fresh = [{"section": "C005", "text": "new penalty amounts"}]
        chunks, preserved = amps.preserve_failed(fresh, [("c005", "404")],
                                                 self.STORED)
        self.assertEqual((chunks, preserved), (fresh, []))

    def test_contravention_we_never_had_cannot_be_preserved(self):
        chunks, preserved = amps.preserve_failed([], [("c999", "404")],
                                                 self.STORED)
        self.assertEqual((chunks, preserved), ([], []))

    def test_a_clean_run_preserves_nothing(self):
        fresh = [{"section": "C005", "text": "x"}]
        chunks, preserved = amps.preserve_failed(fresh, [], self.STORED)
        self.assertEqual((chunks, preserved), (fresh, []))

    def test_total_scrape_failure_restores_the_whole_corpus(self):
        # Every page erroring is the shape of the outage the guard is for: the
        # stored corpus survives intact instead of being written away.
        failed = [("c005", "timeout"), ("c348", "timeout")]
        chunks, preserved = amps.preserve_failed([], failed, self.STORED)
        self.assertEqual(preserved, ["C005", "C348"])
        self.assertEqual(len(chunks), len(self.STORED))


class ChunkIdentityTests(unittest.TestCase):
    """preserve_failed keys on `section`; parse_page must keep filling it."""

    HTML = ("<main><h1>Administrative Monetary Penalty C005</h1>"
            "<p>Person failed to keep the prescribed records for the "
            "prescribed period.</p>"
            "<table><tr><td>First</td><td>$500</td></tr></table></main>"
            '<time property="dateModified">2026-05-01</time>')

    def test_parsed_section_is_the_preserve_key(self):
        chunk = amps.parse_page(self.HTML, "c005", "https://x/c005-eng.html")
        self.assertEqual(chunk["section"], "C005")
        _chunks, preserved = amps.preserve_failed([], [("c005", "404")],
                                                  [chunk])
        self.assertEqual(preserved, ["C005"])

    def test_one_chunk_per_contravention_so_the_key_is_unique(self):
        chunk = amps.parse_page(self.HTML, "c005", "https://x/c005-eng.html")
        self.assertEqual(chunk["id"], "amps-c005")


class BuildWiringTests(unittest.TestCase):
    """build() with the network stubbed out: is the guard actually reached?"""

    INDEX = ('<a href="/trade-commerce/amps/contraventions-infractions/'
             'c005-eng.html">C005</a>')
    PAGE = ("<main><h1>Administrative Monetary Penalty C005</h1>"
            "<p>Person failed to keep the prescribed records for the "
            "prescribed period.</p></main>")

    def setUp(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.dir = Path(tmp.name)
        self.out = self.dir / "amps.json"

    def _build(self, index_html, **kwargs):
        def stub(url, _dest):
            return (index_html if url == amps.INDEX_URL
                    else self.PAGE).encode("utf-8")
        with mock.patch.object(amps, "PROCESSED_DIR", self.dir), \
                mock.patch.object(amps, "_fetch", side_effect=stub), \
                contextlib.redirect_stdout(io.StringIO()) as out:
            return amps.build(**kwargs), out.getvalue()

    def test_an_index_that_scrapes_to_nothing_cannot_empty_the_corpus(self):
        # CBSA rebuilding the MPD index into something _LINK no longer matches.
        self.out.write_text(json.dumps([{"section": f"C{i:03d}"}
                                        for i in range(50)]), encoding="utf-8")
        ok, log = self._build("<table id='mpd'></table>")
        self.assertFalse(ok)
        self.assertIn("REFUSING", log)
        self.assertEqual(
            len(json.loads(self.out.read_text(encoding="utf-8"))), 50)

    def test_allow_shrink_lets_an_understood_drop_through(self):
        self.out.write_text(json.dumps([{"section": f"C{i:03d}"}
                                        for i in range(50)]), encoding="utf-8")
        ok, _log = self._build("<table id='mpd'></table>", allow_shrink=True)
        self.assertTrue(ok)
        self.assertEqual(json.loads(self.out.read_text(encoding="utf-8")), [])

    def test_a_healthy_run_writes_two_space_indent(self):
        # The stored file is indent=2; writing any other width would rewrite
        # every line of a 440KB data file as a spurious diff.
        ok, log = self._build(self.INDEX)
        self.assertTrue(ok)
        self.assertTrue(self.out.read_text(encoding="utf-8").startswith('[\n  {'))
        self.assertIn("1 contraventions -> amps.json", log)


class MainExitCodeTests(unittest.TestCase):
    def test_refused_write_exits_non_zero(self):
        with mock.patch.object(amps, "build", return_value=False), \
                mock.patch.object(amps.sys, "argv", ["amps"]):
            with self.assertRaises(SystemExit) as caught:
                amps.main()
        self.assertEqual(caught.exception.code, 1)

    def test_successful_run_exits_zero(self):
        with mock.patch.object(amps, "build", return_value=True), \
                mock.patch.object(amps.sys, "argv", ["amps"]):
            with self.assertRaises(SystemExit) as caught:
                amps.main()
        self.assertEqual(caught.exception.code, 0)

    def test_allow_shrink_flag_reaches_build(self):
        with mock.patch.object(amps, "build", return_value=True) as build, \
                mock.patch.object(amps.sys, "argv", ["amps", "--allow-shrink"]):
            with self.assertRaises(SystemExit):
                amps.main()
        self.assertTrue(build.call_args.kwargs["allow_shrink"])

    def test_allow_shrink_is_off_by_default(self):
        with mock.patch.object(amps, "build", return_value=True) as build, \
                mock.patch.object(amps.sys, "argv", ["amps"]):
            with self.assertRaises(SystemExit):
                amps.main()
        self.assertFalse(build.call_args.kwargs["allow_shrink"])


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