user-engine/tests/test_service_operations.py

import time
from unittest.mock import patch
from dataclasses import replace
from user_engine.domain import OutboxEvent, utc_now
from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter, DeliveryError
from user_engine.operations_status import check_services
from test_journey_roles import JourneyFixture, TENANT
from test_web import invoke


class ServiceOperations(JourneyFixture):
    def seed(self):
        event = OutboxEvent(
            event_id="p05-one",
            event_type="family_member.invited",
            aggregate_id="person",
            tenant=TENANT,
            correlation_id="p05-case",
            payload={
                "primary_email": "fixture@example.test",
                "invitation_id": "fixture",
            },
        )
        self.app.service.store.append_outbox(event)
        self.calls = []
        self.app.outbox_delivery = lambda event: self.calls.append(event.event_id)
        return event

    def test_one_confirmed_delivery_and_completed_retry_guard(self):
        event = self.seed()
        self.app.service.store.append_outbox(replace(event, event_id="p05-other"))
        response, body = self.post(
            "/platform/operations/deliver", who="operator", event_id=event.event_id
        )
        self.assertIn(b"Confirm change", body)
        self.assertEqual([], self.calls)
        response, _ = self.post(
            "/platform/operations/deliver",
            who="operator",
            event_id=event.event_id,
            confirmed=True,
        )
        self.assertEqual("303 See Other", response["status"])
        self.assertEqual(["p05-one"], self.calls)
        response, _ = self.post(
            "/platform/operations/deliver", who="operator", event_id=event.event_id
        )
        self.assertEqual("409 Conflict", response["status"])
        self.assertIsNone(self.app.service.store.outbox_event("p05-other").delivered_at)

    def test_role_csrf_and_stale_confirmation_cannot_submit(self):
        event = self.seed()
        for who in ["member", "admin"]:
            response, _ = self.post(
                "/platform/operations/deliver", who=who, event_id=event.event_id
            )
            self.assertEqual("403 Forbidden", response["status"])
        response, _ = self.post(
            "/platform/operations/deliver",
            who="operator",
            event_id=event.event_id,
            csrf_token="wrong",
        )
        self.assertEqual("403 Forbidden", response["status"])
        _, body = self.post(
            "/platform/operations/deliver", who="operator", event_id=event.event_id
        )
        token = self.confirm_token(body)
        self.app.service.store.save_outbox(replace(event, delivery_attempts=1))
        response, _ = self.post(
            "/platform/operations/deliver",
            who="operator",
            event_id=event.event_id,
            confirm_token=token,
        )
        self.assertEqual("409 Conflict", response["status"])
        self.assertEqual([], self.calls)

    def test_failure_releases_claim_and_retry_is_available(self):
        event = self.seed()

        def fail(event):
            raise RuntimeError("private-token-in-error")

        self.app.outbox_delivery = fail
        self.post(
            "/platform/operations/deliver",
            who="operator",
            event_id=event.event_id,
            confirmed=True,
        )
        failed = self.app.service.store.outbox_event(event.event_id)
        self.assertIsNone(failed.claimed_by)
        self.assertEqual("delivery_unavailable", failed.failure_reason)
        _, body = invoke(self.app, "/platform/operations", cookie="ue_session=operator")
        self.assertIn(b"Review delivery attempt", body)
        self.assertNotIn(b"private-token", body)
        self.app.outbox_delivery = lambda e: self.calls.append(e.event_id)
        self.post(
            "/platform/operations/deliver",
            who="operator",
            event_id=event.event_id,
            confirmed=True,
        )
        self.assertIsNotNone(
            self.app.service.store.outbox_event(event.event_id).delivered_at
        )

    def test_crashed_claim_reconciles_under_event_guard(self):
        event = self.seed()
        self.app.service.store.save_outbox(
            replace(event, claimed_by="crashed", claimed_at=utc_now())
        )
        self.post(
            "/platform/operations/deliver",
            who="operator",
            event_id=event.event_id,
            confirmed=True,
        )
        self.assertEqual(["p05-one"], self.calls)

    def test_missing_mail_lane_never_marks_an_invitation_accepted(self):
        event = self.seed()
        adapter = HTTPOutboxDeliveryAdapter(
            event_url="http://unused", event_bearer_token="fixture"
        )
        with self.assertRaises(DeliveryError) as error:
            adapter(event)
        self.assertEqual("mail_unconfigured", error.exception.code)

    def test_status_is_checked_bounded_and_never_claims_receipt(self):
        class Mail:
            def mail_status(self):
                return {"checks": {"database": "ok", "smtp_authentication": "failed"}}

        class OIDC:
            backend_url = "http://fixture"

        with patch(
            "user_engine.operations_status.sign_in_status",
            return_value={
                "checks": [
                    {"name": "factor_reader", "status": "failed"},
                    {"name": "lldap", "status": "ok"},
                ]
            },
        ):
            html = check_services(OIDC(), Mail())
        self.assertIn("Factor-read credential", html)
        self.assertIn("Unavailable", html)
        self.assertIn("send no email", html)
        self.assertIn("do not prove inbox receipt", html)

    def test_selected_mail_evidence_is_safe_and_read_only(self):
        event = self.seed()

        class Mail:
            def mail_delivery_status(self, event_id):
                return {"state": "unknown", "private": "secret-provider-detail"}

        self.app.outbox_delivery = Mail()
        _, body = invoke(
            self.app,
            "/platform/operations",
            cookie="ue_session=operator",
            query="event_id=" + event.event_id,
        )
        self.assertIn(b"Automatic resending is blocked", body)
        self.assertNotIn(b"secret-provider-detail", body)
        self.assertEqual(
            0, self.app.service.store.outbox_event(event.event_id).delivery_attempts
        )