Skip to content

Commit f6895af

Browse files
committed
feat(registry-check): implement registry-fetch smoke check module
- GREEN phase: check_registry_urls() performs live GETs against the Pages registry, raising RegistryCheckError naming the first failing path on non-200, HTTPError, or network-unavailable conditions - main() is the python -m cas_evals.registry_check CLI entry point, checking index.json, v0.1/manifest.json, v0.1/common.schema.json, and v0.1/evaluation-result.schema.json by default; exits non-zero on failure - Follows reference_product.py's urllib pattern: capped response reads (MAX_RESPONSE_BYTES), typed exception-to-domain-error translation - REQ-1.4.11
1 parent 3f99291 commit f6895af

1 file changed

Lines changed: 80 additions & 0 deletions

File tree

src/cas_evals/registry_check.py

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
"""Registry-fetch smoke check for the live cas-contracts GitHub Pages registry.
2+
3+
Performs live HTTP GETs against known-published registry paths and asserts
4+
each responds with HTTP 200. This is deliberately network-dependent and
5+
distinct from cas_evals.contracts.verify_vendored_contract(), which only
6+
checks bytes on disk and never touches the network. The two mechanisms are
7+
additive: verify_vendored_contract() proves the pinned offline copy is intact;
8+
this module proves the live registry actually resolves today.
9+
"""
10+
11+
from __future__ import annotations
12+
13+
import sys
14+
from urllib.error import HTTPError, URLError
15+
from urllib.request import Request, urlopen
16+
17+
DEFAULT_BASE_URL = "https://coding-autopilot-system.github.io/cas-contracts/registry"
18+
DEFAULT_PATHS = (
19+
"index.json",
20+
"v0.1/manifest.json",
21+
"v0.1/common.schema.json",
22+
"v0.1/evaluation-result.schema.json",
23+
)
24+
MAX_RESPONSE_BYTES = 2_000_000
25+
26+
27+
class RegistryCheckError(RuntimeError):
28+
"""Raised when a registry-fetch smoke check fails."""
29+
30+
31+
def check_registry_urls(
32+
base_url: str,
33+
paths: list[str] | tuple[str, ...],
34+
timeout_seconds: float = 10,
35+
) -> list[tuple[str, int]]:
36+
"""Fetch each path joined onto base_url and assert HTTP 200.
37+
38+
Returns a list of (path, status_code) tuples when every URL responds
39+
with 200. Raises RegistryCheckError naming the first failing path if any
40+
URL returns non-200, times out, or the network is unavailable.
41+
"""
42+
results: list[tuple[str, int]] = []
43+
for path in paths:
44+
url = f"{base_url.rstrip('/')}/{path.lstrip('/')}"
45+
request = Request(url, method="GET")
46+
try:
47+
with urlopen(request, timeout=timeout_seconds) as response:
48+
status = response.status
49+
response.read(MAX_RESPONSE_BYTES + 1)
50+
except HTTPError as error:
51+
raise RegistryCheckError(f"{path} returned HTTP {error.code}") from error
52+
except (URLError, TimeoutError, OSError) as error:
53+
raise RegistryCheckError(f"{path} is unavailable (network error)") from error
54+
if status != 200:
55+
raise RegistryCheckError(f"{path} returned HTTP {status}")
56+
results.append((path, status))
57+
return results
58+
59+
60+
def main(argv: list[str] | None = None) -> int:
61+
"""CLI entry point: python -m cas_evals.registry_check [--base-url URL]."""
62+
args = list(sys.argv[1:] if argv is None else argv)
63+
base_url = DEFAULT_BASE_URL
64+
if "--base-url" in args:
65+
index = args.index("--base-url")
66+
base_url = args[index + 1]
67+
68+
try:
69+
results = check_registry_urls(base_url, DEFAULT_PATHS)
70+
except RegistryCheckError as error:
71+
print(f"FAIL: {error}")
72+
return 1
73+
74+
for path, status in results:
75+
print(f"{path}: {status}")
76+
return 0
77+
78+
79+
if __name__ == "__main__":
80+
sys.exit(main())

0 commit comments

Comments
 (0)