govoplan-files/tests/test_configuration_provider.py

from __future__ import annotations

from types import SimpleNamespace
import unittest

from govoplan_core.core.configuration_packages import (
    ConfigurationPackageFragment,
    ConfigurationPreflightContext,
)
from govoplan_core.core.infrastructure_capabilities import (
    infrastructure_capability_receipt_from_mapping,
)
from govoplan_files.backend.configuration_provider import (
    FILES_CONFIGURATION_CAPABILITY,
    FilesConfigurationProvider,
)
from govoplan_files.backend.manifest import manifest


def _receipt(*, source: str = "host-local", state: str = "configured"):
    endpoint = (
        {"kind": "filesystem", "reference": "volume:files-data"}
        if source == "host-local"
        else {"scheme": "http", "host": "garage", "port": 3900}
    )
    secret_refs = (
        []
        if source == "host-local"
        else [
            "env:FILE_STORAGE_S3_ACCESS_KEY_ID",
            "env:FILE_STORAGE_S3_SECRET_ACCESS_KEY",
            "env:GARAGE_RPC_SECRET",
        ]
    )
    return infrastructure_capability_receipt_from_mapping(
        {
            "schema_version": 1,
            "installation_id": "files-provider-test",
            "profile": "evaluation",
            "capabilities": [
                {
                    "id": "files.storage",
                    "label": "Managed file content storage",
                    "state": state,
                    "source": source,
                    "detail": "Deployment-owned storage binding.",
                    "endpoint": endpoint,
                    "secret_refs": secret_refs,
                    "dependent_modules": ["files"],
                }
            ],
            "post_install_tasks": [],
        }
    )


def _local_settings(**overrides):
    values = {
        "file_storage_backend": "local",
        "file_storage_local_root": "/var/lib/govoplan/files",
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def _s3_settings(**overrides):
    values = {
        "file_storage_backend": "s3",
        "file_storage_s3_endpoint_url": "http://garage:3900",
        "file_storage_s3_bucket": "files",
        "file_storage_s3_deployment_managed": True,
        "file_storage_s3_endpoint_trusted": False,
    }
    values.update(overrides)
    return SimpleNamespace(**values)


class FilesConfigurationProviderTests(unittest.TestCase):
    def test_provider_is_registered(self) -> None:
        self.assertIn(FILES_CONFIGURATION_CAPABILITY, manifest.capability_factories)

    def test_matching_local_storage_is_an_idempotent_noop(self) -> None:
        provider = FilesConfigurationProvider(
            settings=_local_settings(),
            environment={},
        )
        context = ConfigurationPreflightContext(
            infrastructure_receipt=_receipt(),
        )
        fragment = ConfigurationPackageFragment(
            module_id="files",
            fragment_type="managed_storage",
            fragment_id="files-storage",
            payload={
                "expected_backend": "local",
                "expected_source": "host-local",
            },
        )

        first = provider.preflight(fragment, context)
        applied = provider.apply(fragment, {}, context)
        second = provider.preflight(fragment, context)

        self.assertEqual("skip", first.plan[0].action)
        self.assertEqual("skip", second.plan[0].action)
        self.assertEqual((), applied.diagnostics)
        self.assertEqual({}, applied.created_refs)
        self.assertEqual({}, applied.updated_refs)

    def test_runtime_backend_mismatch_blocks_without_rewriting_settings(self) -> None:
        settings = _local_settings()
        provider = FilesConfigurationProvider(settings=settings, environment={})
        context = ConfigurationPreflightContext(
            infrastructure_receipt=_receipt(source="installer-managed-garage"),
        )
        fragment = ConfigurationPackageFragment(
            module_id="files",
            fragment_type="managed_storage",
            payload={},
        )

        result = provider.preflight(fragment, context)

        self.assertEqual("blocked", result.plan[0].action)
        self.assertIn(
            "files_storage_runtime_mismatch",
            {item.code for item in result.diagnostics},
        )
        self.assertEqual("local", settings.file_storage_backend)

    def test_garage_binding_requires_only_files_secret_references(self) -> None:
        fragment = ConfigurationPackageFragment(
            module_id="files",
            fragment_type="managed_storage",
            payload={},
        )
        context = ConfigurationPreflightContext(
            infrastructure_receipt=_receipt(source="installer-managed-garage"),
        )
        missing = FilesConfigurationProvider(
            settings=_s3_settings(),
            environment={},
        ).preflight(fragment, context)
        available = FilesConfigurationProvider(
            settings=_s3_settings(),
            environment={
                "FILE_STORAGE_S3_ACCESS_KEY_ID": "reference-resolved",
                "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "reference-resolved",
            },
        ).preflight(fragment, context)

        self.assertEqual("blocked", missing.plan[0].action)
        self.assertEqual(
            2,
            sum(
                item.code == "files_storage_secret_reference_unresolved"
                for item in missing.diagnostics
            ),
        )
        self.assertEqual("skip", available.plan[0].action)

    def test_operator_supplied_s3_requires_explicit_endpoint_trust(self) -> None:
        receipt = _receipt(source="operator-supplied-s3", state="externally_supplied")
        fragment = ConfigurationPackageFragment(
            module_id="files",
            fragment_type="managed_storage",
            payload={},
        )
        context = ConfigurationPreflightContext(infrastructure_receipt=receipt)
        environment = {
            "FILE_STORAGE_S3_ACCESS_KEY_ID": "reference-resolved",
            "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "reference-resolved",
        }

        blocked = FilesConfigurationProvider(
            settings=_s3_settings(
                file_storage_s3_deployment_managed=False,
                file_storage_s3_endpoint_trusted=False,
            ),
            environment=environment,
        ).preflight(fragment, context)
        ready = FilesConfigurationProvider(
            settings=_s3_settings(
                file_storage_s3_deployment_managed=False,
                file_storage_s3_endpoint_trusted=True,
            ),
            environment=environment,
        ).preflight(fragment, context)

        self.assertEqual("blocked", blocked.plan[0].action)
        self.assertIn(
            "files_storage_trust_boundary_missing",
            {item.code for item in blocked.diagnostics},
        )
        self.assertEqual("skip", ready.plan[0].action)

    def test_inline_storage_secret_is_rejected(self) -> None:
        provider = FilesConfigurationProvider(
            settings=_local_settings(),
            environment={},
        )
        result = provider.preflight(
            ConfigurationPackageFragment(
                module_id="files",
                fragment_type="managed_storage",
                payload={"secret_access_key": "inline"},
            ),
            ConfigurationPreflightContext(infrastructure_receipt=_receipt()),
        )

        self.assertEqual("blocked", result.plan[0].action)
        self.assertIn(
            "files_configuration_secret_forbidden",
            {item.code for item in result.diagnostics},
        )


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