Files
govoplan-ops/tests/test_infrastructure_capabilities.py
T
zemion ac6b211827
Module Package Release / publish-packages (push) Successful in 12s
feat: expose infrastructure dependency inventory
2026-08-24 14:56:38 +02:00

152 lines
6.0 KiB
Python

from __future__ import annotations
import json
from pathlib import Path
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import patch
from govoplan_core.core.infrastructure_capabilities import InfrastructureDependency
from govoplan_ops.backend.infrastructure import (
OpsInfrastructureDependencyProvider,
deployment_capability_status,
infrastructure_dependency_inventory,
)
class _Provider:
module_id = "mail"
capability_ids = ("mail.smtp",)
def infrastructure_dependencies(self) -> tuple[InfrastructureDependency, ...]:
return (
InfrastructureDependency(
capability_id="mail.smtp",
module_id="mail",
dependency_type="smtp_endpoint",
dependency_ref="mail:server-1",
state="active",
scope="system",
summary="Persisted SMTP endpoint.",
metrics={"credential_binding_count": 1},
required_action="Rebind it before changing SMTP.",
),
)
class _Registry:
def capability_names(self) -> tuple[str, ...]:
return ("infrastructure.dependency_inventory.mail",)
def capability(self, name: str) -> object | None:
return _Provider() if name.endswith(".mail") else None
class InfrastructureCapabilityTests(unittest.TestCase):
def test_collects_authorized_module_dependency_inventory(self) -> None:
with patch(
"govoplan_ops.backend.infrastructure.load_infrastructure_capability_receipt",
return_value=SimpleNamespace(installation_id="govoplan-test"),
):
result = infrastructure_dependency_inventory(
_Registry(),
installation_id="govoplan-test",
).to_dict()
self.assertTrue(result["complete"])
self.assertEqual("mail.smtp", result["dependencies"][0]["capability_id"])
self.assertEqual(1, result["dependencies"][0]["metrics"]["credential_binding_count"])
def test_ops_provider_reports_runtime_bindings_without_endpoint_secrets(self) -> None:
dependencies = OpsInfrastructureDependencyProvider().infrastructure_dependencies()
by_capability = {item.capability_id: item for item in dependencies}
self.assertIn("database.postgresql", by_capability)
self.assertIn("network.ingress", by_capability)
self.assertEqual(
"runtime:database.postgresql",
by_capability["database.postgresql"].dependency_ref,
)
self.assertNotIn("postgresql://", json.dumps([item.to_dict() for item in dependencies]))
def test_reads_bounded_non_secret_capability_receipt(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-ops-capabilities-") as root:
path = Path(root) / "capabilities.json"
path.write_text(
json.dumps(
{
"schema_version": 1,
"installation_id": "govoplan-test",
"profile": "evaluation",
"capabilities": [
{
"id": "mail.smtp",
"label": "SMTP delivery",
"state": "available_unconfigured",
"source": "operator-supplied",
"detail": "Mail needs a profile.",
"endpoint": {},
"secret_refs": ["env:SMTP_CREDENTIAL_REF"],
"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": "Configure Mail.",
"required_inputs": ["credential envelope reference"],
"secret_boundary": "credential-envelope-reference-only",
}
],
}
),
encoding="utf-8",
)
result = deployment_capability_status(path)
self.assertTrue(result["available"])
self.assertEqual("mail.smtp", result["capabilities"][0]["id"])
self.assertEqual("mail.smtp-profile", result["post_install_tasks"][0]["id"])
def test_rejects_inline_secret_instead_of_reference(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-ops-capabilities-") as root:
path = Path(root) / "capabilities.json"
path.write_text(
json.dumps(
{
"schema_version": 1,
"installation_id": "govoplan-test",
"profile": "evaluation",
"capabilities": [
{
"id": "mail.smtp",
"label": "SMTP delivery",
"state": "configured",
"source": "operator-supplied",
"detail": "Configured.",
"endpoint": {},
"secret_refs": ["plaintext-secret"],
"dependent_modules": [],
}
],
"post_install_tasks": [],
}
),
encoding="utf-8",
)
result = deployment_capability_status(path)
self.assertFalse(result["available"])
self.assertIn("environment references", str(result["error"]))
if __name__ == "__main__":
unittest.main()