1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
|
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('<div id="vyos-search" role="search"></div>') is True
assert smoke.search_mount_present('<html><body>no search here</body></html>') 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"<html>no search</html>")]}))
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"]
|
