feat: pass deployment receipts to configuration providers
This commit is contained in:
@@ -152,6 +152,10 @@ from govoplan_core.core.configuration_packages import (
|
||||
export_configuration_package,
|
||||
validate_configuration_package_catalog,
|
||||
)
|
||||
from govoplan_core.core.infrastructure_capabilities import (
|
||||
InfrastructureCapabilityReceiptError,
|
||||
load_infrastructure_capability_receipt,
|
||||
)
|
||||
from govoplan_core.core.configuration_control import (
|
||||
CONFIGURATION_CHANGE_RECORD_RESOURCE,
|
||||
CONFIGURATION_CHANGE_REQUEST_RESOURCE,
|
||||
@@ -1023,6 +1027,12 @@ def _configuration_context(
|
||||
capabilities = {CONFIGURATION_PROVIDER_CAPABILITY, ACCESS_CONFIGURATION_CAPABILITY}
|
||||
external_provider_declarations: dict[str, dict[str, object]] = {}
|
||||
external_provider_states: dict[str, dict[str, object]] = {}
|
||||
infrastructure_receipt = None
|
||||
infrastructure_receipt_error = None
|
||||
try:
|
||||
infrastructure_receipt = load_infrastructure_capability_receipt()
|
||||
except InfrastructureCapabilityReceiptError as exc:
|
||||
infrastructure_receipt_error = str(exc)
|
||||
if registry is not None and hasattr(registry, "manifests"):
|
||||
manifests = registry.manifests()
|
||||
installed_modules = {manifest.id: manifest.version for manifest in manifests}
|
||||
@@ -1048,11 +1058,14 @@ def _configuration_context(
|
||||
return ConfigurationPreflightContext(
|
||||
tenant_id=tenant_id or principal.tenant_id,
|
||||
operator_user_id=principal.user.id,
|
||||
operator_scopes=frozenset(getattr(principal, "scopes", ())),
|
||||
supplied_data=supplied_data or {},
|
||||
installed_modules=installed_modules,
|
||||
capabilities=frozenset(capabilities),
|
||||
external_provider_declarations=external_provider_declarations,
|
||||
external_provider_states=external_provider_states,
|
||||
infrastructure_receipt=infrastructure_receipt,
|
||||
infrastructure_receipt_error=infrastructure_receipt_error,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -6,6 +6,9 @@ import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
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,
|
||||
@@ -13,6 +16,54 @@ from govoplan_core.core.provider_governance import (
|
||||
|
||||
|
||||
class ConfigurationPackageContextTests(unittest.TestCase):
|
||||
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",
|
||||
|
||||
Reference in New Issue
Block a user