generated from coulomb/repo-seed
Add multi-application catalog support
This commit is contained in:
312
tests/test_multi_application_catalogs.py
Normal file
312
tests/test_multi_application_catalogs.py
Normal file
@@ -0,0 +1,312 @@
|
||||
import unittest
|
||||
|
||||
from user_engine.adapters.local import (
|
||||
InMemoryUserEngineStore,
|
||||
LocalAuthorizationCheckPort,
|
||||
)
|
||||
from user_engine.domain import (
|
||||
Application,
|
||||
ApplicationBinding,
|
||||
AttributeDefinition,
|
||||
AuthorizationEffect,
|
||||
Catalog,
|
||||
CatalogLifecycle,
|
||||
Mutability,
|
||||
ProfileScope,
|
||||
ProjectionType,
|
||||
Sensitivity,
|
||||
Visibility,
|
||||
)
|
||||
from user_engine.errors import AuthorizationDenied, ConflictError, ValidationError
|
||||
from user_engine.projections import ClaimsEnrichmentProjectionCache
|
||||
from user_engine.service import UserEngineService
|
||||
from user_engine.testing.fixtures import (
|
||||
FixtureIdentityClaimsAdapter,
|
||||
human_actor_claims,
|
||||
sample_application,
|
||||
sample_application_binding,
|
||||
sample_catalog,
|
||||
)
|
||||
|
||||
|
||||
class MultiApplicationCatalogTests(unittest.TestCase):
|
||||
def test_application_runtime_projection_is_filtered_by_catalog_owner(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_two_apps(service)
|
||||
|
||||
demo_projection = service.projection(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
ProjectionType.APPLICATION_RUNTIME,
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-demo-projection",
|
||||
)
|
||||
reporting_projection = service.projection(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
ProjectionType.APPLICATION_RUNTIME,
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.reporting",
|
||||
correlation_id="corr-reporting-projection",
|
||||
)
|
||||
|
||||
self.assertIn("demo.display_density", demo_projection.values)
|
||||
self.assertNotIn("report.timezone", demo_projection.values)
|
||||
self.assertIn("report.timezone", reporting_projection.values)
|
||||
self.assertNotIn("demo.display_density", reporting_projection.values)
|
||||
|
||||
def test_catalog_namespace_collision_is_rejected(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_demo_app(service)
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
sample_catalog(),
|
||||
correlation_id="corr-demo-catalog",
|
||||
)
|
||||
service.register_application(
|
||||
session.actor,
|
||||
Application(
|
||||
application_id="app.other",
|
||||
display_name="Other",
|
||||
owner="team:other",
|
||||
allowed_profile_scopes=(ProfileScope.GLOBAL,),
|
||||
allowed_projection_types=(ProjectionType.APPLICATION_RUNTIME,),
|
||||
),
|
||||
binding=ApplicationBinding(
|
||||
application_id="app.other",
|
||||
catalog_namespaces=("demo",),
|
||||
oidc_client_id="other-client",
|
||||
),
|
||||
correlation_id="corr-other-app",
|
||||
)
|
||||
|
||||
with self.assertRaises(ConflictError):
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
Catalog(
|
||||
catalog_id="other-demo",
|
||||
namespace="demo",
|
||||
version="0.1.0",
|
||||
owning_application_id="app.other",
|
||||
lifecycle=CatalogLifecycle.ACTIVE,
|
||||
attributes=(
|
||||
AttributeDefinition(
|
||||
key="demo.other",
|
||||
value_type="string",
|
||||
scope=ProfileScope.GLOBAL,
|
||||
sensitivity=Sensitivity.INTERNAL,
|
||||
visibility=(Visibility.APPLICATION,),
|
||||
mutability=(Mutability.ADMIN,),
|
||||
),
|
||||
),
|
||||
),
|
||||
correlation_id="corr-conflict",
|
||||
)
|
||||
|
||||
def test_catalog_migration_rejects_backwards_version_and_sensitivity_downgrade(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_demo_app(service)
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
sample_catalog(),
|
||||
correlation_id="corr-demo-catalog",
|
||||
)
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
_demo_catalog(version="0.0.9", sensitivity=Sensitivity.INTERNAL),
|
||||
correlation_id="corr-backwards",
|
||||
)
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
_demo_catalog(version="0.2.0", sensitivity=Sensitivity.PUBLIC),
|
||||
correlation_id="corr-downgrade",
|
||||
)
|
||||
|
||||
def test_draft_catalog_and_projection_without_application_are_rejected(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_demo_app(service)
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
_demo_catalog(lifecycle=CatalogLifecycle.DRAFT),
|
||||
correlation_id="corr-draft",
|
||||
)
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
service.projection(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
ProjectionType.APPLICATION_RUNTIME,
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-missing-app",
|
||||
)
|
||||
|
||||
def test_projection_denial_does_not_emit_outbox_event(self):
|
||||
service, _, _ = _service(action_effects={"projection.read": AuthorizationEffect.DENY})
|
||||
session = _bootstrap_two_apps(service)
|
||||
before_outbox = len(service.outbox_events())
|
||||
|
||||
with self.assertRaises(AuthorizationDenied):
|
||||
service.projection(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
ProjectionType.APPLICATION_RUNTIME,
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-denied-projection",
|
||||
)
|
||||
|
||||
self.assertEqual(len(service.outbox_events()), before_outbox)
|
||||
|
||||
def test_claims_enrichment_cache_reuses_projection_without_issuing_tokens(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_two_apps(service)
|
||||
cache = ClaimsEnrichmentProjectionCache()
|
||||
|
||||
first = cache.get(
|
||||
service,
|
||||
session.actor,
|
||||
user_id=session.user.user_id,
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-claims-1",
|
||||
)
|
||||
second = cache.get(
|
||||
service,
|
||||
session.actor,
|
||||
user_id=session.user.user_id,
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-claims-2",
|
||||
)
|
||||
|
||||
self.assertIs(first, second)
|
||||
self.assertEqual(first.projection_type, ProjectionType.CLAIMS_ENRICHMENT)
|
||||
self.assertFalse(hasattr(cache, "issue_token"))
|
||||
|
||||
|
||||
def _service(
|
||||
*,
|
||||
action_effects: dict[str, AuthorizationEffect] | None = None,
|
||||
):
|
||||
store = InMemoryUserEngineStore()
|
||||
authz = LocalAuthorizationCheckPort(action_effects=action_effects)
|
||||
service = UserEngineService(
|
||||
store=store,
|
||||
identity_adapter=FixtureIdentityClaimsAdapter(),
|
||||
authorization=authz,
|
||||
)
|
||||
return service, store, authz
|
||||
|
||||
|
||||
def _bootstrap_demo_app(service: UserEngineService):
|
||||
claims = human_actor_claims(subject="app-admin", tenant="tenant:coulomb")
|
||||
claims["roles"] = ["tenant-admin"]
|
||||
session = service.me(claims, correlation_id="corr-me")
|
||||
service.register_application(
|
||||
session.actor,
|
||||
sample_application(),
|
||||
binding=sample_application_binding(),
|
||||
correlation_id="corr-demo-app",
|
||||
)
|
||||
return session
|
||||
|
||||
|
||||
def _bootstrap_two_apps(service: UserEngineService):
|
||||
session = _bootstrap_demo_app(service)
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
sample_catalog(),
|
||||
correlation_id="corr-demo-catalog",
|
||||
)
|
||||
service.register_application(
|
||||
session.actor,
|
||||
_reporting_application(),
|
||||
binding=_reporting_binding(),
|
||||
correlation_id="corr-reporting-app",
|
||||
)
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
_reporting_catalog(),
|
||||
correlation_id="corr-reporting-catalog",
|
||||
)
|
||||
return session
|
||||
|
||||
|
||||
def _reporting_application() -> Application:
|
||||
return Application(
|
||||
application_id="app.reporting",
|
||||
display_name="Reporting",
|
||||
owner="team:reporting",
|
||||
allowed_profile_scopes=(ProfileScope.GLOBAL, ProfileScope.APPLICATION),
|
||||
allowed_projection_types=(ProjectionType.APPLICATION_RUNTIME,),
|
||||
)
|
||||
|
||||
|
||||
def _reporting_binding() -> ApplicationBinding:
|
||||
return ApplicationBinding(
|
||||
application_id="app.reporting",
|
||||
oidc_client_id="reporting-client",
|
||||
protected_system_id="reporting.demo",
|
||||
catalog_namespaces=("report",),
|
||||
event_source="reporting.demo",
|
||||
deployment_ref="local",
|
||||
)
|
||||
|
||||
|
||||
def _reporting_catalog() -> Catalog:
|
||||
return Catalog(
|
||||
catalog_id="report-profile",
|
||||
namespace="report",
|
||||
version="0.1.0",
|
||||
owning_application_id="app.reporting",
|
||||
lifecycle=CatalogLifecycle.ACTIVE,
|
||||
attributes=(
|
||||
AttributeDefinition(
|
||||
key="report.timezone",
|
||||
value_type="string",
|
||||
scope=ProfileScope.APPLICATION,
|
||||
sensitivity=Sensitivity.INTERNAL,
|
||||
visibility=(Visibility.APPLICATION,),
|
||||
mutability=(Mutability.USER,),
|
||||
default="UTC",
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _demo_catalog(
|
||||
*,
|
||||
version: str = "0.1.0",
|
||||
sensitivity: Sensitivity = Sensitivity.INTERNAL,
|
||||
lifecycle: CatalogLifecycle = CatalogLifecycle.ACTIVE,
|
||||
) -> Catalog:
|
||||
return Catalog(
|
||||
catalog_id="demo-profile",
|
||||
namespace="demo",
|
||||
version=version,
|
||||
owning_application_id="app.demo",
|
||||
lifecycle=lifecycle,
|
||||
attributes=(
|
||||
AttributeDefinition(
|
||||
key="demo.display_density",
|
||||
value_type="string",
|
||||
scope=ProfileScope.APPLICATION,
|
||||
sensitivity=sensitivity,
|
||||
visibility=(Visibility.USER, Visibility.APPLICATION),
|
||||
mutability=(Mutability.USER,),
|
||||
default="comfortable",
|
||||
validation={"enum": ["compact", "comfortable"]},
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user