generated from coulomb/repo-seed
362 lines
13 KiB
Python
362 lines
13 KiB
Python
import unittest
|
|
|
|
from user_engine.adapters.local import InMemoryUserEngineStore, LocalAuthorizationCheckPort
|
|
from user_engine.domain import (
|
|
FactorVerification,
|
|
IdentityFactorType,
|
|
OnboardingJourneyStatus,
|
|
OnboardingStepStatus,
|
|
OnboardingTriggerType,
|
|
PreparedEntitlement,
|
|
PreparedEntitlementKind,
|
|
PreparedFactorRequirement,
|
|
WelcomeProtocol,
|
|
WelcomeProtocolStep,
|
|
)
|
|
from user_engine.service import UserEngineService
|
|
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
|
|
|
|
|
|
class OnboardingJourneyTests(unittest.TestCase):
|
|
def test_registration_completion_starts_matching_welcome_protocol(self):
|
|
service, store = _service()
|
|
actor = _actor()
|
|
protocol = service.register_welcome_protocol(
|
|
actor,
|
|
_registration_protocol(),
|
|
correlation_id="corr-register-protocol",
|
|
)
|
|
|
|
completion = _complete_registration(service, actor)
|
|
|
|
journeys = store.onboarding_journeys_for_user(completion.user.user_id)
|
|
context = service.identity_context(
|
|
actor,
|
|
user_id=completion.user.user_id,
|
|
tenant="tenant:coulomb",
|
|
correlation_id="corr-context",
|
|
)
|
|
|
|
self.assertEqual(len(journeys), 1)
|
|
self.assertEqual(journeys[0].protocol_id, protocol.protocol_id)
|
|
self.assertEqual(journeys[0].trigger_type, OnboardingTriggerType.REGISTRATION)
|
|
self.assertEqual(journeys[0].status, OnboardingJourneyStatus.IN_PROGRESS)
|
|
self.assertEqual(journeys[0].steps[0].status, OnboardingStepStatus.IN_PROGRESS)
|
|
self.assertEqual(context.onboarding_journeys[0].journey_id, journeys[0].journey_id)
|
|
self.assertIn(
|
|
"onboarding_journey.started",
|
|
[event.event_type for event in service.outbox_events()],
|
|
)
|
|
self.assertNotIn(
|
|
"sample.user@example.test",
|
|
repr([event.payload for event in service.outbox_events()]),
|
|
)
|
|
|
|
def test_prepared_account_claim_starts_prepared_welcome_protocol(self):
|
|
service, store = _service()
|
|
actor = _actor()
|
|
service.register_welcome_protocol(
|
|
actor,
|
|
_prepared_protocol(),
|
|
correlation_id="corr-prepared-protocol",
|
|
)
|
|
prepared = service.prepare_account(
|
|
actor,
|
|
tenant="tenant:coulomb",
|
|
required_factor_matches=(
|
|
PreparedFactorRequirement(
|
|
factor_type=IdentityFactorType.EMAIL,
|
|
normalized_value="sample.user@example.test",
|
|
),
|
|
),
|
|
entitlements=(
|
|
PreparedEntitlement(
|
|
kind=PreparedEntitlementKind.ONBOARDING_JOURNEY,
|
|
tenant="tenant:coulomb",
|
|
onboarding_journey="welcome-demo",
|
|
),
|
|
),
|
|
correlation_id="corr-prepare",
|
|
)
|
|
completion = _complete_registration(service, actor)
|
|
|
|
service.claim_prepared_account(
|
|
actor,
|
|
completion.session.registration_id,
|
|
prepared_account_id=prepared.prepared_account_id,
|
|
correlation_id="corr-claim",
|
|
)
|
|
|
|
journeys = store.onboarding_journeys_for_user(completion.user.user_id)
|
|
self.assertEqual(len(journeys), 1)
|
|
self.assertEqual(journeys[0].trigger_type, OnboardingTriggerType.PREPARED_ACCOUNT)
|
|
self.assertEqual(journeys[0].source_id, prepared.prepared_account_id)
|
|
self.assertEqual(journeys[0].journey_key, "welcome-demo")
|
|
self.assertIn(
|
|
"prepared_account.onboarding_requested",
|
|
[event.event_type for event in service.outbox_events()],
|
|
)
|
|
self.assertIn(
|
|
"onboarding_journey.started",
|
|
[event.event_type for event in service.outbox_events()],
|
|
)
|
|
|
|
def test_missing_subsystem_callback_blocks_journey_with_gap(self):
|
|
service, store = _service()
|
|
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
|
protocol = service.register_welcome_protocol(
|
|
session.actor,
|
|
WelcomeProtocol(
|
|
tenant="tenant:coulomb",
|
|
name="Blocked Welcome",
|
|
trigger_type=OnboardingTriggerType.MANUAL,
|
|
steps=(
|
|
WelcomeProtocolStep(
|
|
step_key="external-setup",
|
|
title="External Setup",
|
|
subsystem="ops-console",
|
|
requires_subsystem_callback=True,
|
|
),
|
|
),
|
|
),
|
|
correlation_id="corr-blocked-protocol",
|
|
)
|
|
|
|
start = service.start_onboarding_journey(
|
|
session.actor,
|
|
session.user.user_id,
|
|
protocol.protocol_id,
|
|
correlation_id="corr-start-blocked",
|
|
)
|
|
diagnostics = service.onboarding_diagnostics(
|
|
session.actor,
|
|
tenant="tenant:coulomb",
|
|
correlation_id="corr-diagnostics",
|
|
)
|
|
|
|
self.assertEqual(start.journey.status, OnboardingJourneyStatus.BLOCKED)
|
|
self.assertEqual(start.journey.steps[0].status, OnboardingStepStatus.BLOCKED)
|
|
self.assertIn("subsystem-callback-missing", start.journey.steps[0].lifecycle_gap)
|
|
self.assertEqual(store.record_counts()["onboarding_journeys"], 1)
|
|
self.assertEqual(diagnostics.statuses[OnboardingJourneyStatus.BLOCKED.value], 1)
|
|
self.assertTrue(diagnostics.lifecycle_gaps)
|
|
|
|
def test_blocked_journey_can_resume_and_complete(self):
|
|
service, _ = _service()
|
|
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
|
protocol = service.register_welcome_protocol(
|
|
session.actor,
|
|
WelcomeProtocol(
|
|
tenant="tenant:coulomb",
|
|
name="Resume Welcome",
|
|
trigger_type=OnboardingTriggerType.MANUAL,
|
|
steps=(
|
|
WelcomeProtocolStep(
|
|
step_key="callback",
|
|
title="Callback",
|
|
subsystem="crm",
|
|
requires_subsystem_callback=True,
|
|
),
|
|
),
|
|
),
|
|
correlation_id="corr-resume-protocol",
|
|
)
|
|
blocked = service.start_onboarding_journey(
|
|
session.actor,
|
|
session.user.user_id,
|
|
protocol.protocol_id,
|
|
correlation_id="corr-start",
|
|
).journey
|
|
|
|
resumed = service.resume_onboarding_journey(
|
|
session.actor,
|
|
blocked.journey_id,
|
|
callback_refs={"callback": "crm://welcome/callback"},
|
|
correlation_id="corr-resume",
|
|
)
|
|
completed = service.complete_onboarding_step(
|
|
session.actor,
|
|
resumed.journey_id,
|
|
"callback",
|
|
correlation_id="corr-complete-step",
|
|
)
|
|
|
|
self.assertEqual(resumed.status, OnboardingJourneyStatus.IN_PROGRESS)
|
|
self.assertEqual(
|
|
resumed.steps[0].handoff.callback_ref,
|
|
"crm://welcome/callback",
|
|
)
|
|
self.assertIsNone(resumed.steps[0].lifecycle_gap)
|
|
self.assertEqual(completed.status, OnboardingJourneyStatus.COMPLETED)
|
|
self.assertEqual(completed.completed_at is not None, True)
|
|
|
|
def test_progress_skip_and_fail_operations_are_auditable(self):
|
|
service, _ = _service()
|
|
session = service.me(human_actor_claims(), correlation_id="corr-me")
|
|
protocol = service.register_welcome_protocol(
|
|
session.actor,
|
|
WelcomeProtocol(
|
|
tenant="tenant:coulomb",
|
|
name="Two Step Welcome",
|
|
trigger_type=OnboardingTriggerType.MANUAL,
|
|
steps=(
|
|
WelcomeProtocolStep(
|
|
step_key="intro",
|
|
title="Intro",
|
|
subsystem="portal",
|
|
),
|
|
WelcomeProtocolStep(
|
|
step_key="tour",
|
|
title="Tour",
|
|
subsystem="portal",
|
|
),
|
|
),
|
|
),
|
|
correlation_id="corr-two-step-protocol",
|
|
)
|
|
journey = service.start_onboarding_journey(
|
|
session.actor,
|
|
session.user.user_id,
|
|
protocol.protocol_id,
|
|
correlation_id="corr-start-two-step",
|
|
).journey
|
|
|
|
progressed = service.progress_onboarding_step(
|
|
session.actor,
|
|
journey.journey_id,
|
|
"intro",
|
|
correlation_id="corr-progress",
|
|
)
|
|
second_active = service.complete_onboarding_step(
|
|
session.actor,
|
|
progressed.journey_id,
|
|
"intro",
|
|
correlation_id="corr-complete-intro",
|
|
)
|
|
skipped = service.skip_onboarding_step(
|
|
session.actor,
|
|
second_active.journey_id,
|
|
"tour",
|
|
correlation_id="corr-skip-tour",
|
|
)
|
|
|
|
self.assertEqual(skipped.status, OnboardingJourneyStatus.COMPLETED)
|
|
self.assertIn(
|
|
"onboarding_step.skipped",
|
|
[event.event_type for event in service.outbox_events()],
|
|
)
|
|
|
|
failed_protocol = service.register_welcome_protocol(
|
|
session.actor,
|
|
WelcomeProtocol(
|
|
tenant="tenant:coulomb",
|
|
name="Failing Welcome",
|
|
trigger_type=OnboardingTriggerType.MANUAL,
|
|
steps=(
|
|
WelcomeProtocolStep(
|
|
step_key="danger",
|
|
title="Danger",
|
|
subsystem="portal",
|
|
),
|
|
),
|
|
),
|
|
correlation_id="corr-fail-protocol",
|
|
)
|
|
failed_start = service.start_onboarding_journey(
|
|
session.actor,
|
|
session.user.user_id,
|
|
failed_protocol.protocol_id,
|
|
correlation_id="corr-start-fail",
|
|
).journey
|
|
failed = service.fail_onboarding_step(
|
|
session.actor,
|
|
failed_start.journey_id,
|
|
"danger",
|
|
lifecycle_gap="portal-unavailable",
|
|
correlation_id="corr-fail-step",
|
|
)
|
|
|
|
self.assertEqual(failed.status, OnboardingJourneyStatus.FAILED)
|
|
self.assertEqual(failed.steps[0].lifecycle_gap, "portal-unavailable")
|
|
self.assertIn(
|
|
"onboarding_step.failed",
|
|
[event.event_type for event in service.outbox_events()],
|
|
)
|
|
|
|
|
|
def _service():
|
|
store = InMemoryUserEngineStore()
|
|
service = UserEngineService(
|
|
store=store,
|
|
identity_adapter=FixtureIdentityClaimsAdapter(),
|
|
authorization=LocalAuthorizationCheckPort(),
|
|
)
|
|
return service, store
|
|
|
|
|
|
def _actor():
|
|
return FixtureIdentityClaimsAdapter().normalize(
|
|
human_actor_claims(subject="sample-user", tenant="tenant:coulomb")
|
|
)
|
|
|
|
|
|
def _complete_registration(service: UserEngineService, actor):
|
|
session = service.start_registration(actor, correlation_id="corr-start")
|
|
service.attach_registration_factor(
|
|
actor,
|
|
session.registration_id,
|
|
FactorVerification(
|
|
factor_type=IdentityFactorType.EMAIL,
|
|
normalized_value="sample.user@example.test",
|
|
display_value="sample.user@example.test",
|
|
source_system="fixture-email",
|
|
),
|
|
correlation_id="corr-factor",
|
|
)
|
|
return service.complete_registration(
|
|
actor,
|
|
session.registration_id,
|
|
correlation_id="corr-complete",
|
|
)
|
|
|
|
|
|
def _registration_protocol() -> WelcomeProtocol:
|
|
return WelcomeProtocol(
|
|
tenant="tenant:coulomb",
|
|
name="Registration Welcome",
|
|
trigger_type=OnboardingTriggerType.REGISTRATION,
|
|
required_factor_types=(IdentityFactorType.EMAIL,),
|
|
steps=(
|
|
WelcomeProtocolStep(
|
|
step_key="intro",
|
|
title="Intro",
|
|
subsystem="portal",
|
|
callback_ref="portal://welcome/intro",
|
|
requires_subsystem_callback=True,
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _prepared_protocol() -> WelcomeProtocol:
|
|
return WelcomeProtocol(
|
|
tenant="tenant:coulomb",
|
|
name="Prepared Welcome",
|
|
trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT,
|
|
journey_key="welcome-demo",
|
|
prepared_journey="welcome-demo",
|
|
steps=(
|
|
WelcomeProtocolStep(
|
|
step_key="prepared-intro",
|
|
title="Prepared Intro",
|
|
subsystem="portal",
|
|
callback_ref="portal://welcome/prepared",
|
|
requires_subsystem_callback=True,
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|