fireaction-a2a / tests /test_stripe_flow.py
zequn-fireworks's picture
Upgrade to hierarchical schema and build_request() API
4d52c8a
"""Test multi-step Stripe flow: product -> price -> payment link.
This test verifies the runner works for all three Stripe contracts
using a mocked ProviderClient (no real API calls).
"""
import asyncio
from unittest.mock import AsyncMock, MagicMock
import fireaction
from fireaction_a2a.runner import execute_contract
def _mock_stripe_client():
client = AsyncMock()
client.get_context = MagicMock(return_value={})
client.get_base_url = MagicMock(return_value="")
return client
def test_stripe_product_to_price_to_payment_link():
"""Chain three Stripe contracts with ID forwarding between steps."""
mock_client = _mock_stripe_client()
# Step 1: create_product
mock_client.call.return_value = {
"id": "prod_test123",
"name": "Pro Plan",
"active": True,
}
contract = fireaction.load(provider="stripe", action="create_product")
V = "create_product_skeleton"
nodes = [
{"instance_key": "root", "element_key": "root", "variant_key": V,
"data_type": "object", "compile_key": "root", "description": "Root",
"index": 0, "parent_instance_key": None},
{"instance_key": "body_params", "element_key": "body_params", "variant_key": V,
"data_type": "object", "compile_key": "body", "description": "Body",
"index": 1, "parent_instance_key": "root"},
{"instance_key": "name", "element_key": "name", "variant_key": V,
"data_type": "string", "compile_key": "name", "description": "Product name",
"index": 2, "primitive_contents": "Pro Plan", "parent_instance_key": "body_params"},
]
result = asyncio.run(execute_contract(contract, nodes, mock_client))
product_id = result.response["id"]
assert product_id == "prod_test123"
# Step 2: create_price (references product_id)
mock_client.call.return_value = {
"id": "price_test456",
"product": "prod_test123",
"unit_amount": 4900,
"currency": "usd",
}
contract = fireaction.load(provider="stripe", action="create_price")
V = "create_price_skeleton"
nodes = [
{"instance_key": "root", "element_key": "root", "variant_key": V,
"data_type": "object", "compile_key": "root", "description": "Root",
"index": 0, "parent_instance_key": None},
{"instance_key": "body_params", "element_key": "body_params", "variant_key": V,
"data_type": "object", "compile_key": "body", "description": "Body",
"index": 1, "parent_instance_key": "root"},
{"instance_key": "currency", "element_key": "currency", "variant_key": V,
"data_type": "string", "compile_key": "currency", "description": "Currency",
"index": 2, "primitive_contents": "usd", "parent_instance_key": "body_params"},
{"instance_key": "unit_amount", "element_key": "unit_amount", "variant_key": V,
"data_type": "integer", "compile_key": "unit_amount", "description": "Amount in cents",
"index": 3, "primitive_contents": 4900, "parent_instance_key": "body_params"},
{"instance_key": "product", "element_key": "product", "variant_key": V,
"data_type": "string", "compile_key": "product", "description": "Product ID",
"index": 4, "primitive_contents": product_id, "parent_instance_key": "body_params"},
]
result = asyncio.run(execute_contract(contract, nodes, mock_client))
price_id = result.response["id"]
assert price_id == "price_test456"
# Step 3: create_payment_link (references price_id)
mock_client.call.return_value = {
"id": "plink_test789",
"active": True,
"url": "https://buy.stripe.com/test_xxx",
}
contract = fireaction.load(provider="stripe", action="create_payment_link")
V = "create_payment_link_skeleton"
nodes = [
{"instance_key": "root", "element_key": "root", "variant_key": V,
"data_type": "object", "compile_key": "root", "description": "Root",
"index": 0, "parent_instance_key": None},
{"instance_key": "body_params", "element_key": "body_params", "variant_key": V,
"data_type": "object", "compile_key": "body", "description": "Body",
"index": 1, "parent_instance_key": "root"},
{"instance_key": "line_items", "element_key": "line_items_array", "variant_key": V,
"data_type": "array", "compile_key": "line_items", "description": "Line items",
"index": 2, "parent_instance_key": "body_params"},
{"instance_key": "li_1", "element_key": "line_item", "variant_key": V,
"data_type": "object", "compile_key": "li_1", "description": "Line item 1",
"index": 3, "parent_instance_key": "line_items"},
{"instance_key": "li_1_price", "element_key": "line_item_price", "variant_key": V,
"data_type": "string", "compile_key": "price", "description": "Price ID",
"index": 4, "primitive_contents": price_id, "parent_instance_key": "li_1"},
{"instance_key": "li_1_qty", "element_key": "line_item_quantity", "variant_key": V,
"data_type": "integer", "compile_key": "quantity", "description": "Quantity",
"index": 5, "primitive_contents": 1, "parent_instance_key": "li_1"},
]
result = asyncio.run(execute_contract(contract, nodes, mock_client))
assert result.response["id"] == "plink_test789"
assert result.response["url"] == "https://buy.stripe.com/test_xxx"
assert mock_client.call.call_count == 3