Integration Cheatsheet 08 — Testing the Full Stack

import pytest, pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import event
from testcontainers.postgres import PostgresContainer

from src.myapp.main import app
from src.myapp.db import Base
from src.myapp.deps import get_db

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:17") as pg:
        yield pg

@pytest_asyncio.fixture(scope="session")
async def engine(postgres):
    url = postgres.get_connection_url().replace("psycopg2", "asyncpg")
    eng = create_async_engine(url)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    await eng.dispose()

@pytest_asyncio.fixture
async def db_session(engine):
    connection = await engine.connect()
    transaction = await connection.begin()
    async with AsyncSession(bind=connection, expire_on_commit=False) as session:
        nested = await session.begin_nested()
        
        @event.listens_for(session.sync_session, "after_transaction_end")
        def end_savepoint(session, transaction):
            nonlocal nested
            if not nested.is_active:
                nested = session.begin_nested()
        
        yield session
    
    await transaction.rollback()
    await connection.close()

@pytest_asyncio.fixture
async def client(db_session):
    async def override_db():
        yield db_session
    
    app.dependency_overrides[get_db] = override_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

import factory
from src.myapp.models import User

class UserFactory(factory.Factory):
    class Meta:
        model = User
    
    email = factory.Sequence(lambda n: f"user{n}@example.com")
    hashed_password = "fake-hash"
    is_active = True

@pytest.mark.anyio
async def test_create_user(client):
    r = await client.post("/users", json={
        "email": "[email protected]",
        "password": "abcd1234",
    })
    assert r.status_code == 201
    body = r.json()
    assert body["email"] == "[email protected]"
    assert "password" not in body

@pytest.mark.anyio
@pytest.mark.parametrize("payload, status", [
    ({"email": "[email protected]", "password": "abcd1234"}, 201),
    ({"email": "invalid", "password": "abcd1234"}, 422),
    ({"email": "[email protected]", "password": "short"}, 422),
    ({"email": "[email protected]"}, 422),
])
async def test_validation(client, payload, status):
    r = await client.post("/users", json=payload)
    assert r.status_code == status

@pytest.mark.anyio
async def test_login_flow(client, db_session):
    user = UserFactory.build(hashed_password=hash_password("abcd1234"))
    db_session.add(user); await db_session.commit()
    
    r = await client.post("/login", data={"username": user.email, "password": "abcd1234"})
    assert r.status_code == 200
    token = r.json()["access_token"]
    
    r = await client.get("/me", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    assert r.json()["email"] == user.email

@pytest.mark.anyio
async def test_email_unique(client, db_session):
    UserFactory.build(email="[email protected]").__dict__
    db_session.add(UserFactory.build(email="[email protected]"))
    await db_session.commit()
    
    r = await client.post("/users", json={"email": "[email protected]", "password": "abcd1234"})
    assert r.status_code == 409

@pytest.mark.anyio
async def test_no_drift(engine):
    from alembic.autogenerate import compare_metadata
    from alembic.migration import MigrationContext
    
    async with engine.connect() as conn:
        def check(sync_conn):
            ctx = MigrationContext.configure(sync_conn)
            return compare_metadata(ctx, Base.metadata)
        diff = await conn.run_sync(check)
        assert diff == [], f"Drift: {diff}"

def test_migrations_round_trip(postgres):
    from alembic.config import Config
    from alembic import command
    
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", postgres.get_connection_url())
    
    command.upgrade(cfg, "head")
    command.downgrade(cfg, "base")
    command.upgrade(cfg, "head")

import respx

@pytest_asyncio.fixture
async def mock_stripe():
    with respx.mock(base_url="https://api.stripe.com") as m:
        m.post("/v1/charges").respond(json={"id": "ch_test"})
        yield m

@pytest.mark.anyio
async def test_charge(client, mock_stripe):
    r = await client.post("/payments", json={...})
    assert r.status_code == 200
    assert mock_stripe.calls.call_count == 1

@pytest_asyncio.fixture
async def admin_client(client, db_session):
    admin = UserFactory.build(role="admin")
    db_session.add(admin); await db_session.commit()
    
    token = make_access_token(admin.id)
    client.headers["Authorization"] = f"Bearer {token}"
    yield client

pytest --cov=src --cov-report=term --cov-fail-under=80

services:
  postgres:
    image: postgres:17
    env: { POSTGRES_PASSWORD: test }
    ports: ["5432:5432"]

steps:
  - run: uv sync --frozen
  - run: uv run alembic upgrade head
  - run: uv run pytest --cov=src