Testing¶
This guide covers testing strategies and practices for the 2Sigma Backend.
Overview¶
The project uses pytest with async support for testing. Tests ensure functionality, catch regressions, and serve as documentation.
Running Tests¶
Run All Tests¶
Run with Verbose Output¶
Run Specific Tests¶
# Run specific file
pytest tests/test_users.py
# Run specific test
pytest tests/test_users.py::test_create_user
# Run tests matching pattern
pytest -k "user"
Run with Coverage¶
Test Structure¶
Test Organization¶
tests/
├── conftest.py # Shared fixtures
├── test_auth.py # Authentication tests
├── test_users.py # User endpoint tests
├── test_courses.py # Course tests
└── test_chat.py # Chat functionality tests
Basic Test Example¶
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
"""Test user registration"""
response = await client.post(
"/api/v1/auth/register",
json={
"email": "test@example.com",
"password": "testpass123",
"full_name": "Test User"
}
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
Common Test Patterns¶
Testing Authentication¶
@pytest.mark.asyncio
async def test_protected_endpoint(client: AsyncClient, test_user_token: str):
"""Test endpoint requiring authentication"""
response = await client.get(
"/api/v1/users/me",
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 200
Testing Database Operations¶
@pytest.mark.asyncio
async def test_create_course(db_session: AsyncSession):
"""Test course creation in database"""
from app import crud, schemas
course_in = schemas.CourseCreate(
title="Test Course",
code="TEST101",
university_id=1
)
course = await crud.course.create(db_session, obj_in=course_in)
assert course.id is not None
assert course.title == "Test Course"
Testing Validation¶
@pytest.mark.asyncio
async def test_invalid_email(client: AsyncClient):
"""Test registration with invalid email"""
response = await client.post(
"/api/v1/auth/register",
json={
"email": "notanemail",
"password": "testpass123",
"full_name": "Test"
}
)
assert response.status_code == 422 # Validation error
Testing Error Handling¶
@pytest.mark.asyncio
async def test_not_found(client: AsyncClient, test_user_token: str):
"""Test 404 for non-existent resource"""
response = await client.get(
"/api/v1/courses/99999",
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 404
Fixtures¶
Common Fixtures (conftest.py)¶
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from app.main import app
from app.database import get_db
@pytest.fixture
async def client():
"""Async HTTP client for testing"""
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
@pytest.fixture
async def db_session():
"""Database session for testing"""
# Create test database session
engine = create_async_engine("postgresql+asyncpg://...")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSession(engine) as session:
yield session
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
async def test_user(db_session: AsyncSession):
"""Create a test user"""
from app import crud, schemas
user_in = schemas.UserCreate(
email="test@example.com",
password="testpass123",
full_name="Test User"
)
user = await crud.user.create(db_session, obj_in=user_in)
return user
@pytest.fixture
async def test_user_token(test_user):
"""Generate access token for test user"""
from app.core.security import create_access_token
token = create_access_token(subject=str(test_user.id))
return token
Testing Best Practices¶
1. Isolate Tests¶
Each test should be independent:
# Good - creates own data
async def test_delete_user(db_session):
user = await create_test_user(db_session)
await crud.user.delete(db_session, id=user.id)
assert await crud.user.get(db_session, id=user.id) is None
# Bad - depends on other tests
async def test_delete_user(db_session):
# Assumes user with ID 1 exists from previous test
await crud.user.delete(db_session, id=1)
2. Use Descriptive Names¶
3. Test One Thing¶
# Good - focused test
async def test_user_registration_returns_tokens():
response = await register_user()
assert "access_token" in response
assert "refresh_token" in response
# Bad - tests multiple things
async def test_user_flow():
# Tests registration, login, profile update, etc.
...
4. Use Factories for Test Data¶
from faker import Faker
fake = Faker()
def make_user_data():
return {
"email": fake.email(),
"password": fake.password(),
"full_name": fake.name()
}
async def test_many_users(client):
for _ in range(10):
response = await client.post(
"/api/v1/auth/register",
json=make_user_data()
)
assert response.status_code == 200
Mocking External Services¶
Mocking LLM Calls¶
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
@patch('app.services.llm_service.llm')
async def test_chat_message(mock_llm, client, test_user_token):
"""Test chat without calling actual LLM"""
mock_llm.astream.return_value = AsyncMock()
mock_llm.astream.return_value.__aiter__.return_value = [
"Hello, how can I help you?"
]
response = await client.post(
"/api/v1/chat/sessions/1/messages",
json={"content": "Hello"},
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 200
Performance Testing¶
Testing Response Times¶
import time
@pytest.mark.asyncio
async def test_endpoint_performance(client, test_user_token):
"""Ensure endpoint responds within acceptable time"""
start = time.time()
response = await client.get(
"/api/v1/users/me",
headers={"Authorization": f"Bearer {test_user_token}"}
)
duration = time.time() - start
assert response.status_code == 200
assert duration < 0.5 # Should respond in < 500ms
Integration Tests¶
Test complete user flows:
@pytest.mark.asyncio
async def test_enrollment_flow(client):
"""Test complete enrollment process"""
# 1. Register user
register_response = await client.post(
"/api/v1/auth/register",
json={"email": "student@test.com", "password": "pass123", "full_name": "Student"}
)
token = register_response.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
# 2. List available courses
courses_response = await client.get("/api/v1/courses/")
assert courses_response.status_code == 200
# 3. Enroll in course
enroll_response = await client.post(
"/api/v1/enrollments/",
json={"course_offering_id": 1},
headers=headers
)
assert enroll_response.status_code == 200
# 4. Verify enrollment
my_enrollments = await client.get("/api/v1/enrollments/my-enrollments", headers=headers)
assert len(my_enrollments.json()) == 1
Continuous Integration¶
GitHub Actions Example¶
# .github/workflows/test.yml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:14
env:
POSTGRES_PASSWORD: postgres
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v2
- uses: actions/setup-python@v2
with:
python-version: '3.12'
- name: Install dependencies
run: |
pip install -r requirements.txt
- name: Run tests
env:
DATABASE_URL: postgresql+asyncpg://postgres:postgres@localhost/test_db
run: pytest --cov=app --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v2
Troubleshooting¶
Tests Hanging¶
Check for:
- Missing @pytest.mark.asyncio decorator
- Unclosed database sessions
- Unresolved async operations
Database Errors¶
Ensure: - Test database is created - Migrations are applied - Fixtures properly clean up
Import Errors¶
Verify:
- Virtual environment is activated
- Dependencies are installed
- PYTHONPATH includes project root