142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
"""Model-level integration tests for Wallets and Transaction ledger.
|
|
|
|
docs/planning/economy.md US-1 (task 8343bc0a730f).
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.config import settings
|
|
from app.main import app
|
|
from app.models.wallet import Wallet, Transaction, TransactionType
|
|
from app.models.user import User
|
|
|
|
|
|
def _register(client: TestClient, username: str) -> int:
|
|
resp = client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"username": username,
|
|
"email": f"{username}@example.com",
|
|
"password": "secret123",
|
|
"date_of_birth": "1990-01-01",
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
headers = {"Authorization": f"Bearer {resp.json()['access_token']}"}
|
|
me = client.get("/api/users/me", headers=headers)
|
|
assert me.status_code == 200, me.text
|
|
return me.json()["id"]
|
|
|
|
|
|
def _session_factory():
|
|
engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
|
|
return engine, async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wallet_creation_and_defaults():
|
|
client = TestClient(app)
|
|
uid = _register(client, "wallet_user_1")
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
wallet = Wallet(user_id=uid)
|
|
session.add(wallet)
|
|
await session.commit()
|
|
await session.refresh(wallet)
|
|
|
|
assert wallet.id is not None
|
|
assert wallet.user_id == uid
|
|
assert wallet.balance == 0
|
|
assert wallet.created_at is not None
|
|
assert wallet.updated_at is not None
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wallet_unique_user_id():
|
|
client = TestClient(app)
|
|
uid = _register(client, "wallet_user_2")
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
w1 = Wallet(user_id=uid)
|
|
session.add(w1)
|
|
await session.commit()
|
|
|
|
async with sf() as session:
|
|
w2 = Wallet(user_id=uid)
|
|
session.add(w2)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wallet_non_negative_balance_constraint():
|
|
client = TestClient(app)
|
|
uid = _register(client, "wallet_user_3")
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
w = Wallet(user_id=uid, balance=-50)
|
|
session.add(w)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transaction_ledger_and_idempotency():
|
|
client = TestClient(app)
|
|
uid = _register(client, "wallet_user_4")
|
|
engine, sf = _session_factory()
|
|
try:
|
|
async with sf() as session:
|
|
wallet = Wallet(user_id=uid, balance=500)
|
|
session.add(wallet)
|
|
await session.commit()
|
|
await session.refresh(wallet)
|
|
|
|
tx1 = Transaction(
|
|
wallet_id=wallet.id,
|
|
type=TransactionType.credit_purchase,
|
|
amount=500,
|
|
balance_after=500,
|
|
reference_type="purchase_session",
|
|
reference_id="sess_123",
|
|
idempotency_key="idemp_key_001",
|
|
metadata_json={"payment_method": "simulated"},
|
|
)
|
|
session.add(tx1)
|
|
await session.commit()
|
|
await session.refresh(tx1)
|
|
|
|
assert tx1.id is not None
|
|
assert tx1.type == TransactionType.credit_purchase
|
|
assert tx1.amount == 500
|
|
assert tx1.balance_after == 500
|
|
|
|
# Duplicate idempotency_key must raise IntegrityError
|
|
async with sf() as session:
|
|
tx2 = Transaction(
|
|
wallet_id=wallet.id,
|
|
type=TransactionType.credit_purchase,
|
|
amount=500,
|
|
balance_after=1000,
|
|
idempotency_key="idemp_key_001",
|
|
)
|
|
session.add(tx2)
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
finally:
|
|
await engine.dispose()
|