govoplan-core/tests/test_configuration_package_architecture.py

from __future__ import annotations

import unittest

from govoplan_core.core.configuration_packages import (
    ConfigurationApplyResult,
    ConfigurationExportResult,
    ConfigurationExportSelection,
    ConfigurationModuleRequirement,
    ConfigurationPackageFragment,
    ConfigurationPackageEvidence,
    ConfigurationPackageManifest,
    ConfigurationPackageParent,
    ConfigurationPlanItem,
    ConfigurationPreflightContext,
    ConfigurationPreflightResult,
    ConfigurationProviderExpectation,
    ConfigurationRequiredData,
    apply_configuration_package,
    dry_run_configuration_package,
    export_configuration_package,
    validate_configuration_package_derivation,
)


def _evidence(*kinds: str) -> tuple[ConfigurationPackageEvidence, ...]:
    return tuple(
        ConfigurationPackageEvidence(
            kind=kind,  # type: ignore[arg-type]
            reference=f"evidence/{kind}.json",
            summary=f"Evidence for {kind}.",
            checksum=f"sha256:{'a' * 64}",
        )
        for kind in kinds
    )


class ConfigurationPackageArchitectureTests(unittest.TestCase):
    def test_deployment_data_references_are_declared_resolved_and_never_exported(self) -> None:
        class Provider:
            module_id = "forms"

            def __init__(self) -> None:
                self.preflight_payloads: list[dict[str, object]] = []

            def describe(self):
                from govoplan_core.core.configuration_packages import ConfigurationProviderDescription

                return ConfigurationProviderDescription(
                    module_id=self.module_id,
                    fragment_types=("definition",),
                )

            def preflight(self, fragment, context):
                del context
                self.preflight_payloads.append(dict(fragment.payload))
                return ConfigurationPreflightResult(plan=(ConfigurationPlanItem(
                    action="create",
                    module_id=self.module_id,
                    fragment_type=fragment.fragment_type,
                    fragment_id=fragment.fragment_id,
                ),))

            def apply(self, fragment, supplied_data, context):
                del supplied_data, context
                return ConfigurationApplyResult(
                    created_refs={fragment.fragment_id or "definition": "form:resident-parking"}
                )

            def export(self, selection, context):
                del selection, context
                return ConfigurationExportResult(
                    fragments=(ConfigurationPackageFragment(
                        module_id=self.module_id,
                        fragment_type="definition",
                        payload={"name": "Resident parking permit"},
                    ),),
                    data_requirements=(ConfigurationRequiredData(
                        key="payment_credential_ref",
                        label="Payment credential reference",
                        secret=True,
                    ),),
                )

            def health(self, import_result, context):
                del import_result, context
                return ()

        provider = Provider()
        package = ConfigurationPackageManifest(
            package_id="product.resident-parking",
            name="Resident parking permit",
            version="1.0.0",
            required_modules=(ConfigurationModuleRequirement("forms"),),
            data_requirements=({
                "key": "service_name",
                "label": "Public service name",
            },),
            fragments=(ConfigurationPackageFragment(
                module_id="forms",
                fragment_type="definition",
                fragment_id="resident-parking",
                payload={
                    "definition": {
                        "title": {"$data": "service_name"},
                    }
                },
            ),),
        )
        missing_context = ConfigurationPreflightContext(
            installed_modules={"forms": "0.1.0"},
        )

        missing = dry_run_configuration_package(package, (provider,), missing_context)

        self.assertEqual([], provider.preflight_payloads)
        self.assertIn(
            "fragment_data_reference_missing",
            {item.code for item in missing.diagnostics},
        )

        ready_context = ConfigurationPreflightContext(
            installed_modules={"forms": "0.1.0"},
            supplied_data={"service_name": "Anwohnerparkausweis"},
            operator_user_id="operator-1",
        )
        ready = dry_run_configuration_package(package, (provider,), ready_context)
        applied = apply_configuration_package(package, (provider,), ready_context)
        exported = export_configuration_package(
            (provider,),
            ConfigurationExportSelection(
                tenant_id="tenant-1",
                module_ids=("forms",),
            ),
            ready_context,
        )

        self.assertFalse(any(item.severity == "blocker" for item in ready.diagnostics))
        self.assertEqual(
            "Anwohnerparkausweis",
            provider.preflight_payloads[-1]["definition"]["title"],  # type: ignore[index]
        )
        self.assertIsNotNone(applied.rollback)
        assert applied.rollback is not None
        self.assertEqual("database_restore_required", applied.rollback.status)
        self.assertIsNotNone(exported.provenance)
        assert exported.provenance is not None
        self.assertEqual("operator-1", exported.provenance.exporter_id)
        self.assertEqual(
            ("payment_credential_ref",),
            exported.provenance.redacted_secret_keys,
        )

    def test_legacy_package_defaults_to_product_and_round_trips(self) -> None:
        package = ConfigurationPackageManifest.from_mapping(
            {"package_id": "example", "name": "Example", "version": "1.0.0"}
        )

        self.assertEqual(package.package_class, "product")
        self.assertEqual(
            ConfigurationPackageManifest.from_mapping(package.to_dict()),
            package,
        )

    def test_reference_package_requires_operational_evidence(self) -> None:
        with self.assertRaisesRegex(ValueError, "accessibility"):
            ConfigurationPackageManifest(
                package_id="reference.example",
                name="Example reference",
                version="1.0.0",
                package_class="reference",
                evidence=_evidence("target_test", "recovery"),
            )

    def test_reference_evidence_requires_content_checksums(self) -> None:
        with self.assertRaisesRegex(ValueError, "without checksums"):
            ConfigurationPackageManifest(
                package_id="reference.unbound",
                name="Unbound reference",
                version="1.0.0",
                package_class="reference",
                evidence=tuple(
                    ConfigurationPackageEvidence(
                        kind=kind,  # type: ignore[arg-type]
                        reference=f"evidence/{kind}.json",
                        summary=f"{kind} evidence.",
                        checksum=(
                            None
                            if kind == "target_test"
                            else f"sha256:{'b' * 64}"
                        ),
                    )
                    for kind in (
                        "target_test",
                        "recovery",
                        "security",
                        "operations",
                        "accessibility",
                        "privacy",
                        "documentation",
                    )
                ),
            )

    def test_integration_package_preflight_checks_provider_contract_and_health(self) -> None:
        package = ConfigurationPackageManifest(
            package_id="integration.example",
            name="Example integration",
            version="1.0.0",
            package_class="integration",
            evidence=_evidence(
                "target_test", "recovery", "operations", "documentation"
            ),
            provider_expectations=(
                ConfigurationProviderExpectation(
                    provider_id="connectors.example",
                    authority_mode="external_mirror",
                    minimum_maturity="read",
                ),
            ),
        )

        healthy = dry_run_configuration_package(
            package,
            (),
            ConfigurationPreflightContext(
                external_provider_declarations={
                    "connectors.example": {
                        "authority_modes": ["external_mirror"],
                        "maturity": "synchronize",
                    }
                },
                external_provider_states={
                    "connectors.example": {"health": "healthy"}
                },
            ),
        )
        unhealthy = dry_run_configuration_package(
            package,
            (),
            ConfigurationPreflightContext(
                external_provider_declarations={
                    "connectors.example": {
                        "authority_modes": ["external_authoritative"],
                        "maturity": "discover",
                    }
                },
                external_provider_states={
                    "connectors.example": {"health": "error"}
                },
            ),
        )

        self.assertFalse(
            any(item.severity == "blocker" for item in healthy.diagnostics)
        )
        self.assertEqual(
            {
                "external_provider_authority_mode_unsupported",
                "external_provider_maturity_insufficient",
                "external_provider_unhealthy",
            },
            {item.code for item in unhealthy.diagnostics},
        )

    def test_provider_preflight_selects_the_requested_runtime_binding(self) -> None:
        package = ConfigurationPackageManifest(
            package_id="integration.calendar",
            name="Calendar integration",
            version="1.0.0",
            package_class="integration",
            evidence=_evidence(
                "target_test", "recovery", "operations", "documentation"
            ),
            provider_expectations=(
                ConfigurationProviderExpectation(
                    provider_id="calendar.caldav_sync",
                    authority_mode="governed_sync",
                    minimum_maturity="synchronize",
                    binding_ref="calendar:sync-source:required",
                    freshness_expectation="current",
                    recovery_expectation="ready",
                ),
            ),
        )
        context = ConfigurationPreflightContext(
            external_provider_declarations={
                "calendar.caldav_sync": {
                    "authority_modes": ["governed_sync"],
                    "maturity": "synchronize",
                    "behavior": {},
                }
            },
            external_provider_states={
                "calendar.caldav_sync": {
                    "health": "warning",
                    "bindings": [
                        {
                            "binding_ref": "calendar:sync-source:other",
                            "authority_mode": "governed_sync",
                            "health": "error",
                            "freshness": "stale",
                            "recovery": "attention",
                        },
                        {
                            "binding_ref": "calendar:sync-source:required",
                            "authority_mode": "governed_sync",
                            "health": "healthy",
                            "freshness": "current",
                            "recovery": "ready",
                        },
                    ],
                }
            },
        )

        result = dry_run_configuration_package(package, (), context)

        self.assertFalse(
            any(item.severity == "blocker" for item in result.diagnostics)
        )

    def test_derived_package_may_tighten_but_not_remove_parent_constraints(self) -> None:
        parent = ConfigurationPackageManifest(
            package_id="product.base",
            name="Base",
            version="1.0.0",
            required_modules=(
                ConfigurationModuleRequirement("connectors", "1.0.0"),
            ),
            required_capabilities=("connectors.profiles",),
            provider_expectations=(
                ConfigurationProviderExpectation(
                    provider_id="connectors.example",
                    authority_mode="external_mirror",
                    minimum_maturity="read",
                ),
            ),
        )
        child = ConfigurationPackageManifest(
            package_id="sector.example",
            name="Sector example",
            version="1.0.0",
            package_class="sector",
            parents=(
                ConfigurationPackageParent("product.base", "1.0.0"),
            ),
            evidence=_evidence("documentation"),
        )

        codes = {
            item.code
            for item in validate_configuration_package_derivation(child, parent)
        }
        self.assertEqual(
            {
                "package_parent_module_constraint_loosened",
                "package_parent_capability_constraint_loosened",
                "package_parent_provider_constraint_removed",
            },
            codes,
        )


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