Mocking and Test Doubles (unittest.mock)

Updated

September 7, 2026

Mocking and Test Doubles (unittest.mock)

After reading this chapter, you will master Python’s testing and isolation toolkit using the standard library unittest.mock module, replace unreliable external boundaries (databases, HTTP APIs, file systems, clocks) with robust test doubles, intercept functions and classes using @patch and patch.object, enforce API contracts and prevent mock interface drift using autospec=True, simulate multi-step network failures with side_effect, and test asynchronous coroutines using AsyncMock.

Mental model

Unit tests should be deterministic, isolated, and fast. However, production code frequently depends on non-deterministic boundaries: remote webhooks, cloud databases, external clock times, or disk storage.

The unittest.mock library creates test doubles—objects that impersonate real components, record how they were called, and return pre-programmed responses:

┌─────────────────────────────────────────────────────────────┐
│ Production System Flow                                      │
│   Client Code ──▶ [ Real Database / Stripe API / SMTP ]    │
│                   (Slow, unreliable, requires credentials)  │
└─────────────────────────────────────────────────────────────┘
                              │
                      Test Substitution
                              ▼
┌─────────────────────────────────────────────────────────────┐
│ Isolated Unit Test Flow                                     │
│   Client Code ──▶ [ unittest.mock.Mock ]                    │
│                     ├─ return_value: {'status': 'PAID'}     │
│                     ├─ side_effect: ConnectionError()       │
│                     └─ records: call_count, call_args       │
└─────────────────────────────────────────────────────────────┘

The Golden Rule of Patching: “Where to Patch”

The single most common bug when using unittest.mock.patch is patching the module where a class or function was defined, rather than where it was imported and looked up:

WRONG:
  patch('urllib.request.urlopen')  # Does NOT affect service.py!
  service.py imported: from urllib.request import urlopen (local reference)

RIGHT:
  patch('my_app.service.urlopen')  # Patches the reference inside service.py!

Minimal example

Save as mock_overview.py:

# mock_overview.py
from unittest.mock import Mock, patch

def send_welcome_email(email_client, user_email: str) -> bool:
    """Sends a welcome message and returns True on success."""
    response = email_client.send(
        to=user_email,
        subject="Welcome aboard!",
        body="Thank you for joining our platform.",
    )
    return response.get("delivered", False)

def main() -> None:
    # 1. Instantiate a mock client
    mock_client = Mock()

    # Configure return value
    mock_client.send.return_value = {"delivered": True, "message_id": "msg_9012"}

    # 2. Execute target function under test
    result = send_welcome_email(mock_client, "alice@example.com")
    print(f"Function result: {result}")

    # 3. Assert mock interactions
    print(f"send() was called: {mock_client.send.called}")
    print(f"send() call count: {mock_client.send.call_count}")

    # Verify exact keyword arguments passed
    mock_client.send.assert_called_once_with(
        to="alice@example.com",
        subject="Welcome aboard!",
        body="Thank you for joining our platform.",
    )
    print("All mock assertions succeeded!")

if __name__ == "__main__":
    main()

Run via uv run python mock_overview.py:

Function result: True
send() was called: True
send() call count: 1
All mock assertions succeeded!

Worked examples

Case 1: Simulating Flaky Networks with side_effect Retries

A mock’s side_effect attribute can accept an exception class or instance (to raise errors) or an iterable (to yield different return values on consecutive calls). This enables testing exponential backoff and retry loops:

# retry_client_test.py
import time
from unittest.mock import Mock

class TransientNetworkError(Exception):
    pass

def fetch_telemetry_with_retry(api_client, endpoint: str, max_retries: int = 3) -> dict:
    """Fetches data from an API client, retrying up to max_retries on transient errors."""
    for attempt in range(1, max_retries + 1):
        try:
            return api_client.get(endpoint)
        except TransientNetworkError:
            if attempt == max_retries:
                raise
            time.sleep(0.01 * attempt)
    return {}

def test_retry_succeeds_on_third_attempt() -> None:
    mock_api = Mock()

    # Sequence: Fail with error twice, then succeed on third attempt
    mock_api.get.side_effect = [
        TransientNetworkError("Gateway Timeout (504)"),
        TransientNetworkError("Connection Reset"),
        {"nodes_online": 42, "status": "HEALTHY"},
    ]

    result = fetch_telemetry_with_retry(mock_api, "/api/v1/cluster/status")

    print(f"Payload received: {result}")
    print(f"Total API attempts made: {mock_api.get.call_count}")
    assert mock_api.get.call_count == 3
    print("Test passed: Retry loop successfully recovered on attempt 3!")

if __name__ == "__main__":
    test_retry_succeeds_on_third_attempt()

Run:

uv run python retry_client_test.py

Output:

Payload received: {'nodes_online': 42, 'status': 'HEALTHY'}
Total API attempts made: 3
Test passed: Retry loop successfully recovered on attempt 3!

Case 2: Preventing Mock Drift with autospec=True

Standard Mock and MagicMock instances dynamically create any attribute or method accessed on them. If production code refactors a method name or changes parameter signatures, tests using vanilla mocks continue to pass silently. autospec=True binds the mock directly to the real class specification:

# autospec_drift_prevention.py
from unittest.mock import Mock, create_autospec

class BillingGateway:
    def charge_customer(self, customer_id: str, amount_cents: int, currency: str = "USD") -> str:
        """Charges a registered customer's saved payment method."""
        return f"tx_success_{customer_id}_{amount_cents}"

def test_vanilla_mock_drift() -> None:
    # Vanilla Mock permits non-existent methods and wrong signatures!
    vanilla_mock = Mock()
    vanilla_mock.non_existent_method("foo")
    vanilla_mock.charge_customer("cust_01")  # Missing required 'amount_cents'!
    print("Vanilla mock silently succeeded despite invalid method names and missing arguments.")

def test_autospec_drift_detection() -> None:
    # create_autospec introspects BillingGateway methods and signatures
    safe_mock = create_autospec(BillingGateway, instance=True)

    # 1. Calling non-existent method raises AttributeError immediately
    try:
        safe_mock.non_existent_method("foo")
    except AttributeError as err:
        print(f"\nCaught nonexistent method via autospec:\n  {err}")

    # 2. Calling with incorrect signature raises TypeError immediately
    try:
        safe_mock.charge_customer("cust_01")  # Missing required argument!
    except TypeError as err:
        print(f"\nCaught invalid signature via autospec:\n  {err}")

def main() -> None:
    test_vanilla_mock_drift()
    test_autospec_drift_detection()

if __name__ == "__main__":
    main()

Run:

uv run python autospec_drift_prevention.py

Output:

Vanilla mock silently succeeded despite invalid method names and missing arguments.

Caught nonexistent method via autospec:
  Mock object has no attribute 'non_existent_method'

Caught invalid signature via autospec:
  missing a required argument: 'amount_cents'

Case 3: Testing Asynchronous Event Pipelines with AsyncMock

When testing asyncio code, methods that return coroutines cannot be replaced with synchronous Mock objects. AsyncMock ensures that calling the mock returns an awaitable coroutine and records await counts:

# async_worker_test.py
import asyncio
from unittest.mock import AsyncMock

async def process_queue_batch(consumer, alert_service) -> int:
    """Consumes messages from a stream and forwards alerts asynchronously."""
    messages = await consumer.poll_messages(max_items=10)
    processed_count = 0

    for msg in messages:
        if msg.get("severity") == "CRITICAL":
            await alert_service.dispatch_pager(msg["event"])
        await consumer.ack(msg["offset"])
        processed_count += 1

    return processed_count

async def test_async_worker() -> None:
    mock_consumer = AsyncMock()
    mock_alerts = AsyncMock()

    # Setup async return value for poll_messages
    mock_consumer.poll_messages.return_value = [
        {"offset": 101, "severity": "INFO", "event": "User logged in"},
        {"offset": 102, "severity": "CRITICAL", "event": "DB Connection dropped"},
        {"offset": 103, "severity": "WARN", "event": "High memory consumption"},
    ]

    # Execute async function under test
    count = await process_queue_batch(mock_consumer, mock_alerts)

    print(f"Total processed events: {count}")
    print(f"Consumer ack() awaited count: {mock_consumer.ack.await_count}")
    print(f"Alert dispatch_pager() awaited count: {mock_alerts.dispatch_pager.await_count}")

    # Assert async interactions
    mock_consumer.poll_messages.assert_awaited_once_with(max_items=10)
    mock_alerts.dispatch_pager.assert_awaited_once_with("DB Connection dropped")
    assert mock_consumer.ack.await_count == 3
    print("All async assertions verified!")

if __name__ == "__main__":
    asyncio.run(test_async_worker())

Run:

uv run python async_worker_test.py

Output:

Total processed events: 3
Consumer ack() awaited count: 3
Alert dispatch_pager() awaited count: 1
All async assertions verified!

Pitfalls

Pitfall 1: Patching the Definition Site Instead of the Lookup Site

If service.py contains from time import time, patching time.time has no effect because service.py holds a direct pointer to the function bound during import:

# service.py
from time import time

def get_current_epoch() -> float:
    return time()

# test_service.py (THE TRAP):
from unittest.mock import patch

# WRONG: time was imported into service's namespace!
with patch("time.time", return_value=1700000000.0):
    # get_current_epoch() still calls the unpatched original function!
    pass

# THE FIX:
# Patch the symbol where service.py looks it up!
with patch("service.time", return_value=1700000000.0):
    assert get_current_epoch() == 1700000000.0

Pitfall 2: Misspelling Assertion Methods (assert_called_once_with typo)

Because standard Mock instances dynamically create any attribute on demand, misspelling an assertion method name (e.g. assert_called_once_withh) does not raise an error—it merely creates a new child mock, silently passing the test!

# THE TRAP:
from unittest.mock import Mock

mock_obj = Mock()
# Did you notice the extra 'h'?
mock_obj.assert_called_once_withh("bad_arg")  # SILENT PASS! Returns a new Mock object!

# THE FIX:
# Use pytest with unittest.mock or run with PYTHONWARNINGS to catch non-existent mock assertions,
# or use autospec=True.

Exercises

  1. Create a unit test using @patch("os.environ.get") to verify that a configuration loader returns default fallback values when an environment variable is missing.
  2. Build a test for a database transaction context manager (with db.transaction(): ...) using MagicMock to ensure commit() is called on clean exit and rollback() is called on exceptions.
  3. Test a file download function by patching urllib.request.urlopen with a mock context manager that streams 1 KB chunk buffers.
  4. Using side_effect, implement a mock for a rate-limited API that returns HTTP 429 twice before returning HTTP 200 with valid JSON.
  5. Convert an existing unit test that uses vanilla Mock to create_autospec(TargetClass, instance=True) and observe which tests break due to out-of-date method signatures.

Further reading

  • Python Documentation: unittest.mock — Mock object library.
  • Harry Percival & Bob Gregory: Architecture Patterns with Python (Chapter 3: Coupling and Abstractions; Mocks vs Adapters).
  • Martin Fowler: Mocks Aren’t Stubs (Fowler’s taxonomy of test doubles: Dummy, Fake, Stubs, Mocks).
  • Brett Slatkin: Effective Python (Item 80: Consider unittest.mock for Testing Code with Dependencies).