refactor: share deployment receipt validation

src/govoplan_ops/backend/infrastructure.py

@@ -1,171 +1,18 @@
 from __future__ import annotations
 
-import json
-import os
 from pathlib import Path
-from typing import Any, Mapping
 
-
+from govoplan_core.core.infrastructure_capabilities import (
-MAX_CAPABILITY_DOCUMENT_BYTES = 256 * 1024
+    deployment_capability_status as _deployment_capability_status,
-CAPABILITY_STATES = {
+)
-    "configured",
-    "available_unconfigured",
-    "externally_supplied",
-    "unavailable",
-}
 
 
 def deployment_capability_status(
     path: Path | None = None,
 ) -> dict[str, object]:
-    configured_path = path
+    """Project Core's validated non-secret deployment receipt into Ops."""
-    if configured_path is None:
+
-        raw_path = os.getenv("GOVOPLAN_DEPLOYMENT_CAPABILITIES_PATH", "").strip()
+    return _deployment_capability_status(path)
-        if not raw_path:
-            return _unavailable(configured=False, error=None)
-        configured_path = Path(raw_path)
-    try:
-        document = _read_document(configured_path)
-        return {
-            "configured": True,
-            "available": True,
-            "schema_version": 1,
-            "installation_id": document["installation_id"],
-            "profile": document["profile"],
-            "capabilities": document["capabilities"],
-            "post_install_tasks": document["post_install_tasks"],
-            "error": None,
-        }
-    except (OSError, ValueError, json.JSONDecodeError) as exc:
-        return _unavailable(configured=True, error=str(exc))
 
 
-def _read_document(path: Path) -> dict[str, object]:
+__all__ = ["deployment_capability_status"]
-    if path.is_symlink() or not path.is_file():
-        raise ValueError("Deployment capability receipt is not a regular file.")
-    size = path.stat().st_size
-    if size > MAX_CAPABILITY_DOCUMENT_BYTES:
-        raise ValueError("Deployment capability receipt exceeds 256 KiB.")
-    raw = path.read_bytes()
-    if len(raw) != size:
-        raise ValueError("Deployment capability receipt changed while being read.")
-    payload = json.loads(raw.decode("utf-8"))
-    if not isinstance(payload, Mapping) or payload.get("schema_version") != 1:
-        raise ValueError("Deployment capability receipt has an unsupported schema.")
-    installation_id = _required_text(payload, "installation_id", maximum=100)
-    profile = _required_text(payload, "profile", maximum=100)
-    raw_capabilities = payload.get("capabilities")
-    if not isinstance(raw_capabilities, list) or len(raw_capabilities) > 100:
-        raise ValueError("Deployment capability receipt has invalid capabilities.")
-    capabilities = [_capability(item) for item in raw_capabilities]
-    ids = [str(item["id"]) for item in capabilities]
-    if len(ids) != len(set(ids)):
-        raise ValueError("Deployment capability receipt repeats a capability id.")
-    raw_tasks = payload.get("post_install_tasks", [])
-    if not isinstance(raw_tasks, list) or len(raw_tasks) > 100:
-        raise ValueError("Deployment capability receipt has invalid post-install tasks.")
-    tasks = [_task(item) for item in raw_tasks]
-    return {
-        "installation_id": installation_id,
-        "profile": profile,
-        "capabilities": capabilities,
-        "post_install_tasks": tasks,
-    }
-
-
-def _capability(value: object) -> dict[str, object]:
-    if not isinstance(value, Mapping):
-        raise ValueError("Deployment capability entries must be objects.")
-    state = _required_text(value, "state", maximum=40)
-    if state not in CAPABILITY_STATES:
-        raise ValueError(f"Deployment capability state is unsupported: {state!r}.")
-    endpoint = value.get("endpoint", {})
-    if not isinstance(endpoint, Mapping) or len(endpoint) > 10:
-        raise ValueError("Deployment capability endpoint metadata is invalid.")
-    normalized_endpoint: dict[str, object] = {}
-    for key, raw in endpoint.items():
-        if not isinstance(key, str) or len(key) > 50:
-            raise ValueError("Deployment capability endpoint key is invalid.")
-        if isinstance(raw, bool) or raw is None:
-            normalized_endpoint[key] = raw
-        elif isinstance(raw, int):
-            normalized_endpoint[key] = raw
-        elif isinstance(raw, str) and len(raw) <= 500:
-            normalized_endpoint[key] = raw
-        else:
-            raise ValueError("Deployment capability endpoint value is invalid.")
-    secret_refs = _string_list(value.get("secret_refs"), maximum_items=30)
-    if any(not item.startswith("env:") for item in secret_refs):
-        raise ValueError("Deployment capability secrets must use environment references.")
-    return {
-        "id": _required_text(value, "id", maximum=120),
-        "label": _required_text(value, "label", maximum=200),
-        "state": state,
-        "source": _required_text(value, "source", maximum=120),
-        "detail": _required_text(value, "detail", maximum=1000),
-        "endpoint": normalized_endpoint,
-        "secret_refs": secret_refs,
-        "dependent_modules": _string_list(
-            value.get("dependent_modules"),
-            maximum_items=100,
-        ),
-    }
-
-
-def _task(value: object) -> dict[str, object]:
-    if not isinstance(value, Mapping):
-        raise ValueError("Deployment post-install task entries must be objects.")
-    return {
-        "id": _required_text(value, "id", maximum=120),
-        "resume_key": _required_text(value, "resume_key", maximum=240),
-        "capability_id": _required_text(value, "capability_id", maximum=120),
-        "state": _required_text(value, "state", maximum=40),
-        "owner_module": _required_text(value, "owner_module", maximum=120),
-        "summary": _required_text(value, "summary", maximum=1000),
-        "required_inputs": _string_list(
-            value.get("required_inputs"),
-            maximum_items=30,
-        ),
-        "secret_boundary": _required_text(
-            value,
-            "secret_boundary",
-            maximum=120,
-        ),
-    }
-
-
-def _required_text(
-    value: Mapping[str, Any],
-    key: str,
-    *,
-    maximum: int,
-) -> str:
-    raw = value.get(key)
-    text = str(raw).strip() if raw is not None else ""
-    if not text or len(text) > maximum:
-        raise ValueError(f"Deployment capability field {key!r} is invalid.")
-    return text
-
-
-def _string_list(value: object, *, maximum_items: int) -> list[str]:
-    if not isinstance(value, list) or len(value) > maximum_items:
-        raise ValueError("Deployment capability list field is invalid.")
-    result: list[str] = []
-    for item in value:
-        if not isinstance(item, str) or not item.strip() or len(item) > 500:
-            raise ValueError("Deployment capability list item is invalid.")
-        result.append(item.strip())
-    return result
-
-
-def _unavailable(*, configured: bool, error: str | None) -> dict[str, object]:
-    return {
-        "configured": configured,
-        "available": False,
-        "schema_version": None,
-        "installation_id": None,
-        "profile": None,
-        "capabilities": [],
-        "post_install_tasks": [],
-        "error": error,
-    }