9. Testing your API
Tests are the contract. pytest with FastAPI's TestClient lets you exercise every endpoint without spinning up a server. The fixtures pattern keeps test databases clean.
Why testing matters for APIs
An API is a contract that other code depends on, and the cost of breaking that contract is paid by every integration at once. The realistic failure mode isn't a dramatic bug; it's a small change to the auth dependency that quietly starts returning 401 to a client whose retry logic doesn't surface 401 as a user-visible error. A test that exercises every endpoint behind the dependency catches that drift before the deploy.
The suite for this chapter covers the surfaces that get most of the regressions: authentication (missing header, malformed header, unknown key, valid key), the article endpoint's contract (response shape, status codes, cache behaviour), and the failure modes of the external sources (so the tests don't burn NewsAPI quota every time CI runs).
pytest + FastAPI test setup
FastAPI provides excellent testing support through TestClient. Combined with pytest fixtures, you can test your entire API without running a real server.
Install testing dependencies:
pip install pytest pytest-cov httpx
Create conftest.py with pytest fixtures for database and authentication:
import os
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
os.environ["DATABASE_URL"] = "sqlite://"
os.environ["ADMIN_API_KEY"] = "test-admin-key"
os.environ["NEWSAPI_KEY"] = "test-newsapi-key"
os.environ["GUARDIAN_KEY"] = "test-guardian-key"
from main import app
from database import Base, get_db
from auth import create_api_key
from rate_limit import REQUEST_LOGS
# Test database (in-memory SQLite)
TEST_DATABASE_URL = "sqlite://"
engine = create_engine(
TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(autouse=True)
def reset_rate_limit_state():
"""Keep the process-level rate limiter isolated between tests."""
REQUEST_LOGS.clear()
yield
REQUEST_LOGS.clear()
@pytest.fixture
def db():
"""Create test database and return session."""
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def client(db):
"""Create test client with test database."""
def override_get_db():
try:
yield db
finally:
pass
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
@pytest.fixture
def api_key(db):
"""Create test API key and return the raw key."""
result = create_api_key(db, name="Test Key", tier="basic")
return result["api_key"]
@pytest.fixture
def auth_headers(api_key):
"""Create authorization headers with API key."""
return {"Authorization": f"Bearer {api_key}"}
The fixtures are stacked, and the stacking is the point. reset_rate_limit_state clears the in-memory request log around every test because that state belongs to the Python process rather than the database. db then spins up an in-memory SQLite database, creates the schema, hands the session to the test, and drops the tables on teardown. Every test starts without rate-limit or database state bleeding in from the one before it.
Section 5 made the case for PostgreSQL over SQLite in production; the test suite reaches for SQLite specifically because the trade-offs invert under test load. No daemon to manage, instant fixture teardown, and the multi-writer concurrency that drove the production choice doesn't matter when every test runs serially in one process. The production app still requires PostgreSQL; the suite just borrows SQLite as a fast, isolated stand-in.
The environment variables are set before importing main, because database.py creates its engine at import time. client builds on db by overriding the get_db dependency so any endpoint that touches the database goes through the test session, then yields a TestClient that drives the app in-process. api_key creates a valid key against the test database, and auth_headers wraps it in the Bearer format. A test that needs an authenticated request just declares auth_headers as a parameter.
Testing authentication
Create test_auth.py to verify authentication works correctly.
def test_authentication_required(client):
"""Test that protected endpoints require authentication."""
response = client.get("/articles")
assert response.status_code == 401
assert "Missing Authorization" in response.json()["detail"]
def test_invalid_api_key(client):
"""Test that invalid API keys are rejected."""
headers = {"Authorization": "Bearer invalid_key_12345"}
response = client.get("/articles", headers=headers)
assert response.status_code == 401
assert "Invalid or inactive" in response.json()["detail"]
def test_malformed_authorization_header(client):
"""Test that Bearer formatting is required."""
response = client.get("/articles", headers={"Authorization": "invalid_key_12345"})
assert response.status_code == 401
assert "Invalid Authorization format" in response.json()["detail"]
def test_valid_authentication(client, auth_headers):
"""Test that valid API keys are accepted."""
response = client.get("/articles/999", headers=auth_headers)
assert response.status_code == 404
assert response.json()["detail"] == "Article not found"
def test_api_key_generation(client, db):
"""Test API key generation endpoint."""
response = client.post(
"/admin/api-keys",
headers={"X-Admin-Key": "test-admin-key"},
json={"name": "New Test Key", "tier": "premium"}
)
assert response.status_code == 201
data = response.json()
assert "api_key" in data
assert data["tier"] == "premium"
assert len(data["api_key"]) > 20
The five tests cover the auth dependency's full surface. test_authentication_required proves that an unauthenticated request to a protected endpoint comes back as 401 with the expected detail substring; test_invalid_api_key does the same for a Bearer token that doesn't match any stored hash; test_malformed_authorization_header checks the required Bearer format. test_valid_authentication uses a valid key against a missing article and gets the route's 404, which proves authentication passed and the handler body ran. test_api_key_generation exercises the admin endpoint with the admin header; the response carries the raw key (the only place it's ever visible), the requested tier round-trips correctly, and the key string is long enough to actually have entropy.
Mocking external APIs
Tests shouldn't call real external APIs. External APIs are slow, cost money, rate limit you, and can fail independently of your code. Mock them for fast, reliable tests.
from unittest.mock import patch, Mock
@patch('sources.requests.get')
def test_cache_behavior(mock_get, client, auth_headers, db):
"""Test that caching works correctly."""
# Mock external API response
mock_response = Mock()
mock_response.json.return_value = {
"articles": [
{
"title": "Test Article",
"description": "Test description",
"url": "https://example.com/test",
"publishedAt": "2026-05-25T10:00:00Z"
}
]
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
# First request (cache miss)
response1 = client.get("/articles?category=technology&limit=1", headers=auth_headers)
assert response1.status_code == 200
assert response1.json()["cache_status"] == "miss"
assert mock_get.called
# Second request (cache hit - shouldn't call external API)
mock_get.reset_mock()
response2 = client.get("/articles?category=technology&limit=1", headers=auth_headers)
assert response2.status_code == 200
assert response2.json()["cache_status"] == "hit"
assert not mock_get.called # Cache hit - no external API call
The test mocks upstream HTTP, then proves both halves of the cache contract: the first request fetches from the source and reports "miss"; the second request avoids the source and reports "hit".
One thing the test deliberately doesn't do is exercise both fetchers. The mock returns NewsAPI's {"articles": [...]} shape on every call, so fetch_newsapi finds the article and fetch_guardian finds nothing (its branch looks for response.results) and returns []. The cache-status assertion is correct either way; add a Guardian-shape test next:
@patch('sources.requests.get')
def test_guardian_response_is_normalised(mock_get, client, auth_headers):
"""A Guardian response becomes the API's standard article shape."""
mock_response = Mock()
mock_response.json.return_value = {
"response": {
"results": [{
"webTitle": "Guardian test",
"webUrl": "https://example.com/guardian",
"webPublicationDate": "2026-05-25T10:00:00Z",
"sectionName": "Technology",
"fields": {"trailText": "Guardian description"}
}]
}
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
response = client.get(
"/articles?source=guardian&category=technology&limit=1",
headers=auth_headers,
)
assert response.status_code == 200
assert response.json()["articles"][0]["source"] == "guardian"
assert response.json()["articles"][0]["category"] == "technology"
@patch('sources.requests.get')
def test_cache_refresh_updates_category(mock_get, client, auth_headers):
"""Refreshing an existing URL must not return its previous category."""
mock_response = Mock()
mock_response.json.return_value = {
"articles": [{
"title": "Shared article",
"description": "Appears in another category",
"url": "https://example.com/shared",
"publishedAt": "2026-05-25T10:00:00Z"
}]
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
client.get(
"/articles?source=newsapi&category=technology&limit=1",
headers=auth_headers,
)
response = client.get(
"/articles?source=newsapi&category=business&limit=1",
headers=auth_headers,
)
assert response.status_code == 200
assert response.json()["articles"][0]["category"] == "business"
def test_limit_enforces_documented_bounds(client, auth_headers):
"""The route rejects list sizes outside the 1-to-100 contract."""
assert client.get("/articles?limit=0", headers=auth_headers).status_code == 422
assert client.get("/articles?limit=101", headers=auth_headers).status_code == 422
The extra tests close three gaps in the contract: Guardian normalisation is exercised directly, a cached URL refreshed under a new category exposes the new category, and values outside the documented limit range fail validation before any external request can run.
Testing rate limiting
Rate limiting is one of the chapter's headline features, so the suite needs to exercise both visible behaviours: the 429 when a caller is over budget, and the X-RateLimit-* headers on a successful request. Hammering the endpoint 100 times in a test would be slow; the cleaner move is to prime the in-memory counter directly, then issue one request.
from datetime import datetime, timezone
from unittest.mock import patch
from rate_limit import REQUEST_LOGS, RATE_LIMITS
from auth import validate_api_key
def test_rate_limit_exceeded(client, auth_headers, api_key, db):
"""Pre-fill the bucket, then assert the next request gets 429."""
db_key = validate_api_key(db, api_key)
now = datetime.now(timezone.utc)
REQUEST_LOGS[db_key.id] = [now] * RATE_LIMITS["basic"]
response = client.get("/articles", headers=auth_headers)
assert response.status_code == 429
assert "Rate limit exceeded" in response.json()["detail"]
assert "Retry-After" in response.headers
@patch("cache.fetch_from_all_sources", return_value=[])
def test_rate_limit_headers_on_success(mock_fetch, client, auth_headers):
"""A normal request returns the X-RateLimit-* headers describing quota."""
response = client.get("/articles", headers=auth_headers)
assert response.status_code == 200
assert response.headers["X-RateLimit-Limit"] == "100"
assert int(response.headers["X-RateLimit-Remaining"]) <= 99
mock_fetch.assert_called_once()
The first test primes REQUEST_LOGS with one hundred timestamps, then issues a request that lands as request 101: 429 with a Retry-After header. The autouse fixture clears that in-memory bucket before the next test begins. The second test patches the cache module's external-fetch boundary to return no rows, then confirms the quota headers are present on a successful response without making any network request.
Run tests with coverage:
# Run all tests
pytest
# Run with coverage report
pytest --cov=. --cov-report=html
# Output:
# ========================== test session starts ===========================
# collected 11 items
#
# tests/test_auth.py::test_authentication_required PASSED
# tests/test_auth.py::test_invalid_api_key PASSED
# tests/test_auth.py::test_malformed_authorization_header PASSED
# tests/test_auth.py::test_valid_authentication PASSED
# tests/test_auth.py::test_api_key_generation PASSED
# tests/test_articles.py::test_cache_behavior PASSED
# tests/test_articles.py::test_guardian_response_is_normalised PASSED
# tests/test_articles.py::test_cache_refresh_updates_category PASSED
# tests/test_articles.py::test_limit_enforces_documented_bounds PASSED
# tests/test_rate_limit.py::test_rate_limit_exceeded PASSED
# tests/test_rate_limit.py::test_rate_limit_headers_on_success PASSED
#
# ========================= 11 passed in 2.10s =============================
#
# Coverage report saved to htmlcov/index.htmlTest the failure modes, not just the happy path. The bugs that survive into production tend to live in the branches that aren't exercised by a working request: missing auth header, rate-limit exceeded, external source returning a 500, malformed input. Each of those needs its own test.
Keep the suite fast. Mocks for external HTTP and in-memory SQLite for the database mean the full suite runs in seconds. A slow suite gets run less, which is when regressions slip through.
Next, in section 10, we deploy what we've built to Railway, smoke-test the live URL, and run the chapter quiz.