File size: 5,054 Bytes
942b115
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Integration test fixtures (TEST-02).

The configured DB user lacks `CREATE DATABASE` privilege, so a literal
second MySQL schema isn't available here. Instead, each test runs inside a
SAVEPOINT-nested transaction against the real configured database that is
*always rolled back* at the end of the test (SQLAlchemy's documented
"join a Session into an external transaction" recipe) -- route handlers'
internal `db.commit()` calls only release the savepoint, they never commit
the outer transaction, so nothing a test writes is ever actually persisted.
This achieves the requirement's real goal (integration tests never leave
data behind in the database other code reads) without needing a schema the
credentials can't create.

Note: the app's lifespan startup calls `seed_officer_if_missing`, which
runs against the *real* engine (via `session_scope()`), not the
savepoint-wrapped test session -- so the seed officer row is a genuine,
persistent row (idempotent: it no-ops if already present), not test data
that gets rolled back. That's intentional and matches production behavior.
"""

from __future__ import annotations

import app.auth as auth_module
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import event

from app.config import settings
from app.db.session import SessionLocal, engine, get_session
from app.main import app

# TestClient's default base URL is http://testserver -- a "Secure" cookie
# (correct for the real deployment behind an HTTPS reverse proxy) would
# never be sent back over that scheme, breaking every authenticated request
# in tests. Same fix as local dev: force it off for the test process.
auth_module.COOKIE_SECURE = False


@pytest.fixture()
def db_session():
    connection = engine.connect()
    outer_transaction = connection.begin()
    session = SessionLocal(bind=connection, autoflush=False, autocommit=False)

    nested = connection.begin_nested()

    @event.listens_for(session, "after_transaction_end")
    def _restart_savepoint(sess, transaction):
        nonlocal nested
        if not nested.is_active:
            nested = connection.begin_nested()

    try:
        yield session
    finally:
        session.close()
        outer_transaction.rollback()
        connection.close()


@pytest.fixture()
def api_client(db_session):
    def _override_get_session():
        yield db_session

    app.dependency_overrides[get_session] = _override_get_session
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.pop(get_session, None)


def login(api_client: TestClient, email: str, password: str) -> dict:
    """Logs in and returns `{"cookies": {...}}` for use as
    `api_client.get(url, cookies=identity["cookies"])` -- lets a single
    TestClient (one shared DB transaction) act as several logged-in
    identities at once without re-running the (expensive, model-loading)
    lifespan per identity.
    """
    res = api_client.post("/api/auth/login", json={"email": email, "password": password})
    assert res.status_code == 200, res.text
    token = res.cookies.get("session_token")
    return {"cookies": {"session_token": token}, "user": res.json()}


@pytest.fixture()
def officer_identity(api_client):
    return login(api_client, settings.SEED_OFFICER_EMAIL, settings.SEED_OFFICER_PASSWORD)


def create_client(api_client: TestClient, officer_identity: dict, *, email: str, name: str, balance: float = 5000) -> dict:
    res = api_client.post(
        "/api/officer/clients",
        json={
            "name": name,
            "email": email,
            "temp_password": "TempPass123!",
            "starting_balance": balance,
            "account_type": "checking",
        },
        cookies=officer_identity["cookies"],
    )
    assert res.status_code == 200, res.text
    body = res.json()
    identity = login(api_client, email, "TempPass123!")
    identity["client_id"] = body["client"]["id"]
    return identity


@pytest.fixture()
def two_clients(api_client, officer_identity):
    """Two client identities with distinct starting balances, per the
    role-separation testing requirement (one seeded officer, two seeded
    clients)."""
    client_a = create_client(
        api_client, officer_identity, email="fixture.alpha@fakebankmail.com", name="Fixture Alpha", balance=5000
    )
    client_b = create_client(
        api_client, officer_identity, email="fixture.beta@fakebankmail.com", name="Fixture Beta", balance=3000
    )
    return client_a, client_b


TRANSFER_FRAUD_PATTERN = {
    "step": 5,
    "type": "TRANSFER",
    "amount": 181.0,
    "nameOrig": "CTEST_FRAUD_ORIG",
    "oldbalanceOrg": 181.0,
    "newbalanceOrig": 0.0,
    "nameDest": "CTEST_FRAUD_DEST",
    "oldbalanceDest": 0.0,
    "newbalanceDest": 0.0,
}

PAYMENT_LEGIT_PATTERN = {
    "step": 5,
    "type": "PAYMENT",
    "amount": 50.0,
    "nameOrig": "CTEST_LEGIT_ORIG",
    "oldbalanceOrg": 5000.0,
    "newbalanceOrig": 4950.0,
    "nameDest": "MTEST_MERCHANT",
    "oldbalanceDest": 0.0,
    "newbalanceDest": 0.0,
}