Module 6: Mocking, Fixtures and Validation Loops
Mocking External Services in Practice
Mocking External Services in Practice
Capsule overview
You already know the fundamentals of mocking — Mock, MagicMock, patch, side_effect. But in the real world you don't mock "abstract objects": you mock HTTP APIs that return JSON, OpenAI clients that generate text, databases that return rows, files that contain config, environment variables that hold API keys. This capsule takes you into concrete practice: how to mock the most common external services in Python applications, with runnable examples from start to finish.
By the end you'll have mastered mocking HTTP, AI APIs, databases, the filesystem, time and environment variables. You'll also see how to generate mocks with Claude Code to speed up the process. You'll close with a complete example: a weather service that consumes an external API, tested 100% with mocks.
1. Mocking HTTP APIs (requests / httpx)
The typical scenario
Your code calls an external API to send notifications, get data, or sync with a third-party service. In tests you don't want to make real calls: they're slow, they depend on the network, they can fail because of rate limits, and they cost money if the API is paid.
With requests
# mymodule.py
import requests
def send_notification(email: str, message: str) -> dict:
"""Sends a notification via an external API."""
response = requests.post(
"https://api.notify.example.com/send",
json={"email": email, "message": message},
timeout=5,
)
response.raise_for_status()
return response.json()
A test with @patch:
# tests/test_notifications.py
from unittest.mock import patch
from mymodule import send_notification
@patch("mymodule.requests.post")
def test_send_notification_success(mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {"sent": True}
mock_post.return_value.raise_for_status = lambda: None
result = send_notification("user@test.com", "Hello!")
assert result["sent"] is True
mock_post.assert_called_once()
call_args = mock_post.call_args
assert call_args[1]["json"]["email"] == "user@test.com"
assert call_args[1]["json"]["message"] == "Hello!"
Mocking error responses and timeouts
from unittest.mock import patch
import pytest
import requests
from mymodule import send_notification
@patch("mymodule.requests.post")
def test_send_notification_api_error(mock_post):
mock_post.return_value.raise_for_status.side_effect = requests.HTTPError("500")
mock_post.return_value.status_code = 500
with pytest.raises(requests.HTTPError):
send_notification("user@test.com", "Hello!")
@patch("mymodule.requests.post")
def test_send_notification_timeout(mock_post):
import requests as req
mock_post.side_effect = req.exceptions.Timeout("Connection timed out")
with pytest.raises(req.exceptions.Timeout):
send_notification("user@test.com", "Hello!")
With httpx (async and sync)
If you use httpx instead of requests, the patching is similar:
# services/external.py
import httpx
def fetch_user(user_id: int) -> dict:
response = httpx.get(f"https://api.example.com/users/{user_id}")
response.raise_for_status()
return response.json()
from unittest.mock import Mock
from services.external import fetch_user
@patch("services.external.httpx.get")
def test_fetch_user(mock_get):
mock_response = Mock()
mock_response.raise_for_status = Mock()
mock_response.json.return_value = {"id": 1, "name": "Alice"}
mock_get.return_value = mock_response
result = fetch_user(1)
assert result["name"] == "Alice"
mock_get.assert_called_once_with("https://api.example.com/users/1")
Using the responses library (a more realistic alternative)
The responses library intercepts HTTP at the socket level and simulates responses in a way that's closer to real:
# pip install responses
import responses
from mymodule import send_notification
@responses.activate
def test_send_notification_with_responses():
responses.add(
responses.POST,
"https://api.notify.example.com/send",
json={"sent": True},
status=200,
)
result = send_notification("user@test.com", "Hello!")
assert result["sent"] is True
assert len(responses.calls) == 1
assert responses.calls[0].request.url == "https://api.notify.example.com/send"
With responses you don't need to patch — the library intercepts the real HTTP calls and redirects them to your mocked responses. Useful when the code uses requests internally and you want to verify that the correct URL is called with the correct headers.
2. Mocking AI APIs (OpenAI / Anthropic)
Why AI APIs MUST be mocked
The OpenAI, Anthropic, etc. APIs shouldn't be called in tests for three reasons:
- Cost: Every call consumes credits. Thousands of tests = a big bill.
- Non-determinism: The model can return different responses for the same prompt.
- Rate limits and latency: Slow tests and possible intermittent failures.
An example with OpenAI (the modern client)
# services/summarizer.py
from openai import OpenAI
def generate_summary(client: OpenAI, text: str, max_tokens: int = 100) -> str:
"""Generates a summary using the OpenAI API."""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": f"Summarize in at most {max_tokens} words: {text}"}],
max_tokens=max_tokens,
)
return response.choices[0].message.content.strip()
A test with a mock:
# tests/test_summarizer.py
from unittest.mock import Mock, patch
from services.summarizer import generate_summary
def test_generate_summary_returns_content():
mock_client = Mock()
mock_client.chat.completions.create.return_value = Mock(
choices=[Mock(message=Mock(content="This is a summary."))]
)
mock_client.chat.completions.create.return_value.choices[0].message.strip.return_value = "This is a summary."
result = generate_summary(mock_client, "Long text here...")
assert "summary" in result.lower()
mock_client.chat.completions.create.assert_called_once()
If the client is imported internally
# services/summarizer.py
from openai import OpenAI
_client = OpenAI(api_key="sk-...")
def generate_summary(text: str) -> str:
response = _client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": f"Summarize: {text}"}],
max_tokens=100,
)
return response.choices[0].message.content.strip()
@patch("services.summarizer._client")
def test_generate_summary(mock_client):
mock_response = Mock()
mock_response.choices = [Mock()]
mock_response.choices[0].message = Mock()
mock_response.choices[0].message.content = "Short summary."
mock_response.choices[0].message.strip.return_value = "Short summary."
mock_client.chat.completions.create.return_value = mock_response
result = generate_summary("Long text...")
assert "Short" in result
Patch where it's used (services.summarizer._client), not where it's defined (openai.OpenAI).
With Anthropic ( Claude )
The structure is similar: you mock the client or the create method:
# With Anthropic
@patch("services.claude_wrapper.anthropic.Anthropic")
def test_generate_with_claude(mock_anthropic):
mock_client = Mock()
mock_client.messages.create.return_value = Mock(
content=[Mock(text="Mocked response.")]
)
mock_anthropic.return_value = mock_client
result = generate_with_claude("prompt")
assert "Mocked" in result
3. Mocking database operations
The pattern: mock the access layer
You don't mock SQLAlchemy or the driver directly — you mock the function or class that encapsulates the queries (repository, dao, etc.).
# mymodule.py
def get_user_by_email(db, email: str) -> dict | None:
"""Gets a user by email from the database."""
rows = db.query("SELECT * FROM users WHERE email = %s", (email,))
if not rows:
return None
return rows[0]
# tests/test_users.py
import pytest
from mymodule import get_user_by_email
def test_get_user_by_email_found(mocker):
mock_db = mocker.patch("mymodule.db")
mock_db.query.return_value = [{"id": 1, "email": "user@test.com", "name": "Test"}]
user = get_user_by_email(mock_db, "user@test.com")
assert user["email"] == "user@test.com"
mock_db.query.assert_called_once()
def test_get_user_by_email_not_found(mocker):
mock_db = mocker.patch("mymodule.db")
mock_db.query.return_value = []
user = get_user_by_email(mock_db, "nonexistent@test.com")
assert user is None
If db is injected as a parameter, you don't need to patch — you pass the mock directly. If db is an imported global object, you use mocker.patch("mymodule.db") or @patch("mymodule.db").
With a SQLAlchemy session
# repositories/user_repo.py
from sqlalchemy.orm import Session
def get_user(session: Session, user_id: int):
return session.query(User).filter(User.id == user_id).first()
def test_get_user(mocker):
mock_session = mocker.MagicMock()
mock_user = Mock(id=1, email="user@test.com")
mock_session.query.return_value.filter.return_value.first.return_value = mock_user
result = get_user(mock_session, 1)
assert result.email == "user@test.com"
4. Mocking the filesystem
mock_open to read files
# config_loader.py
import json
def read_config(path: str) -> dict:
with open(path, "r") as f:
return json.load(f)
# tests/test_config_loader.py
from unittest.mock import mock_open, patch
from config_loader import read_config
def test_read_config(mocker):
mock_file = mock_open(read_data='{"key": "value", "port": 8080}')
mocker.patch("builtins.open", mock_file)
config = read_config("config.json")
assert config["key"] == "value"
assert config["port"] == 8080
mock_file.assert_called_once_with("config.json", "r")
With unittest.mock (without pytest-mock)
from unittest.mock import mock_open, patch
@patch("builtins.open", mock_open(read_data='{"key": "value"}'))
def test_read_config():
config = read_config("config.json")
assert config["key"] == "value"
Important: You patch builtins.open because open is a built-in function. The path depends on the module where it's used: if config_loader imports and uses open, you could use config_loader.open if you did from builtins import open — but the simplest is builtins.open or the module that contains the function under test.
To be safe, patch in the module that uses open:
@patch("config_loader.open", mock_open(read_data='{"key": "value"}'))
def test_read_config():
config = read_config("config.json")
assert config["key"] == "value"
5. Mocking time and datetime
datetime.utcnow
# auth/utils.py
from datetime import datetime
def is_expired(token: dict) -> bool:
"""Checks whether a JWT token has expired."""
exp_str = token.get("exp")
if not exp_str:
return True
exp = datetime.fromisoformat(exp_str)
return datetime.utcnow() >= exp
# tests/test_auth.py
from unittest.mock import patch
from datetime import datetime
from auth.utils import is_expired
@patch("auth.utils.datetime")
def test_is_expired_true(mock_dt):
mock_dt.utcnow.return_value = datetime(2024, 6, 15, 12, 0, 0)
mock_dt.fromisoformat = datetime.fromisoformat
token = {"exp": "2024-06-15T11:00:00"}
assert is_expired(token) is True
@patch("auth.utils.datetime")
def test_is_expired_false(mock_dt):
mock_dt.utcnow.return_value = datetime(2024, 6, 15, 10, 0, 0)
mock_dt.fromisoformat = datetime.fromisoformat
token = {"exp": "2024-06-15T12:00:00"}
assert is_expired(token) is False
If you mock the whole datetime module, you must preserve fromisoformat because your code uses it. Alternatively, patch only utcnow:
@patch("auth.utils.datetime.utcnow")
def test_is_expired(mock_utcnow):
mock_utcnow.return_value = datetime(2024, 6, 15, 12, 0, 0)
token = {"exp": "2024-06-15T11:00:00"}
assert is_expired(token) is True
6. Mocking environment variables
# config.py
import os
def get_api_key() -> str:
return os.environ.get("API_KEY", "")
def test_get_api_key(mocker):
mocker.patch.dict("os.environ", {"API_KEY": "test-key-123"}, clear=False)
assert get_api_key() == "test-key-123"
Or with patch:
from unittest.mock import patch
@patch.dict("os.environ", {"API_KEY": "test-key-123"})
def test_get_api_key():
from config import get_api_key
assert get_api_key() == "test-key-123"
7. Generating mocks with Claude Code
Effective prompts
You can ask Claude Code to generate mocks for a specific service:
Generate mocks to test this function that calls the weather API. Include:
- A mock of a successful response (200, JSON with temp and condition)
- A mock of a 500 error
- A mock of a timeout
- Tests using pytest and unittest.mock
Or more specifically:
I need tests for
send_notification, which usesrequests.post. Generate:
- test_send_notification_success — the mock returns {"sent": true}
- test_send_notification_api_error — the mock raises HTTPError
- test_send_notification_timeout — the mock raises Timeout Use @patch in the module where requests is used.
Claude Code usually generates tests with the correct structure. Check that the patch path is the right one (where it's used, not where it's defined).
8. A complete example: a Weather Service
A weather service that consumes an external API, with complete tests using mocks.
The service's code
# weather_service.py
"""
A weather service that consumes an external API.
"""
import os
import httpx
from typing import Any
WEATHER_API_URL = "https://api.weather.example.com/v1/current"
WEATHER_API_KEY = os.environ.get("WEATHER_API_KEY", "")
def get_current_weather(city: str) -> dict[str, Any]:
"""
Gets the current weather for a city.
Returns:
a dict with keys: temp_c, condition, humidity, error (optional)
"""
if not WEATHER_API_KEY:
return {"error": "API key not configured"}
try:
response = httpx.get(
f"{WEATHER_API_URL}",
params={"q": city, "key": WEATHER_API_KEY},
timeout=10,
)
response.raise_for_status()
data = response.json()
return {
"temp_c": data["current"]["temp_c"],
"condition": data["current"]["condition"]["text"],
"humidity": data["current"]["humidity"],
}
except httpx.HTTPStatusError as e:
return {"error": f"HTTP error: {e.response.status_code}"}
except httpx.TimeoutException:
return {"error": "Request timeout"}
except Exception as e:
return {"error": str(e)}
The complete tests
# tests/test_weather_service.py
"""
Tests of the weather service using mocks.
"""
import pytest
from unittest.mock import patch, Mock
# Import after a possible patch of os.environ
import weather_service
@patch.dict("os.environ", {"WEATHER_API_KEY": "test-key-123"}, clear=False)
@patch("weather_service.httpx.get")
def test_get_current_weather_success(mock_get, monkeypatch):
monkeypatch.setattr(weather_service, "WEATHER_API_KEY", "test-key-123")
mock_response = Mock()
mock_response.json.return_value = {
"current": {
"temp_c": 22.5,
"condition": {"text": "Partly cloudy"},
"humidity": 65,
}
}
mock_response.raise_for_status = Mock()
mock_get.return_value = mock_response
result = weather_service.get_current_weather("Madrid")
assert result["temp_c"] == 22.5
assert result["condition"] == "Partly cloudy"
assert result["humidity"] == 65
assert "error" not in result
mock_get.assert_called_once()
@patch.dict("os.environ", {"WEATHER_API_KEY": "test-key"}, clear=False)
@patch("weather_service.httpx.get")
def test_get_current_weather_http_error(mock_get, monkeypatch):
monkeypatch.setattr(weather_service, "WEATHER_API_KEY", "test-key")
mock_response = Mock()
mock_response.status_code = 500
mock_get.return_value = mock_response
mock_get.return_value.raise_for_status.side_effect = Exception("500")
import httpx
mock_get.return_value.raise_for_status.side_effect = httpx.HTTPStatusError(
"Server Error", request=Mock(), response=Mock(status_code=500)
)
result = weather_service.get_current_weather("Madrid")
assert "error" in result
assert "500" in result["error"]
@patch.dict("os.environ", {"WEATHER_API_KEY": "test-key"}, clear=False)
@patch("weather_service.httpx.get")
def test_get_current_weather_timeout(mock_get, monkeypatch):
monkeypatch.setattr(weather_service, "WEATHER_API_KEY", "test-key")
import httpx
mock_get.side_effect = httpx.TimeoutException("Connection timed out")
result = weather_service.get_current_weather("Madrid")
assert result["error"] == "Request timeout"
@patch.dict("os.environ", {}, clear=False)
def test_get_current_weather_no_api_key(monkeypatch):
monkeypatch.setattr(weather_service, "WEATHER_API_KEY", "")
result = weather_service.get_current_weather("Madrid")
assert result["error"] == "API key not configured"
A simplified version that avoids problems with the module import and the environment variables:
# tests/test_weather_service.py - the simplified version
import pytest
from unittest.mock import patch, Mock
import httpx
@patch.dict("os.environ", {"WEATHER_API_KEY": "test-key-123"})
@patch("weather_service.httpx.get")
def test_weather_success(mock_get):
import importlib
import weather_service
importlib.reload(weather_service)
mock_get.return_value = Mock(
json=lambda: {"current": {"temp_c": 22, "condition": {"text": "Sunny"}, "humidity": 50}},
raise_for_status=Mock(),
)
result = weather_service.get_current_weather("Madrid")
assert result["temp_c"] == 22
assert result["condition"] == "Sunny"
A cleaner implementation uses dependency injection for the HTTP client and the API key:
# weather_service_v2.py - with dependency injection
from typing import Any, Callable
def get_current_weather(
city: str,
fetch_weather: Callable[[str], dict],
api_key: str = "",
) -> dict[str, Any]:
if not api_key:
return {"error": "API key not configured"}
try:
data = fetch_weather(city)
return {
"temp_c": data["current"]["temp_c"],
"condition": data["current"]["condition"]["text"],
"humidity": data["current"]["humidity"],
}
except Exception as e:
return {"error": str(e)}
def test_weather_v2_success():
def mock_fetch(city):
return {
"current": {"temp_c": 20, "condition": {"text": "Cloudy"}, "humidity": 70}
}
result = get_current_weather("Barcelona", fetch_weather=mock_fetch, api_key="key")
assert result["temp_c"] == 20
assert result["condition"] == "Cloudy"
Injection simplifies the tests: you don't need to patch, you just pass a mock or a fake function.
Exercises
Exercise 1: Mocking HTTP with responses (Easy)
You have a fetch_products() function that does requests.get("https://api.example.com/products") and returns response.json(). Write a test that mocks the call and verifies that it returns a list with at least one product {"id": 1, "name": "Widget"}.
See solution
from unittest.mock import patch
from mymodule import fetch_products
@patch("mymodule.requests.get")
def test_fetch_products_returns_list(mock_get):
mock_get.return_value.json.return_value = [{"id": 1, "name": "Widget"}]
mock_get.return_value.raise_for_status = lambda: None
result = fetch_products()
assert len(result) >= 1
assert result[0]["name"] == "Widget"
mock_get.assert_called_once_with("https://api.example.com/products")
Exercise 2: Mocking OpenAI (Medium)
A classify_sentiment(text: str, client: OpenAI) -> str function calls client.chat.completions.create() and returns choices[0].message.content. Write a test that passes a mock of the client and verifies that it returns "positive".
See solution
from unittest.mock import Mock
from mymodule import classify_sentiment
def test_classify_sentiment_returns_positive():
mock_client = Mock()
mock_response = Mock()
mock_response.choices = [Mock()]
mock_response.choices[0].message = Mock()
mock_response.choices[0].message.content = "positive"
mock_client.chat.completions.create.return_value = mock_response
result = classify_sentiment("I love this!", mock_client)
assert result == "positive"
Exercise 3: Mocking a database (Medium)
A get_user_by_id(repo, user_id: int) function calls repo.get(user_id) and returns the user or None. Write two tests: one where the repo returns a user and another where it returns None.
See solution
from unittest.mock import Mock
from mymodule import get_user_by_id
def test_get_user_by_id_found():
mock_repo = Mock()
mock_repo.get.return_value = {"id": 1, "name": "Alice"}
user = get_user_by_id(mock_repo, 1)
assert user["name"] == "Alice"
mock_repo.get.assert_called_once_with(1)
def test_get_user_by_id_not_found():
mock_repo = Mock()
mock_repo.get.return_value = None
user = get_user_by_id(mock_repo, 999)
assert user is None
Exercise 4: Mocking a file with mock_open (Medium)
A load_secrets(path: str) -> dict function reads a JSON from the filesystem. Write a test that mocks open and verifies that it returns {"db_password": "secret123"}.
See solution
from unittest.mock import mock_open, patch
from mymodule import load_secrets
@patch("mymodule.open", mock_open(read_data='{"db_password": "secret123"}'))
def test_load_secrets():
result = load_secrets("/etc/secrets.json")
assert result["db_password"] == "secret123"
If load_secrets is in mymodule, patch mymodule.open. If it uses open directly, patch builtins.open.
Exercise 5: Mocking datetime (Medium)
An is_weekend() -> bool function uses datetime.now().weekday() and returns True if it's Saturday (5) or Sunday (6). Write a test that mocks datetime to simulate a Saturday and verifies that it returns True.
See solution
from unittest.mock import patch
from datetime import datetime
@patch("mymodule.datetime")
def test_is_weekend_saturday(mock_dt):
mock_dt.now.return_value = datetime(2024, 6, 15) # Saturday
mock_dt.now.return_value.weekday = lambda: 5
from mymodule import is_weekend
assert is_weekend() is True
Or by creating a real datetime with weekday 5:
@patch("mymodule.datetime")
def test_is_weekend_saturday(mock_dt):
saturday = datetime(2024, 6, 15)
mock_dt.now.return_value = saturday
from mymodule import is_weekend
assert is_weekend() is True
Exercise 6: Mocking an environment variable (Easy)
A get_database_url() -> str function returns os.environ.get("DATABASE_URL", "sqlite:///default.db"). Write a test that sets DATABASE_URL=postgres://localhost/test and verifies that the function returns that URL.
See solution
from unittest.mock import patch
from mymodule import get_database_url
@patch.dict("os.environ", {"DATABASE_URL": "postgres://localhost/test"})
def test_get_database_url():
assert get_database_url() == "postgres://localhost/test"
With pytest-mock:
def test_get_database_url(mocker):
mocker.patch.dict("os.environ", {"DATABASE_URL": "postgres://localhost/test"})
from mymodule import get_database_url
assert get_database_url() == "postgres://localhost/test"
Troubleshooting
Problem 1: patch doesn't replace the call — it keeps using the real service
Cause: You're patching in the wrong namespace. If mymodule does from requests import get and uses get(), you must patch mymodule.get, not requests.get.
Solution: Always patch where the object is used. If in doubt, look for the import in the file under test and patch that namespace.
Problem 2: An OpenAI/Anthropic mock — AttributeError on choices[0].message
Cause: The mock doesn't have the nested structure the code expects. Mock() creates automatic mocks for attributes, but sometimes .strip() or index access fails.
Solution: Build the structure explicitly:
mock_response = Mock()
mock_message = Mock()
mock_message.content = "text"
mock_message.strip.return_value = "text"
mock_response.choices = [Mock(message=mock_message)]
mock_client.chat.completions.create.return_value = mock_response
Problem 3: mock_open doesn't apply — it keeps reading real files
Cause: You're patching open in a different module from the one that uses it. Or you're using mock_open incorrectly (for example, without passing read_data).
Solution: Patch in the module that opens the file. Make sure mock_open(read_data='...') receives a valid string (e.g. JSON if the code does json.load).
Problem 4: The environment variable isn't reflected in the test
Cause: The module already imported os.environ or the variable before the patch was applied. Some modules cache os.environ.get("X") at import time.
Solution: Use patch.dict before importing the module under test, or use importlib.reload(module) after the patch. Better yet: inject the configuration as a parameter to avoid depending on os.environ at import time.
Problem 5: A test passes in isolation but fails when it runs with other tests
Cause: A mock or patch persists between tests and contaminates the state. For example, a module-level patch without a decorator or context manager.
Solution: Use @patch as a decorator or with patch(...): so the restoration is automatic. With pytest-mock, mocker.patch cleans up at the end of the test.
Project Connection
In module 6's Validation pipeline project:
- You'll mock the OpenAI API (or a similar one) to simulate text generation responses
- You'll mock the database service to simulate reads/writes without a real PostgreSQL
- You'll use the techniques from this capsule: patching HTTP, mocking AI clients, mocking repositories
- The pipeline will process input data, send it to the mocks, and the tests will verify the complete flow
The next capsule (Capsule 06) is the module's project: building the validation pipeline with mocks, fixtures and validation loops. This capsule gives you the tools to correctly mock the external services the pipeline consumes.
Summary
- HTTP (requests/httpx): Use
@patchon the client or the method (requests.get,httpx.get) in the namespace where it's used. Configurereturn_value.json,raise_for_status, andside_effectfor errors and timeouts. - AI APIs (OpenAI/Anthropic): Always mock — cost, non-determinism and rate limits. Build the response structure (
choices[0].message.content) explicitly. - Databases: Mock the access layer (repository, dao, session), not SQL directly.
- Filesystem: Use
mock_open(read_data='...')and patchopenin the module that uses it. - Time: Patch
datetimeordatetime.utcnowto fix the "current" moment. - Environment variables: Use
patch.dict("os.environ", {...})ormocker.patch.dict. - Claude Code: It can generate complete mocks if you tell it the service and the scenarios (success, error, timeout).
- Dependency injection: It makes tests easier — pass mocks directly instead of patching.
Additional Resources
- unittest.mock — mock_open - mock_open documentation
- responses library - More realistic HTTP mocking for requests
- pytest-mock - The mocker fixture for pytest
- Where to patch - The golden rule of patching
- Testing HTTP with httpx - httpx documentation for testing
- VCR.py - Record and replay real HTTP responses (an alternative to manual mocks)
Module 6, Capsule 05 — Testing with Claude Code Guide