lab_ai/.agents/skills/pytest/references/api-testing.md

API Testing

Table of Contents

• FastAPI Testing
• Flask Testing
• Test Factories
• Authentication Testing
• Database Testing

FastAPI Testing

Setup

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlmodel import SQLModel

from app.main import app
from app.core.database import get_session

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

test_engine = create_async_engine(TEST_DATABASE_URL)
TestSession = async_sessionmaker(test_engine, class_=AsyncSession)


@pytest.fixture(scope="function")
async def db_session():
    """Fresh database for each test."""
    async with test_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    async with TestSession() as session:
        yield session

    async with test_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)


@pytest.fixture
async def client(db_session):
    """Test client with dependency override."""
    async def override_session():
        yield db_session

    app.dependency_overrides[get_session] = override_session

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac

    app.dependency_overrides.clear()

Basic Endpoint Tests

import pytest
from httpx import AsyncClient

class TestUsers:
    @pytest.mark.asyncio
    async def test_create_user(self, client: AsyncClient):
        response = await client.post(
            "/api/users",
            json={"email": "test@example.com", "name": "Test User"},
        )

        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "test@example.com"
        assert "id" in data

    @pytest.mark.asyncio
    async def test_get_user(self, client: AsyncClient, test_user):
        response = await client.get(f"/api/users/{test_user.id}")

        assert response.status_code == 200
        assert response.json()["id"] == test_user.id

    @pytest.mark.asyncio
    async def test_get_user_not_found(self, client: AsyncClient):
        response = await client.get("/api/users/99999")

        assert response.status_code == 404

    @pytest.mark.asyncio
    async def test_update_user(self, client: AsyncClient, test_user, auth_headers):
        response = await client.patch(
            f"/api/users/{test_user.id}",
            json={"name": "Updated Name"},
            headers=auth_headers,
        )

        assert response.status_code == 200
        assert response.json()["name"] == "Updated Name"

    @pytest.mark.asyncio
    async def test_delete_user(self, client: AsyncClient, test_user, admin_headers):
        response = await client.delete(
            f"/api/users/{test_user.id}",
            headers=admin_headers,
        )

        assert response.status_code == 204

Testing Query Parameters

@pytest.mark.asyncio
async def test_list_users_pagination(self, client: AsyncClient, auth_headers):
    response = await client.get(
        "/api/users",
        params={"page": 1, "per_page": 10, "sort": "-created_at"},
        headers=auth_headers,
    )

    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data
    assert data["page"] == 1

@pytest.mark.asyncio
async def test_search_users(self, client: AsyncClient, auth_headers):
    response = await client.get(
        "/api/users",
        params={"q": "john", "active": True},
        headers=auth_headers,
    )

    assert response.status_code == 200

Testing File Uploads

import io

@pytest.mark.asyncio
async def test_upload_file(self, client: AsyncClient, auth_headers):
    file_content = b"Hello, World!"
    files = {"file": ("test.txt", io.BytesIO(file_content), "text/plain")}

    response = await client.post(
        "/api/files/upload",
        files=files,
        headers=auth_headers,
    )

    assert response.status_code == 201
    assert response.json()["filename"] == "test.txt"

Flask Testing

Setup

import pytest
from app import create_app
from app.extensions import db

@pytest.fixture
def app():
    """Create application for testing."""
    app = create_app("testing")
    app.config["TESTING"] = True
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"

    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()

@pytest.fixture
def client(app):
    """Test client."""
    return app.test_client()

@pytest.fixture
def runner(app):
    """CLI test runner."""
    return app.test_cli_runner()

Flask Test Examples

def test_create_user(client):
    response = client.post(
        "/api/users",
        json={"email": "test@example.com", "name": "Test"},
    )

    assert response.status_code == 201

def test_get_user(client, test_user):
    response = client.get(f"/api/users/{test_user.id}")

    assert response.status_code == 200
    assert response.json["email"] == test_user.email

def test_protected_route(client, auth_token):
    response = client.get(
        "/api/protected",
        headers={"Authorization": f"Bearer {auth_token}"},
    )

    assert response.status_code == 200

Test Factories

Using factory_boy

# tests/factories.py
import factory
from faker import Faker
from app.models import User, Project

fake = Faker()


class UserFactory(factory.Factory):
    class Meta:
        model = User

    email = factory.LazyAttribute(lambda _: fake.email())
    name = factory.LazyAttribute(lambda _: fake.name())
    hashed_password = "hashed_password"
    is_active = True


class ProjectFactory(factory.Factory):
    class Meta:
        model = Project

    name = factory.LazyAttribute(lambda _: fake.company())
    description = factory.LazyAttribute(lambda _: fake.sentence())
    owner_id = None


# Async helper
async def create_user(session, **kwargs) -> User:
    user = UserFactory(**kwargs)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user


async def create_users(session, count: int = 5, **kwargs) -> list[User]:
    users = [UserFactory(**kwargs) for _ in range(count)]
    session.add_all(users)
    await session.commit()
    return users

Using Factories in Tests

from tests.factories import create_user, create_users

@pytest.mark.asyncio
async def test_list_users(client, db_session, auth_headers):
    # Create 20 users
    await create_users(db_session, count=20)

    response = await client.get("/api/users", headers=auth_headers)

    assert response.status_code == 200
    assert response.json()["total"] >= 20

@pytest.mark.asyncio
async def test_filter_active_users(client, db_session, auth_headers):
    await create_user(db_session, is_active=True)
    await create_user(db_session, is_active=False)

    response = await client.get(
        "/api/users",
        params={"active": True},
        headers=auth_headers,
    )

    data = response.json()
    assert all(u["is_active"] for u in data["items"])

Authentication Testing

Auth Fixtures

from app.core.security import create_access_token, hash_password

@pytest.fixture
async def test_user(db_session):
    """Create a regular test user."""
    user = User(
        email="test@example.com",
        hashed_password=hash_password("password123"),
        is_active=True,
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user

@pytest.fixture
async def admin_user(db_session):
    """Create an admin user."""
    user = User(
        email="admin@example.com",
        hashed_password=hash_password("admin123"),
        is_active=True,
        is_admin=True,
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user

@pytest.fixture
def auth_headers(test_user):
    """Headers with valid auth token."""
    token = create_access_token({"sub": str(test_user.id)})
    return {"Authorization": f"Bearer {token}"}

@pytest.fixture
def admin_headers(admin_user):
    """Headers with admin auth token."""
    token = create_access_token({"sub": str(admin_user.id)})
    return {"Authorization": f"Bearer {token}"}

Auth Test Examples

class TestAuth:
    @pytest.mark.asyncio
    async def test_login_success(self, client, test_user):
        response = await client.post(
            "/api/auth/login",
            data={"username": test_user.email, "password": "password123"},
        )

        assert response.status_code == 200
        assert "access_token" in response.json()

    @pytest.mark.asyncio
    async def test_login_wrong_password(self, client, test_user):
        response = await client.post(
            "/api/auth/login",
            data={"username": test_user.email, "password": "wrong"},
        )

        assert response.status_code == 401

    @pytest.mark.asyncio
    async def test_protected_endpoint_no_auth(self, client):
        response = await client.get("/api/protected")

        assert response.status_code == 401

    @pytest.mark.asyncio
    async def test_admin_endpoint_regular_user(self, client, auth_headers):
        response = await client.delete(
            "/api/admin/users/1",
            headers=auth_headers,
        )

        assert response.status_code == 403

Database Testing

Transaction Rollback Pattern

@pytest.fixture
async def db_session(engine):
    """Each test runs in a transaction that rolls back."""
    async with engine.connect() as conn:
        await conn.begin()

        async with AsyncSession(bind=conn) as session:
            yield session

        await conn.rollback()

Testing Repository Layer

from app.repositories import UserRepository

class TestUserRepository:
    @pytest.mark.asyncio
    async def test_create(self, db_session):
        repo = UserRepository(db_session)

        user = await repo.create({
            "email": "test@example.com",
            "hashed_password": "hash",
        })

        assert user.id is not None
        assert user.email == "test@example.com"

    @pytest.mark.asyncio
    async def test_get_by_email(self, db_session, test_user):
        repo = UserRepository(db_session)

        user = await repo.get_by_email(test_user.email)

        assert user.id == test_user.id

    @pytest.mark.asyncio
    async def test_soft_delete(self, db_session, test_user):
        repo = UserRepository(db_session)

        await repo.soft_delete(test_user.id)

        user = await repo.get_by_id(test_user.id)
        assert user.status == 0  # Soft deleted