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
|
from __future__ import annotations
import io
import urllib.error
import urllib.request
from email.message import Message
from urllib.parse import urlsplit
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
|
