@mindtdilly / Omnipay-4
Code Commits Issues Pulls Sessions CI/CD
feat/agentic-commerce-x402-local-sim
feat/agentic-commerce-x402-local-sim feat/checkout-funding-ach main
py 217 lines 6.91 KB
Raw
1 """Local x402 happy path, denial, and fabricated-receipt rejection tests."""
2
3 from __future__ import annotations
4
5 from datetime import UTC, datetime, timedelta
6 from decimal import Decimal
7
8 import pytest
9
10 from omnipay_x402.agent import (
11 PurchaseResultRecorder,
12 SupplierResearchOutput,
13 run_supplier_research,
14 validate_supplier_research_output,
15 )
16 from omnipay_x402.application import CommerceApplication
17 from omnipay_x402.aws_stub import live_agentcore_status
18 from omnipay_x402.codec import encode_model
19 from omnipay_x402.errors import AgentResultInvalid, PolicyDenied
20 from omnipay_x402.merchant import (
21 PAYMENT_REQUIRED_HEADER,
22 PAYMENT_SIGNATURE_HEADER,
23 RESOURCE_URL,
24 SyntheticMerchant,
25 )
26 from omnipay_x402.models import (
27 ApprovalGrant,
28 AuditEventType,
29 CommercePolicy,
30 PurchaseRequest,
31 )
32 from omnipay_x402.payments import LocalPaymentProcessor
33 from omnipay_x402.policy import PolicyEngine
34 from omnipay_x402.tool import build_x402_fetch_tool
35 from omnipay_x402 import VALUE_TRANSFERRED
36
37 NOW = datetime(2026, 7, 30, 16, 0, tzinfo=UTC)
38
39
40 def make_system(
41 *,
42 price: Decimal = Decimal("0.25"),
43 request_limit: Decimal = Decimal("0.50"),
44 run_limit: Decimal = Decimal("1.00"),
45 threshold: Decimal = Decimal("0.10"),
46 session_expires_at: datetime | None = None,
47 ) -> tuple[CommerceApplication, SyntheticMerchant, LocalPaymentProcessor]:
48 payments = LocalPaymentProcessor()
49 merchant = SyntheticMerchant(payments, price=price, now=NOW)
50 policy = PolicyEngine(
51 CommercePolicy(
52 allowed_merchants=frozenset({"merchant.invalid"}),
53 allowed_purposes=frozenset({"supplier_due_diligence"}),
54 per_request_limit=request_limit,
55 per_run_limit=run_limit,
56 approval_threshold=threshold,
57 session_expires_at=(session_expires_at or NOW + timedelta(minutes=15)),
58 )
59 )
60 app = CommerceApplication(
61 client=merchant.client(),
62 policy=policy,
63 payments=payments,
64 )
65 return app, merchant, payments
66
67
68 def request(
69 *,
70 request_id: str = "request-001",
71 resource_url: str = RESOURCE_URL,
72 purpose: str = "supplier_due_diligence",
73 idempotency_key: str = "purchase-001",
74 ) -> PurchaseRequest:
75 return PurchaseRequest(
76 request_id=request_id,
77 resource_url=resource_url,
78 purpose=purpose,
79 idempotency_key=idempotency_key,
80 )
81
82
83 def approval(
84 purchase: PurchaseRequest,
85 *,
86 maximum_amount: Decimal = Decimal("0.25"),
87 expires_at: datetime | None = None,
88 ) -> ApprovalGrant:
89 return ApprovalGrant(
90 approval_id=f"approval-{purchase.request_id}",
91 request_id=purchase.request_id,
92 resource_url=purchase.resource_url,
93 purpose=purchase.purpose,
94 maximum_amount=maximum_amount,
95 approved_by="synthetic-reviewer",
96 approved_at=NOW,
97 expires_at=expires_at or NOW + timedelta(minutes=10),
98 )
99
100
101 def test_value_transferred_is_always_false() -> None:
102 assert VALUE_TRANSFERRED is False
103 status = live_agentcore_status()
104 assert status["status"] == "SKIPPED"
105 assert status["reason"] == "NOT_READY"
106 assert status["value_transferred"] is False
107 assert status["network_calls"] is False
108
109
110 def test_approved_purchase_completes_full_402_sequence() -> None:
111 """Happy path: 402 → approve → synthetic proof → receipt/audit."""
112
113 app, merchant, payments = make_system()
114 purchase = request()
115
116 result = app.purchase(purchase, approval=approval(purchase), now=NOW)
117
118 assert result.status == "completed"
119 assert result.receipt.amount == Decimal("0.25")
120 assert result.authorization.requires_human_approval is True
121 assert result.report.report_id == "SYNTH-SUPPLIER-RISK-001"
122 assert merchant.request_count == 2
123 assert merchant.fulfilled_count == 1
124 assert payments.charge_count == 1
125 assert [event.event_type for event in result.audit_events] == [
126 AuditEventType.RESOURCE_REQUESTED,
127 AuditEventType.PAYMENT_REQUIRED,
128 AuditEventType.AUTHORIZATION_CHECKED,
129 AuditEventType.PAYMENT_ATTEMPTED,
130 AuditEventType.PROOF_CREATED,
131 AuditEventType.MERCHANT_RETRY,
132 AuditEventType.CONTENT_RETURNED,
133 ]
134
135
136 def test_missing_human_approval_is_denied_before_payment() -> None:
137 """Denial without approval: policy blocks before synthetic payment."""
138
139 app, merchant, payments = make_system()
140
141 with pytest.raises(PolicyDenied) as exc_info:
142 app.purchase(request(), now=NOW)
143
144 assert exc_info.value.code == "human_approval_required"
145 assert merchant.request_count == 1
146 assert payments.charge_count == 0
147
148
149 def test_application_validation_rejects_fabricated_receipt() -> None:
150 """Fabricated receipt rejection: agent output must match app evidence."""
151
152 app, _, _ = make_system()
153 purchase = request()
154 completed = app.purchase(purchase, approval=approval(purchase), now=NOW)
155 fabricated = SupplierResearchOutput(
156 status=completed.status,
157 report_id=completed.report.report_id,
158 supplier=completed.report.supplier,
159 signals=completed.report.signals,
160 disclaimer=completed.report.disclaimer,
161 receipt_id="receipt-fabricated",
162 amount=str(completed.receipt.amount),
163 currency=completed.receipt.currency,
164 requires_human_approval=completed.authorization.requires_human_approval,
165 )
166
167 with pytest.raises(AgentResultInvalid) as exc_info:
168 validate_supplier_research_output(
169 fabricated,
170 PurchaseResultRecorder(results=[completed]),
171 )
172
173 assert exc_info.value.code == "agent_output_mismatch"
174 assert "receipt_id" in str(exc_info.value)
175
176
177 def test_scripted_agent_run_matches_application_evidence() -> None:
178 app, merchant, payments = make_system()
179 purchase = request(request_id="request-agent-001", idempotency_key="purchase-agent-001")
180
181 result = run_supplier_research(
182 app,
183 request_id=purchase.request_id,
184 idempotency_key=purchase.idempotency_key,
185 approval=approval(purchase),
186 now=NOW,
187 )
188
189 assert result.output.receipt_id == result.purchase.receipt.receipt_id
190 assert merchant.request_count == 2
191 assert payments.charge_count == 1
192
193
194 def test_merchant_rejects_unrecognized_proof() -> None:
195 _, merchant, payments = make_system()
196 response = merchant.client().get(
197 RESOURCE_URL,
198 headers={PAYMENT_SIGNATURE_HEADER: encode_model(merchant.payment_required)},
199 )
200
201 assert response.status_code == 402
202 assert PAYMENT_REQUIRED_HEADER in response.headers
203 assert merchant.invalid_proof_count == 1
204 assert payments.charge_count == 0
205
206
207 def test_agents_tool_has_prebound_authority_and_expected_name() -> None:
208 app, _, _ = make_system()
209 purchase = request()
210 tool = build_x402_fetch_tool(
211 app,
212 request_id=purchase.request_id,
213 idempotency_key=purchase.idempotency_key,
214 approval=approval(purchase),
215 )
216
217 assert tool.name == "x402_fetch"