govoplan-access/tests/test_configuration_package_context.py

from __future__ import annotations

from datetime import UTC, datetime
from types import SimpleNamespace
import unittest
from unittest.mock import patch

from govoplan_access.backend.api.v1.admin_schemas import (
    ConfigurationPackageApplyResponse,
    ConfigurationPackageExportResponse,
)
from govoplan_access.backend.api.v1.routes import _configuration_context
from govoplan_core.core.infrastructure_capabilities import (
    InfrastructureCapabilityReceiptError,
)
from govoplan_core.core.provider_governance import (
    ExternalProviderRuntimeState,
    ExternalProviderStateProviderRegistration,
)


class ConfigurationPackageContextTests(unittest.TestCase):
    def test_api_responses_preserve_rollback_and_redacted_export_provenance(self) -> None:
        applied = ConfigurationPackageApplyResponse(
            rollback={
                "status": "database_restore_required",
                "summary": "Snapshot is the generic rollback boundary.",
                "recovery_action": "Retain the snapshot.",
            }
        )
        exported = ConfigurationPackageExportResponse(
            provenance={
                "exported_at": "2026-08-22T12:00:00+00:00",
                "source_core_version": "0.1.35",
                "module_versions": {"forms": "0.1.20"},
                "tenant_id": "tenant-1",
                "exporter_id": "user-1",
                "selection": {
                    "scopes": ["tenant"],
                    "module_ids": ["forms"],
                    "object_refs": [],
                },
                "redacted_secret_keys": ["credential_ref"],
            }
        )

        self.assertEqual(
            "database_restore_required",
            applied.model_dump()["rollback"]["status"],
        )
        self.assertEqual(
            ["credential_ref"],
            exported.model_dump()["provenance"]["redacted_secret_keys"],
        )

    def test_context_carries_operator_scopes_and_validated_infrastructure_receipt(self) -> None:
        receipt = SimpleNamespace(installation_id="deployment-1")
        principal = SimpleNamespace(
            tenant_id="tenant-1",
            user=SimpleNamespace(id="user-1"),
            scopes=frozenset({"system:settings:write"}),
        )

        with (
            patch(
                "govoplan_access.backend.api.v1.routes.get_registry",
                return_value=None,
            ),
            patch(
                "govoplan_access.backend.api.v1.routes.load_infrastructure_capability_receipt",
                return_value=receipt,
            ),
        ):
            context = _configuration_context(principal)

        self.assertIs(receipt, context.infrastructure_receipt)
        self.assertEqual(
            frozenset({"system:settings:write"}),
            context.operator_scopes,
        )

    def test_context_preserves_invalid_receipt_as_fail_closed_provider_state(self) -> None:
        principal = SimpleNamespace(
            tenant_id="tenant-1",
            user=SimpleNamespace(id="user-1"),
            scopes=frozenset(),
        )

        with (
            patch(
                "govoplan_access.backend.api.v1.routes.get_registry",
                return_value=None,
            ),
            patch(
                "govoplan_access.backend.api.v1.routes.load_infrastructure_capability_receipt",
                side_effect=InfrastructureCapabilityReceiptError("invalid receipt"),
            ),
        ):
            context = _configuration_context(principal)

        self.assertIsNone(context.infrastructure_receipt)
        self.assertEqual("invalid receipt", context.infrastructure_receipt_error)

    def test_context_projects_installed_external_provider_declarations(self) -> None:
        declaration = SimpleNamespace(
            id="connectors.example",
            to_dict=lambda: {
                "id": "connectors.example",
                "maturity": "read",
                "authority_modes": ["external_mirror"],
            },
        )
        registry = SimpleNamespace(
            manifests=lambda: (
                SimpleNamespace(id="access", version="0.1.14"),
                SimpleNamespace(id="connectors", version="0.1.14"),
            ),
            capability_names=lambda: ("connectors.profiles",),
            external_provider_declarations=lambda: (declaration,),
        )
        principal = SimpleNamespace(
            tenant_id="tenant-1",
            user=SimpleNamespace(id="user-1"),
        )

        with patch(
            "govoplan_access.backend.api.v1.routes.get_registry",
            return_value=registry,
        ):
            context = _configuration_context(principal)

        self.assertEqual("0.1.14", context.installed_modules["connectors"])
        self.assertIn("connectors.profiles", context.capabilities)
        self.assertEqual(
            "external_mirror",
            context.external_provider_declarations["connectors.example"][
                "authority_modes"
            ][0],
        )

    def test_context_projects_tenant_runtime_provider_state(self) -> None:
        declaration = SimpleNamespace(
            id="calendar.caldav_sync",
            to_dict=lambda: {
                "id": "calendar.caldav_sync",
                "maturity": "synchronize",
                "authority_modes": ["governed_sync"],
            },
        )
        registration = ExternalProviderStateProviderRegistration(
            module_id="calendar",
            provider_id="calendar.caldav_sync",
            provider=lambda context: (
                ExternalProviderRuntimeState(
                    provider_id="calendar.caldav_sync",
                    binding_ref="calendar:sync-source:one",
                    authority_mode="governed_sync",
                    observed_at=datetime(2026, 8, 1, 12, 0, tzinfo=UTC),
                    configured=True,
                    active=True,
                    health="healthy",
                    freshness="current",
                    conflict="clear",
                    recovery="ready",
                    metrics={"tenant_matches": context.tenant_id == "tenant-1"},
                ),
            ),
        )
        registry = SimpleNamespace(
            manifests=lambda: (SimpleNamespace(id="calendar", version="0.1.8"),),
            capability_names=lambda: (),
            external_provider_declarations=lambda: (declaration,),
            external_provider_state_providers=lambda: (registration,),
        )
        principal = SimpleNamespace(
            tenant_id="tenant-1",
            user=SimpleNamespace(id="user-1"),
        )

        with patch(
            "govoplan_access.backend.api.v1.routes.get_registry",
            return_value=registry,
        ):
            context = _configuration_context(principal, session=object())

        state = context.external_provider_states["calendar.caldav_sync"]
        self.assertEqual("healthy", state["health"])
        self.assertEqual("calendar:sync-source:one", state["binding_ref"])


if __name__ == "__main__":
    unittest.main()