generated from coulomb/repo-seed
Implement isolated user-engine MVP
This commit is contained in:
256
tests/test_isolated_mvp.py
Normal file
256
tests/test_isolated_mvp.py
Normal file
@@ -0,0 +1,256 @@
|
||||
import unittest
|
||||
|
||||
from user_engine.adapters.local import (
|
||||
InMemoryUserEngineStore,
|
||||
LocalAuthorizationCheckPort,
|
||||
)
|
||||
from user_engine.domain import (
|
||||
AccountStatus,
|
||||
AttributeDefinition,
|
||||
AuthorizationEffect,
|
||||
Catalog,
|
||||
CatalogLifecycle,
|
||||
Mutability,
|
||||
ProfileScope,
|
||||
ProjectionType,
|
||||
Sensitivity,
|
||||
Visibility,
|
||||
)
|
||||
from user_engine.errors import AuthorizationDenied, ValidationError
|
||||
from user_engine.service import REDACTED, UserEngineService
|
||||
from user_engine.testing.fixtures import (
|
||||
FixtureIdentityClaimsAdapter,
|
||||
human_actor_claims,
|
||||
sample_application,
|
||||
sample_application_binding,
|
||||
)
|
||||
|
||||
|
||||
class IsolatedMvpTests(unittest.TestCase):
|
||||
def test_me_creates_user_account_and_identity_once(self):
|
||||
service, store, _ = _service()
|
||||
|
||||
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
||||
again = service.me(human_actor_claims(), correlation_id="corr-me-again")
|
||||
|
||||
self.assertEqual(service.health().status, "ok")
|
||||
self.assertTrue(service.readiness().ready)
|
||||
self.assertEqual(session.user.user_id, again.user.user_id)
|
||||
self.assertEqual(session.account.status, AccountStatus.ACTIVE)
|
||||
self.assertEqual(len(store.users), 1)
|
||||
self.assertEqual(len(store.identities), 1)
|
||||
self.assertEqual(len(service.outbox_events()), 1)
|
||||
|
||||
def test_account_lifecycle_and_identity_linking(self):
|
||||
service, _, _ = _service()
|
||||
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
||||
|
||||
disabled = service.set_account_status(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
AccountStatus.DISABLED,
|
||||
correlation_id="corr-disable",
|
||||
)
|
||||
linked = service.link_identity(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
issuer="https://issuer.example.test",
|
||||
subject="alternate-subject",
|
||||
provider="fixture",
|
||||
correlation_id="corr-link",
|
||||
)
|
||||
linked_session = service.me(
|
||||
human_actor_claims(subject="alternate-subject"),
|
||||
correlation_id="corr-linked-me",
|
||||
)
|
||||
|
||||
self.assertEqual(disabled.status, AccountStatus.DISABLED)
|
||||
self.assertEqual(linked.user_id, session.user.user_id)
|
||||
self.assertEqual(linked_session.user.user_id, session.user.user_id)
|
||||
self.assertIn(
|
||||
"identity.linked",
|
||||
[event.event_type for event in service.outbox_events()],
|
||||
)
|
||||
|
||||
def test_catalog_profile_effective_profile_and_projection_flow(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_app_and_catalog(service)
|
||||
before_audit = len(service.audit_records())
|
||||
before_outbox = len(service.outbox_events())
|
||||
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"compact",
|
||||
correlation_id="corr-global-density",
|
||||
)
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"comfortable",
|
||||
scope=ProfileScope.APPLICATION,
|
||||
scope_id="app.demo",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-app-density",
|
||||
)
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.recovery_hint",
|
||||
"first keyboard",
|
||||
correlation_id="corr-sensitive",
|
||||
)
|
||||
|
||||
effective = service.effective_profile(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-effective",
|
||||
)
|
||||
runtime = service.projection(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
ProjectionType.APPLICATION_RUNTIME,
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-runtime",
|
||||
)
|
||||
|
||||
self.assertEqual(effective.values["demo.display_density"], "comfortable")
|
||||
self.assertEqual(
|
||||
effective.source_layers["demo.display_density"],
|
||||
"application:app.demo",
|
||||
)
|
||||
self.assertEqual(runtime.values["demo.recovery_hint"], REDACTED)
|
||||
self.assertEqual(runtime.redactions["demo.recovery_hint"], "sensitivity")
|
||||
self.assertEqual(len(service.audit_records()), before_audit + 3)
|
||||
self.assertEqual(len(service.outbox_events()), before_outbox + 3)
|
||||
|
||||
def test_catalog_validation_rejects_duplicate_keys(self):
|
||||
service, _, _ = _service()
|
||||
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
||||
service.register_application(
|
||||
session.actor,
|
||||
sample_application(),
|
||||
binding=sample_application_binding(),
|
||||
correlation_id="corr-app",
|
||||
)
|
||||
attribute = _display_density_attribute()
|
||||
catalog = Catalog(
|
||||
catalog_id="demo-duplicate",
|
||||
namespace="demo",
|
||||
version="0.1.0",
|
||||
owning_application_id="app.demo",
|
||||
lifecycle=CatalogLifecycle.ACTIVE,
|
||||
attributes=(attribute, attribute),
|
||||
)
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
service.publish_catalog(session.actor, catalog, correlation_id="corr-cat")
|
||||
|
||||
def test_profile_value_validation_rejects_invalid_enum(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap_app_and_catalog(service)
|
||||
before_outbox = len(service.outbox_events())
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"spacious",
|
||||
correlation_id="corr-invalid-density",
|
||||
)
|
||||
|
||||
self.assertEqual(len(service.outbox_events()), before_outbox)
|
||||
|
||||
def test_authorization_denial_does_not_mutate_profile_or_outbox(self):
|
||||
service, store, _ = _service(
|
||||
action_effects={"profile.write": AuthorizationEffect.DENY}
|
||||
)
|
||||
session = _bootstrap_app_and_catalog(service)
|
||||
before_outbox = len(service.outbox_events())
|
||||
before_audit = len(service.audit_records())
|
||||
|
||||
with self.assertRaises(AuthorizationDenied):
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"compact",
|
||||
correlation_id="corr-denied-profile",
|
||||
)
|
||||
|
||||
self.assertEqual(store.profile_values, {})
|
||||
self.assertEqual(len(service.outbox_events()), before_outbox)
|
||||
self.assertEqual(len(service.audit_records()), before_audit + 1)
|
||||
self.assertEqual(service.audit_records()[-1].summary, "authorization denied")
|
||||
|
||||
|
||||
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_app_and_catalog(service: UserEngineService):
|
||||
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
||||
service.register_application(
|
||||
session.actor,
|
||||
sample_application(),
|
||||
binding=sample_application_binding(),
|
||||
correlation_id="corr-app",
|
||||
)
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
_mvp_catalog(),
|
||||
correlation_id="corr-catalog",
|
||||
)
|
||||
return session
|
||||
|
||||
|
||||
def _mvp_catalog() -> Catalog:
|
||||
return Catalog(
|
||||
catalog_id="demo-profile",
|
||||
namespace="demo",
|
||||
version="0.1.0",
|
||||
owning_application_id="app.demo",
|
||||
lifecycle=CatalogLifecycle.ACTIVE,
|
||||
attributes=(
|
||||
_display_density_attribute(),
|
||||
AttributeDefinition(
|
||||
key="demo.recovery_hint",
|
||||
value_type="string",
|
||||
scope=ProfileScope.GLOBAL,
|
||||
sensitivity=Sensitivity.SENSITIVE,
|
||||
visibility=(Visibility.USER, Visibility.APPLICATION, Visibility.ADMIN),
|
||||
mutability=(Mutability.USER, Mutability.ADMIN),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _display_density_attribute() -> AttributeDefinition:
|
||||
return AttributeDefinition(
|
||||
key="demo.display_density",
|
||||
value_type="string",
|
||||
scope=ProfileScope.APPLICATION,
|
||||
sensitivity=Sensitivity.INTERNAL,
|
||||
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