feat: collect infrastructure dependency inventories
Module Package Release / publish-packages (push) Successful in 13s

This commit is contained in:
2026-08-24 15:00:08 +02:00
parent 08c3e47b6d
commit 9cb2080938
8 changed files with 378 additions and 38 deletions
+76
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import json
import os
from datetime import UTC, datetime
from pathlib import Path
import tempfile
import unittest
@@ -9,12 +10,53 @@ from unittest.mock import patch
from govoplan_core.core.infrastructure_capabilities import (
InfrastructureCapabilityReceiptError,
InfrastructureDependency,
collect_infrastructure_dependency_inventory,
deployment_capability_status,
infrastructure_capability_receipt_from_mapping,
load_infrastructure_capability_receipt,
)
class _InventoryProvider:
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:server-1",
state="active",
scope="system",
summary="One active SMTP endpoint uses the deployment relay.",
metrics={"credential_binding_count": 1},
required_action="Rebind or retire the endpoint before removal.",
),
)
class _FailingInventoryProvider:
module_id = "files"
capability_ids = ("files.storage",)
def infrastructure_dependencies(self) -> tuple[InfrastructureDependency, ...]:
raise RuntimeError("database URL must not escape")
class _Registry:
def __init__(self, providers: dict[str, object]) -> None:
self.providers = providers
def capability_names(self) -> tuple[str, ...]:
return tuple(self.providers)
def capability(self, name: str) -> object | None:
return self.providers.get(name)
def _receipt_payload() -> dict[str, object]:
return {
"schema_version": 1,
@@ -48,6 +90,40 @@ def _receipt_payload() -> dict[str, object]:
class InfrastructureCapabilityReceiptTests(unittest.TestCase):
def test_collects_non_secret_provider_dependency_inventory(self) -> None:
inventory = collect_infrastructure_dependency_inventory(
_Registry(
{
"infrastructure.dependency_inventory.mail": _InventoryProvider(),
"unrelated.capability": object(),
}
),
installation_id="govoplan-test",
observed_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC),
)
self.assertTrue(inventory.complete)
self.assertEqual(("mail.smtp",), inventory.inspected_capability_ids)
self.assertEqual("mail-server:server-1", inventory.dependencies[0].dependency_ref)
self.assertEqual("2026-08-24T12:00:00+00:00", inventory.generated_at)
self.assertNotIn("database URL", json.dumps(inventory.to_dict()))
def test_provider_failure_makes_inventory_incomplete_without_leaking_error(self) -> None:
inventory = collect_infrastructure_dependency_inventory(
_Registry(
{
"infrastructure.dependency_inventory.files": (
_FailingInventoryProvider()
)
}
),
installation_id="govoplan-test",
)
self.assertFalse(inventory.complete)
self.assertEqual("error", inventory.providers[0].state)
self.assertNotIn("database URL", str(inventory.providers[0].error))
def test_parses_typed_capability_and_task_lookup(self) -> None:
receipt = infrastructure_capability_receipt_from_mapping(_receipt_payload())