"""VM112 portal infrastructure checks (read-only API).""" from __future__ import annotations from typing import Any import httpx def _result(check_id: str, label: str, ok: bool | None, message: str, evidence: dict | None = None) -> dict[str, Any]: if ok is True: status = "pass" elif ok is False: status = "fail" else: status = "error" return { "check_id": check_id, "label": label, "status": status, "message": message, "evidence": evidence or {}, } def collect(domain: str, api_base: str) -> dict[str, dict[str, Any]]: url = f"{api_base.rstrip('/')}/api/onboarding/infrastructure/status/{domain}" try: with httpx.Client(timeout=15.0) as client: response = client.get(url) if response.status_code != 200: err = _result("carbonio", "Carbonio domain", None, f"Portal API HTTP {response.status_code}") return { "carbonio": err, "nginx_vhost": {**err, "check_id": "nginx_vhost", "label": "carbonio-nginx vhost"}, "cert_le": {**err, "check_id": "cert_le", "label": "Let's Encrypt certificate"}, } data = response.json() except Exception as exc: err = _result("carbonio", "Carbonio domain", None, str(exc)) return { "carbonio": err, "nginx_vhost": {**err, "check_id": "nginx_vhost", "label": "carbonio-nginx vhost"}, "cert_le": {**err, "check_id": "cert_le", "label": "Let's Encrypt certificate"}, } steps = {s.get("id"): s for s in data.get("steps") or [] if isinstance(s, dict)} def from_step(check_id: str, label: str, step_id: str) -> dict[str, Any]: step = steps.get(step_id) or {} return _result( check_id, label, step.get("ok"), step.get("message") or f"Step {step_id}", {"step_id": step_id, "ready": data.get("ready")}, ) cert = from_step("cert_le", "Let's Encrypt certificate", "cert_san") if cert["status"] == "pass": cert["status"] = "pass" return { "carbonio": from_step("carbonio", "Carbonio domain", "carbonio_domain"), "nginx_vhost": from_step("nginx_vhost", "carbonio-nginx vhost", "carbonio_nginx_vhost"), "cert_le": cert, }