Spaces:
Sleeping
Sleeping
File size: 8,414 Bytes
aef804e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 | """
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()
# Save original values
original_api_base = stripe.api_base
original_api_key = stripe.api_key
# Configure stripe to use mock
stripe.api_base = mock_url
stripe.api_key = "sk_test_12345" # Valid-looking test key for stripe-mock
yield
# Restore original values
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
"""
# Use in-memory SQLite for tests
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
echo=False,
)
# Import all models
from accounting.models import (
Account, Transaction, JournalEntry, CategorizationProposal,
Entity, Bill, Invoice, Document, TaxNexus, FinancialClose,
CategorizationRule, Budget
)
from sqlalchemy import Table, Column, String, MetaData
# Create metadata for dummy FK tables
metadata = MetaData()
# Create dummy tables for FK references
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)
)
# Create dummy tables first
for table in [workspaces_table, service_projects_table, service_milestones_table, users_table]:
table.create(engine, checkfirst=True)
# Now create accounting tables
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)
# Create session factory
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create and yield session
session = TestingSessionLocal()
try:
yield session
finally:
session.close()
# Drop all tables to clean up
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
"""
# Import here to avoid import errors if FastAPI routes don't exist
try:
from fastapi.testclient import TestClient
from integrations.stripe_routes import router
# Create test app with 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
# Create Stripe customer
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"},
)
# In a real implementation, you would also store this in your database
# For now, we just return the Stripe customer info
yield customer["id"], customer
# Cleanup: Stripe customer will be cleaned up when mock server stops
# No manual cleanup needed for mock data
__all__ = [
"stripe_mock_container",
"mock_stripe_api",
"payment_client",
"stripe_access_token",
"stripe_test_customer",
"setup_customer",
]
|