| """ |
| Pytest fixtures for payment integration tests. |
| |
| Provides stripe-mock server, database sessions, and test clients |
| for deterministic payment testing. |
| """ |
|
|
| import os |
| import pytest |
| import stripe |
| from datetime import datetime |
| from unittest import mock |
| from typing import Generator |
|
|
| from sqlalchemy import create_engine |
| from sqlalchemy.orm import Session, sessionmaker |
|
|
| from core.models import Base |
| from tests.mocks.stripe_mock_server import ( |
| start_stripe_mock, |
| stop_stripe_mock, |
| get_stripe_mock_url, |
| StripeMockError, |
| ) |
|
|
|
|
| @pytest.fixture(scope="session") |
| def stripe_mock_container(): |
| """ |
| Start stripe-mock Docker container once per test session. |
| |
| Yields: |
| Container: Docker container object |
| |
| Example: |
| def test_charge(stripe_mock_container): |
| # stripe-mock is running on http://localhost:12111 |
| response = stripe.Charge.create(amount=100, currency="usd") |
| assert response["status"] == "succeeded" |
| """ |
| container = None |
| try: |
| container = start_stripe_mock() |
| yield container |
| except StripeMockError as e: |
| pytest.skip(f"Failed to start stripe-mock: {e}") |
| finally: |
| if container: |
| try: |
| stop_stripe_mock(container) |
| except Exception as e: |
| print(f"Warning: Failed to stop stripe-mock: {e}") |
|
|
|
|
| @pytest.fixture(autouse=True) |
| def mock_stripe_api(stripe_mock_container): |
| """ |
| Automatically configure stripe to use mock server. |
| |
| This fixture runs automatically for all tests in this module. |
| It sets stripe.api_base to the mock server URL and provides a test API key. |
| |
| Example: |
| def test_create_charge(): |
| # stripe.api_base is automatically set to mock server |
| charge = stripe.Charge.create(amount=100, currency="usd") |
| assert charge["status"] == "succeeded" |
| """ |
| mock_url = get_stripe_mock_url() |
|
|
| |
| original_api_base = stripe.api_base |
| original_api_key = stripe.api_key |
|
|
| |
| stripe.api_base = mock_url |
| stripe.api_key = "sk_test_12345" |
|
|
| yield |
|
|
| |
| stripe.api_base = original_api_base |
| stripe.api_key = original_api_key |
|
|
|
|
| @pytest.fixture |
| def db_session() -> Generator[Session, None, None]: |
| """ |
| Create an in-memory SQLite database session for testing. |
| |
| Uses SQLAlchemy with in-memory SQLite for fast, isolated tests. |
| Creates accounting tables with necessary foreign key tables. |
| |
| Yields: |
| Session: SQLAlchemy database session |
| |
| Example: |
| def test_create_customer(db_session): |
| customer = Customer(name="Test User") |
| db_session.add(customer) |
| db_session.commit() |
| assert customer.id is not None |
| """ |
| |
| engine = create_engine( |
| "sqlite:///:memory:", |
| connect_args={"check_same_thread": False}, |
| echo=False, |
| ) |
|
|
| |
| from accounting.models import ( |
| Account, Transaction, JournalEntry, CategorizationProposal, |
| Entity, Bill, Invoice, Document, TaxNexus, FinancialClose, |
| CategorizationRule, Budget |
| ) |
| from sqlalchemy import Table, Column, String, MetaData |
|
|
| |
| metadata = MetaData() |
|
|
| |
| workspaces_table = Table( |
| "workspaces", metadata, |
| Column("id", String, primary_key=True) |
| ) |
|
|
| service_projects_table = Table( |
| "service_projects", metadata, |
| Column("id", String, primary_key=True) |
| ) |
|
|
| service_milestones_table = Table( |
| "service_milestones", metadata, |
| Column("id", String, primary_key=True) |
| ) |
|
|
| users_table = Table( |
| "users", metadata, |
| Column("id", String, primary_key=True) |
| ) |
|
|
| |
| for table in [workspaces_table, service_projects_table, service_milestones_table, users_table]: |
| table.create(engine, checkfirst=True) |
|
|
| |
| accounting_tables = [ |
| Account.__table__, Transaction.__table__, JournalEntry.__table__, |
| CategorizationProposal.__table__, Entity.__table__, Bill.__table__, |
| Invoice.__table__, Document.__table__, TaxNexus.__table__, |
| FinancialClose.__table__, CategorizationRule.__table__, Budget.__table__, |
| ] |
|
|
| for table in accounting_tables: |
| table.create(engine, checkfirst=True) |
|
|
| |
| TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) |
|
|
| |
| session = TestingSessionLocal() |
| try: |
| yield session |
| finally: |
| session.close() |
| |
| for table in reversed(accounting_tables): |
| table.drop(engine) |
| for table in [users_table, service_milestones_table, service_projects_table, workspaces_table]: |
| table.drop(engine) |
| engine.dispose() |
|
|
|
|
| @pytest.fixture |
| def payment_client(): |
| """ |
| Create a FastAPI TestClient for Stripe routes. |
| |
| Note: This fixture is a placeholder for future API route testing. |
| For now, we use the stripe Python SDK directly with mock_stripe_api. |
| |
| Returns: |
| TestClient: FastAPI test client |
| |
| Example: |
| def test_payments_endpoint(payment_client): |
| response = payment_client.get("/api/stripe/payments") |
| assert response.status_code == 200 |
| """ |
| |
| try: |
| from fastapi.testclient import TestClient |
| from integrations.stripe_routes import router |
|
|
| |
| from fastapi import FastAPI |
| app = FastAPI() |
| app.include_router(router) |
|
|
| return TestClient(app) |
| except ImportError as e: |
| pytest.skip(f"FastAPI routes not available: {e}") |
|
|
|
|
| @pytest.fixture |
| def stripe_access_token() -> str: |
| """ |
| Provide a fake Stripe access token for testing. |
| |
| Returns: |
| str: Mock access token |
| |
| Example: |
| def test_stripe_service(stripe_access_token): |
| result = stripe_service.get_balance(stripe_access_token) |
| assert "available" in result |
| """ |
| return "sk_test_mock_token_1234567890" |
|
|
|
|
| @pytest.fixture |
| def stripe_test_customer(stripe_mock_container, mock_stripe_api): |
| """ |
| Create a test Stripe customer. |
| |
| Returns: |
| dict: Stripe customer object |
| |
| Example: |
| def test_subscription(stripe_test_customer): |
| # Use the customer object |
| sub = stripe.Subscription.create( |
| customer=stripe_test_customer["id"], |
| items=[{"price": "price_test"}] |
| ) |
| """ |
| customer = stripe.Customer.create( |
| email="test@example.com", |
| name="Test User", |
| description="Test customer for payment integration tests", |
| ) |
| return customer |
|
|
|
|
| @pytest.fixture |
| def setup_customer(): |
| """ |
| Create a test customer in the database for payment flow tests. |
| |
| Creates a test Stripe customer and stores reference in database. |
| Cleans up automatically after test completes. |
| |
| Yields: |
| tuple: (customer_id: str, customer_object: dict) |
| |
| Example: |
| def test_charge_flow(setup_customer): |
| customer_id, customer = setup_customer |
| charge = stripe.Charge.create( |
| amount=1000, |
| currency="usd", |
| customer=customer_id |
| ) |
| """ |
| import stripe |
| from core.decimal_utils import to_decimal |
|
|
| |
| customer = stripe.Customer.create( |
| email=f"test_customer_{id(object())}@example.com", |
| name="Test Payment Flow Customer", |
| description="Customer for payment flow integration tests", |
| metadata={"test": "true"}, |
| ) |
|
|
| |
| |
|
|
| yield customer["id"], customer |
|
|
| |
| |
|
|
|
|
| __all__ = [ |
| "stripe_mock_container", |
| "mock_stripe_api", |
| "payment_client", |
| "stripe_access_token", |
| "stripe_test_customer", |
| "setup_customer", |
| ] |
|
|