"""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"") 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"") 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"), 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\nInvoke-WebRequest : Service UnavailableThe server is busy._x000' b'D__x000A_At line:3 char:5_x000D__x000A_+ Invoke-WebRequest -Uri \'http://127.0.0.1:51908/503' b'-html\' -OutFile ..._x000D__x000A_+ ~~~~~~~~' b'~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000D__' b'x000A_ + CategoryInfo : InvalidOpera' b'tion: (System.Net.HttpWebRequest:HttpWebRequest) [Invoke-WebReque' b'st], WebExc _x000D__x000A_ eption_x000D__x000A' b'_ + FullyQualifiedErrorId : WebCmdletWebRespo' b'nseException,Microsoft.PowerShell.Commands.InvokeWebRequestComman' b'd_x000D__x000A_') PS_404_PHONE = ( b'CANLEX_HTTP_STATUS 404\r\n#< CLIXML\r\nInvoke-WebRequest : Page not found. Call us at (514) 555-0100_x' b'000D__x000A_At line:3 char:5_x000D__x000A_+ Invoke-WebRequest -Uri \'http://127.0.0.1:51908/' b'404-phone\' -OutFil ..._x000D__x000A_+ ~~~~~' b'~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000' b'D__x000A_ + CategoryInfo : InvalidOp' b'eration: (System.Net.HttpWebRequest:HttpWebRequest) [Invoke-WebRe' b'quest], WebExc _x000D__x000A_ eption_x000D__x0' b'00A_ + FullyQualifiedErrorId : WebCmdletWebRe' b'sponseException,Microsoft.PowerShell.Commands.InvokeWebRequestCom' b'mand_x000D__x000A_') PS_404_QUOTING_A_MARKER = ( b'CANLEX_HTTP_STATUS 404\r\n#< CLIXML\r\nInvoke-WebRequest : _x000D__x000A_CANLEX_HTTP_' b'STATUS 503_x000D__x000A_At line:3 char:5_x000D__' b'x000A_+ Invoke-WebRequest -Uri \'http://127.' b'0.0.1:51908/404-marker\' -OutFi ..._x000D__x000A_+ ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~' b'~~~~~~~_x000D__x000A_ + CategoryInfo ' b' : InvalidOperation: (System.Net.HttpWebRequest:HttpWebRequest) [' b'Invoke-WebRequest], WebExc _x000D__x000A_ epti' b'on_x000D__x000A_ + FullyQualifiedErrorId : We' b'bCmdletWebResponseException,Microsoft.PowerShell.Commands.InvokeW' b'ebRequestCommand_x000D__x000A_') PS_REFUSED = ( b'#< CLIXML\r\nInvoke-WebRequest : Unab' b'le to connect to the remote server_x000D__x000A_' b'At line:3 char:5_x000D__x000A_+ Invoke-WebRe' b"quest -Uri 'http://127.0.0.1:51909/refused' -OutFile ..._x000D__" b'x000A_+ ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~' b'~~~~~~~~~~~~~~~~~~~~~~~~~~~~~_x000D__x000A_ +' b' CategoryInfo : InvalidOperation: (System.Net.HttpWebReq' b'uest:HttpWebRequest) [Invoke-WebRequest], WebExc _x000D__x000A_ eption_x000D__x000A_ + Full' b'yQualifiedErrorId : WebCmdletWebResponseException,Microsoft.Power' b'Shell.Commands.InvokeWebRequestCommand_x000D__x000A_') 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 ' 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"

Service Unavailable

" b"

The server is busy.

"), "/404-phone": (404, b"Page not found. Call us at " b"(514) 555-0100")} 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"") self.assertEqual(self.path.read_bytes(), b"") 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"])