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",
]