Mastering Mock Objects in Python¶

Overview

This guide takes you from the fundamentals of mocking all the way to production-grade patterns — covering unittest.mock, patching strategies, mocking external HTTP APIs, and AWS services with moto.

What is Mocking?¶

When unit testing, you want to test one piece of code in isolation. But real code communicates with databases, REST APIs, file systems, and cloud services. These collaborators make tests slow, flaky, and expensive.

A mock object is a fake replacement that stands in for a real dependency. You control exactly what it returns and verify exactly how it was called.

Why Mock?¶

Benefit	Description
⚡ Speed	No real network calls — tests run in milliseconds
🔒 Isolation	Test only your logic, not your dependencies
🎯 Control	Simulate errors, timeouts, edge cases on demand
💰 Cost	No AWS charges, no rate limits, no side effects

Test Double Glossary¶

Type	What it does	When to use
Stub	Returns pre-canned data	When you need a fixed response
Fake	Lightweight working implementation	In-memory DB instead of real one
Mock	Records calls, verifiable later	When you need to assert how it was used

No install needed

unittest.mock is built into Python 3.3+. Just import it: python from unittest.mock import Mock, MagicMock, patch

MagicMock Basics¶

MagicMock is the most commonly used mock class. It accepts any attribute access or method call without raising errors, auto-creating child mocks on the fly.

Creating Your First Mock¶

from unittest.mock import MagicMock

# Create a mock — it accepts anything
db = MagicMock()

# Calling methods returns another MagicMock by default
result = db.get_user(42)
print(result)        # <MagicMock name='mock.get_user()' id='...'>

# Set a return value
db.get_user.return_value = {"id": 42, "name": "Alice"}

user = db.get_user(42)
print(user["name"])  # "Alice"

Mock vs MagicMock¶

Feature	`Mock`	`MagicMock`
Basic call recording	✅	✅
Magic methods (`__len__`, `__iter__`, etc.)	❌	✅
Context manager (`with` statement)	❌	✅
Iteration support	❌	✅

Rule of thumb

Use MagicMock by default. Only drop down to Mock if you want magic method access to raise AttributeError intentionally.

Configuring Return Values¶

from unittest.mock import MagicMock

mock_service = MagicMock()

# Simple return value
mock_service.get_price.return_value = 9.99

# Nested attribute
mock_service.config.timeout = 30

# Chained calls
mock_service.db.session.query.return_value.first.return_value = {"id": 1}

print(mock_service.get_price())                          # 9.99
print(mock_service.config.timeout)                       # 30
print(mock_service.db.session.query().first())           # {'id': 1}

Using `configure_mock()`¶

mock = MagicMock()
mock.configure_mock(**{
    "name": "payment-service",
    "charge.return_value": {"status": "ok", "amount": 50.0},
    "refund.return_value": {"status": "refunded"},
})

print(mock.name)            # payment-service
print(mock.charge())        # {'status': 'ok', 'amount': 50.0}

The `patch()` Decorator¶

patch() temporarily replaces an object in the module under test with a mock for the duration of the test.

Critical rule: patch where it's used

Always patch the name as it's imported in the module under test — not where it's originally defined.

myapp/orders.py

import requests

# ✅ Correct — patch where it's used
@patch("myapp.orders.requests.get")

# ❌ Wrong — patching the source doesn't affect the already-imported name
@patch("requests.get")

Basic Usage¶

myapp/weather.py

import requests

def get_temperature(city: str) -> float:
    response = requests.get(f"https://api.weather.com/{city}")
    return response.json()["temp"]

tests/test_weather.py

import unittest
from unittest.mock import patch, MagicMock
from myapp.weather import get_temperature

class TestWeather(unittest.TestCase):

    @patch("myapp.weather.requests.get")
    def test_get_temperature(self, mock_get):
        # Arrange
        mock_response = MagicMock()
        mock_response.json.return_value = {"temp": 22.5}
        mock_get.return_value = mock_response

        # Act
        result = get_temperature("London")

        # Assert
        self.assertEqual(result, 22.5)
        mock_get.assert_called_once_with("https://api.weather.com/London")

`patch` as a Context Manager¶

def test_get_temperature_context_manager():
    with patch("myapp.weather.requests.get") as mock_get:
        mock_get.return_value.json.return_value = {"temp": 15.0}
        result = get_temperature("Paris")
        assert result == 15.0

`patch.object()` — Patch a Specific Attribute¶

from unittest.mock import patch
from myapp import payment_service

class PaymentClient:
    def charge(self, amount):
        return payment_service.process(amount)

def test_charge():
    client = PaymentClient()
    with patch.object(payment_service, "process", return_value={"status": "ok"}) as mock_proc:
        result = client.charge(100)
        assert result == {"status": "ok"}
        mock_proc.assert_called_once_with(100)

`patch.dict()` — Patch Dictionaries / Environment Variables¶

import os
from unittest.mock import patch

def get_db_url():
    return os.environ.get("DATABASE_URL", "sqlite:///:memory:")

def test_db_url_from_env():
    with patch.dict(os.environ, {"DATABASE_URL": "postgresql://localhost/testdb"}):
        assert get_db_url() == "postgresql://localhost/testdb"

Stacking Multiple Patches¶

@patch("myapp.orders.send_email")
@patch("myapp.orders.charge_card")
@patch("myapp.orders.save_to_db")
def test_place_order(mock_save, mock_charge, mock_email):
    # Note: decorators apply bottom-up, arguments inject top-down
    mock_charge.return_value = {"transaction_id": "txn_123"}
    mock_save.return_value = True
    mock_email.return_value = None

    from myapp.orders import place_order
    place_order(user_id=1, amount=50.0)

    mock_charge.assert_called_once()
    mock_save.assert_called_once()
    mock_email.assert_called_once()

Assertions on Mocks¶

Mocks record every interaction. Use built-in assertion methods to verify behaviour.

Call Assertions¶

from unittest.mock import MagicMock, call

mock = MagicMock()
mock("hello")
mock("world")
mock(x=42)

# Was it called at all?
mock.assert_called()

# Called exactly once?
# mock.assert_called_once()  # would fail — called 3 times

# Called with specific args?
mock.assert_any_call("hello")
mock.assert_any_call(x=42)

# Check all calls in order
mock.assert_has_calls([call("hello"), call("world")], any_order=False)

Inspecting Call History¶

mock = MagicMock()
mock(1, 2, key="val")
mock(3, 4)

print(mock.call_count)        # 2
print(mock.call_args)         # call(3, 4)  — last call
print(mock.call_args_list)    # [call(1, 2, key='val'), call(3, 4)]
print(mock.called)            # True

# Reset recorded calls (not return_value)
mock.reset_mock()
print(mock.call_count)        # 0

Asserting NOT Called¶

mock = MagicMock()

mock.assert_not_called()  # ✅ passes

mock()
mock.assert_not_called()  # ❌ raises AssertionError

Side Effects¶

side_effect lets you go beyond a fixed return value — raise exceptions, return different values on successive calls, or run custom logic.

Raising Exceptions¶

from unittest.mock import MagicMock

mock_db = MagicMock()
mock_db.connect.side_effect = ConnectionError("DB is down")

try:
    mock_db.connect()
except ConnectionError as e:
    print(e)  # "DB is down"

Returning Different Values on Each Call¶

mock = MagicMock()
mock.side_effect = [10, 20, 30]

print(mock())  # 10
print(mock())  # 20
print(mock())  # 30
# mock()       # raises StopIteration

Using a Function as a Side Effect¶

def dynamic_response(user_id):
    if user_id == 0:
        raise ValueError("Invalid user")
    return {"id": user_id, "name": f"User {user_id}"}

mock_db = MagicMock()
mock_db.get_user.side_effect = dynamic_response

print(mock_db.get_user(5))    # {'id': 5, 'name': 'User 5'}
mock_db.get_user(0)           # raises ValueError: Invalid user

Simulating Intermittent Failures (Retry Logic)¶

from unittest.mock import patch, MagicMock
import requests

def fetch_with_retry(url, retries=3):
    for i in range(retries):
        try:
            return requests.get(url).json()
        except requests.Timeout:
            if i == retries - 1:
                raise
    return None

def test_retry_succeeds_on_third_attempt():
    mock_response = MagicMock()
    mock_response.json.return_value = {"data": "ok"}

    with patch("requests.get") as mock_get:
        mock_get.side_effect = [
            requests.Timeout,
            requests.Timeout,
            mock_response,   # third attempt succeeds
        ]
        result = fetch_with_retry("https://api.example.com/data")
        assert result == {"data": "ok"}
        assert mock_get.call_count == 3

Spec & Autospec¶

By default, mocks accept any attribute or call — even invalid ones. This means typos in your tests go undetected. spec and autospec fix this.

`spec` — Restrict to Real Interface¶

from unittest.mock import MagicMock

class UserService:
    def get_user(self, user_id: int) -> dict: ...
    def delete_user(self, user_id: int) -> bool: ...

# Without spec — typo goes unnoticed
mock = MagicMock()
mock.get_usre(42)   # ✅ no error — but it's a typo!

# With spec — typo caught immediately
mock = MagicMock(spec=UserService)
mock.get_usre(42)   # ❌ AttributeError: Mock object has no attribute 'get_usre'
mock.get_user(42)   # ✅ works fine

`autospec` — Full Signature Enforcement¶

autospec goes further: it validates argument signatures, not just attribute names.

from unittest.mock import create_autospec

class EmailService:
    def send(self, to: str, subject: str, body: str) -> bool: ...

mock_email = create_autospec(EmailService)

# ✅ Correct signature
mock_email.send("alice@example.com", "Hello", "Hi there")

# ❌ Wrong — missing required args
mock_email.send("alice@example.com")
# TypeError: missing a required argument: 'subject'

`autospec` with `patch`¶

@patch("myapp.notifications.EmailService", autospec=True)
def test_send_welcome_email(mock_email_cls):
    instance = mock_email_cls.return_value
    instance.send.return_value = True

    from myapp.notifications import send_welcome_email
    send_welcome_email("alice@example.com")

    instance.send.assert_called_once_with(
        "alice@example.com",
        subject="Welcome!",
        body=unittest.mock.ANY,
    )

Use autospec=True in production tests

It catches API drift — if the real class changes its signature, your tests will break immediately rather than silently passing with the wrong interface.

Mocking External HTTP APIs¶

Using `responses` (for `requests`)¶

pip install responses

myapp/github.py

import requests

def get_repo_stars(owner: str, repo: str) -> int:
    url = f"https://api.github.com/repos/{owner}/{repo}"
    data = requests.get(url).json()
    return data["stargazers_count"]

tests/test_github.py

import responses
import unittest
from myapp.github import get_repo_stars

class TestGitHub(unittest.TestCase):

    @responses.activate
    def test_get_repo_stars(self):
        responses.add(
            responses.GET,
            "https://api.github.com/repos/psf/requests",
            json={"stargazers_count": 50000, "name": "requests"},
            status=200,
        )

        stars = get_repo_stars("psf", "requests")
        self.assertEqual(stars, 50000)

    @responses.activate
    def test_handles_api_error(self):
        responses.add(
            responses.GET,
            "https://api.github.com/repos/psf/requests",
            json={"message": "Not Found"},
            status=404,
        )

        with self.assertRaises(KeyError):
            get_repo_stars("psf", "requests")

Using `httpx` + `respx` (async-friendly)¶

pip install httpx respx

myapp/async_client.py

import httpx

async def fetch_user(user_id: int) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.example.com/users/{user_id}")
        r.raise_for_status()
        return r.json()

tests/test_async_client.py

import pytest
import respx
import httpx
from myapp.async_client import fetch_user

@pytest.mark.anyio
@respx.mock
async def test_fetch_user():
    respx.get("https://api.example.com/users/1").mock(
        return_value=httpx.Response(200, json={"id": 1, "name": "Alice"})
    )

    user = await fetch_user(1)
    assert user["name"] == "Alice"

Using `responses` with Callbacks¶

import responses
import json

def request_callback(request):
    payload = json.loads(request.body)
    if payload.get("amount", 0) > 10000:
        return (400, {}, json.dumps({"error": "Limit exceeded"}))
    return (200, {}, json.dumps({"status": "charged"}))

@responses.activate
def test_payment_limit():
    responses.add_callback(
        responses.POST,
        "https://payments.example.com/charge",
        callback=request_callback,
        content_type="application/json",
    )

    import requests
    r = requests.post(
        "https://payments.example.com/charge",
        json={"amount": 99999}
    )
    assert r.status_code == 400
    assert r.json()["error"] == "Limit exceeded"

Mocking AWS with moto¶

moto intercepts boto3 calls and runs a fully functional fake AWS in-process.

pip install moto[s3,sqs,dynamodb,iam,sts]  # install only what you need

S3 — Upload and Read Files¶

myapp/storage.py

import boto3

def upload_report(bucket: str, key: str, content: str) -> str:
    s3 = boto3.client("s3", region_name="us-east-1")
    s3.put_object(Bucket=bucket, Key=key, Body=content.encode())
    return f"s3://{bucket}/{key}"

def download_report(bucket: str, key: str) -> str:
    s3 = boto3.client("s3", region_name="us-east-1")
    obj = s3.get_object(Bucket=bucket, Key=key)
    return obj["Body"].read().decode()

tests/test_storage.py

import boto3
import pytest
from moto import mock_aws
from myapp.storage import upload_report, download_report

@mock_aws
def test_upload_and_download_report():
    # Create the bucket first (moto gives you a blank slate)
    s3 = boto3.client("s3", region_name="us-east-1")
    s3.create_bucket(Bucket="reports")

    url = upload_report("reports", "q3/summary.txt", "Revenue: $1M")
    assert url == "s3://reports/q3/summary.txt"

    content = download_report("reports", "q3/summary.txt")
    assert content == "Revenue: $1M"

DynamoDB — Full Table Operations¶

myapp/users_table.py

import boto3

TABLE_NAME = "Users"

def put_user(user_id: str, name: str, email: str):
    ddb = boto3.resource("dynamodb", region_name="us-east-1")
    table = ddb.Table(TABLE_NAME)
    table.put_item(Item={"user_id": user_id, "name": name, "email": email})

def get_user(user_id: str) -> dict:
    ddb = boto3.resource("dynamodb", region_name="us-east-1")
    table = ddb.Table(TABLE_NAME)
    response = table.get_item(Key={"user_id": user_id})
    return response.get("Item")

tests/test_users_table.py

import boto3
import pytest
from moto import mock_aws
from myapp.users_table import put_user, get_user, TABLE_NAME

@pytest.fixture
def dynamodb_table():
    with mock_aws():
        ddb = boto3.resource("dynamodb", region_name="us-east-1")
        table = ddb.create_table(
            TableName=TABLE_NAME,
            KeySchema=[{"AttributeName": "user_id", "KeyType": "HASH"}],
            AttributeDefinitions=[{"AttributeName": "user_id", "AttributeType": "S"}],
            BillingMode="PAY_PER_REQUEST",
        )
        table.meta.client.get_waiter("table_exists").wait(TableName=TABLE_NAME)
        yield table

def test_put_and_get_user(dynamodb_table):
    put_user("u1", "Alice", "alice@example.com")
    user = get_user("u1")
    assert user["name"] == "Alice"
    assert user["email"] == "alice@example.com"

def test_get_missing_user_returns_none(dynamodb_table):
    user = get_user("does-not-exist")
    assert user is None

SQS — Send and Consume Messages¶

myapp/queue.py

import boto3

def send_message(queue_url: str, body: str) -> str:
    sqs = boto3.client("sqs", region_name="us-east-1")
    r = sqs.send_message(QueueUrl=queue_url, MessageBody=body)
    return r["MessageId"]

def receive_messages(queue_url: str, max_msgs: int = 10) -> list:
    sqs = boto3.client("sqs", region_name="us-east-1")
    r = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=max_msgs)
    return r.get("Messages", [])

tests/test_queue.py

import boto3
from moto import mock_aws
from myapp.queue import send_message, receive_messages

@mock_aws
def test_send_and_receive():
    sqs = boto3.client("sqs", region_name="us-east-1")
    queue = sqs.create_queue(QueueName="jobs")
    url = queue["QueueUrl"]

    msg_id = send_message(url, '{"job": "export_csv"}')
    assert msg_id  # non-empty string

    messages = receive_messages(url)
    assert len(messages) == 1
    assert messages[0]["Body"] == '{"job": "export_csv"}'

SSM Parameter Store¶

myapp/config.py

import boto3

def get_secret(name: str) -> str:
    ssm = boto3.client("ssm", region_name="us-east-1")
    r = ssm.get_parameter(Name=name, WithDecryption=True)
    return r["Parameter"]["Value"]

@mock_aws
def test_get_secret():
    ssm = boto3.client("ssm", region_name="us-east-1")
    ssm.put_parameter(
        Name="/myapp/db/password",
        Value="supersecret",
        Type="SecureString",
    )

    from myapp.config import get_secret
    value = get_secret("/myapp/db/password")
    assert value == "supersecret"

Always set AWS credentials for moto

moto requires boto3 to be configured, even with fake values:

conftest.py

import os
import pytest

@pytest.fixture(autouse=True)
def aws_credentials():
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"

Advanced Patterns¶

Reusable Mock Fixtures with `pytest`¶

conftest.py

import pytest
from unittest.mock import MagicMock, create_autospec
from myapp.services import EmailService, PaymentService

@pytest.fixture
def mock_email():
    mock = create_autospec(EmailService)
    mock.send.return_value = True
    return mock

@pytest.fixture
def mock_payment():
    mock = create_autospec(PaymentService)
    mock.charge.return_value = {"status": "ok", "transaction_id": "txn_001"}
    return mock

tests/test_checkout.py

def test_checkout_sends_confirmation(mock_email, mock_payment):
    from myapp.checkout import process_checkout
    process_checkout(user_id=1, amount=99.0,
                     email_svc=mock_email, payment_svc=mock_payment)

    mock_payment.charge.assert_called_once_with(user_id=1, amount=99.0)
    mock_email.send.assert_called_once()
    _, kwargs = mock_email.send.call_args
    assert "confirmation" in kwargs.get("subject", "").lower()

Mocking Class Instances with `patch`¶

myapp/processor.py

from myapp.clients import S3Client

class ReportProcessor:
    def __init__(self):
        self.client = S3Client()

    def run(self, bucket, key):
        data = self.client.download(bucket, key)
        return data.upper()

@patch("myapp.processor.S3Client")
def test_report_processor(mock_s3_cls):
    # mock_s3_cls is the class — configure its instance
    instance = mock_s3_cls.return_value
    instance.download.return_value = "hello world"

    from myapp.processor import ReportProcessor
    processor = ReportProcessor()
    result = processor.run("my-bucket", "report.txt")

    assert result == "HELLO WORLD"
    instance.download.assert_called_once_with("my-bucket", "report.txt")

Mocking `datetime.now()`¶

myapp/scheduler.py

from datetime import datetime

def is_business_hours() -> bool:
    now = datetime.now()
    return 9 <= now.hour < 17

from unittest.mock import patch
from datetime import datetime

def test_is_business_hours_at_noon():
    fake_now = datetime(2024, 6, 15, 12, 0, 0)
    with patch("myapp.scheduler.datetime") as mock_dt:
        mock_dt.now.return_value = fake_now
        from myapp.scheduler import is_business_hours
        assert is_business_hours() is True

def test_is_not_business_hours_at_midnight():
    fake_now = datetime(2024, 6, 15, 0, 0, 0)
    with patch("myapp.scheduler.datetime") as mock_dt:
        mock_dt.now.return_value = fake_now
        from myapp.scheduler import is_business_hours
        assert is_business_hours() is False

Property Mocking with `PropertyMock`¶

from unittest.mock import MagicMock, PropertyMock, patch

class Config:
    @property
    def debug(self) -> bool:
        return False

def test_debug_mode():
    with patch.object(Config, "debug", new_callable=PropertyMock) as mock_debug:
        mock_debug.return_value = True
        cfg = Config()
        assert cfg.debug is True

Capturing and Asserting on Complex Call Args¶

from unittest.mock import MagicMock, call, ANY

mock_notify = MagicMock()

# Calling with a complex payload
mock_notify(
    event="user_created",
    payload={"user_id": 99, "timestamp": "2024-01-01T00:00:00Z"}
)

# Use ANY to ignore fields you don't care about
mock_notify.assert_called_once_with(
    event="user_created",
    payload={"user_id": 99, "timestamp": ANY},
)

Async Mock with `AsyncMock`¶

myapp/async_service.py

import httpx

async def get_exchange_rate(from_ccy: str, to_ccy: str) -> float:
    async with httpx.AsyncClient() as client:
        r = await client.get(
            "https://fx.example.com/rate",
            params={"from": from_ccy, "to": to_ccy}
        )
        return r.json()["rate"]

tests/test_async_service.py

import pytest
from unittest.mock import AsyncMock, patch, MagicMock

@pytest.mark.asyncio
async def test_get_exchange_rate():
    mock_response = MagicMock()
    mock_response.json.return_value = {"rate": 1.25}

    mock_client = AsyncMock()
    mock_client.__aenter__.return_value.get = AsyncMock(return_value=mock_response)

    with patch("myapp.async_service.httpx.AsyncClient", return_value=mock_client):
        from myapp.async_service import get_exchange_rate
        rate = await get_exchange_rate("GBP", "USD")
        assert rate == 1.25

Quick Reference¶

Cheat Sheet¶

Task	Code
Create a mock	`m = MagicMock()`
Set return value	`m.method.return_value = value`
Raise on call	`m.method.side_effect = ValueError("oops")`
Return sequence	`m.method.side_effect = [1, 2, 3]`
Patch a function	`@patch("module.function")`
Patch an attribute	`@patch.object(obj, "attr", value)`
Patch env variable	`@patch.dict(os.environ, {"KEY": "val"})`
Enforce interface	`create_autospec(MyClass)`
Assert called once	`m.assert_called_once_with(arg)`
Assert not called	`m.assert_not_called()`
Count calls	`m.call_count`
Last call args	`m.call_args`
All calls	`m.call_args_list`
Reset mock	`m.reset_mock()`
Ignore an arg	`unittest.mock.ANY`
Mock async func	`AsyncMock()`
Mock property	`PropertyMock()`
Mock AWS S3	`@mock_aws` from `moto`

Common Pitfalls¶

Pitfall 1: Patching the wrong location

python # myapp/orders.py does: from datetime import datetime # ✅ Correct @patch("myapp.orders.datetime") # ❌ Wrong — too late, already imported @patch("datetime.datetime")

Pitfall 2: Forgetting return_value for instance methods

python mock_cls = MagicMock() # ❌ This configures the class itself mock_cls.get_user.return_value = user # ✅ This configures the *instance* returned by calling the class mock_cls.return_value.get_user.return_value = user

Pitfall 3: Not resetting mocks between tests

Use reset_mock() or create a fresh MagicMock() per test. With pytest, use fixtures to get a fresh mock each time.

Pitfall 4: moto requires a real-ish region and credentials

python # Always set before any boto3 call under moto os.environ["AWS_DEFAULT_REGION"] = "us-east-1" os.environ["AWS_ACCESS_KEY_ID"] = "testing" os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"