feat: preserve governed configuration context
This commit is contained in:
@@ -0,0 +1,104 @@
|
||||
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.routes import _configuration_context
|
||||
from govoplan_core.core.provider_governance import (
|
||||
ExternalProviderRuntimeState,
|
||||
ExternalProviderStateProviderRegistration,
|
||||
)
|
||||
|
||||
|
||||
class ConfigurationPackageContextTests(unittest.TestCase):
|
||||
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()
|
||||
Reference in New Issue
Block a user