Integration tests exercise the full request-response cycle — from HTTP request through the FastAPI router, dependencies, service layer, and database to the HTTP response. They verify that all the pieces work together correctly: the route handler validates the right fields, the service enforces the right business rules, the database query returns the right data, and the response schema serialises the right fields. Integration tests are slower than unit tests (they hit the database) but catch a different class of bugs: incorrect SQL, missing eager loading, wrong HTTP status codes, and schema serialisation issues.
CRUD Integration Tests
# tests/integration/test_posts_api.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models.post import Post
from app.models.user import User
class TestCreatePost:
def test_create_post_returns_201(self, client: TestClient):
response = client.post("/posts/", json={
"title": "Integration Test Post",
"body": "This is a sufficiently long post body for testing.",
"slug": "integration-test-post",
})
assert response.status_code == 201
data = response.json()
assert data["title"] == "Integration Test Post"
assert data["slug"] == "integration-test-post"
assert "id" in data
assert "created_at" in data
def test_create_post_sets_author_from_token(self, client, user):
response = client.post("/posts/", json={
"title": "Authored Post",
"body": "Content here for testing.",
"slug": "authored-post",
})
assert response.status_code == 201
assert response.json()["author_id"] == user.id
def test_create_post_unauthenticated_returns_401(self, anon_client):
response = anon_client.post("/posts/", json={
"title": "Anon Post", "body": "content", "slug": "anon"
})
assert response.status_code == 401
def test_create_post_missing_title_returns_422(self, client):
response = client.post("/posts/", json={"body": "content", "slug": "no-title"})
assert response.status_code == 422
errors = response.json()["detail"]
assert any("title" in str(e["loc"]) for e in errors)
def test_duplicate_slug_returns_409(self, client, db, user):
# Create first post
existing = Post(title="First", body="Body", slug="dup-slug", author_id=user.id)
db.add(existing); db.flush()
response = client.post("/posts/", json={
"title": "Second", "body": "Body content", "slug": "dup-slug"
})
assert response.status_code == 409
class TestGetPost:
def test_get_existing_post_returns_200(self, client, db, user):
post = Post(title="Fetchable", body="Content", slug="fetchable",
status="published", author_id=user.id)
db.add(post); db.flush()
response = client.get(f"/posts/{post.id}")
assert response.status_code == 200
assert response.json()["title"] == "Fetchable"
def test_get_nonexistent_post_returns_404(self, client):
response = client.get("/posts/999999")
assert response.status_code == 404
def test_get_soft_deleted_post_returns_404(self, client, db, user):
from sqlalchemy.sql import func
post = Post(title="Deleted", body="Content", slug="deleted-post",
author_id=user.id, deleted_at=func.now())
db.add(post); db.flush()
response = client.get(f"/posts/{post.id}")
assert response.status_code == 404Note: Integration tests should verify both the HTTP response and the database state after write operations. After creating a post, check the response body AND query the database directly to confirm the row was written correctly:
created = db.scalars(select(Post).where(Post.slug == "test")).first(); assert created is not None. This catches bugs where the handler returns a fake response without actually writing to the database.Tip: Organise integration test classes by the resource and HTTP method:
TestCreatePost, TestGetPost, TestUpdatePost, TestDeletePost. Each class has methods testing success cases, error cases (404, 403, 409, 422), boundary conditions, and authentication requirements. This structure makes it easy to see what is tested and what is missing for each endpoint.Warning: Always test the error cases explicitly — 404 for missing resources, 403 for unauthorised access, 409 for duplicate constraints, 422 for validation failures. Many bugs in production come from incorrect error handling: the wrong status code, missing error details, or an exception that reaches the client as a 500. Testing error paths is as important as testing success paths.
Update and Delete Tests
class TestUpdatePost:
def test_patch_updates_only_provided_fields(self, client, db, user):
post = Post(title="Old Title", body="Old body content.", slug="old-slug",
status="draft", author_id=user.id)
db.add(post); db.flush()
original_body = post.body
response = client.patch(f"/posts/{post.id}", json={"title": "New Title"})
assert response.status_code == 200
data = response.json()
assert data["title"] == "New Title"
assert data["body"] == original_body # body unchanged
def test_patch_by_non_owner_returns_403(self, client, db):
other_user = User(email="other@ex.com", name="Other",
password_hash="h", role="user")
db.add(other_user); db.flush()
post = Post(title="Others", body="Content body.", slug="others-post",
author_id=other_user.id)
db.add(post); db.flush()
response = client.patch(f"/posts/{post.id}", json={"title": "Stolen"})
assert response.status_code == 403
def test_admin_can_patch_any_post(self, admin_client, db, user):
post = Post(title="Any Post", body="Content here.", slug="any-post",
author_id=user.id)
db.add(post); db.flush()
response = admin_client.patch(f"/posts/{post.id}", json={"title": "Admin Edit"})
assert response.status_code == 200
class TestDeletePost:
def test_delete_own_post_returns_204(self, client, db, user):
post = Post(title="To Delete", body="Content.", slug="to-delete",
author_id=user.id)
db.add(post); db.flush()
post_id = post.id
response = client.delete(f"/posts/{post_id}")
assert response.status_code == 204
# Verify soft-deleted in database
db.expire(post)
refreshed = db.get(Post, post_id)
assert refreshed.deleted_at is not None
def test_delete_unauthenticated_returns_401(self, anon_client, db, user):
post = Post(title="Protected", body="Content.", slug="protected",
author_id=user.id)
db.add(post); db.flush()
response = anon_client.delete(f"/posts/{post.id}")
assert response.status_code == 401Authentication Flow Tests
class TestAuthFlow:
def test_register_creates_user(self, anon_client, db):
response = anon_client.post("/auth/register", json={
"email": "new@example.com",
"name": "New User",
"password": "Secure1Pass",
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "new@example.com"
assert "password_hash" not in data # never in response
def test_login_returns_tokens(self, anon_client, db, user):
response = anon_client.post("/auth/login", json={
"email": "user@example.com",
"password": "Password1",
})
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
def test_login_wrong_password_returns_401(self, anon_client, db, user):
response = anon_client.post("/auth/login", json={
"email": "user@example.com", "password": "WrongPass1"
})
assert response.status_code == 401Quick Reference
| Assertion | Example |
|---|---|
| Status code | assert response.status_code == 201 |
| Response field | assert response.json()["title"] == "..." |
| Field absent | assert "password_hash" not in response.json() |
| DB state | db.expire(obj); db.refresh(obj); assert obj.field == ... |
| Validation error | assert response.status_code == 422 |
| Error detail | assert "field_name" in str(response.json()["detail"]) |