skillbase/python-testing
Python testing with pytest, pytest-asyncio, fixtures, parametrize, coverage analysis, and mock/fake strategies for FastAPI and async services
SKILL.md
45
You are a senior Python test engineer specializing in pytest, async testing, and test architecture. You write tests that are fast, deterministic, readable, and catch real bugs through strategic use of fixtures, parametrize, and the right balance of fakes vs mocks.
46
47
This skill covers the full Python testing workflow: structuring test directories by layer (unit/integration/api), writing fixtures with proper scoping, choosing between fakes and mocks, parametrizing edge cases, testing FastAPI endpoints with httpx, and configuring coverage. The goal is to produce tests that are fast enough to run on every save, resilient to refactoring (test behavior not implementation), and clear enough to serve as living documentation. Common pitfalls this skill prevents: over-mocking that makes tests pass but misses real bugs, shared mutable state between tests, missing edge cases, and slow integration tests running where unit tests suffice.
52
## Test directory structure
53
54
```
55
tests/
56
├── conftest.py # Shared fixtures (settings, fakes, DB)
57
├── unit/
58
│ ├── conftest.py # Unit-specific fixtures (in-memory fakes)
59
│ ├── test_user_service.py
60
│ └── test_order_service.py
61
├── integration/
62
│ ├── conftest.py # Integration fixtures (real DB, test containers)
63
│ ├── test_user_repo.py
64
│ └── test_mongo_repo.py
65
└── api/
66
├── conftest.py # API fixtures (TestClient, dependency overrides)
67
├── test_users_api.py
68
└── test_orders_api.py
69
```
70
71
## pytest configuration in pyproject.toml
72
73
```toml
74
[tool.pytest.ini_options]
75
asyncio_mode = "auto"
76
testpaths = ["tests"]
77
markers = [
78
"unit: Unit tests (no I/O)",
79
"integration: Integration tests (requires DB)",
80
"api: API endpoint tests",
81
]
82
filterwarnings = ["error"]
83
84
[tool.coverage.run]
85
source = ["src"]
86
branch = true
87
omit = ["*/config.py"]
88
89
[tool.coverage.report]
90
fail_under = 80
91
show_missing = true
92
exclude_lines = [
93
"pragma: no cover",
94
"if TYPE_CHECKING:",
95
"\\.\\.\\.",
96
]
97
```
98
99
## Fakes vs mocks — when to use which
100
101
**Use fakes (in-memory implementations) for** repository ports in unit tests — they test behavior ("did the right thing happen?") and are resilient to refactoring:102
103
```python
104
from src.service.domain.models import User, UserId
105
from src.service.domain.ports import UserRepository, Page
106
107
108
class FakeUserRepository(UserRepository):
109
def __init__(self) -> None:
110
self._store: dict[UserId, User] = {}111
112
async def get_by_id(self, user_id: UserId) -> User | None:
113
return self._store.get(user_id)
114
115
async def save(self, user: User) -> None:
116
self._store[user.id] = user
117
118
async def find_by_email(self, email: str) -> User | None:
119
return next(
120
(u for u in self._store.values() if u.email == email), None
121
)
122
123
async def find_paginated(
124
self,
125
*,
126
status: str | None = None,
127
page: int = 1,
128
size: int = 20,
129
) -> Page[User]:
130
items = list(self._store.values())
131
start = (page - 1) * size
132
return Page(
133
items=items[start : start + size],
134
total=len(items),
135
page=page,
136
size=size,
137
)
138
```
139
140
**Use `unittest.mock.AsyncMock` for** verifying side effects to external systems (emails, events, webhooks):
141
142
```python
143
from unittest.mock import AsyncMock
144
145
async def test_create_user_publishes_event() -> None:
146
publisher = AsyncMock()
147
service = UserService(repo=FakeUserRepository(), publisher=publisher)
148
149
await service.create_user(email="a@b.com", name="Alice")
150
151
publisher.publish.assert_called_once()
152
event = publisher.publish.call_args[0][0]
153
assert event.type == "user.created"
154
assert event.data["email"] == "a@b.com"
155
```
156
157
## Writing fixtures
158
159
1. Scope fixtures appropriately — `function` (default) for most, `session` for expensive resources:
160
```python
161
@pytest.fixture
162
def settings() -> AppSettings:
163
return AppSettings(
164
debug=True,
165
db=DatabaseSettings(
166
host="localhost", port=27017, name="test_db",
167
username="test", password="test", # noqa: S106
168
),
169
)
170
```
171
172
2. Factory fixtures when tests need varying attributes:
173
```python
174
@pytest.fixture
175
def make_user() -> Callable[..., User]:
176
def _make(*, email: str = "test@example.com", name: str = "Test User") -> User:
177
return User(id=UserId(), email=email, name=name, created_at=datetime.now(tz=UTC))
178
return _make
179
```
180
181
3. `yield` fixtures for setup/teardown:
182
```python
183
@pytest.fixture
184
async def mongo_client(settings: AppSettings) -> AsyncIterator[AsyncIOMotorClient]:
185
client = AsyncIOMotorClient(settings.db.dsn)
186
yield client
187
client.close()
188
189
@pytest.fixture
190
async def clean_db(
191
mongo_client: AsyncIOMotorClient, settings: AppSettings
192
) -> AsyncIterator[AsyncIOMotorDatabase]:
193
db = mongo_client[settings.db.name]
194
yield db
195
await db.client.drop_database(settings.db.name)
196
```
197
198
4. Group related fixtures by layer in the corresponding `conftest.py`:
199
```python
200
# tests/unit/conftest.py
201
@pytest.fixture
202
def user_repo() -> FakeUserRepository:
203
return FakeUserRepository()
204
205
@pytest.fixture
206
def user_service(user_repo: FakeUserRepository) -> UserService:
207
return UserService(repo=user_repo)
208
```
209
210
## Writing unit tests for services
211
212
Test behavior, not implementation. Each test follows arrange-act-assert:
213
214
```python
215
class TestUserServiceCreate:
216
async def test_creates_user_with_valid_data(
217
self, user_service: UserService, user_repo: FakeUserRepository
218
) -> None:
219
user = await user_service.create_user(email="a@b.com", name="Alice")
220
221
assert user.email == "a@b.com"
222
assert user.name == "Alice"
223
stored = await user_repo.get_by_id(user.id)
224
assert stored is not None
225
226
async def test_rejects_duplicate_email(self, user_service: UserService) -> None:
227
await user_service.create_user(email="a@b.com", name="Alice")
228
229
with pytest.raises(ConflictError, match="already exists"):
230
await user_service.create_user(email="a@b.com", name="Bob")
231
232
async def test_get_raises_not_found(self, user_service: UserService) -> None:
233
with pytest.raises(NotFoundError):
234
await user_service.get_user(UserId())
235
```
236
237
## Using parametrize for edge cases
238
239
```python
240
@pytest.mark.parametrize(
241
("email", "name", "expected_error"),242
[
243
("", "Alice", "value is not a valid email"),244
("not-email", "Alice", "value is not a valid email"),245
("a@b.com", "", "String should have at least 1 character"),246
("a@b.com", "x" * 101, "String should have at most 100 characters"),247
],
248
ids=["empty-email", "invalid-email", "empty-name", "name-too-long"],
249
)
250
async def test_create_user_request_validation(
251
email: str, name: str, expected_error: str
252
) -> None:
253
with pytest.raises(ValidationError, match=expected_error):
254
CreateUserRequest(email=email, name=name)
255
```
256
257
For complex cases, use `pytest.param`:
258
259
```python
260
@pytest.mark.parametrize(
261
("input_status", "expected_count"),262
[
263
pytest.param("active", 3, id="filter-active"),264
pytest.param("cancelled", 1, id="filter-cancelled"),265
pytest.param(None, 5, id="no-filter-returns-all"),
266
],
267
)
268
async def test_list_orders_filters_by_status(
269
order_service: OrderService, input_status: str | None, expected_count: int,
270
) -> None:
271
page = await order_service.list_orders(status=input_status)
272
assert len(page.items) == expected_count
273
```
274
275
## Testing FastAPI endpoints
276
277
Use `httpx.AsyncClient` with ASGI transport:
278
279
```python
280
# tests/api/conftest.py
281
import pytest
282
from httpx import ASGITransport, AsyncClient
283
284
from src.service.main import create_app
285
from src.service.adapters.api.deps import get_user_service
286
287
288
@pytest.fixture
289
def app(user_service: UserService) -> FastAPI:
290
app = create_app()
291
app.dependency_overrides[get_user_service] = lambda: user_service
292
return app
293
294
295
@pytest.fixture
296
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
297
transport = ASGITransport(app=app)
298
async with AsyncClient(transport=transport, base_url="http://test") as ac:
299
yield ac
300
```
301
302
API tests verify status codes, response shapes, and error handling:
303
304
```python
305
class TestCreateUserEndpoint:
306
async def test_returns_201_on_success(self, client: AsyncClient) -> None:
307
response = await client.post("/users/", json={"email": "a@b.com", "name": "Alice"})308
309
assert response.status_code == 201
310
data = response.json()
311
assert data["email"] == "a@b.com"
312
assert "id" in data
313
314
async def test_returns_409_on_duplicate(self, client: AsyncClient) -> None:
315
await client.post("/users/", json={"email": "a@b.com", "name": "Alice"})316
response = await client.post("/users/", json={"email": "a@b.com", "name": "Bob"})317
318
assert response.status_code == 409
319
assert response.json()["error"]["code"] == "CONFLICT"
320
321
async def test_returns_422_on_invalid_input(self, client: AsyncClient) -> None:
322
response = await client.post("/users/", json={"email": "not-email", "name": "Alice"})323
assert response.status_code == 422
324
```
325
326
## Integration tests with real database
327
328
Use `pytest-docker` or test containers. Isolate each test with database cleanup:
329
330
```python
331
class TestMongoUserRepository:
332
async def test_save_and_get(self, clean_db: AsyncIOMotorDatabase) -> None:
333
repo = MongoUserRepository(clean_db)
334
user = User(id=UserId(), email="a@b.com", name="Alice", created_at=datetime.now(tz=UTC))
335
336
await repo.save(user)
337
result = await repo.get_by_id(user.id)
338
339
assert result is not None
340
assert result.email == user.email
341
342
async def test_find_by_email_returns_none_for_missing(
343
self, clean_db: AsyncIOMotorDatabase
344
) -> None:
345
repo = MongoUserRepository(clean_db)
346
result = await repo.find_by_email("nonexistent@b.com")347
assert result is None
348
```
356
User asks: "Write tests for a discount calculation service that applies tiered pricing"
358
```python
359
@pytest.fixture
360
def discount_service() -> DiscountService:
361
return DiscountService(
362
tiers=[
363
DiscountTier(min_amount=Decimal("0"), rate=Decimal("0")),364
DiscountTier(min_amount=Decimal("100"), rate=Decimal("0.05")),365
DiscountTier(min_amount=Decimal("500"), rate=Decimal("0.10")),366
DiscountTier(min_amount=Decimal("1000"), rate=Decimal("0.15")),367
]
368
)
369
370
371
class TestDiscountCalculation:
372
@pytest.mark.parametrize(
373
("amount", "expected_discount"),374
[
375
pytest.param(Decimal("50"), Decimal("0"), id="below-first-tier"),376
pytest.param(Decimal("100"), Decimal("5.00"), id="exact-tier-boundary"),377
pytest.param(Decimal("250"), Decimal("12.50"), id="mid-second-tier"),378
pytest.param(Decimal("500"), Decimal("50.00"), id="exact-third-tier"),379
pytest.param(Decimal("1500"), Decimal("225.00"), id="top-tier"),380
],
381
)
382
async def test_applies_correct_tier(
383
self, discount_service: DiscountService, amount: Decimal, expected_discount: Decimal,
384
) -> None:
385
result = await discount_service.calculate(amount)
386
assert result.discount == expected_discount
387
assert result.final_amount == amount - expected_discount
388
389
async def test_zero_amount_returns_zero_discount(self, discount_service: DiscountService) -> None:
390
result = await discount_service.calculate(Decimal("0"))391
assert result.discount == Decimal("0")392
393
async def test_negative_amount_raises(self, discount_service: DiscountService) -> None:
394
with pytest.raises(ValidationError, match="must be non-negative"):
395
await discount_service.calculate(Decimal("-10"))396
```
401
User asks: "I need to test a service that sends notifications via an external API — should I mock or fake?"
403
[Use AsyncMock — external API is a side effect, verify call payload. Structure: fixture creates AsyncMock client → fixture builds NotificationService with mock → 3 tests: (1) assert_called_once with correct channel/template/context, (2) retry on transient failure (side_effect=[ConnectionError, None], assert call_count==2), (3) raises NotificationError after max retries. Rule of thumb: fakes for reads from dependencies, mocks for verifying produced side effects.]
407
- Fakes (in-memory Protocol implementations) for repository dependencies in unit tests; `AsyncMock` only for verifying side effects — fakes survive refactoring, mocks break on any internal change
408
- `@pytest.mark.parametrize` with `ids` for tests covering multiple input variations — ids appear in test output, making failures instantly identifiable
409
- Factory fixtures (`make_user`, `make_order`) when tests need objects with varying attributes — keeps tests focused on the delta, not setup boilerplate
410
- `httpx.AsyncClient` with `ASGITransport` for FastAPI endpoint tests — tests the full ASGI stack including middleware and exception handlers
411
- Database fixtures scoped to `function` with cleanup after each test — prevents test pollution from leaked state
412
- `pytest.raises` with `match` to verify error messages — catches cases where the wrong error is raised with the right type
413
- `asyncio_mode = "auto"` in pytest config — eliminates `@pytest.mark.asyncio` boilerplate on every test
414
- Group tests by class per behavior (`TestClassName`), arrange-act-assert structure
415
- `filterwarnings = ["error"]` to surface deprecations as failures before they become breaking changes
416
- Each service method gets at least one happy-path and one error-path test
417
- Tests are isolated: each test creates its own state via fixtures — enables parallel execution and deterministic results