Learning Platform
Урок 10.03 · 26 мин
Начальный
mockingmonkeypatchunittest.mockhttpxMockTransport
pytest и mocking HTTP: тестируем API-клиенты без сети ACID: почему in-memory SQLite подходит для тестов транзакций

Зачем мокать вообще

Допустим, у вас есть функция, которая забирает данные из GitHub API и пишет в Postgres:

def sync_repo_stars(owner: str, name: str, db: Connection) -> int:
    response = httpx.get(f"https://api.github.com/repos/{owner}/{name}")
    response.raise_for_status()
    data = response.json()
    db.execute(
        "INSERT INTO repos (name, stars) VALUES (?, ?)",
        (data["full_name"], data["stargazers_count"]),
    )
    return data["stargazers_count"]

Хочется написать тест. Как?

Вариант 1 — звонить в настоящий GitHub. Тест становится медленным (200 мс на запрос), флэкающим (упал интернет — упал тест), зависит от внешнего сервиса (GitHub поменял схему — все тесты красные), требует токен (его секретно нужно прокидывать в CI). Плохая идея.

Вариант 2 — поднять локальный mock-сервер. Сложно настроить, ещё одна точка отказа.

Вариант 3 — подменить httpx.get на функцию-заглушку, которая вместо реального запроса вернёт заранее подготовленный ответ. Это и есть

мок
.

Цели мокирования:

  1. Скорость — тест выполняется за миллисекунды, а не за секунды.
  2. Детерминизм — один и тот же тест всегда даёт один и тот же результат. Никаких «у меня иногда падает».
  3. Изоляция — упал интернет, упала БД, GitHub за пейволом — тесты идут.
  4. Контроль — мы можем легко проверить, как функция реагирует на 404, на 500, на таймаут.

Цена — тест проверяет только нашу логику, а не «оно правда работает с настоящим GitHub». Поэтому моки — для unit-тестов; для проверки «работает целиком» нужны integration/e2e тесты (см. урок 1).

Уровни мокирования в pytest

В pytest есть три инструмента мокирования, и они дополняют друг друга. По возрастанию мощности:

  1. monkeypatch — встроенный, простой, для подмены переменных окружения, атрибутов модулей, функций. Лучший выбор для 80% случаев.
  2. unittest.mock — stdlib, мощный: Mock, MagicMock, patch. Подменяет вообще что угодно, в том числе атрибуты внутри сторонних объектов.
  3. Спец-инструменты для HTTP — httpx.MockTransport, respx. Перехватывают сетевой слой, не зная про конкретную функцию.

Разберём по очереди.

monkeypatch: точечная подмена

monkeypatch — это built-in fixture (из предыдущего урока). Главная фишка: всё, что вы поменяете, откатится автоматически после теста.

Подмена переменных окружения

Самый частый кейс — наша функция читает токен из os.environ:

import os
import httpx


def github_client() -> httpx.Client:
    token = os.environ["GITHUB_TOKEN"]
    return httpx.Client(
        base_url="https://api.github.com",
        headers={"Authorization": f"Bearer {token}"},
    )

Тест:

def test_github_client_reads_env(monkeypatch) -> None:
    monkeypatch.setenv("GITHUB_TOKEN", "ghp_test_123")
    client = github_client()
    assert client.headers["Authorization"] == "Bearer ghp_test_123"


def test_github_client_missing_env(monkeypatch) -> None:
    monkeypatch.delenv("GITHUB_TOKEN", raising=False)
    with pytest.raises(KeyError):
        github_client()

Главное — после теста GITHUB_TOKEN восстановится к тому, что было до. Если у вас локально стояла настоящая переменная — она вернётся.

Подмена атрибута модуля

# my_etl/clock.py
from datetime import datetime, timezone


def now() -> datetime:
    return datetime.now(timezone.utc)
from datetime import datetime, timezone
from my_etl import clock


def test_now_can_be_frozen(monkeypatch) -> None:
    frozen = datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)
    monkeypatch.setattr(clock, "now", lambda: frozen)

    assert clock.now() == frozen

monkeypatch.setattr(clock, "now", lambda: frozen) — на время теста clock.now указывает на нашу лямбду. После — оригинал на месте.

WARNING

Тонкий момент: monkeypatch.setattr должен патчить имя там, где оно используется, а не там, где определено. Если внутри функции do_etl написано from my_etl import clock и clock.now() — патчим clock.now. Если from my_etl.clock import now и now() — патчим my_etl.do_etl.now (имя там, куда импортировали). Это типичная ловушка новичков.

Подмена sys.argv

Полезно для CLI-тестов:

def test_cli_parses_args(monkeypatch) -> None:
    monkeypatch.setattr("sys.argv", ["my-etl", "--mode", "dry-run"])
    args = parse_cli()
    assert args.mode == "dry-run"

unittest.mock: Mock и patch

Когда monkeypatch тесноват — берём unittest.mock из stdlib. Он более выразительный, но и побольше «магии».

Mock как объект-заглушка

Mock — это объект, у которого автоматически создаются любые атрибуты и любые методы:

from unittest.mock import Mock


m = Mock()
m.foo                # Mock-атрибут
m.bar.baz()          # Mock-метод, возвращает другой Mock
m.return_value = 42  # m() теперь возвращает 42
m()                  # 42

Это удобно, когда нужно подсунуть «что-то с такими-то методами», не имитируя весь интерфейс реального объекта.

MagicMock

MagicMock — то же самое, что Mock, но дополнительно поддерживает dunder-методы (__len__, __iter__, __enter__, etc.). Если ваш код использует with x: или for item in x — нужен MagicMock:

from unittest.mock import MagicMock


db = MagicMock()
db.cursor.return_value.fetchall.return_value = [(1, "Anna"), (2, "Bob")]

# симулируем:  with db.cursor() as cur: cur.fetchall()
with db.cursor() as cur:
    rows = cur.fetchall()
assert rows == [(1, "Anna"), (2, "Bob")]

patch как контекст-менеджер и декоратор

patch — самый частый способ. Подменяет имя на время блока кода:

from unittest.mock import patch


# как context manager
def test_with_patch() -> None:
    with patch("my_etl.api.httpx.get") as mock_get:
        mock_get.return_value.json.return_value = {"id": 42}
        mock_get.return_value.status_code = 200

        result = fetch_user(42)

        assert result == {"id": 42}
        mock_get.assert_called_once()


# как decorator
@patch("my_etl.api.httpx.get")
def test_with_decorator(mock_get) -> None:
    mock_get.return_value.json.return_value = {"id": 42}
    ...

Главное правило (то же, что у monkeypatch.setattr): патчите имя там, где оно используется. Если my_etl/api.py делает import httpx и зовёт httpx.get(...) — патчим "my_etl.api.httpx.get", не "httpx.get".

side_effect: исключения и последовательности

return_value — это «всегда возвращай вот это». side_effect — мощнее:

Бросать исключение:

mock_get.side_effect = httpx.TimeoutException("timeout!")
# при первом же вызове mock_get(...) бросится TimeoutException

Возвращать разные значения на каждый вызов:

mock_get.side_effect = [
    Response(json_data={"page": 1}),
    Response(json_data={"page": 2}),
    Response(json_data={"page": 3}),
]
# три первых вызова отдадут эти три, четвёртый — StopIteration

Запускать функцию вместо мока:

def fake_get(url, **kwargs):
    if "users" in url:
        return Response(json_data={"users": [...]})
    return Response(status_code=404)

mock_get.side_effect = fake_get

Это бесценно для теста retry-логики: первый вызов — таймаут, второй — 500, третий — 200. Проверяем, что функция переживает оба фейла и в итоге возвращает корректный результат.

httpx.MockTransport: правильный путь для HTTP

С httpx есть встроенный способ мокать HTTP-запросы, не дёргая внутренности через patch. Это

httpx.MockTransport
— заглушка транспортного слоя, которая не идёт в сеть, а зовёт вашу функцию-хэндлер.

import httpx


def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/repos/python/cpython":
        return httpx.Response(
            200,
            json={"full_name": "python/cpython", "stargazers_count": 60000},
        )
    return httpx.Response(404, json={"message": "Not Found"})


def test_sync_repo_stars() -> None:
    transport = httpx.MockTransport(handler)
    with httpx.Client(transport=transport, base_url="https://api.github.com") as client:
        response = client.get("/repos/python/cpython")
        assert response.json()["stargazers_count"] == 60000

Что произошло: вместо реального сетевого вызова httpx взял наш handler, передал ему request, получил response. С точки зрения кода клиента — он сделал HTTP-запрос; с точки зрения теста — никакой сети не было.

Этот подход рекомендован httpx в их документации и применяется в уроке про requests/httpx. Он чище, чем patch("...httpx.get") — потому что:

  1. Работает на уровне транспорта, не лезет во внутренности библиотеки.
  2. Видит всё, что код делает с клиентом, включая правильные headers и URL.
  3. Не сломается при обновлении httpx (mock через patch — может сломаться).

Интеграция с fixture

В DE-проекте обычно делают фикстуру, которая отдаёт httpx.Client с MockTransport:

# tests/conftest.py
import httpx
import pytest


def _github_handler(request: httpx.Request) -> httpx.Response:
    path = request.url.path
    if path == "/repos/python/cpython":
        return httpx.Response(
            200,
            json={"full_name": "python/cpython", "stargazers_count": 60000},
        )
    if path == "/repos/missing/repo":
        return httpx.Response(404, json={"message": "Not Found"})
    if path == "/repos/server/error":
        return httpx.Response(500, json={"message": "boom"})
    return httpx.Response(404, json={"message": f"unhandled: {path}"})


@pytest.fixture
def github_client() -> httpx.Client:
    transport = httpx.MockTransport(_github_handler)
    return httpx.Client(transport=transport, base_url="https://api.github.com")
# tests/test_sync.py
def test_existing_repo(github_client) -> None:
    response = github_client.get("/repos/python/cpython")
    assert response.status_code == 200
    assert response.json()["stargazers_count"] == 60000


def test_missing_repo(github_client) -> None:
    response = github_client.get("/repos/missing/repo")
    assert response.status_code == 404


def test_handles_5xx(github_client) -> None:
    response = github_client.get("/repos/server/error")
    assert response.status_code == 500

Альтернатива — пакет

respx
(uv add --group dev respx) — более декларативный синтаксис, регистрация ответов через builder-API. Для большинства DE-задач MockTransport достаточно; respx стоит подключать, если у вас десятки эндпоинтов и хочется компактнее.

Мокирование БД: in-memory SQLite

С базой данных самый простой подход — не мокать. Используйте in-memory SQLite (см. предыдущий урок, фикстура db). Это реальная БД, реальные SQL-запросы, реальные ошибки. Скорость — миллисекунды на тест.

@pytest.fixture
def db() -> Iterator[sqlite3.Connection]:
    conn = sqlite3.connect(":memory:")
    conn.executescript("""
        CREATE TABLE repos (
            name TEXT PRIMARY KEY,
            stars INTEGER NOT NULL
        );
    """)
    yield conn
    conn.close()


def test_inserts_repo(db) -> None:
    db.execute("INSERT INTO repos (name, stars) VALUES (?, ?)", ("py/cpy", 60000))
    assert db.execute("SELECT stars FROM repos").fetchone()[0] == 60000

Когда SQLite не годится (специфичные для PostgreSQL JSONB, оконные функции, расширения) — берите testcontainers:

# tests/conftest.py — фикстура с реальным временным PostgreSQL
import pytest
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:17-alpine") as pg:
        yield pg.get_connection_url()

scope="session" — поднимаем контейнер один раз за весь прогон, иначе медленно.

А если хочу замокать psycopg-курсор

Бывает оправдано: например, проверяете обработку DB-ошибок, а не сам SQL. Тогда MagicMock:

from unittest.mock import MagicMock
import psycopg


def test_retries_on_serialization_failure() -> None:
    cursor = MagicMock()
    # первый вызов — ошибка сериализации, второй — успех
    cursor.execute.side_effect = [
        psycopg.errors.SerializationFailure("conflict"),
        None,
    ]
    cursor.fetchone.return_value = (42,)

    conn = MagicMock()
    conn.cursor.return_value.__enter__.return_value = cursor

    result = insert_with_retry(conn, "value")
    assert result == 42
    assert cursor.execute.call_count == 2

Anti-pattern: мокать всё подряд

Главный грех начинающего — замокать вообще всё, что код делает. Получается тест, который проверяет, что mock возвращает то, что ему сказали возвращать. Никакого реального покрытия:

# плохой тест
def test_sync_repo() -> None:
    with patch("my_etl.sync.parse_response") as mock_parse:
        mock_parse.return_value = {"stars": 1}
        with patch("my_etl.sync.insert_repo") as mock_insert:
            with patch("my_etl.sync.fetch_repo") as mock_fetch:
                mock_fetch.return_value = {"raw": "data"}
                result = sync_repo("py/cpy")
                assert result == 1   # это просто прокинули mock_parse через функцию

Этот тест ничего не проверил: вся логика замокана. Если функция реально не работает — тест зелёный.

Правило: мокайте только внешний мир — сеть, БД, файлы вне tmp_path, текущее время. Собственный код проекта не мокайте. Если функция А вызывает функцию Б из этого же модуля, и Б сложная — выделите Б в отдельный класс/модуль и протестируйте её отдельным тестом.

TIP

Эвристика: если вы патчите больше двух вещей в одном тесте — скорее всего, тестируете слишком много за раз. Разбейте на два теста или вытащите часть логики в отдельную функцию, которую можно тестировать чисто.

DE-кейс: тест ETL-функции

Соберём всё вместе. Функция тянет данные с GitHub API и сохраняет в БД:

# src/my_etl/sync.py
import httpx
import sqlite3
from datetime import datetime, timezone


def sync_repo(client: httpx.Client, db: sqlite3.Connection, full_name: str) -> int:
    """Загружает stars-count для репозитория и пишет в БД.

    Возвращает количество звёзд. Бросает HTTPStatusError при 4xx/5xx.
    """
    response = client.get(f"/repos/{full_name}")
    response.raise_for_status()
    data = response.json()
    db.execute(
        "INSERT OR REPLACE INTO repos (name, stars, synced_at) VALUES (?, ?, ?)",
        (
            data["full_name"],
            data["stargazers_count"],
            datetime.now(timezone.utc).isoformat(),
        ),
    )
    db.commit()
    return data["stargazers_count"]

Тесты:

# tests/test_sync.py
import httpx
import pytest
import sqlite3
from my_etl.sync import sync_repo


def _handler(request: httpx.Request) -> httpx.Response:
    path = request.url.path
    if path == "/repos/python/cpython":
        return httpx.Response(200, json={
            "full_name": "python/cpython",
            "stargazers_count": 60123,
        })
    if path == "/repos/missing/repo":
        return httpx.Response(404, json={"message": "Not Found"})
    return httpx.Response(500, json={"message": "boom"})


@pytest.fixture
def client() -> httpx.Client:
    return httpx.Client(
        transport=httpx.MockTransport(_handler),
        base_url="https://api.github.com",
    )


@pytest.fixture
def db():
    conn = sqlite3.connect(":memory:")
    conn.execute("""
        CREATE TABLE repos (
            name TEXT PRIMARY KEY,
            stars INTEGER NOT NULL,
            synced_at TEXT NOT NULL
        )
    """)
    yield conn
    conn.close()


class TestSyncRepo:
    def test_happy_path(self, client, db) -> None:
        stars = sync_repo(client, db, "python/cpython")
        assert stars == 60123
        row = db.execute("SELECT name, stars FROM repos").fetchone()
        assert row == ("python/cpython", 60123)

    def test_not_found_raises(self, client, db) -> None:
        with pytest.raises(httpx.HTTPStatusError):
            sync_repo(client, db, "missing/repo")
        # БД не должна содержать запись
        assert db.execute("SELECT COUNT(*) FROM repos").fetchone()[0] == 0

    def test_server_error_raises(self, client, db) -> None:
        with pytest.raises(httpx.HTTPStatusError):
            sync_repo(client, db, "server/error")

Что в итоге:

  • HTTP мокается через MockTransport (никакой сети).
  • БД — реальная in-memory SQLite (видим, что INSERT работает, что транзакция коммитится).
  • Время мокать не пришлось, потому что мы не проверяем его конкретное значение. Если бы проверяли — monkeypatch.setattr или фикстура freezegun.
  • 3 теста, каждый — один сценарий: happy / 404 / 500.

Это и есть рабочий стиль unit-тестирования ETL. Быстрые, детерминированные, проверяющие логику, а не библиотеку httpx.

Что мы получили

  • Мокаем внешний мир: сеть, файлы (вне tmp), время. Собственный код не мокаем.
  • monkeypatch — лёгкий, идеален для env vars и атрибутов модулей. Auto-cleanup.
  • unittest.mock (Mock, MagicMock, patch, side_effect) — мощнее, но требует осторожности с именованием.
  • Для HTTP — httpx.MockTransport: чище и стабильнее, чем patch.
  • БД лучше не мокать: in-memory SQLite или testcontainers. Если мокаете курсор — только для тестов специфичных DB-ошибок.
  • Anti-pattern: мокать всё подряд. Тогда тест ничего не проверяет.

В следующем уроке — как прогонять эти тесты автоматически на каждый коммит через GitHub Actions, и как измерять coverage.

Закончили урок?

Отметьте его как пройденный, чтобы отслеживать свой прогресс

Войдите чтобы оценить урок
Вернуться к курсу
Прогресс модуля
0 из 4