feat: implement 'requires_escalation' triggered by risk signals

.cspell/custom-words.txt

@@ -1,78 +1,80 @@
 # cspell-specific custom words related to UCP
+absl
+absltest
+adyen
 Adyen
+agentic
 Alam
 Amex
 Ant
 Anytown
+atok
+backorder
 Backordered
 Braintree
 Carrefour
 Centricity
+checkout
 Chewy
 Commerce
-Credentialless
-Depot
-EWALLET
-Etsy
-Flipkart
-Gap
-GitHub
-Google
-Gpay
-Kroger
-Lowe's
-Macy's
-Mastercard
-Paymentech
-Paypal
-Preorders
-Queensway
-Sephora
-Shopify
-Shopee
-Stripe
-Target
-UCP
-Ulta
-Visa
-Wayfair
-Worldpay
-Zalando
-adyen
-agentic
-atok
-backorder
-checkout
-credentialless
 credentialization
+credentialless
+Credentialless
 datamodel
+Depot
 dpan
+Etsy
 ewallet
+EWALLET
+Flipkart
 fontawesome
 fpan
 fulfillable
+Gap
+GitHub
+Google
 gpay
+Gpay
 ingestions
 inlinehilite
+Kroger
 linenums
 llmstxt
+Lowe's
+Macy's
 mastercard
+Mastercard
 mkdocs
 mtok
 openapi
 openrpc
+Paymentech
 paypal
+Paypal
 permissionless
 preorders
+Preorders
 proto
 protobuf
 pymdownx
+Queensway
 renderable
 repudiable
 schemas
 sdjwt
+Sephora
+Shopee
 shopify
+Shopify
+Stripe
 superfences
+Target
+UCP
+Ulta
+Visa
 vulnz
+Wayfair
+Worldpay
 yaml
-yml
+yml
+Zalando

checkout_lifecycle_test.py

@@ -413,6 +413,46 @@ def test_cannot_cancel_completed_checkout(self):
       msg="Should not be able to cancel a completed checkout.",
     )
 
+  def test_escalation_flow(self):
+    """Test checkout escalation flow.
+
+    Given a checkout session,
+    When completion is attempted with a risk signal triggering escalation,
+    Then the status should be 'requires_escalation' and continue_url should be
+    present.
+    """
+    response_json = self.create_checkout_session()
+    checkout_obj = checkout.Checkout(**response_json)
+    checkout_id = checkout_obj.id
+
+    # Complete with risk signal
+    payment_payload = integration_test_utils.get_valid_payment_payload()
+    payment_payload["risk_signals"] = {
+      "simulation_trigger": "escalation_required"
+    }
+
+    response = self.client.post(
+      f"/checkout-sessions/{checkout_id}/complete",
+      json=payment_payload,
+      headers=integration_test_utils.get_headers(),
+    )
+
+    self.assert_response_status(response, 200)
+    updated_checkout = checkout.Checkout(**response.json())
+
+    self.assertEqual(
+      updated_checkout.status,
+      "requires_escalation",
+      msg="Status should be requires_escalation",
+    )
+    self.assertIsNotNone(
+      updated_checkout.continue_url, "continue_url should be present"
+    )
+    self.assertTrue(updated_checkout.messages, "Messages should be present")
+    self.assertEqual(
+      updated_checkout.messages[0].root.severity, "requires_buyer_input"
+    )
+
 
 if __name__ == "__main__":
   absltest.main()

protocol_test.py

@@ -110,17 +110,17 @@ def test_version_negotiation(self):
     profile = UcpDiscoveryProfile(**discovery_resp.json())
     shopping_service = profile.ucp.services.root["dev.ucp.shopping"]
     self.assertIsNotNone(
-        shopping_service, "Shopping service not found in discovery"
+      shopping_service, "Shopping service not found in discovery"
     )
     self.assertIsNotNone(
-        shopping_service.rest, "REST config not found for shopping service"
+      shopping_service.rest, "REST config not found for shopping service"
     )
     self.assertIsNotNone(
-        shopping_service.rest.endpoint,
-        "Endpoint not found for shopping service",
+      shopping_service.rest.endpoint,
+      "Endpoint not found for shopping service",
     )
     checkout_sessions_url = (
-        f"{str(shopping_service.rest.endpoint).rstrip('/')}/checkout-sessions"
+      f"{str(shopping_service.rest.endpoint).rstrip('/')}/checkout-sessions"
     )
 
     create_payload = self.create_checkout_payload()