tegwick
b8506ef2e3
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
79 lines
4.8 KiB
Python
79 lines
4.8 KiB
Python
"""Platform support journeys with synthetic records and provider failures."""
|
|
from dataclasses import replace
|
|
from urllib.parse import urlencode
|
|
from user_engine.domain import AuditRecord, OutboxEvent, utc_now
|
|
from test_journey_roles import JourneyFixture, TENANT, OTHER
|
|
from test_web import invoke
|
|
|
|
class PlatformSupportJourneys(JourneyFixture):
|
|
def seed(self):
|
|
self.app.service.store.append_audit(AuditRecord(audit_id="audit-support", actor=self.actor,
|
|
action="tenant.account.update", subject="person", tenant=TENANT,
|
|
correlation_id="support-123", summary="private-audit-summary"))
|
|
self.app.service.store.append_outbox(OutboxEvent(event_id="support-delivery", event_type="notification.requested",
|
|
aggregate_id="person", tenant=TENANT, correlation_id="support-123",
|
|
payload={"private":"private-payload"}, failed_at=utc_now(), failure_reason="private-provider-error"))
|
|
|
|
def test_operator_correlates_actions_and_delivery_without_raw_content(self):
|
|
self.seed()
|
|
self.app.service.store.append_outbox(OutboxEvent(event_id="unrelated", event_type="private-unrelated",
|
|
aggregate_id="other", tenant=OTHER, correlation_id="different", payload={}))
|
|
response,body=invoke(self.app,"/platform/activity",cookie="ue_session=operator",query="reference=support-123")
|
|
self.assertEqual("200 OK",response["status"])
|
|
for value in [b"tenant.account.update",b"notification.requested",b"Inspect delivery",b"support-delivery",b"Delivery failed"]:
|
|
self.assertIn(value,body)
|
|
for value in [b"private-audit-summary",b"private-payload",b"private-provider-error",b"private-unrelated"]:
|
|
self.assertNotIn(value,body)
|
|
self.assertIn(b"external outcome is unverified",body)
|
|
|
|
def test_activity_denies_nonoperators_and_mutations(self):
|
|
self.seed()
|
|
for who in ["member","admin"]:
|
|
response,body=invoke(self.app,"/platform/activity",cookie="ue_session="+who)
|
|
self.assertEqual("403 Forbidden",response["status"])
|
|
self.assertNotIn(b"support-delivery",body)
|
|
response,_=self.post("/platform/activity",who="operator")
|
|
self.assertEqual("404 Not Found",response["status"])
|
|
|
|
def test_exact_filters_apply_before_display_limit_and_missing_is_explicit(self):
|
|
self.seed()
|
|
for i in range(105):
|
|
self.app.service.store.append_outbox(OutboxEvent(event_id="noise-"+str(i),event_type="noise",
|
|
aggregate_id="other",tenant=OTHER,correlation_id="noise",payload={}))
|
|
response,body=invoke(self.app,"/platform/activity",cookie="ue_session=operator",
|
|
query=urlencode({"reference":"support-123","tenant":TENANT}))
|
|
self.assertEqual("200 OK",response["status"])
|
|
self.assertIn(b"support-delivery",body)
|
|
self.assertIn(b"Showing 2 of 2",body)
|
|
for query in ["reference=support",urlencode({"reference":"support-123","tenant":OTHER})]:
|
|
_,body=invoke(self.app,"/platform/activity",cookie="ue_session=operator",query=query)
|
|
self.assertIn(b"No matching records",body)
|
|
self.assertNotIn(b"support-delivery",body)
|
|
|
|
def test_filter_values_are_escaped_and_bounded(self):
|
|
response,body=invoke(self.app,"/platform/activity",cookie="ue_session=operator",query=urlencode({"reference":"<script>alert(1)</script>"}))
|
|
self.assertEqual("200 OK",response["status"])
|
|
self.assertIn(b"<script>",body);self.assertNotIn(b"<script>",body)
|
|
response,_=invoke(self.app,"/platform/activity",cookie="ue_session=operator",query="reference="+"x"*201)
|
|
self.assertEqual("400 Bad Request",response["status"])
|
|
|
|
def test_recovery_denied_before_preview_and_operator_sees_factor_boundary(self):
|
|
user=self.member(role="tenant-admin")
|
|
path=f"/admin/{TENANT}/users/{user.user_id}/recover"
|
|
response,body=self.post(path,who="admin")
|
|
self.assertEqual("403 Forbidden",response["status"])
|
|
self.assertNotIn(b"Confirm change",body)
|
|
response,body=self.post(path,who="operator")
|
|
self.assertEqual("200 OK",response["status"])
|
|
self.assertIn(b"does not reset a password",body)
|
|
self.assertIn(b"cannot bypass",body)
|
|
self.assertEqual([],self.app.provisioning.actions)
|
|
|
|
def test_service_capabilities_distinguish_configuration_from_health(self):
|
|
self.app.provisioning=None;self.app.tenant_management=None;self.app.outbox_delivery=None
|
|
response,body=invoke(self.app,"/platform/operations",cookie="ue_session=operator")
|
|
self.assertEqual("200 OK",response["status"])
|
|
self.assertIn(b"Configured; live health unverified",body)
|
|
self.assertIn(b"Unavailable in this portal",body)
|
|
self.assertIn(b"assisted password setup",body)
|
|
self.assertIn(b"authentication policy changes are unavailable",body)
|