| name | pytest |
| description | pytest deep dive: fixtures, parametrize, markers, mocking, async tests, coverage, conftest patterns, and CI integration |
pytest Skill
When to activate
- Writing Python tests beyond basic
assert statements
- Setting up fixtures for DB sessions, HTTP clients, or external services
- Parametrising tests to run the same logic across multiple inputs
- Mocking external services or environment variables
- Running async tests (FastAPI, asyncpg, aiohttp)
- Configuring pytest for a project (
pyproject.toml, conftest.py)
- Setting up coverage reporting
When NOT to use
- unittest-style tests in an existing codebase — use pytest's unittest compatibility mode
- Load testing — use Locust or k6 instead
- Property-based testing — use Hypothesis alongside pytest
Instructions
Project setup
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_functions = ["test_*"]
asyncio_mode = "auto"
addopts = [
"--strict-markers",
"-ra",
"--tb=short",
]
markers = [
"integration: marks tests that require a real database or external service",
"slow: marks tests that take more than 1 second",
"smoke: marks critical path tests for quick verification",
]
[tool.coverage.run]
source = ["src"]
omit = ["tests/*", "*/migrations/*"]
[tool.coverage.report]
fail_under = 80
pip install pytest pytest-asyncio pytest-cov pytest-mock httpx
Fixtures — the core concept
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
@pytest.fixture(scope="session")
def engine():
"""One engine per test session — expensive to create."""
return create_async_engine("postgresql+asyncpg://postgres:postgres@localhost/testdb")
@pytest.fixture(scope="function")
async def db(engine):
"""Fresh transaction per test — rolls back after each test."""
async with engine.begin() as conn:
async with AsyncSession(bind=conn) as session:
yield session
await session.rollback()
@pytest.fixture
async def client(db):
"""Async HTTP client for FastAPI with test DB injected."""
from app.main import app
from app.api.deps import get_db
async def ():
db
app.dependency_overrides[get_db] = override_get_db
AsyncClient(transport=ASGITransport(app=app), base_url=) c:
c
app.dependency_overrides.clear()
():
tests.factories UserFactory
UserFactory(session=db)
Async tests
import pytest
async def test_create_user(client: AsyncClient):
resp = await client.post("/users", json={"email": "alice@example.com", "name": "Alice"})
assert resp.status_code == 201
assert resp.json()["email"] == "alice@example.com"
@pytest.mark.asyncio
async def test_fetch_user(db: AsyncSession):
user = await db.get(User, 1)
assert user is not None
Parametrize — test multiple cases cleanly
import pytest
@pytest.mark.parametrize("email,expected_status", [
("valid@example.com", 201),
("not-an-email", 422),
("", 422),
("a" * 500 + "@b.com", 422),
])
async def test_signup_validates_email(client, email, expected_status):
resp = await client.post("/auth/signup", json={"email": email, "password": "password123"})
assert resp.status_code == expected_status
@pytest.mark.parametrize("discount,total,expected", [
(0, 100.0, 100.0),
(10, 100.0, 90.0),
(100, 100.0, 0.0),
], ids=["no-discount", "10pct", "full"])
def test_apply_discount(discount, total, expected):
assert apply_discount(total, discount) == expected
Mocking with pytest-mock
def test_send_email_on_signup(mocker, client):
mock_send = mocker.patch("app.services.email.send_email")
client.post("/auth/signup", json={"email": "a@b.com", "password": "pass123"})
mock_send.assert_called_once_with(
to="a@b.com",
subject="Welcome!",
)
def test_feature_flag(monkeypatch):
monkeypatch.setenv("ENABLE_BETA", "true")
assert is_beta_enabled() is True
def test_stripe_charge(mocker):
mock_charge = mocker.patch("stripe.PaymentIntent.create")
mock_charge.return_value = {"id": "pi_test", "status": "succeeded"}
result = charge_card("tok_visa", 1000)
assert result.payment_intent_id == "pi_test"
mock_charge.assert_called_once_with(amount=1000, currency="usd")
async def test_external_api(mocker):
mocker.patch("app.clients.github.fetch_repo", return_value={"stars": 100})
result = await get_repo_stats()
result.stars ==
Markers and selective test runs
@pytest.mark.integration
async def test_real_database_operation(db):
...
@pytest.mark.slow
def test_complex_algorithm():
...
@pytest.mark.smoke
async def test_health_check(client):
resp = await client.get("/health")
assert resp.status_code == 200
pytest -m smoke
pytest -m "not slow"
pytest tests/test_users.py
pytest tests/test_users.py::test_create_user
pytest -k "email"
pytest -x
pytest --durations=10
Coverage
pytest --cov=src --cov-report=html --cov-report=term-missing
pytest --cov=src --cov-fail-under=80
pytest --cov=src/services/users tests/test_users.py
def legacy_function():
...
if TYPE_CHECKING:
from mymodule import MyType
conftest.py patterns
@pytest.fixture(autouse=True)
def reset_redis(redis_client):
yield
redis_client.flushdb()
@pytest.fixture(scope="module")
def seed_data(db):
"""Seed DB once per module — faster than per-test."""
UserFactory.create_batch(10, session=db)
yield
db.query(User).delete()
@pytest.fixture
def auth_headers(user):
token = create_token(user.id)
return {"Authorization": f"Bearer {token}"}
Snapshot testing
from syrupy import SnapshotAssertion
def test_user_response(snapshot: SnapshotAssertion, client):
resp = client.get("/users/1")
assert resp.json() == snapshot
Example
User: Write tests for a FastAPI POST /auth/login endpoint: correct credentials return 200 + JWT, wrong password returns 401, inactive account returns 403, rate limiting returns 429 after 5 attempts.
Expected output:
@pytest.mark.parametrize("password,expected_status,expected_code", [
("correctpassword", 200, None),
("wrongpassword", 401, "INVALID_CREDENTIALS"),
])
async def test_login(client, user, password, expected_status, expected_code):
...
async def test_login_inactive_account(client, inactive_user):
...
async def test_login_rate_limit(client, user):
for _ in range(5):
await client.post("/auth/login", json={...})
resp = await client.post("/auth/login", json={...})
assert resp.status_code == 429