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
|
"""URL-parity corpus + alias assertions (spec §11).
Modes:
--sitemap-host X pull per-version sitemaps from X (RTD pre-cutover)
--probe-host Y probe every URL on Y (canary via Access, or production)
Fails (exit 1) on any status mismatch (non-200 for corpus rows; wrong
Location for alias rows).
"""
from __future__ import annotations
import argparse
import json
import re
import sys
import urllib.request
from pathlib import Path
SITEMAP_LOC = re.compile(r"<loc>([^<]+)</loc>")
# Sitemap sweep covers only the versions THIS repo builds on CF (rolling/1.5/1.4).
# 1.3/1.2 have NO RTD sitemaps (spec §15a.5) — their parity is the legacy snapshot
# repo's crawl-inventory job. Their alias/PDF redirect rows below stay in scope.
DEFAULT_SLUGS = "rolling,1.5,1.4"
ALIASES = [("latest", "rolling"), ("stable", "1.5"), ("lts", "1.5"),
("circinus", "1.5"), ("sagitta", "1.4"), ("equuleus", "1.3"), ("crux", "1.2")]
def urls_from_sitemap(xml: str) -> list[str]:
out = []
for loc in SITEMAP_LOC.findall(xml):
path = re.sub(r"^https?://[^/]+", "", loc)
if path.startswith("/en/"):
out.append(path)
return out
def alias_corpus() -> list[tuple[str, int, str]]:
rows = [(f"/en/{a}/", 301, f"/en/{s}/") for a, s in ALIASES]
# 1.2 excluded: never had an RTD PDF artifact (Phase-0 finding, spec §15a) — pdf: null
rows += [(f"/_/downloads/en/{s}/pdf/", 301, f"/en/{s}/vyos-documentation.pdf")
for s in ["rolling", "1.5", "1.4", "1.3"]]
rows.append(("/_/downloads/en/latest/pdf/", 301, "/en/rolling/vyos-documentation.pdf"))
return rows
class _NoRedirect(urllib.request.HTTPRedirectHandler):
"""The parity checker must SEE 301s, not follow them (alias assertions)."""
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D401
return None
_OPENER = urllib.request.build_opener(_NoRedirect)
# Overridable in tests (monkeypatched to "http") so fetch() can be exercised end-to-end
# against a real local http.server instead of requiring TLS for a unit test.
_SCHEME = "https"
def fetch(host: str, path: str, access: tuple[str, str] | None, method: str = "HEAD"):
req = urllib.request.Request(f"{_SCHEME}://{host}{path}", method=method)
if access:
req.add_header("CF-Access-Client-Id", access[0])
req.add_header("CF-Access-Client-Secret", access[1])
try:
with _OPENER.open(req, timeout=30) as r:
return r.status, r.headers.get("Location")
except urllib.error.HTTPError as e: # 3xx land here with the no-redirect handler
return e.code, e.headers.get("Location")
except Exception: # noqa: BLE001 — DNS blip/timeout fails THIS probe, not the run
return 0, None
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--sitemap-host", required=True)
ap.add_argument("--probe-host", required=True)
ap.add_argument("--slugs", default=DEFAULT_SLUGS)
ap.add_argument("--access-id")
ap.add_argument("--access-secret")
ap.add_argument("--report", type=Path, default=Path("parity-report.json"))
a = ap.parse_args()
access = (a.access_id, a.access_secret) if a.access_id else None
failures: list[dict] = []
checked = 0
for slug in a.slugs.split(","):
status, _ = fetch(a.sitemap_host, f"/en/{slug}/sitemap.xml", None, "GET")
if status != 200:
failures.append({"path": f"/en/{slug}/sitemap.xml", "reason": f"sitemap {status}"})
continue
try:
with urllib.request.urlopen(f"https://{a.sitemap_host}/en/{slug}/sitemap.xml",
timeout=30) as r:
urls = urls_from_sitemap(r.read().decode())
except Exception as e: # noqa: BLE001 — record per-slug, keep sweeping; report ALWAYS written
failures.append({"path": f"/en/{slug}/sitemap.xml",
"reason": f"sitemap fetch error: {e}"})
continue
for path in urls:
checked += 1
st, _ = fetch(a.probe_host, path, access)
if st != 200:
failures.append({"path": path, "reason": f"status {st}"})
for path, want_status, want_loc in alias_corpus():
checked += 1
st, loc = fetch(a.probe_host, path, access)
if st != want_status or (loc or "") != want_loc:
failures.append({"path": path, "reason": f"got {st} → {loc}, want {want_status} → {want_loc}"})
a.report.write_text(json.dumps({"checked": checked, "failures": failures}, indent=2))
for f in failures:
print(f"PARITY-FAIL {f['path']}: {f['reason']}", file=sys.stderr)
print(f"checked={checked} failures={len(failures)}")
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())
|
