coulomb/user-engine
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement role-based account journeys with database and browser acceptance suites
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details
Build and Publish Container Image / build-and-push (push) Successful in 20s
Details
Account journey acceptance / journeys (push) Failing after 0s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
This commit is contained in:
tegwick 2026-09-13 12:20:02 +02:00
parent 75750c0036
commit 1127f852dd
24 changed files with 1554 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

29
tests/test_web.py
View file

@ -88,6 +88,7 @@ class PortalApplicationTests(unittest.TestCase):
login_url="https://kc.example/login",
)
self.claims = human_actor_claims(tenant="tenant:friendly:binky")
self.claims["roles"] = ["tenant-admin"]
def platform_claims(self):
claims = human_actor_claims(subject="platform-operator", tenant="platform:root")
@ -521,7 +522,7 @@ class PortalApplicationTests(unittest.TestCase):
body={
"display_name": "Ada Admin",
"primary_email": "ada@example.test",
"role": "tenant-admin",
"role": "user",
},
)
self.assertEqual("201 Created", created["status"])
@ -550,7 +551,7 @@ class PortalApplicationTests(unittest.TestCase):
)
self.assertEqual("200 OK", changed["status"])
self.assertEqual("suspended", json.loads(payload)["status"])
self.assertIn(("suspend", "ada"), self.app.provisioning.actions)
self.assertIn(("tenant_disabled", "ada", "tenant:friendly:binky"), self.app.provisioning.actions)
removed, payload = invoke_with_idempotency(
self.app,
f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}",
@ -558,7 +559,7 @@ class PortalApplicationTests(unittest.TestCase):
)
self.assertEqual("200 OK", removed["status"])
self.assertEqual("removed", json.loads(payload)["status"])
self.assertIn(("deprovision", "ada"), self.app.provisioning.actions)
self.assertIn(("tenant_disabled", "ada", "tenant:friendly:binky"), self.app.provisioning.actions)
def test_invitation_lifecycle_is_versioned_and_replay_safe(self):
created, payload = invoke(
@ -874,7 +875,7 @@ class PortalApplicationTests(unittest.TestCase):
self.assertEqual("200 OK", renamed["status"])
self.assertIn(b"Renamed In Browser", html)
retired, html = invoke(
retired, html = invoke_confirmed(
self.app, f"/platform/tenants/{quoted}", method="POST",
cookie="ue_session=platform", form={
"csrf_token": "platform-csrf", "operation": "retire",
@ -886,7 +887,7 @@ class PortalApplicationTests(unittest.TestCase):
# A retired tenant offers no metadata form, matching the authority.
self.assertNotIn(b"Save metadata", html)
resubmitted, html = invoke(
resubmitted, html = invoke_confirmed(
self.app, f"/platform/tenants/{quoted}", method="POST",
cookie="ue_session=platform", form={
"csrf_token": "platform-csrf", "operation": "retire",
@ -976,7 +977,7 @@ class PortalApplicationTests(unittest.TestCase):
self.assertEqual("200 OK", admin_page["status"])
self.assertIn(b"Lifecycle diagnostics", html)
self.assertIn(b"Recover identity", html)
recovered, _ = invoke(
recovered, _ = invoke_confirmed(
self.app,
f"/admin/tenant:friendly:browser/users/{memberships[0].user_id}/recover",
method="POST", cookie="ue_session=platform",
@ -1150,7 +1151,7 @@ class PortalApplicationTests(unittest.TestCase):
"consent_accepted": "yes",
},
)
self.assertEqual("303 See Other", saved["status"])
self.assertEqual("200 OK", saved["status"])
user = self.app.service.store.user(invitation.user_id)
self.assertEqual("Updated Invitee", user.display_name)
self.assertEqual("invitee@example.test", user.primary_email)
@ -1202,6 +1203,10 @@ class FakeProvisioning:
password_setup_url=self.password_setup_url,
)
def tenant_access(self, *, external_subject, tenant, roles, enabled, idempotency_key, correlation_id):
self.actions.append(("tenant_active" if enabled else "tenant_disabled", external_subject, tenant))
return ProvisioningResult("netkingdom-lldap", external_subject, "tenant_active" if enabled else "tenant_disabled")
def suspend(self, *, external_subject, idempotency_key, correlation_id):
self.actions.append(("suspend", external_subject))
return ProvisioningResult("netkingdom-lldap", external_subject, "suspended")
@ -1340,3 +1345,13 @@ def invoke_with_idempotency(app, path, claims, *, method="POST", body=None):
if __name__ == "__main__":
unittest.main()
def invoke_confirmed(app, path, **kwargs):
from html import unescape
response, body = invoke(app, path, **kwargs)
token = re.search(rb'name="confirm_token" value="([^"]+)"', body)
if token:
kwargs['form'] = dict(kwargs['form'], confirm_token=unescape(token.group(1).decode()))
return invoke(app, path, **kwargs)
return response, body