pytest и mocking HTTP: тестируем API-клиенты без сети

API-клиент в data engineering — это код, который висит между чужим сервером и вашим хранилищем. Если он сломан, у вас нет данных. Если он сломан тихо — у вас есть неправильные данные, что хуже. Тесты для такого кода нужны не для галочки в CI, а потому что вы не контролируете внешний API: завтра он отдаст 500, послезавтра поменяет схему ответа, через неделю поднимет rate limit. Хороший тестовый набор гарантирует, что ваш клиент знает, как себя вести в каждом из этих случаев.

Проблема в том, что реально вызывать внешний API в тестах нельзя: тесты должны быть быстрыми, повторяемыми и не требовать интернета. Решение — mocking: подменяем HTTP-слой так, что наш клиент думает, что разговаривает с настоящим сервером, а на самом деле получает заранее заготовленный ответ.

В этом уроке: pytest как инструмент (fixtures, parametrize, AAA), две главные библиотеки для mocking HTTP (responses для requests, respx для httpx), что именно проверять и что не нужно мокать.

Зачем тестировать API-клиент

Junior DE часто думает: «зачем тестировать вызов API? Если URL правильный — вернёт данные, если нет — упадёт с ошибкой, и так понятно». Это так, ровно до момента, пока что-то не пошло не так в продакшене.

Без тестов код «работает» только пока внешний API ведёт себя так же, как когда вы его впервые вызвали. Любое отклонение — невидимый баг, который вылезет через неделю в проде.

pytest за 10 минут: только то, что нужно для тестов API-клиентов

pytest — стандарт де-факто для тестирования в Python. У него три основных механизма, которые нам нужны: ассерты, фикстуры и parametrize.

Ассерт — это просто assert

В отличие от unittest, в pytest нет self.assertEqual — есть обычный Python assert. Pytest перехватывает его и красиво печатает обе части при провале.

# tests/test_basic.py
def test_addition():
    result = 2 + 3
    assert result == 5

def test_dict_subset():
    user = {"id": 1, "name": "Alice", "role": "admin"}
    assert user["role"] == "admin"
    assert "email" not in user

Запуск: pytest -v tests/. Если упадёт — pytest распакует выражение и покажет, что было слева, что справа.

Fixtures — переиспользуемая подготовка

Fixture — функция с декоратором @pytest.fixture, которая возвращает объект. Тест получает её через параметр с тем же именем. Pytest сам вызывает фикстуру и передаёт результат.

# tests/conftest.py
import pytest

@pytest.fixture
def sample_user():
    return {"id": 1, "name": "Alice", "email": "[email protected]"}

@pytest.fixture
def api_base_url():
    return "https://api.example.com/v1"

# tests/test_user.py
def test_user_has_email(sample_user):
    assert "@" in sample_user["email"]

def test_url_uses_https(api_base_url):
    assert api_base_url.startswith("https://")

conftest.py — файл, в котором pytest автоматически собирает фикстуры. Никаких импортов не нужно: pytest сам найдёт фикстуру по имени параметра.

Scope фикстуры контролирует, как часто она вызывается:

@pytest.fixture(scope="module")
def http_session():
    import requests
    s = requests.Session()
    yield s
    s.close()

yield вместо return означает: всё после yield — teardown, выполнится после того, как все тесты, использующие фикстуру, закончатся.

Parametrize — один тест, много входов

Когда нужно проверить функцию на 10 разных входах, не пишите 10 тестов. Используйте @pytest.mark.parametrize.

import pytest

def is_success_status(code: int) -> bool:
    return 200 <= code < 300

@pytest.mark.parametrize("code,expected", [
    (200, True),
    (201, True),
    (299, True),
    (300, False),
    (404, False),
    (500, False),
])
def test_is_success_status(code, expected):
    assert is_success_status(code) == expected

Pytest сгенерирует 6 отдельных тестов с понятными именами: test_is_success_status[200-True], test_is_success_status[404-False] и так далее. Если один упадёт, остальные пройдут.

AAA: структура каждого теста

AAA — Arrange, Act, Assert. Три блока в каждом тесте, разделённые пустой строкой:

1. Arrange — готовим входные данные, моки, фикстуры.
2. Act — вызываем тестируемую функцию.
3. Assert — проверяем результат.

def test_get_user_returns_parsed_dict(responses_mock):
    # Arrange
    responses_mock.add(
        responses.GET,
        "https://api.example.com/users/1",
        json={"id": 1, "name": "Alice"},
        status=200,
    )

    # Act
    user = get_user(user_id=1)

    # Assert
    assert user["name"] == "Alice"
    assert user["id"] == 1

Это не догма, а способ читать чужие тесты быстрее. Когда видишь блок Arrange — понимаешь setup. Видишь Act — понимаешь, что тестируется. Видишь Assert — понимаешь, какое поведение проверяется.

Mocking HTTP для requests: библиотека responses

responses — самая популярная библиотека для мокинга HTTP-вызовов в requests. Она перехватывает вызовы на уровне адаптера и возвращает заранее заданные ответы. Никакой реальной сети, никакого реального сокета.

Установка: pip install responses. Версия 0.25.x.

Минимальный пример

Допустим, у нас есть клиент:

# myapp/github.py
import requests

def get_user(username: str) -> dict:
    response = requests.get(
        f"https://api.github.com/users/{username}",
        timeout=10,
        headers={"Accept": "application/vnd.github+json"},
    )
    response.raise_for_status()
    return response.json()

Тест:

# tests/test_github.py
import pytest
import responses
from myapp.github import get_user

@responses.activate
def test_get_user_success():
    # Arrange
    responses.add(
        responses.GET,
        "https://api.github.com/users/octocat",
        json={"login": "octocat", "id": 583231, "type": "User"},
        status=200,
    )

    # Act
    user = get_user("octocat")

    # Assert
    assert user["login"] == "octocat"
    assert user["id"] == 583231

@responses.activate — декоратор, который включает мокинг на время теста. Любой вызов requests.*, который не был замокан, упадёт с ConnectionError. Это полезно: если клиент полез не туда, тест сразу скажет.

Pytest fixture для responses

Декоратор удобен, но в больших тестах хочется фикстуру, чтобы responses_mock был доступен сразу через параметр.

# tests/conftest.py
import pytest
import responses

@pytest.fixture
def responses_mock():
    with responses.RequestsMock() as rsps:
        yield rsps

RequestsMock() без аргументов — strict mode: упадёт, если был зарегистрирован mock, который не вызвали (assert_all_requests_are_fired по умолчанию True). Это ловит мёртвые моки.

def test_get_user_success(responses_mock):
    responses_mock.add(
        responses.GET,
        "https://api.github.com/users/octocat",
        json={"login": "octocat"},
        status=200,
    )

    user = get_user("octocat")

    assert user["login"] == "octocat"

Проверка отправленного запроса

Mocking — это не только «вернуть ответ», но и «проверить, что клиент отправил правильный запрос». Без этого тест проверяет только парсинг ответа, а не корректность самого вызова.

def test_get_user_sends_correct_headers(responses_mock):
    # Arrange
    rsp = responses_mock.add(
        responses.GET,
        "https://api.github.com/users/octocat",
        json={"login": "octocat"},
        status=200,
    )

    # Act
    get_user("octocat")

    # Assert: что отправили?
    assert len(responses_mock.calls) == 1
    request = responses_mock.calls[0].request
    assert request.headers["Accept"] == "application/vnd.github+json"
    assert request.url == "https://api.github.com/users/octocat"

responses_mock.calls — список всех перехваченных вызовов. У каждого есть .request (что отправили) и .response (что вернули).

POST с body

def create_issue(repo: str, title: str, body: str, token: str) -> dict:
    response = requests.post(
        f"https://api.github.com/repos/{repo}/issues",
        json={"title": title, "body": body},
        headers={"Authorization": f"Bearer {token}"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

def test_create_issue_sends_correct_body(responses_mock):
    # Arrange
    responses_mock.add(
        responses.POST,
        "https://api.github.com/repos/owner/repo/issues",
        json={"number": 42, "title": "Bug", "url": "..."},
        status=201,
    )

    # Act
    issue = create_issue("owner/repo", "Bug", "Steps to reproduce", "ghp_xxx")

    # Assert
    assert issue["number"] == 42
    request = responses_mock.calls[0].request
    import json
    body = json.loads(request.body)
    assert body == {"title": "Bug", "body": "Steps to reproduce"}
    assert request.headers["Authorization"] == "Bearer ghp_xxx"

Mocking HTTP для httpx: библиотека respx

httpx — современный HTTP-клиент с поддержкой sync/async и HTTP/2. Для него responses не работает (другой transport layer). Вместо этого — respx.

Установка: pip install respx. Версия 0.21.x.

Минимальный пример (sync)

# myapp/coingecko.py
import httpx

def get_price(coin_id: str) -> float:
    with httpx.Client(timeout=10.0) as client:
        response = client.get(
            f"https://api.coingecko.com/api/v3/simple/price",
            params={"ids": coin_id, "vs_currencies": "usd"},
        )
        response.raise_for_status()
        return response.json()[coin_id]["usd"]

# tests/test_coingecko.py
import respx
import httpx
from myapp.coingecko import get_price

@respx.mock
def test_get_price():
    # Arrange
    respx.get("https://api.coingecko.com/api/v3/simple/price").mock(
        return_value=httpx.Response(200, json={"bitcoin": {"usd": 65000.0}})
    )

    # Act
    price = get_price("bitcoin")

    # Assert
    assert price == 65000.0

Заметьте: в respx мок-ответ строится через httpx.Response(...) — настоящий объект ответа, не словарь конфигурации.

Pytest fixture для respx

# tests/conftest.py
import pytest
import respx

@pytest.fixture
def respx_mock():
    with respx.mock(assert_all_called=True) as router:
        yield router

assert_all_called=True — то же самое, что у responses: упадёт, если зарегистрировали мок, который не дёрнули.

Async-тесты

Для тестирования async-кода нужен pytest-asyncio (pip install pytest-asyncio) и пометка @pytest.mark.asyncio.

# myapp/coingecko_async.py
import httpx

async def get_price_async(coin_id: str) -> float:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            f"https://api.coingecko.com/api/v3/simple/price",
            params={"ids": coin_id, "vs_currencies": "usd"},
        )
        response.raise_for_status()
        return response.json()[coin_id]["usd"]

# tests/test_coingecko_async.py
import pytest
import respx
import httpx
from myapp.coingecko_async import get_price_async

@pytest.mark.asyncio
async def test_get_price_async(respx_mock):
    respx_mock.get("https://api.coingecko.com/api/v3/simple/price").mock(
        return_value=httpx.Response(200, json={"bitcoin": {"usd": 65000.0}})
    )

    price = await get_price_async("bitcoin")

    assert price == 65000.0

Покрытие edge cases: 4xx, 5xx, network errors, timeouts

Большинство багов в API-клиентах — это не «ответ распарсился неправильно», а «не подумали, что сервер может вернуть X». Хороший тест-набор перебирает все категории ошибок.

Тест 4xx — клиентские ошибки

import pytest
import requests
import responses
from myapp.github import get_user, UserNotFoundError

def test_get_user_404_raises_specific_error(responses_mock):
    responses_mock.add(
        responses.GET,
        "https://api.github.com/users/ghost",
        json={"message": "Not Found"},
        status=404,
    )

    with pytest.raises(UserNotFoundError):
        get_user("ghost")

Здесь важная мысль: клиент должен бросать доменное исключение (UserNotFoundError), а не голый HTTPError из requests. Тест фиксирует этот контракт.

Тест 5xx — серверные ошибки

def test_get_user_500_raises_server_error(responses_mock):
    responses_mock.add(
        responses.GET,
        "https://api.github.com/users/octocat",
        json={"message": "Internal Server Error"},
        status=500,
    )

    with pytest.raises(requests.HTTPError):
        get_user("octocat")

Тест timeout

responses поддерживает выбрасывание исключений вместо ответа:

import requests

def test_get_user_timeout_raises(responses_mock):
    responses_mock.add(
        responses.GET,
        "https://api.github.com/users/octocat",
        body=requests.exceptions.ReadTimeout("timed out"),
    )

    with pytest.raises(requests.exceptions.ReadTimeout):
        get_user("octocat")

Для respx:

import httpx

@respx.mock
def test_get_price_connect_error():
    respx.get("https://api.coingecko.com/api/v3/simple/price").mock(
        side_effect=httpx.ConnectError("DNS failure")
    )

    with pytest.raises(httpx.ConnectError):
        get_price("bitcoin")

Тест retry-логики через несколько ответов подряд

responses позволяет регистрировать несколько ответов на один URL — они вернутся по порядку:

def test_retry_succeeds_on_third_attempt(responses_mock):
    url = "https://api.github.com/users/octocat"
    responses_mock.add(responses.GET, url, status=500)
    responses_mock.add(responses.GET, url, status=500)
    responses_mock.add(
        responses.GET,
        url,
        json={"login": "octocat"},
        status=200,
    )

    # Предполагаем, что get_user внутри использует tenacity или ручной retry
    user = get_user_with_retry("octocat", max_attempts=3)

    assert user["login"] == "octocat"
    assert len(responses_mock.calls) == 3

Это один из самых ценных тестов: он проверяет реальное поведение retry, а не просто факт, что код «как-то работает».

Parametrize для edge cases

Когда тестируете категорию ошибок — статус-коды, разные тела ответов, разные параметры — parametrize экономит время и делает покрытие явным.

import pytest
import responses

@pytest.mark.parametrize("status,raises_class", [
    (401, AuthError),
    (403, AuthError),
    (404, UserNotFoundError),
    (429, RateLimitError),
    (500, ServerError),
    (503, ServerError),
])
def test_error_status_codes_raise_specific_errors(
    responses_mock, status, raises_class
):
    responses_mock.add(
        responses.GET,
        "https://api.github.com/users/test",
        json={"message": "error"},
        status=status,
    )

    with pytest.raises(raises_class):
        get_user("test")

Шесть тестов одной функцией. Добавили новый статус — добавили строчку в parametrize.

Что НЕ нужно мокать

Главная ошибка джунов в тестах — мокать всё подряд, в том числе свою собственную бизнес-логику.

Правило: мокайте границу с внешним миром, тестируйте всё что внутри. Если вы мокаете parse_response() — вы написали тест на mock, а не на код.

# Плохо: мокаем свою функцию
def test_get_user_calls_parser(mocker):
    mock_parse = mocker.patch("myapp.github.parse_user")
    mock_parse.return_value = {"id": 1}
    # ...
    # Этот тест ничего полезного не проверяет

# Хорошо: мокаем только HTTP, проверяем результат через всю цепочку
def test_get_user_returns_parsed_user(responses_mock):
    responses_mock.add(
        responses.GET,
        "https://api.github.com/users/octocat",
        json={"login": "octocat", "id": 583231, "created_at": "2011-01-25T18:44:36Z"},
        status=200,
    )

    user = get_user("octocat")

    assert user.id == 583231
    assert user.login == "octocat"
    assert user.created_at.year == 2011  # parse работает

Чеклист готового теста

Когда пишете тест на API-клиент, проверьте:

1. Happy path — корректный 200 + валидный JSON, проверка что вернулось.
2. Headers/body — клиент отправляет правильный Authorization, правильный Content-Type, правильный body.
3. 4xx — 404 (not found), 401/403 (auth), 422 (validation).
4. 5xx — 500/502/503, retry если есть.
5. 429 — rate limit, обработка Retry-After.
6. Network errors — ConnectError, ReadTimeout.
7. Pagination — если есть, тест с двумя страницами проверяет обход.
8. Edge cases — пустой массив, null в обязательных полях, очень большой response.

Если хотя бы первые 5 категорий покрыты — клиент production-ready на 80%. Без них — рулетка.

Итог

Тесты для API-клиентов — это не «правильно ли парсится один JSON», а карта поведения клиента в условиях, когда внешний мир ведёт себя не идеально. pytest даёт инструменты (assert, fixtures, parametrize), responses и respx — мокинг HTTP-слоя. AAA-структура делает тесты читаемыми. Покрытие edge cases (4xx, 5xx, timeouts, network) превращает «работает у меня» в «работает в проде».

В следующем уроке разберём VCR.py — альтернативный подход, в котором вместо ручного описания моков мы записываем реальные HTTP-обмены один раз и проигрываем в тестах.