tests/test_common.py

"""Unit tests for the shared ingestion helpers (canlex/_common.py).

Offline only. The corpus-write guard exists because every ingester rebuilds its
whole processed file from a scrape: on 2026-07-27 a rebuilt upstream index
scraped to zero items and the write emptied 1,870 good chunks.
"""
import contextlib
import io
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
import unittest.mock
import urllib.error
from pathlib import Path

from canlex import _common


class SafeToWriteTests(unittest.TestCase):
    def test_refuses_to_replace_a_corpus_with_nothing(self):
        self.assertFalse(_common.safe_to_write(0, 1870))

    def test_refuses_a_large_collapse(self):
        self.assertFalse(_common.safe_to_write(500, 1870))

    def test_allows_normal_churn(self):
        self.assertTrue(_common.safe_to_write(1866, 1870))
        self.assertTrue(_common.safe_to_write(1900, 1870))

    def test_first_ever_run_is_allowed(self):
        self.assertTrue(_common.safe_to_write(0, 0))

    def test_ratio_is_tunable(self):
        self.assertTrue(_common.safe_to_write(500, 1000, ratio=0.5))
        self.assertFalse(_common.safe_to_write(499, 1000, ratio=0.5))


class WriteCorpusTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.path = Path(self.tmp.name) / "corpus.json"
        self.addCleanup(self.tmp.cleanup)

    def _write(self, chunks):
        self.path.write_text(json.dumps(chunks), encoding="utf-8")

    def test_writes_when_there_is_nothing_to_lose(self):
        self.assertTrue(_common.write_corpus(self.path, [{"id": "a"}]))
        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
                         [{"id": "a"}])

    def test_refuses_and_leaves_the_file_untouched(self):
        self._write([{"id": str(i)} for i in range(100)])
        with contextlib.redirect_stdout(io.StringIO()) as out:
            self.assertFalse(_common.write_corpus(self.path, []))
        self.assertIn("REFUSING", out.getvalue())
        self.assertEqual(len(json.loads(self.path.read_text(encoding="utf-8"))),
                         100)

    def test_allow_shrink_overrides(self):
        self._write([{"id": str(i)} for i in range(100)])
        self.assertTrue(_common.write_corpus(self.path, [], allow_shrink=True))
        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")), [])

    def test_unreadable_stored_file_does_not_block_a_write(self):
        # A truncated/corrupt corpus reads as empty, so recovery is possible.
        self.path.write_text("{not json", encoding="utf-8")
        self.assertTrue(_common.write_corpus(self.path, [{"id": "a"}]))

    def test_creates_missing_parent_directories(self):
        nested = Path(self.tmp.name) / "deep" / "corpus.json"
        self.assertTrue(_common.write_corpus(nested, [{"id": "a"}]))
        self.assertTrue(nested.exists())

    def test_non_ascii_survives_the_round_trip(self):
        chunks = [{"id": "a", "text": "détention à la frontière"}]
        _common.write_corpus(self.path, chunks)
        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
                         chunks)


class StoredChunksTests(unittest.TestCase):
    def test_missing_file_is_empty(self):
        self.assertEqual(_common.stored_chunks(Path("nope-does-not-exist.json")),
                         [])


class PreserveDroppedTests(unittest.TestCase):
    STORED = [{"section": "A", "text": "one"},
              {"section": "A", "text": "two"},
              {"section": "B", "text": "three"}]

    def test_reattaches_a_dropped_item(self):
        chunks, dropped = _common.preserve_dropped([], self.STORED, only={"A"})
        self.assertEqual(dropped, ["A"])
        self.assertEqual(len(chunks), 2)

    def test_only_restricts_to_genuine_failures(self):
        # B vanished deliberately (cancelled upstream); it must not come back.
        chunks, dropped = _common.preserve_dropped([], self.STORED, only={"A"})
        self.assertNotIn("B", dropped)
        self.assertNotIn("three", [c["text"] for c in chunks])

    def test_without_only_every_missing_item_is_restored(self):
        _chunks, dropped = _common.preserve_dropped([], self.STORED)
        self.assertEqual(dropped, ["A", "B"])

    def test_freshly_scraped_items_are_not_overwritten(self):
        fresh = [{"section": "A", "text": "new"}]
        chunks, dropped = _common.preserve_dropped(fresh, self.STORED,
                                                   only={"A"})
        self.assertEqual((chunks, dropped), (fresh, []))

    def test_a_stored_chunk_without_the_identity_field_is_skipped(self):
        # Not a crash: an older-schema corpus must not take down the rebuild
        # the guard exists to protect.
        stored = self.STORED + [{"id": "legacy-1", "text": "no section key"}]
        chunks, dropped = _common.preserve_dropped([], stored, only={"A"})
        self.assertEqual(dropped, ["A"])
        self.assertNotIn("legacy-1", [c.get("id") for c in chunks])

    def test_custom_key(self):
        stored = [{"case": "Khosa", "text": "x"}]
        chunks, dropped = _common.preserve_dropped(
            [], stored, key=lambda c: c["case"], only={"Khosa"})
        self.assertEqual((len(chunks), dropped), (1, ["Khosa"]))


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


class UniquifyIdsTests(unittest.TestCase):
    """Repeats past 'z' used to run on into '{', '|', DEL and C1 control
    characters (65 ENF ids at 76db310). The continuation must leave every
    id that already exists untouched."""

    def test_first_repeats_keep_their_single_letter_ids(self):
        chunks = [{"id": "x"} for _ in range(26)]
        _common.uniquify_ids(chunks)
        self.assertEqual([c["id"] for c in chunks[:3]], ["x", "x-b", "x-c"])
        self.assertEqual(chunks[25]["id"], "x-z")

    def test_repeats_past_z_continue_with_two_letters(self):
        chunks = [{"id": "x"} for _ in range(30)]
        _common.uniquify_ids(chunks)
        self.assertEqual([c["id"] for c in chunks[26:]],
                         ["x-aa", "x-ab", "x-ac", "x-ad"])

    def test_ids_stay_ascii_and_unique_at_any_depth(self):
        chunks = [{"id": "x"} for _ in range(800)]
        _common.uniquify_ids(chunks)
        ids = [c["id"] for c in chunks]
        self.assertEqual(len(set(ids)), len(ids))
        self.assertTrue(all(re.fullmatch(r"x(-[a-z]+)?", i) for i in ids))

    def test_dup_suffix_is_bijective_base_26(self):
        self.assertEqual(
            [_common.dup_suffix(n) for n in (1, 25, 26, 27, 51, 52, 701, 702)],
            ["b", "z", "aa", "ab", "az", "ba", "zz", "aaa"])
        with self.assertRaises(ValueError):
            _common.dup_suffix(0)


class FetchCachedTests(unittest.TestCase):
    """The cache is a politeness measure, but it silently defeats refreshes:
    on 2026-08-31 a stale CIU Regulations PDF survived a full `canlex.ciu`
    run because the cached copy was returned instead of the revised upstream
    file. force=True is the only way past it."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.dest = Path(self.tmp.name) / "doc.pdf"
        self.addCleanup(self.tmp.cleanup)
        # a transient failure is retried after FETCH_RETRY_DELAY (2026-09-22);
        # the tests below must not wait for it
        sleep = unittest.mock.patch.object(_common.time, "sleep")
        sleep.start()
        self.addCleanup(sleep.stop)
        quiet = contextlib.redirect_stdout(io.StringIO())   # the 'retrying' line
        quiet.__enter__()
        self.addCleanup(quiet.__exit__, None, None, None)

    @contextlib.contextmanager
    def _server(self, payload):
        """Stub urlopen; records how many times the network was hit."""
        calls = []

        class _Resp(io.BytesIO):
            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *exc):
                return False

        def _urlopen(req, timeout=None):
            calls.append(getattr(req, "full_url", req))
            return _Resp(payload)

        with unittest.mock.patch.object(_common.urllib.request, "urlopen",
                                        _urlopen):
            yield calls

    def test_downloads_once_then_serves_the_cache(self):
        with self._server(b"first") as calls:
            a = _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0)
            b = _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0)
        self.assertEqual(a, b"first")
        self.assertEqual(b, b"first")
        self.assertEqual(len(calls), 1, "second call should not hit the network")

    def test_force_refetches_and_overwrites_a_revised_upstream(self):
        with self._server(b"old") as calls:
            _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0)
        with self._server(b"revised") as calls:
            got = _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0,
                                       force=True)
        self.assertEqual(got, b"revised")
        self.assertEqual(len(calls), 1)
        self.assertEqual(self.dest.read_bytes(), b"revised")

    def test_without_force_a_revised_upstream_is_invisible(self):
        # The exact failure mode behind the CIU staleness bug.
        with self._server(b"old"):
            _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0)
        with self._server(b"revised") as calls:
            got = _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0)
        self.assertEqual(got, b"old")
        self.assertEqual(calls, [])

    # 2026-09-21 audit: a download that fails half-way must leave NOTHING at
    # dest, or every later plain run serves the truncated file as the cache.

    def test_a_failed_urllib_fetch_leaves_no_cache_file(self):
        def _urlopen(req, timeout=None):
            raise OSError("simulated: connection reset")
        with unittest.mock.patch.object(_common.urllib.request, "urlopen", _urlopen):
            with self.assertRaises(OSError):
                _common.fetch_cached("http://x/doc.pdf", self.dest, pause=0)
        self.assertFalse(self.dest.exists())
        self.assertEqual(list(self.dest.parent.iterdir()), [])
        # and the next call fetches again
        with self._server(b"good") as calls:
            self.assertEqual(_common.fetch_cached("http://x/doc.pdf", self.dest, pause=0),
                             b"good")
        self.assertEqual(len(calls), 1)

    def test_a_powershell_fetch_that_times_out_leaves_no_cache_file(self):
        # Invoke-WebRequest -OutFile writes as it streams; a stalled server
        # left a 0-byte page at dest that the next run reused.
        def _run(cmd, **kw):
            part = self.dest.with_name(self.dest.name + ".part")
            part.write_bytes(b"")
            raise _common.subprocess.TimeoutExpired(cmd, kw.get("timeout"))
        with unittest.mock.patch.object(_common.subprocess, "run", _run):
            with self.assertRaises(_common.subprocess.TimeoutExpired):
                _common.fetch_cached("http://x/page.html", self.dest, pause=0,
                                     powershell=True)
        self.assertFalse(self.dest.exists())
        self.assertEqual(list(self.dest.parent.iterdir()), [])

    def test_a_powershell_fetch_that_fails_leaves_no_cache_file(self):
        def _run(cmd, **kw):
            raise _common.subprocess.CalledProcessError(1, cmd, stderr=b"404")
        with unittest.mock.patch.object(_common.subprocess, "run", _run):
            with self.assertRaises(_common.subprocess.CalledProcessError):
                _common.fetch_cached("http://x/page.html", self.dest, pause=0,
                                     powershell=True)
        self.assertFalse(self.dest.exists())

    def test_a_successful_powershell_fetch_lands_at_dest(self):
        def _run(cmd, **kw):
            self.dest.with_name(self.dest.name + ".part").write_bytes(b"<html>")
            return unittest.mock.Mock(returncode=0)
        with unittest.mock.patch.object(_common.subprocess, "run", _run):
            got = _common.fetch_cached("http://x/page.html", self.dest, pause=0,
                                       powershell=True)
        self.assertEqual(got, b"<html>")
        self.assertEqual(sorted(p.name for p in self.dest.parent.iterdir()), ["doc.pdf"])


class PowerShellQuotingTests(unittest.TestCase):
    """2026-09-21 audit: the request was spliced into a single-quoted
    -Command string, so an apostrophe in the URL or the destination path
    (C:\\Users\\O'Brien) was a PowerShell parse error. The script now goes
    as -EncodedCommand with every value quoted as a PowerShell literal."""

    def _script(self, url, dest):
        seen = {}

        def _run(cmd, **kw):
            seen["cmd"] = cmd
            return unittest.mock.Mock(returncode=0)
        with unittest.mock.patch.object(_common.subprocess, "run", _run):
            _common.powershell_download(url, dest)
        cmd = seen["cmd"]
        self.assertEqual(cmd[:4], ["powershell", "-NoProfile", "-NonInteractive",
                                   "-EncodedCommand"])
        import base64
        return base64.b64decode(cmd[4]).decode("utf-16-le")

    def test_an_apostrophe_is_doubled_inside_the_literal(self):
        script = self._script("https://x/it's.html", r"C:\Users\O'Brien\p.html")
        self.assertIn("-Uri 'https://x/it''s.html'", script)
        self.assertIn(r"-OutFile 'C:\Users\O''Brien\p.html'", script)
        self.assertIn("-UserAgent '", script)

    def test_a_plain_value_is_quoted_as_is(self):
        script = self._script("https://x/a.html", "C:/tmp/a.html")
        self.assertIn("Invoke-WebRequest -Uri 'https://x/a.html' -OutFile 'C:/tmp/a.html' "
                      "-UseBasicParsing", script)

    def test_ps_literal(self):
        self.assertEqual(_common._ps_literal("a'b"), "'a''b'")
        self.assertEqual(_common._ps_literal("plain"), "'plain'")

    def test_refresh_shares_the_helper(self):
        import inspect
        from canlex import refresh
        self.assertIn("powershell_download", inspect.getsource(refresh._fetch_powershell))
        self.assertNotIn("-Command", inspect.getsource(refresh._fetch_powershell))


class CacheInvalidationContractTests(unittest.TestCase):
    """Every ingester must be able to bypass its own cache.

    fetch_cached reuses whatever is already on disk, so an ingester that
    calls it without force= can never pick up a document that upstream
    revised in place. That is not hypothetical: on 2026-08-31 a stale CIU
    Regulations PDF survived a full `canlex.ciu` run, and the refresh
    check's STALE row was unclearable because the cache mtime never moved.
    These tests hold the line for every ingester, not just that one.
    """

    # fetch_cached and the one-level wrappers the ingesters put over it
    # (dmemo._fetch, enf._fetch_html, ...): the argument list is matched with
    # one level of nested parentheses, so `url.rsplit("/", 1)` inside a call
    # does not end the match early (2026-09-21 audit: the old `fetch_cached\(`
    # pattern never saw dmemo.parse_pdf_memo's `_fetch(...)` with no force=).
    _CALL = re.compile(r"\b_?fetch(?:_cached|_html|_doc)?\(((?:[^()]|\([^()]*\))*)\)", re.S)

    def _ingesters(self):
        pkg = Path(_common.__file__).parent
        for path in sorted(pkg.glob("*.py")):
            if path.name in {"_common.py", "__init__.py"}:
                continue
            src = path.read_text(encoding="utf-8")
            if "fetch_cached(" in src:
                yield path.name, src

    def test_every_fetch_cached_call_threads_force(self):
        offenders = []
        for name, src in self._ingesters():
            for m in self._CALL.finditer(src):
                if src[max(0, m.start() - 4):m.start()] == "def ":
                    continue                      # the wrapper's own definition
                call = m.group(1)
                if "force=" not in call:
                    offenders.append(f"{name}: {m.group(0).strip()[:70]}...")
        self.assertEqual(offenders, [], "fetch_cached (or a wrapper) called without force=")

    def test_the_contract_pattern_sees_a_wrapper_call(self):
        src = ('pdf_bytes = _fetch(pdf_url, DMEMO_DIR / "pdf" / pdf_url.rsplit("/", 1)[-1])\n'
               'html = _fetch(url, DMEMO_DIR / url.rsplit("/", 1)[-1], force=force)\n')
        calls = [m.group(1) for m in self._CALL.finditer(src)]
        self.assertEqual(len(calls), 2)
        self.assertNotIn("force=", calls[0])
        self.assertIn("force=", calls[1])

    def test_every_ingester_exposes_a_force_flag(self):
        missing = [name for name, src in self._ingesters()
                   if '"--force" in sys.argv' not in src]
        self.assertEqual(missing, [], "ingester has no --force CLI flag")

    def test_every_build_accepts_force(self):
        import importlib
        import inspect
        bad = []
        for name, _src in self._ingesters():
            mod = importlib.import_module(f"canlex.{name[:-3]}")
            build = getattr(mod, "build", None)
            if build is None:
                continue
            if "force" not in inspect.signature(build).parameters:
                bad.append(name)
        self.assertEqual(bad, [], "build() does not accept force=")


class FormatCharTests(unittest.TestCase):
    """Audit follow-up (2026-09-22), missing item 6: norm_ws drops the invisible
    in-word format characters. Before, 'assist\\xadance' (C-24.5 s. 8.1) and
    'f\\u200borm' (IRB guideline 8) stayed two tokens each -- \\s never matched
    them -- in 237 chunks of the 2026-09-21 corpus."""

    def test_each_of_the_six_is_dropped_inside_a_word(self):
        for ch in "\u00ad\u200b\u200c\u200d\u2060\ufeff":
            with self.subTest(char=f"U+{ord(ch):04X}"):
                self.assertEqual(_common.norm_ws(f"assist{ch}ance"), "assistance")

    def test_before_the_change_the_word_stayed_split(self):
        # the old norm_ws, verbatim: the format character survived
        old = re.sub(r"\s+", " ", "assist\xadance").strip()
        self.assertEqual(old, "assist\xadance")
        self.assertNotEqual(_common.norm_ws("assist\xadance"), old)

    def test_whitespace_still_collapses_around_a_dropped_character(self):
        self.assertEqual(_common.norm_ws("IRB\n\u200b processes \ufeff"),
                         "IRB processes")
        self.assertEqual(_common.norm_ws(None), "")

    def test_ordinary_text_is_unchanged(self):
        text = "Loi sur l’immigration — art. 36(1)a)\xa0: non-résident"
        self.assertEqual(_common.norm_ws(text), re.sub(r"\s+", " ", text).strip())

    def test_strip_format_chars_keeps_newlines(self):
        self.assertEqual(_common.strip_format_chars("a\u200b\nb\u00ad"), "a\nb")


class DehyphenateTests(unittest.TestCase):
    """Audit follow-up (2026-09-22), missing item 6: the shared line-break
    dehyphenation for the PDF paths. Only the line break goes by default;
    the hyphen goes only when the document itself uses the joined word; a
    URL always keeps it."""

    def test_by_default_only_the_line_break_goes(self):
        self.assertEqual(_common.dehyphenate("to cross-\nexamine the witness\nnext"),
                         "to cross-examine\nthe witness\nnext")

    def test_the_hyphen_goes_when_the_document_uses_the_joined_word(self):
        words = _common.document_words(["immigration and refugee protection"])
        self.assertEqual(_common.dehyphenate("under immi-\ngration law", words),
                         "under immigration\nlaw")

    def test_an_unattested_join_keeps_the_hyphen(self):
        words = _common.document_words(["cross and examine"])
        self.assertEqual(_common.dehyphenate("cross-\nexamine", words), "cross-examine")

    def test_a_url_always_keeps_its_hyphen(self):
        words = _common.document_words(["lawslois"])
        self.assertEqual(
            _common.dehyphenate("see http://laws-\nlois.justice.gc.ca/eng/ now", words),
            "see http://laws-lois.justice.gc.ca/eng/\nnow")

    def test_a_capitalised_continuation_keeps_the_hyphen(self):
        words = _common.document_words(["antiterrorism"])
        self.assertEqual(_common.dehyphenate("Anti-\nTerrorism Act", words),
                         "Anti-Terrorism\nAct")

    def test_a_soft_hyphen_break_reads_as_a_hyphen(self):
        self.assertEqual(_common.dehyphenate("spouse or common\xad\nlaw partner"),
                         "spouse or common-law\npartner")

    def test_a_line_left_empty_is_dropped(self):
        self.assertEqual(_common.dehyphenate("be re-\nestablished.\nNext"),
                         "be re-established.\nNext")

    def test_dashes_numbers_and_blank_lines_are_left_alone(self):
        for text in ("Overview -\nPart two", "COVID-\n19 measures", "word-\n\nnext",
                     "1969-\n70", "- item\n- item"):
            with self.subTest(text=text):
                self.assertEqual(_common.dehyphenate(text), text)

    def test_line_count_never_grows(self):
        text = "a cross-\nexamine b\nimmi-\ngration c\nd"
        self.assertLessEqual(_common.dehyphenate(text).count("\n"), text.count("\n"))

    def test_mark_soft_breaks_only_touches_a_line_end(self):
        self.assertEqual(_common.mark_soft_breaks("common\xad\nlaw assist\xadance"),
                         "common-\nlaw assist\xadance")

    def test_document_words_counts_lower_cased_letter_runs(self):
        words = _common.document_words(["Cross-examine the witness", "the 2 Witness"])
        self.assertEqual((words["cross"], words["witness"], words["the"]), (1, 2, 2))
        self.assertNotIn("2", words)


class TransientFetchTests(unittest.TestCase):
    """Audit follow-up (2026-09-22), missing item 12: one 5xx, reset or
    timeout used to fail an ingest item outright; now it is retried once. A
    4xx never is: enf.is_missing_upstream reads a 404 as a retired chapter."""

    def setUp(self):
        self.sleep = unittest.mock.patch.object(_common.time, "sleep").start()
        self.addCleanup(unittest.mock.patch.stopall)
        quiet = contextlib.redirect_stdout(io.StringIO())
        quiet.__enter__()
        self.addCleanup(quiet.__exit__, None, None, None)

    def _http(self, code):
        exc = urllib.error.HTTPError("http://x", code, "status", {}, io.BytesIO(b""))
        self.addCleanup(exc.close)
        return exc

    @staticmethod
    def _ps(stderr):
        return subprocess.CalledProcessError(1, "powershell", stderr=stderr)

    def test_is_transient(self):
        cases = [(self._http(503), True), (self._http(500), True),
                 (self._http(404), False), (self._http(403), False),
                 (self._http(429), False),
                 (urllib.error.URLError("reset"), True), (TimeoutError("slow"), True),
                 (ConnectionResetError("reset"), True),
                 (subprocess.TimeoutExpired("powershell", 180), True),
                 # the status comes from powershell_download's own marker line
                 # (D-2, verification 2026-09-22) ...
                 (self._ps(b"CANLEX_HTTP_STATUS 503\r\n#< CLIXML\r\n<Objs/>"), True),
                 (self._ps("CANLEX_HTTP_STATUS 404\n(503) Server Unavailable."), False),
                 # ... never from prose, which can be the error page's text
                 # (True before the marker; see PowerShellStatusMarkerTests)
                 (self._ps(b"The remote server returned an error: (503) Server Unavailable."), False),
                 (self._ps("The remote server returned an error: (404) Not Found."), False),
                 (self._ps("The operation has timed out."), True),
                 (self._ps(b"404"), False),
                 (ValueError("not Justice Laws XML"), False), (OSError("disk"), False)]
        for exc, expected in cases:
            with self.subTest(exc=repr(exc)[:60]):
                self.assertIs(_common.is_transient(exc), expected)

    def test_a_503_is_retried_once_and_the_second_answer_wins(self):
        calls = []

        def fetch():
            calls.append(1)
            if len(calls) == 1:
                raise self._http(503)
            return b"ok"
        self.assertEqual(_common.with_retry(fetch, what="u"), b"ok")
        self.assertEqual(len(calls), 2)
        self.sleep.assert_called_once_with(_common.FETCH_RETRY_DELAY)

    def test_a_404_is_not_retried(self):
        calls = []

        def fetch():
            calls.append(1)
            raise self._http(404)
        with self.assertRaises(urllib.error.HTTPError):
            _common.with_retry(fetch)
        self.assertEqual(len(calls), 1)
        self.sleep.assert_not_called()

    def test_two_transient_failures_give_up_with_the_last_error(self):
        calls = []

        def fetch():
            calls.append(1)
            raise urllib.error.URLError(f"reset {len(calls)}")
        with self.assertRaises(urllib.error.URLError) as cm:
            _common.with_retry(fetch)
        self.assertEqual(len(calls), _common.FETCH_ATTEMPTS)
        self.assertIn("reset 2", str(cm.exception))

    def _dest(self, name):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        return Path(tmp.name) / name

    def test_fetch_cached_retries_a_dropped_connection(self):
        dest = self._dest("doc.pdf")
        calls = []

        class _Resp(io.BytesIO):
            def __enter__(self_inner):
                return self_inner

            def __exit__(self_inner, *exc):
                return False

        def _urlopen(req, timeout=None):
            calls.append(1)
            if len(calls) == 1:
                raise ConnectionResetError("reset by peer")
            return _Resp(b"body")
        with unittest.mock.patch.object(_common.urllib.request, "urlopen", _urlopen):
            got = _common.fetch_cached("http://x/doc.pdf", dest, pause=0)
        self.assertEqual((got, len(calls)), (b"body", 2))
        self.assertEqual([p.name for p in dest.parent.iterdir()], ["doc.pdf"])

    def test_fetch_cached_does_not_retry_a_powershell_404(self):
        dest = self._dest("page.html")
        calls = []

        def _run(cmd, **kw):
            calls.append(1)
            raise self._ps("The remote server returned an error: (404) Not Found.")
        with unittest.mock.patch.object(_common.subprocess, "run", _run):
            with self.assertRaises(subprocess.CalledProcessError):
                _common.fetch_cached("http://x/p.html", dest, pause=0, powershell=True)
        self.assertEqual(len(calls), 1)
        self.assertFalse(dest.exists())


# Real Windows PowerShell 5.1 stderr from powershell_download's script, captured
# on 2026-09-22 against a local http.server on 127.0.0.1 (the critic's p_ps.py
# routes): a 503 page with a body, a 404 page quoting a phone number, a 404
# page whose body says 'CANLEX_HTTP_STATUS 503', and a refused connection.
PS_503_PAGE = (
    b'CANLEX_HTTP_STATUS 503\r\n#< CLIXML\r\n<Objs Version="1.1.0.1" xm'
    b'lns="http://schemas.microsoft.com/powershell/2004/04"><S S="Error'
    b'">Invoke-WebRequest : Service UnavailableThe server is busy._x000'
    b'D__x000A_</S><S S="Error">At line:3 char:5_x000D__x000A_</S><S S='
    b'"Error">+     Invoke-WebRequest -Uri \'http://127.0.0.1:51908/503'
    b'-html\' -OutFile ..._x000D__x000A_</S><S S="Error">+     ~~~~~~~~'
    b'~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000D__'
    b'x000A_</S><S S="Error">    + CategoryInfo          : InvalidOpera'
    b'tion: (System.Net.HttpWebRequest:HttpWebRequest) [Invoke-WebReque'
    b'st], WebExc _x000D__x000A_</S><S S="Error">   eption_x000D__x000A'
    b'_</S><S S="Error">    + FullyQualifiedErrorId : WebCmdletWebRespo'
    b'nseException,Microsoft.PowerShell.Commands.InvokeWebRequestComman'
    b'd_x000D__x000A_</S></Objs>')
PS_404_PHONE = (
    b'CANLEX_HTTP_STATUS 404\r\n#< CLIXML\r\n<Objs Version="1.1.0.1" xm'
    b'lns="http://schemas.microsoft.com/powershell/2004/04"><S S="Error'
    b'">Invoke-WebRequest : Page not found. Call us at (514) 555-0100_x'
    b'000D__x000A_</S><S S="Error">At line:3 char:5_x000D__x000A_</S><S'
    b' S="Error">+     Invoke-WebRequest -Uri \'http://127.0.0.1:51908/'
    b'404-phone\' -OutFil ..._x000D__x000A_</S><S S="Error">+     ~~~~~'
    b'~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000'
    b'D__x000A_</S><S S="Error">    + CategoryInfo          : InvalidOp'
    b'eration: (System.Net.HttpWebRequest:HttpWebRequest) [Invoke-WebRe'
    b'quest], WebExc _x000D__x000A_</S><S S="Error">   eption_x000D__x0'
    b'00A_</S><S S="Error">    + FullyQualifiedErrorId : WebCmdletWebRe'
    b'sponseException,Microsoft.PowerShell.Commands.InvokeWebRequestCom'
    b'mand_x000D__x000A_</S></Objs>')
PS_404_QUOTING_A_MARKER = (
    b'CANLEX_HTTP_STATUS 404\r\n#< CLIXML\r\n<Objs Version="1.1.0.1" xm'
    b'lns="http://schemas.microsoft.com/powershell/2004/04"><S S="Error'
    b'">Invoke-WebRequest : _x000D__x000A_</S><S S="Error">CANLEX_HTTP_'
    b'STATUS 503_x000D__x000A_</S><S S="Error">At line:3 char:5_x000D__'
    b'x000A_</S><S S="Error">+     Invoke-WebRequest -Uri \'http://127.'
    b'0.0.1:51908/404-marker\' -OutFi ..._x000D__x000A_</S><S S="Error"'
    b'>+     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
    b'~~~~~~~_x000D__x000A_</S><S S="Error">    + CategoryInfo         '
    b' : InvalidOperation: (System.Net.HttpWebRequest:HttpWebRequest) ['
    b'Invoke-WebRequest], WebExc _x000D__x000A_</S><S S="Error">   epti'
    b'on_x000D__x000A_</S><S S="Error">    + FullyQualifiedErrorId : We'
    b'bCmdletWebResponseException,Microsoft.PowerShell.Commands.InvokeW'
    b'ebRequestCommand_x000D__x000A_</S></Objs>')
PS_REFUSED = (
    b'#< CLIXML\r\n<Objs Version="1.1.0.1" xmlns="http://schemas.micros'
    b'oft.com/powershell/2004/04"><S S="Error">Invoke-WebRequest : Unab'
    b'le to connect to the remote server_x000D__x000A_</S><S S="Error">'
    b'At line:3 char:5_x000D__x000A_</S><S S="Error">+     Invoke-WebRe'
    b"quest -Uri 'http://127.0.0.1:51909/refused' -OutFile  ..._x000D__"
    b'x000A_</S><S S="Error">+     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
    b'~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000D__x000A_</S><S S="Error">    +'
    b' CategoryInfo          : InvalidOperation: (System.Net.HttpWebReq'
    b'uest:HttpWebRequest) [Invoke-WebRequest], WebExc _x000D__x000A_</'
    b'S><S S="Error">   eption_x000D__x000A_</S><S S="Error">    + Full'
    b'yQualifiedErrorId : WebCmdletWebResponseException,Microsoft.Power'
    b'Shell.Commands.InvokeWebRequestCommand_x000D__x000A_</S></Objs>')


def _old_is_transient_ps(stderr):
    """is_transient's PowerShell branch before D-2, verbatim: '(NNN)'
    anywhere in stderr, else the connection-level words."""
    status = re.search(r"\((\d{3})\)", stderr)
    if status:
        return status.group(1).startswith("5")
    return bool(re.search(r"timed out|timeout|unable to connect|connection was closed"
                          r"|forcibly closed|temporar", stderr, re.I))


class PowerShellStatusMarkerTests(unittest.TestCase):
    """Verification 2026-09-22, D-2: Windows PowerShell 5.1 reports a failed
    Invoke-WebRequest as CLIXML carrying the error PAGE's text, so the old
    '(NNN)' read missed a 503 with a body (not retried: is_transient False)
    and retried a 404 page quoting '(514) 555-0100' (True). The script now
    writes 'CANLEX_HTTP_STATUS <n>' from the exception's response and
    is_transient reads that line alone: 503 page True, 404 page False."""

    @staticmethod
    def _ps(stderr):
        return subprocess.CalledProcessError(1, "powershell", stderr=stderr)

    def test_before_the_marker_the_page_text_decided(self):
        self.assertFalse(_old_is_transient_ps(PS_503_PAGE.decode("utf-8")))
        self.assertTrue(_old_is_transient_ps(PS_404_PHONE.decode("utf-8")))

    def test_real_clixml_stderr_is_read_by_its_marker(self):
        cases = [(PS_503_PAGE, True), (PS_404_PHONE, False),
                 (PS_404_QUOTING_A_MARKER, False),   # a marker in the page is not a line
                 (PS_REFUSED, True)]                 # no HTTP answer: connection words
        for stderr, expected in cases:
            with self.subTest(stderr=stderr[:40]):
                self.assertIs(_common.is_transient(self._ps(stderr)), expected)
                self.assertIs(_common.is_transient(self._ps(stderr.decode("utf-8"))),
                              expected)

    def test_the_script_writes_the_marker_from_the_response_and_rethrows(self):
        seen = {}

        def _run(cmd, **kw):
            seen["cmd"] = cmd
            return unittest.mock.Mock(returncode=0)
        with unittest.mock.patch.object(_common.subprocess, "run", _run):
            _common.powershell_download("https://x/a.html", "C:/tmp/a.html")
        import base64
        script = base64.b64decode(seen["cmd"][4]).decode("utf-16-le")
        self.assertIn("try {\n    Invoke-WebRequest -Uri 'https://x/a.html'", script)
        self.assertIn("[Console]::Error.WriteLine('CANLEX_HTTP_STATUS ' + "
                      "[int]$_.Exception.Response.StatusCode.value__)", script)
        self.assertTrue(script.rstrip().endswith("throw\n}"), script)


@unittest.skipUnless(sys.platform == "win32" and shutil.which("powershell"),
                     "Windows PowerShell 5.1 is the path under test")
class PowerShellLiveStatusTests(unittest.TestCase):
    """D-2 end to end: powershell_download against a local http.server on
    127.0.0.1 (no other host), through fetch_cached's retry. Before the
    marker: the 503 page was fetched once and the 404 page twice; now twice
    and once."""

    ROUTES = {"/503-html": (503, b"<html><body><h1>Service Unavailable</h1>"
                                 b"<p>The server is busy.</p></body></html>"),
              "/404-phone": (404, b"<html><body>Page not found. Call us at "
                                  b"(514) 555-0100</body></html>")}

    def setUp(self):
        import http.server
        import threading
        hits = self.hits = {}
        routes = self.ROUTES

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_GET(self):
                hits[self.path] = hits.get(self.path, 0) + 1
                code, body = routes[self.path]
                self.send_response(code)
                self.send_header("Content-Type", "text/html")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, *args):
                pass
        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=server.serve_forever, daemon=True).start()
        self.addCleanup(server.server_close)
        self.addCleanup(server.shutdown)
        self.base = f"http://127.0.0.1:{server.server_address[1]}"
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.dir = Path(tmp.name)

    def _fetch(self, route):
        import functools
        bounded = functools.partial(_common.powershell_download, timeout=60)
        with unittest.mock.patch.object(_common, "powershell_download", bounded), \
                unittest.mock.patch.object(_common, "FETCH_RETRY_DELAY", 0.0), \
                contextlib.redirect_stdout(io.StringIO()):
            with self.assertRaises(subprocess.CalledProcessError) as cm:
                _common.fetch_cached(self.base + route, self.dir / route[1:], pause=0,
                                     powershell=True)
        return cm.exception

    def test_a_503_page_is_retried_and_a_404_page_is_not(self):
        exc = self._fetch("/503-html")
        self.assertIn(b"CANLEX_HTTP_STATUS 503", exc.stderr)
        self.assertTrue(_common.is_transient(exc))
        self.assertEqual(self.hits.get("/503-html"), 2)
        exc = self._fetch("/404-phone")
        self.assertIn(b"(514)", exc.stderr)               # the page text is there ...
        self.assertFalse(_common.is_transient(exc))       # ... and is not read
        self.assertEqual(self.hits.get("/404-phone"), 1)
        self.assertEqual(list(self.dir.iterdir()), [])


class AtomicWriteTests(unittest.TestCase):
    """Audit follow-up (2026-09-22), missing item 12: the marker and cache
    files are written through a temporary file renamed into place."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.path = Path(self.tmp.name) / "sub" / "last-seen.json"

    def test_writes_and_leaves_no_temporary_file(self):
        _common.atomic_write_text(self.path, '{"u": "d"}\n')
        self.assertEqual(self.path.read_text(encoding="utf-8"), '{"u": "d"}\n')
        self.assertEqual([p.name for p in self.path.parent.iterdir()], ["last-seen.json"])

    def test_a_failed_write_keeps_the_old_file_whole(self):
        _common.atomic_write_text(self.path, "old")
        with unittest.mock.patch.object(_common.os, "replace",
                                        side_effect=OSError("killed")):
            with self.assertRaises(OSError):
                _common.atomic_write_text(self.path, "new and longer")
        self.assertEqual(self.path.read_text(encoding="utf-8"), "old")
        self.assertEqual([p.name for p in self.path.parent.iterdir()], ["last-seen.json"])

    def test_bytes(self):
        _common.atomic_write_bytes(self.path, b"<Statute/>")
        self.assertEqual(self.path.read_bytes(), b"<Statute/>")


class AtomicWriteHeldOpenTests(unittest.TestCase):
    """Verification 2026-09-22, D-3: on Windows the rename onto a target
    another process holds open (Python's open(), a scanner, an indexer)
    fails with [WinError 5] Access is denied, where the plain write_text the
    helpers replaced succeeds. Before: atomic_write_text raised at once
    (p_atomic.py: 'PermissionError ... | target now {"old": 1}'). Now the
    rename is retried REPLACE_ATTEMPTS times, REPLACE_RETRY_DELAY apart: a
    brief hold is waited out; one that outlasts the retries still raises,
    the old file whole and no temporary file left."""

    HOLDER = ("import sys\n"
              "f = open(sys.argv[1], 'rb')\n"
              "print('open', flush=True)\n"
              "sys.stdin.readline()\n"
              "f.close()\n"
              "print('closed', flush=True)\n")

    def setUp(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.path = Path(tmp.name) / "guidance_marker.json"
        self.path.write_text('{"old": 1}\n', encoding="utf-8")

    def _hold(self):
        """A child process holding self.path open (as p_atomic.py does) until
        a line arrives on its stdin."""
        holder = subprocess.Popen([sys.executable, "-c", self.HOLDER, str(self.path)],
                                  stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)

        def _stop():
            holder.kill()
            holder.wait(10)
            holder.stdin.close()
            holder.stdout.close()
        self.addCleanup(_stop)
        self.assertEqual(holder.stdout.readline().strip(), "open")
        return holder

    def _left(self):
        return sorted(p.name for p in self.path.parent.iterdir())

    @unittest.skipUnless(sys.platform == "win32", "POSIX renames onto an open file")
    def test_a_brief_hold_is_waited_out(self):
        holder = self._hold()
        pauses = []

        def _release(seconds):
            # the first refused rename's pause: the other process lets go
            pauses.append(seconds)
            if len(pauses) == 1:
                holder.stdin.write("go\n")
                holder.stdin.flush()
                self.assertEqual(holder.stdout.readline().strip(), "closed")
        with unittest.mock.patch.object(_common.time, "sleep", _release):
            _common.atomic_write_text(self.path, '{"new": 2}\n')
        self.assertEqual(self.path.read_text(encoding="utf-8"), '{"new": 2}\n')
        self.assertEqual(pauses, [_common.REPLACE_RETRY_DELAY])
        self.assertEqual(self._left(), ["guidance_marker.json"])

    @unittest.skipUnless(sys.platform == "win32", "POSIX renames onto an open file")
    def test_a_hold_outlasting_the_retries_still_raises(self):
        self._hold()
        with unittest.mock.patch.object(_common.time, "sleep") as sleep:
            with self.assertRaises(PermissionError):
                _common.atomic_write_bytes(self.path, b'{"new": 2}\n')
        self.assertEqual(sleep.call_count, _common.REPLACE_ATTEMPTS - 1)
        self.assertEqual(self.path.read_text(encoding="utf-8"), '{"old": 1}\n')
        self.assertEqual(self._left(), ["guidance_marker.json"])

    def test_only_a_permission_error_is_retried(self):
        real = os.replace
        calls = []

        def _replace(src, dst):
            calls.append(1)
            if len(calls) <= 2:
                raise PermissionError(13, "Access is denied")
            real(src, dst)
        with unittest.mock.patch.object(_common.os, "replace", _replace), \
                unittest.mock.patch.object(_common.time, "sleep") as sleep:
            _common.atomic_write_text(self.path, "new")
        self.assertEqual((len(calls), sleep.call_count), (3, 2))
        self.assertEqual(self.path.read_text(encoding="utf-8"), "new")
        with unittest.mock.patch.object(_common.os, "replace",
                                        side_effect=OSError("disk full")) as rep, \
                unittest.mock.patch.object(_common.time, "sleep") as sleep:
            with self.assertRaises(OSError):
                _common.atomic_write_text(self.path, "newer")
        self.assertEqual((rep.call_count, sleep.call_count), (1, 0))
        self.assertEqual(self.path.read_text(encoding="utf-8"), "new")
        self.assertEqual(self._left(), ["guidance_marker.json"])