Skip to main content

Testing

Test Setup​

Basic Test Configuration​

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import pytest

from app.main import app
from app.database import get_db, Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

@pytest.fixture
def db():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

Test Database with pytest​

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="session")
def db_engine():
    engine = create_engine("sqlite:///./test.db")
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)

@pytest.fixture
def db_session(db_engine):
    SessionLocal = sessionmaker(bind=db_engine)
    session = SessionLocal()
    yield session
    session.close()

@pytest.fixture
def client(db_session):
    def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

Unit Tests​

Testing Endpoints​

def test_create_user(client):
    response = client.post(
        "/users/",
        json={"username": "testuser", "email": "test@example.com", "password": "testpass"}
    )
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"
    assert data["email"] == "test@example.com"
    assert "id" in data

def test_read_user(client, db_session):
    # Create user first
    user = User(username="testuser", email="test@example.com")
    db_session.add(user)
    db_session.commit()
    db_session.refresh(user)

    response = client.get(f"/users/{user.id}")
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "testuser"

def test_user_not_found(client):
    response = client.get("/users/999")
    assert response.status_code == 404
    assert response.json()["detail"] == "User not found"

Testing Authentication​

def test_login_success(client):
    # Create user first
    client.post("/users/", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "testpass"
    })

    response = client.post("/token", data={
        "username": "testuser",
        "password": "testpass"
    })
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

def test_login_invalid_credentials(client):
    response = client.post("/token", data={
        "username": "wronguser",
        "password": "wrongpass"
    })
    assert response.status_code == 401

def test_protected_route_without_token(client):
    response = client.get("/users/me")
    assert response.status_code == 401

def test_protected_route_with_token(client):
    # Login and get token
    login_response = client.post("/token", data={
        "username": "testuser",
        "password": "testpass"
    })
    token = login_response.json()["access_token"]

    # Use token to access protected route
    response = client.get(
        "/users/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200

Integration Tests​

Testing Database Operations​

def test_user_crud_operations(client, db_session):
    # Create
    create_response = client.post("/users/", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "testpass"
    })
    assert create_response.status_code == 201
    user_id = create_response.json()["id"]

    # Read
    read_response = client.get(f"/users/{user_id}")
    assert read_response.status_code == 200

    # Update
    update_response = client.put(f"/users/{user_id}", json={
        "username": "updateduser",
        "email": "updated@example.com"
    })
    assert update_response.status_code == 200
    assert update_response.json()["username"] == "updateduser"

    # Delete
    delete_response = client.delete(f"/users/{user_id}")
    assert delete_response.status_code == 200

    # Verify deletion
    read_after_delete = client.get(f"/users/{user_id}")
    assert read_after_delete.status_code == 404

Testing Relationships​

def test_user_items_relationship(client, db_session):
    # Create user
    user_response = client.post("/users/", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "testpass"
    })
    user_id = user_response.json()["id"]

    # Create item for user
    item_response = client.post(f"/users/{user_id}/items/", json={
        "title": "Test Item",
        "description": "A test item"
    })
    assert item_response.status_code == 201

    # Get user items
    items_response = client.get(f"/users/{user_id}/items/")
    assert items_response.status_code == 200
    items = items_response.json()
    assert len(items) == 1
    assert items[0]["title"] == "Test Item"

Test Fixtures​

Data Fixtures​

@pytest.fixture
def sample_user(db_session):
    user = User(
        username="testuser",
        email="test@example.com",
        hashed_password="hashedpass"
    )
    db_session.add(user)
    db_session.commit()
    db_session.refresh(user)
    return user

@pytest.fixture
def sample_items(db_session, sample_user):
    items = [
        Item(title="Item 1", description="Description 1", owner_id=sample_user.id),
        Item(title="Item 2", description="Description 2", owner_id=sample_user.id)
    ]
    db_session.add_all(items)
    db_session.commit()
    return items

def test_get_user_items(client, sample_user, sample_items):
    response = client.get(f"/users/{sample_user.id}/items/")
    assert response.status_code == 200
    data = response.json()
    assert len(data) == 2

Authentication Fixtures​

@pytest.fixture
def auth_headers(client, sample_user):
    response = client.post("/token", data={
        "username": sample_user.username,
        "password": "testpass"
    })
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

def test_protected_endpoint(client, auth_headers):
    response = client.get("/users/me", headers=auth_headers)
    assert response.status_code == 200

Mocking​

Mocking External Services​

import pytest
from unittest.mock import patch, Mock

@patch('app.services.email_service.send_email')
def test_user_registration_sends_email(mock_send_email, client):
    mock_send_email.return_value = True

    response = client.post("/users/", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "testpass"
    })

    assert response.status_code == 201
    mock_send_email.assert_called_once()
    call_args = mock_send_email.call_args
    assert call_args[0][0] == "test@example.com"  # recipient
    assert "Welcome" in call_args[0][1]  # subject

Mocking Database Operations​

@patch('app.crud.get_user')
def test_user_endpoint_with_mock(mock_get_user, client):
    mock_user = Mock()
    mock_user.id = 1
    mock_user.username = "testuser"
    mock_user.email = "test@example.com"
    mock_get_user.return_value = mock_user

    response = client.get("/users/1")
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "testuser"

Async Testing​

Testing Async Endpoints​

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/users/")
    assert response.status_code == 200

@pytest.mark.asyncio
async def test_async_user_creation():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/users/", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "testpass"
        })
    assert response.status_code == 201

Performance Testing​

Load Testing with pytest-benchmark​

import pytest

def test_user_creation_performance(client, benchmark):
    def create_user():
        return client.post("/users/", json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "testpass"
        })

    result = benchmark(create_user)
    assert result.status_code == 201

def test_user_list_performance(client, benchmark):
    # Create some users first
    for i in range(100):
        client.post("/users/", json={
            "username": f"user{i}",
            "email": f"user{i}@example.com",
            "password": "testpass"
        })

    result = benchmark(lambda: client.get("/users/"))
    assert result.status_code == 200

Test Coverage​

Coverage Configuration​

# pytest.ini
[tool:pytest]
addopts = --cov=app --cov-report=html --cov-report=term-missing
testpaths = tests

Running Tests with Coverage​

# Install coverage
pip install pytest-cov

# Run tests with coverage
pytest --cov=app --cov-report=html

# Generate coverage report
coverage html