govoplan-core/tests/test_infrastructure_capabilities.py

from __future__ import annotations

import json
import os
from datetime import UTC, datetime
from pathlib import Path
import tempfile
import unittest
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,
        "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_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())

        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()