from __future__ import annotations import asyncio from typing import Any import httpx from a2a_pack import LocalRunContext, LocalWorkspaceClient, NoAuth, WorkspaceAccess, WorkspaceMode from agent import ( LEDGER_PATH, ReceivesSupportEmailChecks1919313, _action_digest, _approval_phrase, _normalize_refund_action, ) class FakeResponse: def __init__(self, status_code: int, payload: dict[str, Any]): self.status_code = status_code self._payload = payload def json(self) -> dict[str, Any]: return self._payload class MockAsyncClient: queue: list[Any] = [] calls: list[dict[str, Any]] = [] def __init__(self, *args: Any, **kwargs: Any) -> None: pass async def __aenter__(self) -> "MockAsyncClient": return self async def __aexit__(self, *args: Any) -> None: return None async def get(self, url: str, **kwargs: Any) -> FakeResponse: self.calls.append({"method": "GET", "url": url, **kwargs}) item = self.queue.pop(0) if isinstance(item, BaseException): raise item return item async def post(self, url: str, **kwargs: Any) -> FakeResponse: self.calls.append({"method": "POST", "url": url, **kwargs}) item = self.queue.pop(0) if isinstance(item, BaseException): raise item return item def make_ctx() -> LocalRunContext[NoAuth]: workspace = LocalWorkspaceClient( {LEDGER_PATH: b'{"emails": {}, "refunds": {}}'}, access=WorkspaceAccess.dynamic( max_files=8, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, ), ) workspace.outputs_prefix = "outputs" workspace.write_prefixes = ("outputs/",) workspace.current_mode = WorkspaceMode.READ_WRITE_OVERLAY workspace._apply_patches_directly = True return LocalRunContext( auth=NoAuth(), workspace=workspace, consumer_secrets={"STRIPE_SECRET_KEY": "sk_test_x", "SLACK_BOT_TOKEN": "xoxb-test"}, consumer_config={"SLACK_CHANNEL_ID": "C123", "STRIPE_BASE_URL": "https://stripe.test", "SLACK_BASE_URL": "https://slack.test/api"}, ) def test_missing_setup_returns_clear_result() -> None: async def run() -> None: agent = ReceivesSupportEmailChecks1919313() ctx = LocalRunContext(auth=NoAuth(), consumer_secrets={}, consumer_config={}) result = await agent.process_support_email( ctx, sender_email="customer@example.com", subject="Refund", body="refund pi_abc123", message_id="m1", stripe_payment_intent_id="pi_abc123", requested_refund_amount_cents=1000, ) assert result["status"] == "setup_required" assert "STRIPE_SECRET_KEY" in result["missing"] assert "SLACK_BOT_TOKEN" in result["missing"] assert "SLACK_CHANNEL_ID" in result["missing"] assert "sk_test" not in str(result) asyncio.run(run()) def test_process_email_proposes_refund_and_dedupes(monkeypatch: Any) -> None: async def run() -> None: MockAsyncClient.queue = [ FakeResponse(200, {"id": "pi_abc123", "status": "succeeded", "amount_received": 3000, "currency": "usd", "livemode": False, "latest_charge": {"amount_refunded": 0}}), FakeResponse(200, {"ok": True, "ts": "1.2", "channel": "C123"}), ] MockAsyncClient.calls = [] monkeypatch.setattr("agent.httpx.AsyncClient", MockAsyncClient) agent = ReceivesSupportEmailChecks1919313() ctx = make_ctx() kwargs = dict( sender_email="Customer ", subject="Refund request", body="Please refund payment pi_abc123", message_id="msg-1", stripe_payment_intent_id="pi_abc123", requested_refund_amount_cents=2500, currency="usd", customer_name="Customer", ) first = await agent.process_support_email(ctx, **kwargs) second = await agent.process_support_email(ctx, **kwargs) assert first["status"] == "proposed" assert first["approval_required"] is True assert first["executed"] is False assert first["slack"]["status"] == "posted" assert "APPROVE-REFUND-" in first["approval_instructions"] assert second["status"] == "duplicate_email" assert second["previous_action_digest"] == first["action_digest"] assert "sk_test_x" not in str(first) assert "xoxb-test" not in str(first) asyncio.run(run()) def test_execute_requires_digest_bound_approval(monkeypatch: Any) -> None: async def run() -> None: MockAsyncClient.calls = [] monkeypatch.setattr("agent.httpx.AsyncClient", MockAsyncClient) agent = ReceivesSupportEmailChecks1919313() ctx = make_ctx() action = _normalize_refund_action( payment_intent_id="pi_abc123", amount_cents=1200, currency="usd", customer_email="customer@example.com", reason="requested_by_customer", ) digest = _action_digest(action) wrong = await agent.execute_approved_refund( ctx, action_digest=digest, approval_token="looks good", stripe_payment_intent_id="pi_abc123", amount_cents=1200, customer_email="customer@example.com", ) assert wrong["status"] == "approval_required" MockAsyncClient.queue = [ FakeResponse(200, {"id": "re_123", "status": "succeeded"}), FakeResponse(200, {"ok": True, "ts": "2.3", "channel": "C123"}), ] approved = await agent.execute_approved_refund( ctx, action_digest=digest, approval_token=_approval_phrase(digest), stripe_payment_intent_id="pi_abc123", amount_cents=1200, customer_email="customer@example.com", ) assert approved["status"] == "executed" assert approved["provider_refund_id"] == "re_123" assert approved["idempotency_key"].startswith("a2a-refund-") refund_call = next(call for call in MockAsyncClient.calls if call["url"].endswith("/v1/refunds")) assert refund_call["headers"]["Idempotency-Key"] == approved["idempotency_key"] assert "sk_test_x" not in str(approved) assert "xoxb-test" not in str(approved) asyncio.run(run()) def test_provider_timeout_returns_reconciliation(monkeypatch: Any) -> None: async def run() -> None: MockAsyncClient.queue = [ httpx.TimeoutException("slow stripe"), FakeResponse(200, {"ok": True, "ts": "3.4", "channel": "C123"}), ] MockAsyncClient.calls = [] monkeypatch.setattr("agent.httpx.AsyncClient", MockAsyncClient) agent = ReceivesSupportEmailChecks1919313() ctx = make_ctx() result = await agent.process_support_email( ctx, sender_email="customer@example.com", subject="Refund request", body="Please refund payment pi_timeout", message_id="msg-timeout", stripe_payment_intent_id="pi_timeout", requested_refund_amount_cents=1000, ) assert result["status"] == "needs_reconciliation" assert result["policy"]["approval_required"] is True assert result["executed"] is False asyncio.run(run())