from __future__ import annotations import io import sys import urllib.error import urllib.request from email.message import Message from urllib.parse import urlsplit import pytest from scripts.docs_gates import smoke from scripts.docs_gates.conftest import REDIRECT_LOCATION, REDIRECT_PATH def test_probe_plan_scoped_to_slug(): plan = smoke.probe_plan("1.5", pdf="/en/1.5/vyos-documentation.pdf", critical=["index.html", "cli/index.html"]) urls = [p.path for p in plan] assert "/en/1.5/index.html" in urls assert "/en/1.5/cli/index.html" in urls assert "/en/1.5/vyos-documentation.pdf" in urls assert "/en/1.5/definitely-missing-page-xyz.html" in urls # 404-status probe assert "/versions.json" in urls and "/healthz" in urls # apex specials assert not any(u.startswith("/en/rolling/") for u in urls) # scoped (§7.1.2) def test_assertions_follow_header_contract(): plan = smoke.probe_plan("1.5", pdf=None, critical=["index.html"]) content = next(p for p in plan if p.path == "/en/1.5/index.html") assert content.expect_status == 200 and content.assert_docs_build is True apex = next(p for p in plan if p.path == "/versions.json") assert apex.assert_docs_build is False and apex.assert_apex_build is True missing = next(p for p in plan if "definitely-missing" in p.path) assert missing.expect_status == 404 def test_skip_sentinel_relaxes_sha_to_presence_only(): # expect_sha == "SKIP" (nightly sweep, Task 3.5): header must be PRESENT but any value passes assert smoke.docs_build_ok("anything", expect_sha="SKIP") is True assert smoke.docs_build_ok(None, expect_sha="SKIP") is False assert smoke.docs_build_ok("abc", expect_sha="abc") is True assert smoke.docs_build_ok("abc", expect_sha="def") is False # --- Phase-2 obligation (authorized addition, not in the original brief): the # index.html probe must assert the CF-built HTML carries the `#vyos-search` # mount div, so CI catches a build that silently forgot to set # DOCS_VERSION_SLUG (which would ship stock RTD search instead of Pagefind). --- def test_probe_plan_asserts_search_mount_on_index_only(): plan = smoke.probe_plan("1.5", pdf=None, critical=["index.html", "cli/index.html"]) index = next(p for p in plan if p.path == "/en/1.5/index.html") assert index.assert_search_mount is True other_content = [p for p in plan if p.path != "/en/1.5/index.html" and p.assert_docs_build] assert other_content and all(p.assert_search_mount is False for p in other_content) # --- CodeRabbit finding: smoke's probe requests must mirror parity.py's _NoRedirect # opener — an exact-status probe (200/404) that silently followed a 3xx would report # whatever the redirect target returns instead of the redirect itself. --- def test_opener_observes_redirect_directly_not_followed(redirect_http_server): # End-to-end: a real local HTTP server returns a 301, opened through smoke.py's # module-level _OPENER (the exact object `run()` uses) — proves it's actually # wired to refuse the redirect, matching run()'s HTTPError-catch handling of 3xx. req = urllib.request.Request(f"http://{redirect_http_server}{REDIRECT_PATH}", method="GET") try: smoke._OPENER.open(req, timeout=5) raise AssertionError("expected HTTPError for a 301 with the no-redirect opener") except urllib.error.HTTPError as e: assert e.code == 301 assert e.headers.get("Location") == REDIRECT_LOCATION def test_search_mount_present(): assert smoke.search_mount_present('') is True assert smoke.search_mount_present('no search here') is False # --- Hardening: explicit UA + round-based retry with an overall deadline. Mock at the _OPENER # boundary (the object _probe_once() opens through, mirroring the redirect test above which # drives smoke._OPENER directly). run()-level round tests inject a small plan via probe_plan and # a path-keyed opener; sleeps are spied (or constants shrunk) so nothing wall-clocks. --- class _FakeResp: """Stand-in for what _OPENER.open() yields: a context manager exposing .status / .headers / .read().""" def __init__(self, status: int, headers: dict[str, str], body: bytes = b""): self.status = status self.headers = headers self._body = body def read(self) -> bytes: return self._body def __enter__(self) -> "_FakeResp": return self def __exit__(self, *exc: object) -> bool: return False class _FakeOpener: """Yields queued responses in order; an Exception item is raised (models a transport error, or a non-2xx delivered as HTTPError). Records each opened Request + the timeout it was called with, for assertions.""" def __init__(self, responses: list[object]): self._responses = list(responses) self.calls: list[urllib.request.Request] = [] self.timeouts: list[float | None] = [] def open(self, req: urllib.request.Request, timeout: float | None = None) -> object: self.calls.append(req) self.timeouts.append(timeout) item = self._responses.pop(0) if isinstance(item, Exception): raise item return item class _RecordingBody(io.BytesIO): """HTTPError.fp stand-in: records close() (so tests can assert the response stream is closed) and can optionally raise on read (a transport error DURING body read). urllib's addbase binds read through to fp and closes fp on close(), so overriding them here is what e.read() / closing e actually hit.""" def __init__(self, data: bytes = b"", raise_on_read: bool = False): super().__init__(data) self.close_calls = 0 self._raise_on_read = raise_on_read def read(self, *args: object) -> bytes: # noqa: D401 if self._raise_on_read: raise OSError("reset during body read") return super().read(*args) def close(self) -> None: self.close_calls += 1 super().close() def _http_error_read_boom(code: int) -> urllib.error.HTTPError: return urllib.error.HTTPError( "https://host.invalid/x", code, "msg", Message(), _RecordingBody(raise_on_read=True)) class _PathOpener: """Opener keyed by request path: each path maps to a queue consumed one item per probe of that path (item = _FakeResp, or an Exception that is raised). Records probed paths in order, so round scoping / retry counts are assertable across rounds that re-probe only failures.""" def __init__(self, by_path: dict[str, list[object]]): self._by_path = {p: list(v) for p, v in by_path.items()} self.calls: list[str] = [] self.timeouts: list[float | None] = [] def open(self, req: urllib.request.Request, timeout: float | None = None) -> object: path = urlsplit(req.full_url).path self.calls.append(path) self.timeouts.append(timeout) item = self._by_path[path].pop(0) if isinstance(item, Exception): raise item return item def _plan(paths: list[str]) -> list[smoke.Probe]: """Minimal status-only plan (no build/search assertions) for run() round tests.""" return [smoke.Probe(p, 200, assert_docs_build=False, assert_apex_build=False) for p in paths] def test_probe_sends_explicit_user_agent(monkeypatch): opener = _FakeOpener([_FakeResp(200, {"X-Docs-Build": "sha1"})]) monkeypatch.setattr(smoke, "_OPENER", opener) probe = smoke.Probe("/en/1.5/index.html", 200, assert_docs_build=True, assert_apex_build=False) ok, *_ = smoke._probe_once("host.example", probe, "sha1", "cf-id", "cf-secret") assert ok is True req = opener.calls[0] assert req.get_header("User-agent") == smoke.USER_AGENT # urllib capitalizes the key assert req.get_header("Cf-access-client-id") == "cf-id" # CF-Access headers still sent def test_transport_error_recovers_in_second_round(monkeypatch): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/en/rolling/index.html"])) monkeypatch.setattr(smoke, "RETRY_SLEEP_SECONDS", 0) opener = _PathOpener({ "/en/rolling/index.html": [OSError("dns hiccup"), _FakeResp(200, {})], }) monkeypatch.setattr(smoke, "_OPENER", opener) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 0 # round 1 transport error, round 2 success — the same path is re-probed assert opener.calls == ["/en/rolling/index.html", "/en/rolling/index.html"] def test_httperror_read_crash_is_contained_as_failure(monkeypatch): # agy-critical: a transport error DURING e.read() must not escape as a traceback. monkeypatch.setattr(smoke, "_OPENER", _FakeOpener([_http_error_read_boom(502)])) probe = smoke.Probe("/en/rolling/index.html", 200, assert_docs_build=False, assert_apex_build=False) ok, status, docs_build, detail = smoke._probe_once("host", probe, "sha", "id", "sec") assert ok is False assert status is None and docs_build is None assert detail is not None and "reset during body read" in detail def test_sleeps_once_per_inter_round_gap_not_per_probe(monkeypatch): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a", "/b", "/c"])) sleeps: list[float] = [] monkeypatch.setattr(smoke.time, "sleep", lambda s: sleeps.append(s)) # /a and /b fail round 1 then pass round 2; /c passes round 1 opener = _PathOpener({ "/a": [_FakeResp(500, {}), _FakeResp(200, {})], "/b": [_FakeResp(500, {}), _FakeResp(200, {})], "/c": [_FakeResp(200, {})], }) monkeypatch.setattr(smoke, "_OPENER", opener) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 0 assert sleeps == [smoke.RETRY_SLEEP_SECONDS] # ONE inter-round sleep despite 2 failing probes def test_second_round_reprobes_only_the_failed_path(monkeypatch): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a", "/b", "/c"])) monkeypatch.setattr(smoke.time, "sleep", lambda s: None) opener = _PathOpener({ "/a": [_FakeResp(200, {})], # passes round 1 "/b": [_FakeResp(500, {}), _FakeResp(200, {})], # fails r1, passes r2 "/c": [_FakeResp(200, {})], # passes round 1 }) monkeypatch.setattr(smoke, "_OPENER", opener) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 0 assert opener.calls == ["/a", "/b", "/c", "/b"] # only the failed path is re-probed def test_run_reports_failure_count_and_nonzero_exit(monkeypatch, capsys): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a", "/b"])) monkeypatch.setattr(smoke, "MAX_ROUNDS", 1) # single round, no retries monkeypatch.setattr(smoke, "_OPENER", _PathOpener({ "/a": [_FakeResp(200, {})], "/b": [_FakeResp(500, {})], })) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 1 assert '{"failures": 1}' in capsys.readouterr().out def test_run_reports_clean_and_zero_exit(monkeypatch, capsys): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a", "/b"])) monkeypatch.setattr(smoke, "_OPENER", _PathOpener({ "/a": [_FakeResp(200, {})], "/b": [_FakeResp(200, {})], })) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 0 assert '{"failures": 0}' in capsys.readouterr().out def test_deadline_counts_unresolved_as_failed(monkeypatch, capsys): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a", "/b"])) monkeypatch.setattr(smoke, "DEADLINE_SECONDS", 0) # trips before the first probe opener = _PathOpener({"/a": [_FakeResp(200, {})], "/b": [_FakeResp(200, {})]}) monkeypatch.setattr(smoke, "_OPENER", opener) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 1 captured = capsys.readouterr() assert "SMOKE-DEADLINE" in captured.err assert '{"failures": 2}' in captured.out assert opener.calls == [] # deadline reached before any probe ran def test_probe_plan_dedups_index_html(): plan = smoke.probe_plan("rolling", None, ["index.html", "cli.html"]) index_probes = [p for p in plan if p.path == "/en/rolling/index.html"] assert len(index_probes) == 1 # not duplicated by the critical list assert index_probes[0].assert_search_mount is True # still the single search-mount probe assert sum(1 for p in plan if p.path == "/en/rolling/cli.html") == 1 # cli.html preserved # --- CR round 2: close the HTTPError response stream (it owns a socket) + name the failed # assertion in retry/fail logs. --- def test_httperror_response_is_closed(monkeypatch): # HTTPError owns the response socket; the expected-404 path must close it, not leak. body = _RecordingBody(b"not found") monkeypatch.setattr(smoke, "_OPENER", _FakeOpener([ urllib.error.HTTPError("https://host.invalid/x", 404, "msg", Message(), body)])) probe = smoke.Probe("/en/rolling/missing.html", 404, assert_docs_build=False, assert_apex_build=False) ok, *_ = smoke._probe_once("host", probe, "sha", "id", "sec") assert ok is True # 404 expected → passes assert body.close_calls >= 1 # response stream closed (no socket leak / ResourceWarning) def test_httperror_response_is_closed_even_when_read_raises(monkeypatch): body = _RecordingBody(raise_on_read=True) monkeypatch.setattr(smoke, "_OPENER", _FakeOpener([ urllib.error.HTTPError("https://host.invalid/x", 502, "msg", Message(), body)])) probe = smoke.Probe("/en/rolling/index.html", 200, assert_docs_build=False, assert_apex_build=False) ok, _, _, detail = smoke._probe_once("host", probe, "sha", "id", "sec") assert ok is False assert detail is not None and "reset during body read" in detail assert body.close_calls >= 1 # close still attempted despite the read raising def test_apex_build_failure_is_named_in_logs(monkeypatch, capsys): # 200 but no X-Apex-Build: without a named detail the log read "status=200 docs-build=None". probe = smoke.Probe("/versions.json", 200, assert_docs_build=False, assert_apex_build=True) monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: [probe]) monkeypatch.setattr(smoke, "MAX_ROUNDS", 1) monkeypatch.setattr(smoke, "_OPENER", _PathOpener({"/versions.json": [_FakeResp(200, {})]})) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 1 err = capsys.readouterr().err assert "SMOKE-FAIL /versions.json:" in err assert "detail=apex-build" in err # names the failed check assert "status=200" in err # existing fields preserved def test_search_mount_failure_is_named_in_logs(monkeypatch, capsys): # 200 + correct build, but the HTML lacks the #vyos-search mount div. probe = smoke.Probe("/en/rolling/index.html", 200, assert_docs_build=True, assert_apex_build=False, assert_search_mount=True) monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: [probe]) monkeypatch.setattr(smoke, "MAX_ROUNDS", 1) monkeypatch.setattr(smoke, "_OPENER", _PathOpener({ "/en/rolling/index.html": [_FakeResp(200, {"X-Docs-Build": "goodsha"}, b"no search")]})) assert smoke.run("host", "rolling", "goodsha", "id", "sec", None, []) == 1 assert "detail=search-mount" in capsys.readouterr().err def test_probe_once_detail_joins_multiple_failed_checks(monkeypatch): # wrong status AND missing X-Apex-Build → detail "status+apex-build" monkeypatch.setattr(smoke, "_OPENER", _FakeOpener([_FakeResp(500, {})])) probe = smoke.Probe("/versions.json", 200, assert_docs_build=False, assert_apex_build=True) ok, _, _, detail = smoke._probe_once("host", probe, "sha", "id", "sec") assert ok is False assert detail == "status+apex-build" # --- Adversarial round: DEADLINE_SECONDS is a HARD bound — the per-probe socket timeout and the # inter-round sleep are both capped to the remaining budget so neither can overshoot it. --- def test_probe_timeout_capped_to_remaining_budget(monkeypatch): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a"])) monkeypatch.setattr(smoke, "DEADLINE_SECONDS", 5) # << PROBE_TIMEOUT_SECONDS (30) opener = _PathOpener({"/a": [_FakeResp(200, {})]}) monkeypatch.setattr(smoke, "_OPENER", opener) smoke.run("host", "rolling", "sha", "id", "sec", None, []) assert opener.timeouts, "the probe recorded the timeout it was opened with" t = opener.timeouts[0] assert 1 <= t <= smoke.DEADLINE_SECONDS # capped DOWN to the ~5s remaining budget assert t < smoke.PROBE_TIMEOUT_SECONDS # NOT the full 30s socket timeout def test_inter_round_sleep_capped_to_remaining_budget(monkeypatch): monkeypatch.setattr(smoke, "probe_plan", lambda slug, pdf, critical: _plan(["/a"])) monkeypatch.setattr(smoke, "DEADLINE_SECONDS", 10) # << RETRY_SLEEP_SECONDS (30) sleeps: list[float] = [] monkeypatch.setattr(smoke.time, "sleep", lambda s: sleeps.append(s)) opener = _PathOpener({"/a": [_FakeResp(500, {}), _FakeResp(200, {})]}) # fail r1, pass r2 monkeypatch.setattr(smoke, "_OPENER", opener) assert smoke.run("host", "rolling", "sha", "id", "sec", None, []) == 0 assert len(sleeps) == 1 assert 0 < sleeps[0] <= smoke.DEADLINE_SECONDS # capped to the ~10s remaining budget assert sleeps[0] < smoke.RETRY_SLEEP_SECONDS # NOT the full 30s sleep # --- The PDF probe was structurally unpassable: it asserted X-Docs-Build, but 1.3's PDF is # served by the APEX Worker straight from R2 (spec §5, 29.2 MiB > the 25 MiB asset cap) and # that path sets only etag / accept-ranges / content-type / content-length. Observed nightly: # "/en/1.3/vyos-documentation.pdf: status=206 docs-build=None detail=status+docs-build". --- def test_pdf_probe_does_not_assert_docs_build(): plan = smoke.probe_plan("1.3", pdf="/en/1.3/vyos-documentation.pdf", critical=["index.html"]) pdf = next(p for p in plan if p.path.endswith(".pdf")) assert pdf.assert_docs_build is False # apex's R2 path legitimately never sets it assert pdf.assert_apex_build is False # nor does the content Worker set X-Apex-Build # ...but the build SHA is still gated for this version, via the HTML probes: assert next(p for p in plan if p.path.endswith("/index.html")).assert_docs_build is True def test_pdf_probe_still_demands_an_exact_200(): # The 206 seen alongside the docs-build failure was an apex defect (a 206 answered to a # request carrying no Range header), fixed in workers/apex/src/index.ts — NOT something # this gate should learn to tolerate. plan = smoke.probe_plan("1.3", pdf="/en/1.3/vyos-documentation.pdf", critical=[]) assert next(p for p in plan if p.path.endswith(".pdf")).expect_status == 200 # --- CF Access credentials: env by default (argv publishes secrets to the process table), # flags as a manual fallback, and an empty value is rejected rather than sent as a blank # header (every probe would then 403 and the report would blame the wrong thing). --- def _argv(monkeypatch, tmp_path, *extra): crit = tmp_path / "critical.txt" crit.write_text("index.html\n") monkeypatch.setattr(sys, "argv", ["smoke", "--host", "h", "--slug", "rolling", "--expect-sha", "SKIP", "--critical-list", str(crit), *extra]) return crit def test_access_credentials_default_from_environment(monkeypatch, tmp_path): _argv(monkeypatch, tmp_path) monkeypatch.setenv("CF_ACCESS_CLIENT_ID", "env-id") monkeypatch.setenv("CF_ACCESS_CLIENT_SECRET", "env-secret") seen: dict[str, str] = {} monkeypatch.setattr(smoke, "run", lambda host, slug, sha, aid, asec, pdf, critical: seen.update(id=aid, secret=asec) or 0) assert smoke.main() == 0 assert seen == {"id": "env-id", "secret": "env-secret"} def test_secret_bearing_flags_are_rejected_not_silently_ignored(monkeypatch, tmp_path): # --access-id/--access-secret are GONE, not deprecated: an argv-passed secret is readable # from the process table and captured verbatim by `set -x` traces. argparse must reject # them so the old muscle-memory invocation errors out instead of silently ignoring the # credential the operator passed and then 403ing on every probe. for flag, value in (("--access-id", "an-id"), ("--access-secret", "a-secret")): _argv(monkeypatch, tmp_path, flag, value) monkeypatch.setenv("CF_ACCESS_CLIENT_ID", "env-id") monkeypatch.setenv("CF_ACCESS_CLIENT_SECRET", "env-secret") monkeypatch.setattr(smoke, "run", lambda *a, **k: pytest.fail("must not probe")) with pytest.raises(SystemExit) as exc: smoke.main() assert exc.value.code == 2 def test_critical_list_is_read_through_a_path_without_resource_warnings(monkeypatch, tmp_path): # `open(a.critical_list).read()` left the descriptor to be closed by GC; --critical-list # is now `type=Path` and read via Path.read_text(), which closes deterministically. # HONEST SCOPE: this is not a strict regression test for the close itself — CPython's # refcounting also closes the bare-open form immediately, so no ResourceWarning fires # either way and this test passes against the pre-fix source (verified). What it DOES # pin is the argparse `type=Path` change (a str would have no .read_text()) plus # warning-free reading on interpreters without refcounting, e.g. PyPy, where the # bare-open form genuinely leaks until GC. import warnings crit = _argv(monkeypatch, tmp_path) monkeypatch.setenv("CF_ACCESS_CLIENT_ID", "id") monkeypatch.setenv("CF_ACCESS_CLIENT_SECRET", "sec") seen: list[str] = [] monkeypatch.setattr(smoke, "run", lambda host, slug, sha, aid, asec, pdf, critical: seen.extend(critical) or 0) with warnings.catch_warnings(): warnings.simplefilter("error", ResourceWarning) assert smoke.main() == 0 assert seen == ["index.html"] assert crit.exists() def test_missing_access_credentials_fail_loudly_without_probing(monkeypatch, tmp_path, capsys): _argv(monkeypatch, tmp_path) monkeypatch.delenv("CF_ACCESS_CLIENT_ID", raising=False) monkeypatch.delenv("CF_ACCESS_CLIENT_SECRET", raising=False) monkeypatch.setattr(smoke, "run", lambda *a, **k: pytest.fail("must not probe")) assert smoke.main() == 2 err = capsys.readouterr().err assert "CF_ACCESS_CLIENT_ID" in err and "CF_ACCESS_CLIENT_SECRET" in err def test_critical_list_strips_before_testing_for_comments(monkeypatch, tmp_path): # An INDENTED comment used to survive the `line.startswith("#")` test (applied to the # unstripped line) and become a live critical page — which can never exist as a file. crit = tmp_path / "critical.txt" crit.write_text("# leading comment\n # indented comment\n\n cli.html \n") monkeypatch.setenv("CF_ACCESS_CLIENT_ID", "id") monkeypatch.setenv("CF_ACCESS_CLIENT_SECRET", "sec") monkeypatch.setattr(sys, "argv", ["smoke", "--host", "h", "--slug", "rolling", "--expect-sha", "SKIP", "--critical-list", str(crit)]) seen: list[str] = [] monkeypatch.setattr(smoke, "run", lambda host, slug, sha, aid, asec, pdf, critical: seen.extend(critical) or 0) assert smoke.main() == 0 assert seen == ["cli.html"]