feat: validate receipt-bound storage configuration
This commit is contained in:
@@ -0,0 +1,219 @@
|
||||
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()
|
||||
@@ -4,6 +4,7 @@ import unittest
|
||||
|
||||
|
||||
STATIC_TOPIC_IDS = {
|
||||
"files.configuration-package.managed-storage",
|
||||
"files.quick-access-and-product-area",
|
||||
"files.search.managed-content",
|
||||
"files.workflow.organize-managed-files",
|
||||
|
||||
Reference in New Issue
Block a user