"""Agent financial authorization, uncertainty, and exactly-once fulfillment.""" from datetime import timedelta from types import SimpleNamespace from unittest.mock import AsyncMock from uuid import uuid4 import pytest from src.core.exceptions import ForbiddenError from src.domain.agent_access.service import utcnow from src.domain.billing.agent_purchase_processor import AgentPurchaseProcessor from src.domain.billing.agent_purchases import AgentPurchaseService, require_purchase_budget def operation(**kwargs): return SimpleNamespace( **{ "id": uuid4(), "workspace_id": uuid4(), "status": "queued", "action": "purchase_credits", "result": {}, "created_at": utcnow(), "updated_at": utcnow(), "arguments": { "amount_minor": 500, "currency": "usd", "units": 100, "customer_id": "cus_owner", "package_id": str(uuid4()), "package_slug": "small", }, **kwargs, } ) @pytest.fixture def processor(): access, stripe = AsyncMock(), AsyncMock() access.settings = SimpleNamespace(app_url="https://dashboard.p2hs.com") access.repo.add.side_effect = lambda row: row return AgentPurchaseProcessor(access, stripe) @pytest.mark.parametrize( "amount,currency,reserved", [(501, "usd", 0), (500, "eur", 0), (500, "usd", 501)] ) def test_rejects_per_purchase_currency_and_monthly_limit(amount, currency, reserved): with pytest.raises(ForbiddenError): require_purchase_budget( { "saved_payment_consent": True, "currency": "usd", "per_purchase_minor": 500, "monthly_minor": 1000, }, amount, currency, reserved, ) def test_budget_counts_pending_spend_and_allows_exact_boundary(): require_purchase_budget( { "saved_payment_consent": True, "currency": "usd", "per_purchase_minor": 500, "monthly_minor": 1000, }, 500, "usd", 500, ) async def test_persist_method_then_intent_before_confirming(processor): row = operation() processor.stripe.customer.return_value = { "invoice_settings": {"default_payment_method": "pm_saved"} } await processor.process(row, object()) assert row.arguments["payment_method"] == "pm_saved" processor.stripe.create_intent.assert_not_awaited() processor.stripe.create_intent.return_value = {"id": "pi_one"} await processor.process(row, object()) assert row.arguments["provider_id"] == "pi_one" processor.stripe.confirm_intent.assert_not_awaited() async def test_never_recreate_past_stripe_idempotency_retention(processor): row = operation(created_at=utcnow() - timedelta(hours=24)) await processor.process(row, object()) assert row.status == "manual_review" processor.stripe.create_intent.assert_not_awaited() processor.stripe.create_checkout.assert_not_awaited() async def test_retrieves_uncertain_payment_before_any_new_charge(processor): row = operation() row.arguments["provider_id"] = "pi_one" processor.stripe.retrieve_intent.return_value = {"status": "processing"} await processor.process(row, object()) processor.stripe.retrieve_intent.assert_awaited_once_with("pi_one") processor.stripe.confirm_intent.assert_not_awaited() processor.stripe.create_intent.assert_not_awaited() assert row.status == "queued" async def test_paid_payment_reconciles_even_after_grant_revocation(processor): row = operation() row.arguments["provider_id"] = "pi_one" processor.stripe.retrieve_intent.return_value = {"status": "succeeded"} processor.purchases.fulfill = AsyncMock() processor.access.require_scope.side_effect = ForbiddenError() await processor.process(row, None) processor.purchases.fulfill.assert_awaited_once() processor.access.require_scope.assert_not_awaited() processor.stripe.confirm_intent.assert_not_awaited() async def test_disabling_agents_blocks_new_charges(processor): row = operation() processor.access.settings.agent_access_enabled = False await processor.process(row, object()) assert row.status == "canceled" processor.stripe.create_intent.assert_not_awaited() processor.stripe.create_checkout.assert_not_awaited() async def test_disabling_agents_still_reconciles_paid_money(processor): row = operation() row.arguments["provider_id"] = "pi_one" processor.access.settings.agent_access_enabled = False processor.stripe.retrieve_intent.return_value = {"status": "succeeded"} processor.purchases.fulfill = AsyncMock() await processor.process(row, None) processor.purchases.fulfill.assert_awaited_once() async def test_revocation_cancels_known_unpaid_intent_before_releasing_cap(processor): row = operation() row.arguments["provider_id"] = "pi_one" processor.stripe.retrieve_intent.return_value = {"status": "requires_confirmation"} processor.access.require_scope.side_effect = ForbiddenError() await processor.process(row, None) processor.stripe.cancel_intent.assert_awaited_once() assert row.status == "canceled" processor.stripe.confirm_intent.assert_not_awaited() async def test_uncertain_cancellation_keeps_reservation(processor): row = operation() row.arguments["provider_id"] = "pi_one" processor.stripe.retrieve_intent.return_value = {"status": "requires_action"} processor.stripe.cancel_intent.side_effect = TimeoutError() with pytest.raises(TimeoutError): await processor.process(row, object()) assert row.status == "queued" @pytest.mark.parametrize("mismatch", ["amount_received", "currency", "customer", "metadata"]) async def test_fulfillment_verifies_immutable_snapshot(processor, mismatch): row = operation() payment = { "id": "pi_one", "status": "succeeded", "amount_received": 500, "currency": "usd", "customer": "cus_owner", "metadata": {"agent_operation_id": str(row.id)}, } payment[mismatch] = {} if mismatch == "metadata" else "wrong" await processor.purchases.fulfill(row, payment, checkout=False) assert row.status == "manual_review" processor.access.repo.add.assert_not_awaited() async def test_duplicate_fulfillment_has_no_second_ledger_entry(processor): row = operation(status="completed") await AgentPurchaseService(processor.access).fulfill(row, {}, checkout=False) processor.access.repo.add.assert_not_awaited() async def test_valid_purchase_uses_snapshot_units_and_same_transaction(processor, monkeypatch): row = operation() ledger = AsyncMock() monkeypatch.setattr( "src.domain.billing.agent_purchases.WorkspaceCreditLedgerRepository", lambda db: ledger ) payment = { "id": "pi_one", "status": "succeeded", "amount_received": 500, "currency": "usd", "customer": "cus_owner", "metadata": {"agent_operation_id": str(row.id)}, } await processor.purchases.fulfill(row, payment, checkout=False) assert row.status == "completed" assert row.result["credits_added"] == 100 ledger.add_entry.assert_awaited_once() assert ledger.add_entry.call_args.args == (row.workspace_id, "purchase", 100) processor.access.repo.db.commit.assert_not_awaited()