""" Stripe API simulator. Provides realistic mock responses for Stripe API actions. """ from typing import Optional from .base import BaseSimulator from datetime import datetime import time import random import string class StripeSimulator(BaseSimulator): """Simulator for Stripe API.""" def __init__(self): super().__init__('stripe') def load_mock_responses(self): """Load Stripe mock response templates.""" self.mock_responses = { 'create_payment_intent': self._create_payment_intent_template, 'retrieve_customer': self._retrieve_customer_template, 'create_subscription': self._create_subscription_template, 'create_customer': self._create_customer_template, 'list_charges': self._list_charges_template, 'refund_payment': self._refund_payment_template, } def get_required_permissions(self, action: str) -> set[str]: """Get required Stripe permissions for an action.""" permissions_map = { 'create_payment_intent': {'payment_intents_write'}, 'retrieve_customer': {'customers_read'}, 'create_subscription': {'subscriptions_write'}, 'create_customer': {'customers_write'}, 'list_charges': {'charges_read'}, 'refund_payment': {'refunds_write'}, } return permissions_map.get(action, set()) def validate_params(self, action: str, params: dict) -> tuple[bool, Optional[str]]: """Validate parameters for Stripe actions.""" if action == 'create_payment_intent': required = {'amount', 'currency'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" if not isinstance(params['amount'], int): return False, "Amount must be an integer (in cents)" elif action == 'retrieve_customer': required = {'customer_id'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'create_subscription': required = {'customer', 'items'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'create_customer': # Email is optional but recommended pass elif action == 'list_charges': # All params optional pass elif action == 'refund_payment': required = {'payment_intent'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" else: return False, f"Unknown action: {action}" return True, None def generate_mock_response(self, action: str, params: dict) -> dict: """Generate realistic Stripe API response.""" if action not in self.mock_responses: raise ValueError(f"Unknown action: {action}") template_func = self.mock_responses[action] return template_func(params) def _generate_id(self, prefix: str) -> str: """Generate a Stripe-style ID.""" random_part = ''.join(random.choices(string.ascii_lowercase + string.digits, k=24)) return f"{prefix}_{random_part}" def _create_payment_intent_template(self, params: dict) -> dict: """Mock response for creating a payment intent.""" pi_id = self._generate_id('pi') now = int(time.time()) return { "id": pi_id, "object": "payment_intent", "amount": params['amount'], "amount_capturable": 0, "amount_received": 0, "application": None, "application_fee_amount": None, "canceled_at": None, "cancellation_reason": None, "capture_method": params.get('capture_method', 'automatic'), "client_secret": f"{pi_id}_secret_{''.join(random.choices(string.ascii_letters + string.digits, k=32))}", "confirmation_method": "automatic", "created": now, "currency": params['currency'], "customer": params.get('customer'), "description": params.get('description'), "invoice": None, "last_payment_error": None, "livemode": False, "metadata": params.get('metadata', {}), "next_action": None, "on_behalf_of": None, "payment_method": None, "payment_method_options": {}, "payment_method_types": ["card"], "processing": None, "receipt_email": params.get('receipt_email'), "review": None, "setup_future_usage": None, "shipping": None, "statement_descriptor": params.get('statement_descriptor'), "statement_descriptor_suffix": None, "status": "requires_payment_method", "transfer_data": None, "transfer_group": None } def _retrieve_customer_template(self, params: dict) -> dict: """Mock response for retrieving a customer.""" customer_id = params['customer_id'] return { "id": customer_id, "object": "customer", "address": None, "balance": 0, "created": int(time.time()) - 86400 * 30, # 30 days ago "currency": "usd", "default_source": None, "delinquent": False, "description": "Mock customer for testing", "discount": None, "email": "customer@example.com", "invoice_prefix": "INV", "invoice_settings": { "custom_fields": None, "default_payment_method": None, "footer": None, "rendering_options": None }, "livemode": False, "metadata": {}, "name": "Test Customer", "phone": None, "preferred_locales": [], "shipping": None, "tax_exempt": "none", "test_clock": None } def _create_subscription_template(self, params: dict) -> dict: """Mock response for creating a subscription.""" sub_id = self._generate_id('sub') now = int(time.time()) # Create items from params items_data = [] for item in params['items']: items_data.append({ "id": self._generate_id('si'), "object": "subscription_item", "created": now, "metadata": {}, "price": { "id": item.get('price', 'price_1234'), "object": "price", "active": True, "currency": "usd", "product": "prod_1234", "type": "recurring", "unit_amount": 2000, "recurring": { "interval": "month", "interval_count": 1 } }, "quantity": item.get('quantity', 1), "subscription": sub_id }) return { "id": sub_id, "object": "subscription", "application": None, "application_fee_percent": None, "automatic_tax": { "enabled": False }, "billing_cycle_anchor": now, "billing_thresholds": None, "cancel_at": None, "cancel_at_period_end": False, "canceled_at": None, "collection_method": "charge_automatically", "created": now, "currency": "usd", "current_period_end": now + 2592000, # 30 days later "current_period_start": now, "customer": params['customer'], "days_until_due": None, "default_payment_method": None, "default_source": None, "default_tax_rates": [], "description": None, "discount": None, "ended_at": None, "items": { "object": "list", "data": items_data, "has_more": False, "total_count": len(items_data), "url": f"/v1/subscription_items?subscription={sub_id}" }, "latest_invoice": self._generate_id('in'), "livemode": False, "metadata": params.get('metadata', {}), "next_pending_invoice_item_invoice": None, "pause_collection": None, "payment_settings": { "payment_method_options": None, "payment_method_types": None, "save_default_payment_method": "off" }, "pending_invoice_item_interval": None, "pending_setup_intent": None, "pending_update": None, "schedule": None, "start_date": now, "status": "active", "test_clock": None, "transfer_data": None, "trial_end": None, "trial_start": None } def _create_customer_template(self, params: dict) -> dict: """Mock response for creating a customer.""" customer_id = self._generate_id('cus') return { "id": customer_id, "object": "customer", "address": params.get('address'), "balance": 0, "created": int(time.time()), "currency": None, "default_source": None, "delinquent": False, "description": params.get('description'), "discount": None, "email": params.get('email'), "invoice_prefix": "INV", "invoice_settings": { "custom_fields": None, "default_payment_method": None, "footer": None, "rendering_options": None }, "livemode": False, "metadata": params.get('metadata', {}), "name": params.get('name'), "phone": params.get('phone'), "preferred_locales": [], "shipping": None, "tax_exempt": "none", "test_clock": None } def _list_charges_template(self, params: dict) -> dict: """Mock response for listing charges.""" # Generate a few mock charges charges = [] for i in range(3): charge_id = self._generate_id('ch') charges.append({ "id": charge_id, "object": "charge", "amount": 5000 + (i * 1000), "amount_captured": 5000 + (i * 1000), "amount_refunded": 0, "application": None, "balance_transaction": self._generate_id('txn'), "billing_details": { "address": None, "email": "customer@example.com", "name": f"Customer {i+1}", "phone": None }, "captured": True, "created": int(time.time()) - (i * 3600), "currency": "usd", "customer": self._generate_id('cus'), "description": f"Charge {i+1}", "disputed": False, "failure_code": None, "failure_message": None, "fraud_details": {}, "invoice": None, "livemode": False, "metadata": {}, "outcome": { "network_status": "approved_by_network", "reason": None, "risk_level": "normal", "seller_message": "Payment complete.", "type": "authorized" }, "paid": True, "payment_intent": self._generate_id('pi'), "payment_method": self._generate_id('pm'), "receipt_email": "customer@example.com", "receipt_url": f"https://pay.stripe.com/receipts/{charge_id}", "refunded": False, "status": "succeeded" }) return { "object": "list", "data": charges, "has_more": False, "url": "/v1/charges" } def _refund_payment_template(self, params: dict) -> dict: """Mock response for refunding a payment.""" refund_id = self._generate_id('re') return { "id": refund_id, "object": "refund", "amount": params.get('amount', 5000), "balance_transaction": self._generate_id('txn'), "charge": None, "created": int(time.time()), "currency": "usd", "metadata": params.get('metadata', {}), "payment_intent": params['payment_intent'], "reason": params.get('reason'), "receipt_number": None, "source_transfer_reversal": None, "status": "succeeded", "transfer_reversal": None }