OverviewHistoryStatsSecurity
npx skills add ...
Documentation
SKILL.md
npx skills add hieutrtr/ai1-skills --skill pytest-patterns
Python backend testing patterns with pytest for FastAPI applications. Use when writing Python tests: unit tests for services and repositories, integration tests for API endpoints with httpx.AsyncClient, fixture creation, factory setup with factory_boy, async testing with pytest-asyncio, mocking strategies, and parametrized tests. Covers test organization (tests/unit, tests/integration), conftest hierarchy, and coverage requirements. Does NOT cover frontend tests (use react-testing-patterns) or E2E browser tests (use e2e-testing).
npx skills add hieutrtr/ai1-skills --skill pytest-patterns
Activate this skill when:
Do NOT use this skill for:
react-testing-patterns)e2e-testing)tdd-workflow)python-backend-expert)Naming conventions:
test_<module>.pyTest<Feature> (group related tests, no __init__)test_<action>_<expected_outcome> or test_<scenario>db_session, authenticated_client, sample_user)Marker conventions:
Run subsets: pytest -m unit, pytest -m integration, pytest -m "not slow".
Fixtures cascade: root conftest.py provides shared fixtures; subdirectory conftest files add layer-specific fixtures.
Root conftest (tests/conftest.py):
| Scope | Use For | Example |
|---|---|---|
function (default) | Isolated per-test data | db_session, sample_user |
class | Shared across test class | service_instance |
module | Shared across test file | seeded_database |
session | Shared across entire run | engine, anyio_backend |
Rules:
function scope for data isolationsession scope only for expensive, stateless resources (engine, event loop)session scope for mutable data -- tests will interfere with each otherUse factory_boy for consistent, overridable test data.
Usage in tests:
SQLAlchemy integration (for integration tests that persist to DB):
Set session in conftest:
Test FastAPI endpoints with httpx.AsyncClient against the real app, but with a test database.
Key patterns:
authenticated_client for protected endpoints, plain client for auth testing"id" in data)With asyncio_mode = "auto" in pyproject.toml, all async def test_* functions run automatically.
Common pitfalls:
sync and async fixtures carelessly -- an async fixture can only be used by async testspytest-asyncio (not anyio directly) for consistencyasync keywordsMock external services -- YES:
Mock the database -- NO:
What to mock:
| Mock | Do Not Mock |
|---|---|
HTTP APIs (use respx or unittest.mock) | Database queries |
| Email/SMS services | SQLAlchemy sessions |
| File storage (S3, GCS) | Repository methods (in integration tests) |
| Message queues (Redis, RabbitMQ) | Pydantic validation |
Time/datetime (freezegun) | FastAPI dependency injection |
| Random/UUID generation | ORM relationships |
respx for HTTP mocking:
Use @pytest.mark.parametrize for testing multiple input/output combinations:
Parametrize with IDs for readable output:
Parametrize multiple fixtures:
Minimum thresholds:
pyproject.toml configuration:
Running coverage:
Use scripts/check-test-coverage.sh to automate coverage checks with report output.
1. Testing implementation details:
2. Shared mutable state between tests:
3. Overly broad assertions:
4. Missing error path tests: Every endpoint should have tests for at least: success, not found, validation error, and unauthorized.
See references/conftest-template.py for production conftest setup.
See references/factory-template.py for factory_boy patterns.
See references/api-test-template.py for API integration test patterns.
See references/service-test-template.py for service unit test patterns.
See references/integration-test-template.py for full integration test patterns.
# pyproject.toml
[tool.pytest.ini_options]
markers = [
"unit: Unit tests (no DB, no network)",
"integration: Integration tests (real DB, real HTTP)",
"slow: Tests that take > 1 second",
]
asyncio_mode = "auto"import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from app.main import app
from app.database import get_db
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="session")
async def engine():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest.fixture
async def db_session(engine):
async with async_sessionmaker(engine, class_=AsyncSession)() as session:
yield session
await session.rollback()
@pytest.fixture
async def client(db_session):
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
app.dependency_overrides.clear()@pytest.fixture
def auth_headers():
"""Return authorization headers for a standard test user."""
token = create_test_token(user_id=1, role="member")
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
async def authenticated_client(client, auth_headers):
"""AsyncClient pre-configured with auth headers."""
client.headers.update(auth_headers)
return client
@pytest.fixture
def admin_headers():
"""Return authorization headers for an admin user."""
token = create_test_token(user_id=99, role="admin")
return {"Authorization": f"Bearer {token}"}import factory
from app.models import User, Order
class UserFactory(factory.Factory):
class Meta:
model = User
id = factory.Sequence(lambda n: n + 1)
email = factory.LazyAttribute(lambda o: f"user{o.id}@example.com")
display_name = factory.Faker("name")
role = "member"
is_active = True
class OrderFactory(factory.Factory):
class Meta:
model = Order
id = factory.Sequence(lambda n: n + 1)
user_id = factory.LazyAttribute(lambda o: UserFactory().id)
total_cents = factory.Faker("random_int", min=100, max=100000)
status = "pending"def test_user_defaults():
user = UserFactory()
assert user.is_active is True
assert user.role == "member"
def test_user_override():
admin = UserFactory(role="admin", display_name="Admin User")
assert admin.role == "admin"
def test_user_batch():
users = UserFactory.build_batch(5)
assert len(users) == 5class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
class Meta:
model = User
sqlalchemy_session = None # Set per-test via conftest
# ... fields same as above@pytest.fixture(autouse=True)
def set_factory_session(db_session):
UserFactory._meta.sqlalchemy_session = db_session
OrderFactory._meta.sqlalchemy_session = db_sessionimport pytest
from httpx import AsyncClient
class TestUsersAPI:
"""Integration tests for /api/v1/users endpoints."""
async def test_create_user_success(self, authenticated_client: AsyncClient):
response = await authenticated_client.post("/api/v1/users", json={
"email": "new@example.com",
"display_name": "New User",
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "new@example.com"
assert "id" in data
async def test_create_user_duplicate_email(self, authenticated_client, sample_user):
response = await authenticated_client.post("/api/v1/users", json={
"email": sample_user.email,
"display_name": "Duplicate",
})
assert response.status_code == 409
assert "already exists" in response.json()["detail"]
async def test_list_users_pagination(self, authenticated_client):
response = await authenticated_client.get("/api/v1/users?limit=10&cursor=0")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "next_cursor" in data
async def test_get_user_not_found(self, authenticated_client):
response = await authenticated_client.get("/api/v1/users/99999")
assert response.status_code == 404
async def test_unauthenticated_request(self, client):
response = await client.get("/api/v1/users")
assert response.status_code == 401async def test_async_service_call(db_session):
service = UserService(db_session)
user = await service.create_user(email="test@example.com", display_name="Test")
assert user.id is not None
async def test_concurrent_operations(db_session):
service = UserService(db_session)
import asyncio
results = await asyncio.gather(
service.get_user(1),
service.get_user(2),
service.get_user(3),
)
assert len(results) == 3from unittest.mock import AsyncMock, patch
async def test_send_notification(db_session):
with patch("app.services.notification.EmailClient") as mock_email:
mock_email.return_value.send = AsyncMock(return_value=True)
service = NotificationService(db_session)
result = await service.notify_user(user_id=1, message="Hello")
assert result is True
mock_email.return_value.send.assert_called_once()# BAD: Mocking the database hides real query issues
async def test_user_service(mock_db):
mock_db.execute.return_value = MockResult([user_dict]) # Don't do this
# GOOD: Use a real test database (SQLite or PostgreSQL in Docker)
async def test_user_service(db_session):
service = UserService(db_session)
user = await service.create_user(email="test@example.com", display_name="Test")
fetched = await service.get_user(user.id)
assert fetched.email == "test@example.com"import respx
from httpx import Response
@respx.mock
async def test_external_api_call():
respx.get("https://api.example.com/data").mock(
return_value=Response(200, json={"key": "value"})
)
service = ExternalDataService()
result = await service.fetch_data()
assert result["key"] == "value"@pytest.mark.parametrize("email,is_valid", [
("user@example.com", True),
("user@sub.domain.com", True),
("user+tag@example.com", True),
("", False),
("not-an-email", False),
("@missing-local.com", False),
("user@", False),
])
def test_email_validation(email, is_valid):
if is_valid:
assert validate_email(email) is True
else:
with pytest.raises(ValidationError):
validate_email(email)@pytest.mark.parametrize("status,expected_code", [
pytest.param("active", 200, id="active-user-ok"),
pytest.param("suspended", 403, id="suspended-user-forbidden"),
pytest.param("deleted", 404, id="deleted-user-not-found"),
])
async def test_user_access_by_status(authenticated_client, status, expected_code):
...@pytest.mark.parametrize("role,can_delete", [
("admin", True),
("member", False),
("viewer", False),
])
async def test_delete_permission(client, role, can_delete):
headers = {"Authorization": f"Bearer {create_test_token(role=role)}"}
response = await client.delete("/api/v1/users/1", headers=headers)
if can_delete:
assert response.status_code == 204
else:
assert response.status_code == 403[tool.coverage.run]
source = ["app"]
omit = ["app/migrations/*", "app/main.py", "app/__init__.py"]
[tool.coverage.report]
fail_under = 80
show_missing = true
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"if __name__ ==",
"@overload",
]pytest --cov=app --cov-report=term-missing --cov-report=html --cov-fail-under=80# BAD: Tests internal method calls
def test_service_calls_repo(mock_repo):
service.create_user(data)
mock_repo.insert.assert_called_once_with(...)
# GOOD: Tests observable behavior
async def test_create_user(db_session):
service = UserService(db_session)
user = await service.create_user(email="a@b.com", display_name="A")
fetched = await service.get_user(user.id)
assert fetched is not None# BAD: Module-level mutable data
users = []
def test_add_user():
users.append(User(id=1)) # Leaks to other tests
# GOOD: Use fixtures with function scope
@pytest.fixture
def users():
return [UserFactory()]# BAD
assert response.status_code == 200 # Only checks status
# GOOD
assert response.status_code == 200
data = response.json()
assert data["email"] == "test@example.com"
assert data["role"] == "member"