feat: expose typed deployment capability receipts

2026-08-07 11:15:49 +02:00
parent 40cc012124
commit eb0c01c5d2
4 changed files with 508 additions and 1 deletions
@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from govoplan_core.core.infrastructure_capabilities import (
+    InfrastructureCapabilityReceiptError,
+    deployment_capability_status,
+    infrastructure_capability_receipt_from_mapping,
+    load_infrastructure_capability_receipt,
+)
+
+
+def _receipt_payload() -> dict[str, object]:
+    return {
+        "schema_version": 1,
+        "installation_id": "govoplan-test",
+        "profile": "evaluation",
+        "capabilities": [
+            {
+                "id": "mail.smtp",
+                "label": "SMTP delivery",
+                "state": "available_unconfigured",
+                "source": "installer-managed-test",
+                "detail": "Mail owns the profile binding.",
+                "endpoint": {"scheme": "smtp", "host": "test-mail", "port": 3025},
+                "secret_refs": [],
+                "dependent_modules": ["mail"],
+            }
+        ],
+        "post_install_tasks": [
+            {
+                "id": "mail.smtp-profile",
+                "resume_key": "govoplan-test:mail.smtp-profile:v1",
+                "capability_id": "mail.smtp",
+                "state": "pending",
+                "owner_module": "mail",
+                "summary": "Create a Mail SMTP profile.",
+                "required_inputs": ["credential envelope reference when required"],
+                "secret_boundary": "credential-envelope-reference-only",
+            }
+        ],
+    }
+
+
+class InfrastructureCapabilityReceiptTests(unittest.TestCase):
+    def test_parses_typed_capability_and_task_lookup(self) -> None:
+        receipt = infrastructure_capability_receipt_from_mapping(_receipt_payload())
+
+        capability = receipt.capability("mail.smtp")
+        self.assertIsNotNone(capability)
+        assert capability is not None
+        self.assertEqual("test-mail", capability.endpoint["host"])
+        self.assertEqual(
+            "govoplan-test:mail.smtp-profile:v1",
+            receipt.tasks_for(capability_id="mail.smtp", owner_module="mail")[0].resume_key,
+        )
+        self.assertEqual(_receipt_payload(), receipt.to_dict())
+
+    def test_loader_uses_configured_path_and_status_remains_non_secret(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-capability-receipt-") as root:
+            path = Path(root) / "capabilities.json"
+            path.write_text(json.dumps(_receipt_payload()), encoding="utf-8")
+            with patch.dict(
+                os.environ,
+                {"GOVOPLAN_DEPLOYMENT_CAPABILITIES_PATH": str(path)},
+                clear=False,
+            ):
+                receipt = load_infrastructure_capability_receipt()
+                status = deployment_capability_status()
+
+        self.assertIsNotNone(receipt)
+        self.assertTrue(status["available"])
+        self.assertNotIn("password", json.dumps(status).casefold())
+
+    def test_rejects_inline_secret_endpoint_and_non_environment_secret_reference(self) -> None:
+        for endpoint, secret_refs in (
+            ({"password": "inline"}, []),
+            ({}, ["plaintext-secret"]),
+        ):
+            with self.subTest(endpoint=endpoint, secret_refs=secret_refs):
+                payload = _receipt_payload()
+                capability = dict(payload["capabilities"][0])  # type: ignore[index]
+                capability["endpoint"] = endpoint
+                capability["secret_refs"] = secret_refs
+                payload["capabilities"] = [capability]
+
+                with self.assertRaises(InfrastructureCapabilityReceiptError):
+                    infrastructure_capability_receipt_from_mapping(payload)
+
+    def test_rejects_post_install_task_for_unknown_capability(self) -> None:
+        payload = _receipt_payload()
+        task = dict(payload["post_install_tasks"][0])  # type: ignore[index]
+        task["capability_id"] = "unknown.capability"
+        payload["post_install_tasks"] = [task]
+
+        with self.assertRaises(InfrastructureCapabilityReceiptError):
+            infrastructure_capability_receipt_from_mapping(payload)
+
+
+if __name__ == "__main__":
+    unittest.main()