generated from coulomb/repo-seed
112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
"""Probe whether Inter-Hub production exposes the ops-hub bootstrap API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from dataclasses import asdict, dataclass
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urljoin
|
|
from urllib.request import Request, urlopen
|
|
|
|
REQUIRED_OPENAPI_PATHS = (
|
|
"/hubs",
|
|
"/hub-capability-manifests",
|
|
"/api-consumers",
|
|
"/policy-scopes",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HttpObservation:
|
|
url: str
|
|
status: int | None
|
|
ok: bool
|
|
error: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GateResult:
|
|
base_url: str
|
|
passed: bool
|
|
hubs_status: int | None
|
|
required_paths_present: list[str]
|
|
required_paths_missing: list[str]
|
|
observations: list[HttpObservation]
|
|
|
|
|
|
def normalize_base_url(base_url: str) -> str:
|
|
return base_url.rstrip("/") + "/"
|
|
|
|
|
|
def api_url(base_url: str, path: str) -> str:
|
|
return urljoin(normalize_base_url(base_url), path.lstrip("/"))
|
|
|
|
|
|
def fetch_json(url: str, timeout: float) -> tuple[HttpObservation, dict[str, Any] | None]:
|
|
request = Request(url, headers={"Accept": "application/json", "User-Agent": "ops-hub-gate-probe/0.1"})
|
|
try:
|
|
with urlopen(request, timeout=timeout) as response:
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
return HttpObservation(url=url, status=response.status, ok=True), payload
|
|
except HTTPError as exc:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
return HttpObservation(url=url, status=exc.code, ok=False, error=body[:240]), None
|
|
except (URLError, TimeoutError, json.JSONDecodeError) as exc:
|
|
return HttpObservation(url=url, status=None, ok=False, error=str(exc)), None
|
|
|
|
|
|
def observe_status(url: str, timeout: float) -> HttpObservation:
|
|
request = Request(url, headers={"Accept": "application/json", "User-Agent": "ops-hub-gate-probe/0.1"})
|
|
try:
|
|
with urlopen(request, timeout=timeout) as response:
|
|
response.read()
|
|
return HttpObservation(url=url, status=response.status, ok=True)
|
|
except HTTPError as exc:
|
|
exc.read()
|
|
return HttpObservation(url=url, status=exc.code, ok=False, error=exc.reason)
|
|
except (URLError, TimeoutError) as exc:
|
|
return HttpObservation(url=url, status=None, ok=False, error=str(exc))
|
|
|
|
|
|
def evaluate_openapi_paths(openapi: dict[str, Any] | None) -> tuple[list[str], list[str]]:
|
|
paths = set((openapi or {}).get("paths", {}).keys())
|
|
present = [path for path in REQUIRED_OPENAPI_PATHS if path in paths]
|
|
missing = [path for path in REQUIRED_OPENAPI_PATHS if path not in paths]
|
|
return present, missing
|
|
|
|
|
|
def probe_interhub_gate(base_url: str, timeout: float = 10.0) -> GateResult:
|
|
normalized = normalize_base_url(base_url)
|
|
hubs = observe_status(api_url(normalized, "/api/v2/hubs"), timeout)
|
|
openapi_observation, openapi = fetch_json(api_url(normalized, "/api/v2/openapi.json"), timeout)
|
|
present, missing = evaluate_openapi_paths(openapi)
|
|
passed = hubs.status == 401 and not missing
|
|
return GateResult(
|
|
base_url=normalized.rstrip("/"),
|
|
passed=passed,
|
|
hubs_status=hubs.status,
|
|
required_paths_present=present,
|
|
required_paths_missing=missing,
|
|
observations=[hubs, openapi_observation],
|
|
)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--base", default="https://hub.coulomb.social", help="Inter-Hub base URL")
|
|
parser.add_argument("--timeout", default=10.0, type=float, help="HTTP timeout in seconds")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
result = probe_interhub_gate(args.base, timeout=args.timeout)
|
|
print(json.dumps(asdict(result), indent=2, sort_keys=True))
|
|
return 0 if result.passed else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|