Внешние HTTP — мок через pytest-httpserver и respx

Как мокировать внешние REST-сервисы в интеграционных тестах FastAPI: реальный локальный HTTP-сервер через pytest-httpserver или transport-мок через respx. Стабы пишем прямо в тесте, а не в общих файлах — тест сам объявляет, на что опирается. base-url внешнего клиента переключается через dependency_overrides на адрес заглушки.

Опубликовано 23 июня 2026 ~6 мин чтения Автор: Вадим Викулин

Опирается на правила: PYTS-23, PYTS-24, PYTS-25, PYTS-X5 из Python Test Strategy → раздел 7. Внешние HTTP — мок.

Важно знать

Внешний REST (платёж, каталог, логистика) — локальный HTTP-сервер (pytest-httpserver) или transport-мок (respx); реальные endpoints не трогаем.

pytest-httpserver предпочтительнее: поднимает настоящий TCP-сокет, проверяет сериализацию, заголовки, retry и timeout поведение клиента.

respx допустим как лёгкий вариант, когда нужно проверить только логику, а не HTTP-механику.

Стабы — в самом тесте, не в conftest.py и не в общих файлах: тест должен сам объявлять, на что опирается.

base_url внешнего клиента переключается через dependency_overrides на адрес локальной заглушки; хардкодить адреса нельзя.

MagicMock на порт-репозиторий в интеграционном тесте запрещён: интеграционный тест проверяет реальный путь; мок — только для внешней системы.

Реальный внешний сервис в тесте — только @pytest.mark.e2e, отдельный CI-этап.

UCP-подход: интеграционный тест охватывает полный путь внутри сервиса — от HTTP-запроса до строки в PostgreSQL. Внешние REST-сервисы к этому пути не относятся; они замокированы так, чтобы тест оставался детерминированным и не зависящим от доступности платёжного шлюза или каталога.

Архитектура мока

Внешний клиент (SberPaymentClient, CatalogClient, LogisticsClient) конструируется с настраиваемым base_url. В тесте этот URL заменяется на адрес локальной заглушки через dependency_overrides.

┌─────────────────────────────┐
│ FastAPI app                 │
│  ┌──────────────┐           │
│  │ Handler       │           │
│  │  └─ PaymentPort          │
│  │      └─ SberPaymentClient│──► http://127.0.0.1:<port> (заглушка)
│  └──────────────┘           │          │
└─────────────────────────────┘    pytest-httpserver
                                   (реальный TCP)

Сервис не знает, что «Sber» — это заглушка. Клиент делает настоящий HTTP-запрос; pytest-httpserver принимает его и возвращает заданный ответ.

pytest-httpserver — предпочтительный способ

PYTS-23: реальный локальный HTTP-сервер в фикстуре.

Фикстура клиента с переопределением base_url

# conftest.py
import pytest
from pytest_httpserver import HTTPServer
from httpx import AsyncClient, ASGITransport

from app.main import app
from app.payments.client import get_payment_client, SberPaymentClient


@pytest.fixture
def payment_server(httpserver: HTTPServer) -> HTTPServer:
    return httpserver


@pytest.fixture
async def client(payment_server: HTTPServer) -> AsyncClient:
    app.dependency_overrides[get_payment_client] = lambda: SberPaymentClient(
        base_url=payment_server.url_for("").rstrip("/"),
    )
    async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

httpserver — built-in фикстура pytest-httpserver (session или function scope по конфигурации). payment_server.url_for("") возвращает http://127.0.0.1:<port>.

Тест с реальным HTTP-стабом

PYTS-24: стаб — в самом тесте.

# tests/payments/test_create_payment.py
import uuid
import pytest
from httpx import AsyncClient
from pytest_httpserver import HTTPServer

from tests.conftest import success_token


async def test_create_payment_when_sber_accepts_returns_201(
    client: AsyncClient,
    payment_server: HTTPServer,
    db_preparer,
):
    """BR-PAY-1: успешный запрос к Sber создаёт платёж со статусом PENDING."""
    order_id = uuid.UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
    payment_id = uuid.UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")

    # Arrange
    db_preparer.clear_payments().create_order(order_id=order_id, amount=500_00).prepare()

    payment_server.expect_request(
        "/v2/payments",
        method="POST",
        json={"orderId": str(order_id), "amount": 500_00, "currency": "RUB"},
    ).respond_with_json(
        {"paymentId": str(payment_id), "status": "ACCEPTED"},
        status=201,
    )

    # Act
    response = await client.post(
        "/v1/payments",
        json={"orderId": str(order_id), "amount": 500_00},
        headers=success_token(),
    )

    # Assert
    assert response.status_code == 201
    body = response.json()
    assert body["paymentId"] == str(payment_id)
    assert body["status"] == "PENDING"

    payments = await db_preparer.find_payments(order_id=order_id)
    assert payments[0]["external_id"] == str(payment_id)
    assert payments[0]["status"] == "PENDING"

expect_request(...) задаёт ожидание: метод, путь, тело. respond_with_json(...) — ответ. Если клиент шлёт другой запрос — pytest-httpserver выдаёт ошибку в конце теста. Это проверяет не только логику, но и корректность сериализации исходящего запроса.

Проверка retry и timeout

Одно из преимуществ pytest-httpserver: можно имитировать сбои и проверить поведение клиента.

async def test_create_payment_when_sber_times_out_returns_503(
    client: AsyncClient,
    payment_server: HTTPServer,
    db_preparer,
):
    """BR-PAY-4: таймаут платёжного шлюза → 503 для caller-а."""
    order_id = uuid.UUID("cccccccc-cccc-cccc-cccc-cccccccccccc")

    # Arrange
    db_preparer.clear_payments().create_order(order_id=order_id, amount=100_00).prepare()

    payment_server.expect_request("/v2/payments", method="POST").respond_with_handler(
        lambda req: __import__("time").sleep(10) or __import__("flask").Response(status=200)
    )

    # Act
    response = await client.post(
        "/v1/payments",
        json={"orderId": str(order_id), "amount": 100_00},
        headers=success_token(),
    )

    # Assert
    assert response.status_code == 503
    assert response.json()["code"] == "PAYMENT_GATEWAY_UNAVAILABLE"

Или через respond_with_data с нужным HTTP-статусом:

payment_server.expect_request("/v2/payments", method="POST").respond_with_data(
    status=500,
    content_type="application/json",
    response_data='{"error": "internal"}',
)

Несколько внешних сервисов

Если сервис обращается к нескольким внешним системам — каждой свой httpserver-инстанс. pytest-httpserver поддерживает несколько серверов через фабрику.

# conftest.py
from pytest_httpserver import HTTPServer
import threading


@pytest.fixture(scope="session")
def catalog_server() -> HTTPServer:
    server = HTTPServer(host="127.0.0.1")
    server.start()
    yield server
    server.clear()
    if server.is_running():
        server.stop()


@pytest.fixture(scope="session")
def logistics_server() -> HTTPServer:
    server = HTTPServer(host="127.0.0.1")
    server.start()
    yield server
    server.clear()
    if server.is_running():
        server.stop()

В каждом тесте — server.expect_request(...).respond_with_json(...) в блоке Arrange. Стаб живёт до конца теста, потом server.clear() сбрасывает ожидания.

respx — лёгкий transport-мок

PYTS-25: respx перехватывает httpx на уровне transport, без реального TCP.

Когда выбирать respx

Критерий	pytest-httpserver	respx
Проверяет сериализацию тела	да (реальный HTTP)	частично (через `content`)
Проверяет заголовки запроса	да	да
Проверяет retry/timeout клиента	да	нет (transport перехватывает до timeout)
Скорость фикстуры	~3–10 мс (сокет)	< 1 мс
Поднимает TCP	да	нет
Подходит для smoke	нет	да

respx уместен, когда важно проверить логику Handler-а (ветвление по статусу ответа), а не правильность сериализации клиента.

Фикстура с respx

# conftest.py
import respx
import pytest
from httpx import AsyncClient, ASGITransport

from app.main import app


@pytest.fixture
async def client() -> AsyncClient:
    async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as c:
        yield c

respx используется как контекстный менеджер прямо в тесте.

Тест с respx

# tests/catalog/test_get_product.py
import uuid
import respx
import pytest
from httpx import AsyncClient, Response


async def test_get_product_when_catalog_unavailable_returns_503(client: AsyncClient):
    """BR-PROD-7: недоступность каталога → 503, не 500."""
    product_id = uuid.UUID("dddddddd-dddd-dddd-dddd-dddddddddddd")

    # Arrange
    with respx.mock(base_url="https://catalog.internal") as mock:
        mock.get(f"/v1/products/{product_id}").mock(
            return_value=Response(503, json={"error": "service unavailable"})
        )

        # Act
        response = await client.get(
            f"/v1/products/{product_id}",
            headers=success_token(),
        )

    # Assert
    assert response.status_code == 503
    assert response.json()["code"] == "CATALOG_UNAVAILABLE"

respx.mock(base_url=...) перехватывает все запросы к указанному адресу внутри блока with. Запрос за пределы блока — ошибка.

respx + dependency_overrides

Если base_url клиента не зафиксирован — переключаем через dependency_overrides:

async def test_place_order_when_catalog_returns_product_reserved(
    client: AsyncClient,
    db_preparer,
):
    """BR-ORD-3: товар зарезервирован в каталоге — заказ переходит в CONFIRMED."""
    product_id = uuid.UUID("11111111-1111-1111-1111-111111111111")
    order_id = uuid.UUID("22222222-2222-2222-2222-222222222222")

    # Arrange
    db_preparer.clear_orders().create_customer("c1").prepare()
    app.dependency_overrides[get_id_generator] = lambda: FixedIdGenerator(next_id=order_id)

    with respx.mock(base_url="http://catalog-stub") as catalog_mock:
        app.dependency_overrides[get_catalog_client] = lambda: CatalogClient(
            base_url="http://catalog-stub"
        )
        catalog_mock.post("/v1/reservations").mock(
            return_value=Response(200, json={"reservationId": "r1", "status": "RESERVED"})
        )

        # Act
        response = await client.post(
            "/v1/orders",
            json={"customerId": "c1", "productId": str(product_id), "qty": 2},
            headers=success_token(),
        )

    # Assert
    assert response.status_code == 201
    orders = await db_preparer.find_orders(order_id=order_id)
    assert orders[0]["status"] == "CONFIRMED"

Проверка исходящих заголовков

Часто нужно убедиться, что клиент передаёт Authorization, X-Request-Id, Content-Type. pytest-httpserver проверяет это напрямую:

payment_server.expect_request(
    "/v2/payments",
    method="POST",
    headers={"Content-Type": "application/json", "X-Client-Id": "order-service"},
).respond_with_json({"paymentId": "p1", "status": "ACCEPTED"}, status=201)

В respx — через match:

catalog_mock.get(f"/v1/products/{product_id}").mock(
    side_effect=lambda req: Response(200, json={"productId": str(product_id), "price": 500})
    if req.headers.get("X-Service-Token") == "expected-token"
    else Response(401)
)

Домен Customer — пример с несколькими ответами

Сервис Customer вызывает SberID для верификации и NotificationService для отправки кода.

async def test_verify_customer_when_sber_id_ok_sends_notification(
    client: AsyncClient,
    payment_server: HTTPServer,
    notification_server: HTTPServer,
    db_preparer,
):
    """BR-CUST-2: успешная верификация SberID → отправить SMS-код."""
    customer_id = uuid.UUID("33333333-3333-3333-3333-333333333333")

    # Arrange
    db_preparer.clear_customers().create_customer(customer_id=customer_id).prepare()

    payment_server.expect_request(
        f"/v1/identity/{customer_id}/verify", method="GET"
    ).respond_with_json({"verified": True}, status=200)

    notification_server.expect_request(
        "/v1/sms/send", method="POST"
    ).respond_with_json({"messageId": "msg-1"}, status=200)

    # Act
    response = await client.post(
        f"/v1/customers/{customer_id}/verify",
        headers=success_token(),
    )

    # Assert
    assert response.status_code == 200
    notification_server.check_assertions()

check_assertions() явно проверяет, что все expect_request(...) были вызваны. Без этого незатронутый стаб не вызовет ошибку.

Что запрещено

Антипаттерн	Правило	Что взамен
`MagicMock` на порт-репозиторий / Handler в integration-тесте	`PYTS-X5`	реальный путь; мок — только для внешней системы
Стаб в `conftest.py` (общий для всех тестов)	`PYTS-24`	стаб прямо в тесте в блоке Arrange
Реальный endpoint внешнего сервиса в integration-тесте	`PYTS-23`	`pytest-httpserver` / `respx`
Хардкод `base_url` внешнего клиента в тесте	`PYTS-23`	`dependency_overrides[get_payment_client]`
Один стаб на несколько сценариев без сброса	`PYTS-24`	`server.clear()` между тестами или function-scope фикстура
`unittest.mock.patch` на httpx в integration-тесте	`PYTS-25`	`respx` или `pytest-httpserver`
Реальный внешний сервис без `@pytest.mark.e2e`	`PYTS-28`	отдельный CI-этап, ≤ 10 тестов на сервис

Куда дальше

Базовые правила — детерминизм, AAA-структура, dependency_overrides.
Фикстуры и конфигурация тестов — conftest.py, PostgresContainer, session-scope.
Один тест — полный пример, имена, docstring с BR-кодом.
Что НЕ покрывается интеграционными — unit агрегата, unit контроллера, E2E.
Resilience — retry и circuit breaker — как правильно настроить клиент, чтобы было что тестировать через заглушку.