Пишем быстрые API-автотесты без флаков, стендов и боли: изоляционный подход в CI/CD

Вступление

В этой статье я хочу показать, как на практике писать изоляционные API-автотесты на моках. Тема очень актуальная, но при этом вокруг неё много мифов и лишней сложности.

Самое важное — такие тесты не сложные. Они выглядят максимально просто, запускаются быстро и при этом дают высокую стабильность. Я бы даже сказал, что это эталон современной автоматизации тестирования: минимальный код, предсказуемое окружение и запуск в CI/CD буквально в пару десятков строк.

Такой подход максимально приближен к идее left shift testing и при этом хорошо масштабируется. Без боли, без флаков и без зависимости от внешних окружений.

Сразу дам определение, потому что термин «изоляционные тесты» пока не очень распространён. Изоляционные тесты — это тесты, которые выполняются в полностью изолированной среде. Если у нас микросервисная архитектура, все внешние сервисы для тестируемого сервиса мокаются. Сам сервис поднимается локально, как и необходимая инфраструктура: базы данных, брокеры сообщений, кэши и всё остальное. В результате сервис полностью изолирован от внешнего мира. Делается это не ради абстрактной «красоты», а ради стабильности и предсказуемости тестового окружения.

Примеры в статье будут на Python, но важно понимать: это не «питоновская магия». Всё, что здесь показано, одинаково хорошо ложится на Java, Go, TS/JS и любой другой стек. Ограничений по языку здесь нет — есть только архитектурное мышление и желание сделать нормально.

Ранее я уже писал про принципы стабильных автотестов и про left shift testing:

• «Лучшие практики автоматизации тестирования: 9 принципов стабильных автотестов»

• «Left Shift Testing: как выстроить процесс, чтобы тесты реально помогали»

После этих статей мне регулярно задавали один и тот же вопрос: «Окей, звучит разумно. А как это выглядит на практике?» В этой статье я как раз и показываю — без абстракций и без усложнений.

Также сразу обозначу границы. Я не буду подробно объяснять, как устроен FastAPI, как работает httpx, зачем нужен Pydantic и что происходит «под капотом». На эти темы у меня уже есть достаточно материалов — при желании можно спокойно с ними ознакомиться: Habr, Stepik

Здесь мы фокусируемся не на фреймворках, а на подходе.

Контекст

Тестировать мы будем классический gateway-сервис онлайн-магазина. Архитектура максимально типовая: есть gateway, через который проходят все внешние запросы, и три внутренних сервиса — users, orders и billing. Gateway агрегирует данные от этих сервисов и отдает единый ответ клиенту.

Продовая схема выглядит ровно так, как показано на изображении ниже.

Важно сразу зафиксировать один момент. В рамках этой статьи внутреннее устройство сервисов users, orders и billing нас не интересует вообще. Мы не разбираем, как они реализованы, на каком языке написаны, какие у них базы данных и сколько там слоёв абстракций. Для нас это чёрные ящики. Единственное, что у нас есть и что нам действительно нужно, — это их контракты.

Наша задача простая и инженерная: понять контракты → на основе контрактов сделать моки → написать изоляционные тесты для gateway.

На реализацию микросервисов мы сознательно не завязываемся, чтобы тесты не зависели от того, как именно эти сервисы имплементированы сегодня и как они будут переписаны завтра.

Контракты сервисов

Все контракты в проекте описаны явно — через схемы и HTTP-клиенты. Для нас это идеальная точка опоры: именно эти контракты gateway использует в рантайме, и именно их поведение мы буде�� воспроизводить в моках.

Ниже — фактическая API-спека сервисов в том виде, в котором её видит gateway.

Users service

Endpoint

GET /api/v1/users/{user_id}

Описание

Возвращает данные пользователя по его идентификатору. Gateway использует этот эндпоинт для получения основной информации о пользователе, без какой-либо дополнительной логики.

Ответ

{
  "user": {
    "id": UUID,
    "name": string,
    "email": string,
    "status": "ACTIVE | BLOCKED | UNSPECIFIED"
  }
}

Здесь нет скрытых состояний или бизнес-логики. Статус — перечисление, структура фиксированная. Gateway просто принимает эти данные и прокидывает их дальше в агрегированный ответ.

Orders service

Endpoint

GET /api/v1/orders?userId={user_id}

Описание

Возвращает агрегированную информацию по заказам пользователя. Gateway передаёт user_id в query-параметрах и ожидает summary по заказам.

Ответ

{
  "summary": {
    "total_amount": float >= 0,
    "total_orders": int >= 0,
    "active_orders": int >= 0
  }
}

Важно, что все значения валидированы на уровне контракта: отрицательные значения невозможны. Gateway не считает эти данные сам и не применяет к ним бизнес-логику — он работает строго с тем, что пришло по контракту.

Billing service

Endpoint

GET /api/v1/billing?userId={user_id}

Описание

Возвращает финансовую информацию по пользователю. Gateway запрашивает summary по user_id и использует эти данные в итоговом ответе.

Ответ

{
  "summary": {
    "debt": float >= 0,
    "balance": float,
    "currency": string (ISO, 3 символа)
  }
}

Контракт жёстко определён: валюта фиксированного формата, отрицательные значения долга запрещены. Именно эти ограничения и являются частью ожидаемого поведения сервиса, которое мы будем воспроизводить в моках.

Gateway service

Endpoint

GET /api/v1/gateway/users/{user_id}/summary

Описание

Агрегирующий эндпоинт. Gateway:

• запрашивает пользователя из users,

• summary заказов из orders,

• summary биллинга из billing,

• собирает всё в единый ответ.

Ответ

{
  "user": { ... },
  "orders_summary": { ... },
  "billing_summary": { ... }
}

Это и есть наша точка тестирования. Мы проверяем, что gateway:

• корректно вызывает внешние сервисы,

• правильно прокидывает параметры,

• корректно агрегирует ответы,

• возвращает ожидаемую структуру.

Почему нам достаточно контрактов

Обратите внимание: во всей этой схеме нет ни слова про базы данных, очереди, транзакции или внутренние алгоритмы сервисов. И это не упущение — это принципиально.

Gateway взаимодействует с остальной системой исключительно через HTTP-контракты. Если контракт соблюдён — gateway работает корректно. Если контракт нарушен — это либо ошибка upstream-сервиса, либо отдельный сценарий, который мы можем явно смоделировать в тесте.

Именно поэтому дальше в статье мы будем:

• мокать не сервисы целиком, а их HTTP-контракты,

• не поднимать реальные users / orders / billing,

• не зависеть от данных, состояния и доступности этих сервисов.

В результате тесты получаются изолированными, быстрыми и детерминированными, и при этом проверяют ровно то, за что gateway реально отвечает.

Дальше перейдём к самому мок-сервису и посмотрим, как эта схема реализуется на практике.

Делаем мок

Мок в этом примере будет максимально простым. Без overengineering, без «универсального решения на все случаи жизни». Ровно настолько сложным, насколько это нужно для изоляционных автотестов.

У мок-сервиса будет всего два административных эндпоинта:

• POST /admin/rules — создать правила мокирования

• DELETE /admin/rules — удалить все правила мокирования

И один универсальный эндпоинт-диспетчер, который будет перехватывать все остальные запросы и отдавать ответы на основе заранее заданных правил.

Почему именно так.

Можно было пойти по пути персистентных моков: описать ответы в JSON-файлах, положить их рядом с мок-сервисом и просто раздавать по маршрутам. Такой подход вполне валиден, но он ближе к стабам. Он хорошо подходит, например, для нагрузочных тестов, когда нам не принципиально, какой именно ответ вернётся — главное, чтобы он был и соответствовал контракту.

Но здесь мы пишем автотесты. А в автотестах нам нужно:

• динамически формировать разные бизнес-сценарии,

• легко моделировать ошибки,

• управлять поведением сервисов прямо из теста.

Для этого и нужен динамический мок, которым можно управлять во время выполнения теста. Именно поэтому правила мокирования создаются и удаляются через API.

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

Ниже — реализация. Она нарочно сделана минималистичной, чтобы было видно саму идею, а не обвязку.

Схема правил мокирования

./services/mock/schema.py

from http import HTTPMethod, HTTPStatus
from typing import Any

from pydantic import Field

from libs.schema.base import BaseSchema


class MockRuleSchema(BaseSchema):
    # Query-параметры запроса, по которым будет происходить матчинг
    # Если пусто — запрос без query
    query: dict[str, str] = Field(default_factory=dict)

    # Полный путь запроса (например: /api/v1/users/{id})
    route: str

    # HTTP-метод, по умолчанию GET
    method: HTTPMethod = HTTPMethod.GET

    # Тело ответа, которое мок вернёт клиенту
    # Тип Any, так как мок не накладывает ограничений на структуру
    response: Any

    # HTTP-статус ответа
    status_code: HTTPStatus = HTTPStatus.OK


class CreateMockRulesRequest(BaseSchema):
    # Список правил, которые будут добавлены в мок за один запрос
    rules: list[MockRuleSchema]

Одно правило мокирования описывает:

• HTTP-метод,

• путь запроса,

• query-параметры,

• тело ответа,

• HTTP-статус.

Никакой магии. Если входящий запрос полностью совпадает с правилом — мок отдаёт заданный ответ.

Хранилище правил

./services/mock/rules.py

import asyncio

from fastapi import Request

from services.mock.schema import MockRuleSchema


class MockRulesStore:
    def __init__(self):
        # Lock нужен, так как правила могут изменяться во время обработки запросов
        self.lock = asyncio.Lock()
        self.rules: list[MockRuleSchema] = []

    async def create(self, rules: list[MockRuleSchema]) -> None:
        # Добавляем новые правила в общее хранилище
        async with self.lock:
            self.rules.extend(rules)

    async def find(self, request: Request) -> MockRuleSchema | None:
        # Извлекаем параметры входящего запроса
        request_query = dict(request.query_params)
        request_route = request.url.path
        request_method = request.method

        # Последовательно ищем правило, полностью совпадающее с запросом
        async with self.lock:
            for rule in self.rules:
                if rule.method.value != request_method:
                    continue

                if rule.route != request_route:
                    continue

                if rule.query != request_query:
                    continue

                return rule

        # Если подходящего правила нет — возвращаем None
        return None

    async def clear(self) -> None:
        # Полная очистка всех правил (обычно используется между тестами)
        async with self.lock:
            self.rules.clear()

Здесь всё предельно прямолинейно:

• правила хранятся в памяти,

• доступ защищён asyncio.Lock,

• поиск правила — это обычное последовательное сравнение метода, пути и query-параметров.

Да, это не самый оптимальный алгоритм. И да, здесь нет индексов, кэшей и прочих оптимизаций. Но для изоляционных автотестов это вообще не проблема. Правил мало, тесты быстрые, читаемость и предсказуемость важнее микросекунд.

API мок-сервиса

./services/mock/api.py

from fastapi import APIRouter, Request, HTTPException, status
from fastapi.responses import JSONResponse

from services.mock.rules import MockRulesStore
from services.mock.schema import CreateMockRulesRequest

mock_router = APIRouter()
mock_rules_store = MockRulesStore()


@mock_router.post("/admin/rules", status_code=status.HTTP_201_CREATED)
async def create_mock_rule_view(request: CreateMockRulesRequest):
    # Создаём новые правила мокирования
    await mock_rules_store.create(request.rules)


@mock_router.delete("/admin/rules", status_code=status.HTTP_204_NO_CONTENT)
async def delete_mock_rule_view():
    # Полностью очищаем правила (обычно вызывается в teardown тестов)
    await mock_rules_store.clear()


@mock_router.api_route(
    "/{full_path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
)
async def dispatch_mock_rule_view(request: Request):
    # Универсальный обработчик всех запросов к мок-сервису
    rule = await mock_rules_store.find(request=request)

    if not rule:
        # Если правило не найдено — явно сигнализируем об этом
        raise HTTPException(status_code=404, detail="no mock rule")

    # Возвращаем заранее заданный ответ
    return JSONResponse(content=rule.response, status_code=rule.status_code)

Здесь три ключевых момента.

Первое — административные эндпоинты. Они позволяют из теста:

• задать нужное поведение сервисов,

• полностью очистить состояние мока между тестами.

Второе — универсальный dispatcher. Он принимает любой HTTP-запрос и пытается сопоставить его с правилами. Если правило найдено — возвращается нужный ответ. Если нет — 404. Никаких «молчаливых» фолбеков, всё максимально явно.

Третье — отсутствие логики. Мок ничего не считает, ничего не трансформирует и ничего не «угадывает». Он либо отдаёт заданный ответ, либо падает. Именно это делает тесты детерминированными.

Мок готов. Как видно, всё максимально просто и прозрачно — порядка ста строк кода. И это осознанно. Цель этого примера — не написать «идеальный мок на все случаи жизни», а показать сам подход. Дальше вы уже сами решаете: усложнять его, расширять или заменить на стороннее решение.

В следующем шаге мы подключим этот мок к gateway и посмотрим, как переключить сервисы через конфигурацию, не меняя ни строчки кода приложения.

Переключаем gateway на мок

Теперь самое время подключить мок и сказать gateway, что вместо реальных сервисов он должен ходить в него. Делается это исключительно через конфигурацию, без единого изменения в коде приложения.

Сначала посмотрим, как конфигурация выглядит в проде.

Продовая конфигурация

# users-service
USERS_HTTP_CLIENT.HOST=https://company.users.org

# orders-service
ORDERS_HTTP_CLIENT.HOST=https://company.orders.org

# billing-service
BILLING_HTTP_CLIENT.HOST=https://company.billing.org

# gateway-service
GATEWAY_HTTP_CLIENT.HOST=https://company.gateway.org
GATEWAY_HTTP_SERVER.PORT=443
GATEWAY_HTTP_SERVER.HOST=0.0.0.0
GATEWAY_HTTP_SERVER.WORKERS=3

Здесь всё стандартно: gateway ходит в реальные сервисы по их продовым адресам и сам слушает HTTPS-трафик.

Конфигурация для изоляционных тестов

Локально ситуация меняется. Вся инфраструктура поднимается через docker-compose.yaml, а все внешние зависимости мы полностью мокаем.

.env.ci

# mock-service
MOCK_HTTP_CLIENT.HOST=http://localhost:8000
MOCK_HTTP_SERVER.PORT=8000
MOCK_HTTP_SERVER.HOST=0.0.0.0
MOCK_HTTP_SERVER.WORKERS=1

# users-service
USERS_HTTP_CLIENT.HOST=http://mock:8000

# orders-service
ORDERS_HTTP_CLIENT.HOST=http://mock:8000

# billing-service
BILLING_HTTP_CLIENT.HOST=http://mock:8000

# gateway-service
GATEWAY_HTTP_CLIENT.HOST=http://localhost:9000
GATEWAY_HTTP_SERVER.PORT=9000
GATEWAY_HTTP_SERVER.HOST=0.0.0.0
GATEWAY_HTTP_SERVER.WORKERS=3

Ключевой момент здесь в том, что мы не меняем код вообще. Мы просто говорим gateway на уровне конфигурации: «Теперь users, orders и billing находятся по другому хосту».

Этим хостом становится мок-сервис. Gateway ходит к нему back-to-back внутри Docker-сети, а тесты обращаются к самому gateway снаружи по localhost.

Никакой магии. Обычная подмена адресов.

Расширяем конфигурацию сервиса

Чтобы gateway знал про мок, достаточно расширить его конфигурацию. Никаких отдельных режимов или специальных флагов.

config.py

import os

from pydantic_settings import BaseSettings, SettingsConfigDict

from libs.config.http import HTTPServerConfig, HTTPClientConfig


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        extra='allow',
        env_file=os.environ.get('ENV_FILE', '.env'),
        env_file_encoding='utf-8',
        env_nested_delimiter='.'
    )

    # mock-service
    mock_http_client: HTTPClientConfig
    mock_http_server: HTTPServerConfig

    # external services
    users_http_client: HTTPClientConfig
    orders_http_client: HTTPClientConfig
    billing_http_client: HTTPClientConfig

    # gateway
    gateway_http_client: HTTPClientConfig
    gateway_http_server: HTTPServerConfig


settings = Settings()

Можно было бы завести отдельный конфиг специально под автотесты, и это было бы валидно. Но здесь мы идём по самому простому и честному пути — встраиваемся в существующую инфраструктуру gateway, не плодя параллельные конфигурации.

Запуск мок-сервиса

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

./services/mock/server.py

from fastapi import FastAPI

from config import settings
from libs.http.server.base import build_http_server
from services.mock.api import mock_router

app = FastAPI(title="mock-service")
app.include_router(mock_router)

if __name__ == "__main__":
    build_http_server("services.mock.server:app", settings.mock_http_server)

Это самый обычный FastAPI-сервис:

• создаётся приложение,

• подключается роутер мока,

• сервер поднимается через общий util для запуска HTTP-сервисов.

Функция запуска выглядит так:

import uvicorn

from libs.config.http import HTTPServerConfig


def build_http_server(app: str, config: HTTPServerConfig):
    uvicorn.run(
        app=app,
        host=str(config.host),
        port=config.port,
        workers=config.workers
    )

Здесь есть очень важный момент, который нельзя игнорировать.

Для мок-сервиса в конфигурации всегда должен быть установлен workers = 1. И это сделано осознанно.

Мок хранит правила в памяти процесса — в обычном Python-объекте. Если запустить несколько воркеров, каждый из них будет жить в своём процессе и иметь собственное хранилище правил. В результате:

• правила, созданные из теста, попадут в один воркер,

• запросы от gateway могут улететь в другой,

• и тесты начнут вести себя недетерминированно.

В контексте изоляционных автотестов нам не нужна параллельность внутри мок-сервиса. Он должен быть максимально предсказуемым и детерминированным. Один воркер — ровно то, что нужно.

Если в будущем появится необходимость в масштабировании или параллельном выполнении тестов — это решается отдельно (через изоляцию по сценариям, заголовкам, отдельным инстансам и т.д.). Но для базового и честного подхода один процесс — это правильный и осознанный выбор.

Что в итоге получилось?

После переключения конфигурации схема взаимодействия сильно упрощается. Остаётся всего два компонента:

• gateway-сервис,

• mock-сервис.

Gateway больше не ходит напрямую в users, orders и billing. Он ходит в мок, а мок уже динамически притворяется любым из этих сервисов, в зависимости от правил, заданных из теста.

Отдельно подчеркну: мок в этой статье — максимально минималистичный. Это не «god mock» и не универсальное решение на все случаи жизни. Цель статьи — показать концепцию. Дальше вы уже сами решаете: писать мок самостоятельно, использовать стороннее решение, расширять функциональность или оставить всё как есть. Базу я показал.

API-клиенты

Перед тем как писать тесты, нам нужны API-клиенты для взаимодействия с gateway и с мок-сервисом. И здесь есть важный момент: мы не пишем какие-то специальные тестовые клиенты. Мы используем те же самые клиентские абстракции, которые уже существуют в проекте.

Это принципиально. Таким образом:

• тесты используют тот же сетевой слой, что и прод-код,

• ошибки сериализации, маршрутов и контрактов ловятся сразу,

• нет расхождения между тем, «как ходит код» и «как ходят тесты».

Начнём с клиента gateway.

Клиент gateway-сервиса

./services/gateway/client.py

import uuid

from httpx import Response

from config import settings
from libs.http.client.base import HTTPClient, get_http_client
from libs.http.client.handlers import handle_http_error, HTTPClientError
from libs.logger import get_logger
from libs.routes import APIRoutes
from services.gateway.schema import GetUserSummaryResponseSchema


class GatewayHTTPClientError(HTTPClientError):
    # Кастомное исключение клиента gateway
    pass


class GatewayHTTPClient(HTTPClient):
    @handle_http_error(
        client='GatewayHTTPClient',
        exception=GatewayHTTPClientError
    )
    async def get_user_summary_api(self, user_id: uuid.UUID) -> Response:
        # Низкоуровневый вызов HTTP-эндпоинта gateway
        return await self.get(
            f'{APIRoutes.GATEWAY}/users/{user_id}/summary'
        )

    async def get_user_summary(
        self,
        user_id: uuid.UUID
    ) -> GetUserSummaryResponseSchema:
        # Высокоуровневый метод:
        # выполняет HTTP-запрос и валидирует ответ по схеме
        response = await self.get_user_summary_api(user_id)
        return GetUserSummaryResponseSchema.model_validate_json(response.text)


def get_gateway_http_client() -> GatewayHTTPClient:
    # Фабрика клиента с уже настроенным httpx-клиентом и логгером
    logger = get_logger("GATEWAY_SERVICE_HTTP_CLIENT")
    client = get_http_client(
        logger=logger,
        config=settings.gateway_http_client
    )

    return GatewayHTTPClient(client=client)

Здесь нет ничего специфичного для тестов. Это обычный клиент, который:

• ходит по HTTP,

• обрабатывает ошибки,

• валидирует ответы через схемы.

Тесты используют его ровно так же, как его мог бы использовать любой другой код в системе.

Клиент mock-сервиса

Теперь клиент для управления моками. Он нужен только для тестов, но реализован в том же стиле и на тех же базовых абстракциях.

./services/mock/client.py

from httpx import Response

from config import settings
from libs.http.client.base import HTTPClient, get_http_client
from libs.http.client.handlers import handle_http_error, HTTPClientError
from libs.logger import get_logger
from services.mock.schema import CreateMockRulesRequest


class MockHTTPClientError(HTTPClientError):
    # Кастомное исключение клиента мок-сервиса
    pass


class MockHTTPClient(HTTPClient):
    @handle_http_error(
        client='MockHTTPClient',
        exception=MockHTTPClientError
    )
    async def create_mock_rule_api(
        self,
        request: CreateMockRulesRequest
    ) -> Response:
        # Создание правил мокирования через admin API
        return await self.post(
            '/admin/rules',
            json=request.model_dump(
                mode='json',
                by_alias=True
            )
        )

    @handle_http_error(
        client='MockHTTPClient',
        exception=MockHTTPClientError
    )
    async def delete_mock_rule_api(self) -> Response:
        # Полная очистка всех правил мокирования
        return await self.delete('/admin/rules')


def get_mock_http_client() -> MockHTTPClient:
    # Фабрика клиента мок-сервиса
    logger = get_logger("MOCK_SERVICE_HTTP_CLIENT")
    client = get_http_client(
        logger=logger,
        config=settings.mock_http_client
    )

    return MockHTTPClient(client=client)

Этот клиент позволяет из тестов:

• динамически задавать поведение сервисов,

• полностью сбрасывать состояние мока между тестами.

При этом он остаётся таким же HTTP-клиентом, как и все остальные в проекте.

Что здесь важно

Все остальные клиенты, через которые gateway в реальности взаимодействует с users, orders и billing, нас больше не интересуют. Они остаются в прод-коде, но в изоляционных тестах gateway с ними напрямую не общается — он ходит только в мок.

В результате:

• тесты работают через реальные HTTP-клиенты,

• но при этом полностью контролируют внешнее поведение системы,

• без сложной подготовки данных и без зависимости от окружения.

Дальше перейдём к фикстурам и посмотрим, как именно эти клиенты используются для сборки тестовых сценариев.

Фикстуры

Что в этой архитектуре делают фикстуры? По сути — две вещи.

Первая — инициализируют нужные API-клиенты.
Вторая — динамически формируют поведение внешних сервисов через мок.

Именно это и отличает такой подход от классических «тяжёлых» фикстур с подготовкой данных, баз, сидов и прочего. Мы не готовим данные — мы описываем сценарий.

Посмотрим на реализацию.

./tests/conftest.py

import faker
import pytest
import pytest_asyncio

from libs.routes import APIRoutes
from services.billing.schema import (
    BillingSummarySchema,
    GetBillingSummaryQuerySchema,
    GetBillingSummaryResponseSchema
)
from services.gateway.client import GatewayHTTPClient, get_gateway_http_client
from services.mock.client import MockHTTPClient, get_mock_http_client
from services.mock.schema import MockRuleSchema, CreateMockRulesRequest
from services.orders.schema import (
    OrdersSummarySchema,
    GetOrdersSummaryQuerySchema,
    GetOrdersSummaryResponseSchema
)
from services.users.schema import UserStatus, UserSchema, GetUserResponseSchema


@pytest.fixture
def fake() -> faker.Faker:
    # Faker используется для генерации реалистичных тестовых данных
    # Это позволяет не хардкодить значения и при этом сохранять читаемость сценариев
    return faker.Faker()


@pytest.fixture
def mock_http_client() -> MockHTTPClient:
    # HTTP-клиент для управления мок-сервисом:
    # создание и удаление правил мокирования
    return get_mock_http_client()


@pytest.fixture
def gateway_http_client() -> GatewayHTTPClient:
    # HTTP-клиент для обращения к gateway-сервису
    # Используется непосредственно в автотестах
    return get_gateway_http_client()


@pytest_asyncio.fixture
async def mock_user_service_get_user(
        fake: faker.Faker,
        mock_http_client: MockHTTPClient
) -> GetUserResponseSchema:
    # Формируем ответ users-service так, как если бы он пришёл из реального сервиса
    response = GetUserResponseSchema(
        user=UserSchema(
            id=fake.uuid4(),
            name=fake.user_name(),
            email=fake.email(),
            status=fake.enum(UserStatus)
        )
    )

    # Описываем правило мокирования:
    # конкретный HTTP-маршрут -> заранее заданный ответ
    mock_request = CreateMockRulesRequest(
        rules=[
            MockRuleSchema(
                route=f'{APIRoutes.USERS}/{response.user.id}',
                response=response.model_dump(by_alias=True),
            )
        ]
    )

    # Регистрируем правило в мок-сервисе
    await mock_http_client.create_mock_rule_api(mock_request)

    # Возвращаем данные в тест для последующих проверок
    yield response

    # После завершения теста полностью очищаем правила мокирования
    await mock_http_client.delete_mock_rule_api()


@pytest_asyncio.fixture
async def mock_orders_service_get_orders_summary(
        fake: faker.Faker,
        mock_http_client: MockHTTPClient,
        mock_user_service_get_user: GetUserResponseSchema
) -> GetOrdersSummaryResponseSchema:
    # Формируем query-параметры точно так же,
    # как это делает gateway при вызове orders-service
    query = GetOrdersSummaryQuerySchema(
        user_id=mock_user_service_get_user.user.id
    )

    # Формируем ответ orders-service
    response = GetOrdersSummaryResponseSchema(
        summary=OrdersSummarySchema(
            total_amount=fake.pyfloat(min_value=0, max_value=1000),
            total_orders=fake.pyint(min_value=0, max_value=1000),
            active_orders=fake.pyint(min_value=0, max_value=1000)
        )
    )

    # Регистрируем правило мокирования с учётом query-параметров
    mock_request = CreateMockRulesRequest(
        rules=[
            MockRuleSchema(
                route=APIRoutes.ORDERS,
                query=query.model_dump(mode='json', by_alias=True),
                response=response.model_dump(by_alias=True),
            )
        ]
    )

    await mock_http_client.create_mock_rule_api(mock_request)
    yield response
    await mock_http_client.delete_mock_rule_api()


@pytest_asyncio.fixture
async def mock_billing_service_get_billing_summary(
        fake: faker.Faker,
        mock_http_client: MockHTTPClient,
        mock_user_service_get_user: GetUserResponseSchema
) -> GetBillingSummaryResponseSchema:
    # Формируем query-параметры для billing-service
    query = GetBillingSummaryQuerySchema(
        user_id=mock_user_service_get_user.user.id
    )

    # Формируем ответ billing-service
    response = GetBillingSummaryResponseSchema(
        summary=BillingSummarySchema(
            debt=fake.pyfloat(min_value=0, max_value=100),
            balance=fake.pyfloat(min_value=0, max_value=100),
            currency="USD"
        )
    )

    # Регистрируем правило мокирования для billing-service
    mock_request = CreateMockRulesRequest(
        rules=[
            MockRuleSchema(
                route=APIRoutes.BILLING,
                query=query.model_dump(mode='json', by_alias=True),
                response=response.model_dump(by_alias=True),
            )
        ]
    )

    await mock_http_client.create_mock_rule_api(mock_request)
    yield response
    await mock_http_client.delete_mock_rule_api()

Почему это работает так хорошо

И посмотрите, насколько это удобно. Вместо сотен тяжёлых фикстур на подготовку данных мы декларативно собираем сценарий: какой пользователь существует, какие у него заказы, какой у него биллинг.

Это не подготовка окружения — это описание бизнес-кейса. И да, моки здесь простые. Но никто не мешает сделать их сложнее, если это потребуется.

Две важные оговорки

Первая — про параллельность.

Тесты в этом подходе запускаются синхронно, и это осознанное решение. Мы держим общее состояние на стороне мок-сервиса и можем себе это позволить.

Да, при желании можно прокидывать scenario_id, чистить и добавлять моки по нему, передавать его через заголовки и т.д. Но на практике это не нужно. Тесты здесь сверхбыстрые, и запускать их синхронно — абсолютно валидно.

Запуск через pytest-xdist в таком сценарии почти ничего не даёт. Он ускоряет только длинные тесты. Когда каждый тест выполняется за доли секунды, выигрыш съедается сетапом воркеров и распределением задач. В итоге лучше иметь чёткое и детерминированное окружение, чем 5 потоков параллельности и выигрыш в 20 секунд.

Вторая — про async / await.

Часто спрашивают:

«А зачем тут async/await, чтобы ускорить тесты?» Нет. Async/await здесь ничего не ускоряет — я подробно писал об этом отдельно в статье: «Асинхронные тесты для UI и API на Python: примеры, подводные камни и трезвый вывод»

Async здесь нужен по другой причине: весь проект живёт в async-экосистеме. Клиенты, серверный код, мок — всё async. В такой ситуации проще и честнее писать async-тесты, чем городить синхронные костыли и адаптеры.

Если вы изначально живёте в async-мире — async в тестах оправдан.

Дальше остаётся самое простое — написать сами тесты и посмотреть, насколько они в итоге получаются тонкими, быстрыми и стабильными.

Тесты

Теперь финально напишем автотест. И здесь происходит самое показательное.

После всей инфраструктуры, мока, клиентов и фикстур сам тест получается максимально тонким. В нём нет подготовки данных, нет сетапа окружения, нет сложной логики. Он читаетcя как описание бизнес-сценария.

./tests/test_gateway.py

import pytest

from services.billing.schema import GetBillingSummaryResponseSchema
from services.gateway.client import GatewayHTTPClient
from services.orders.schema import GetOrdersSummaryResponseSchema
from services.users.schema import GetUserResponseSchema


@pytest.mark.gateway
@pytest.mark.regression
async def test_get_user_summary(
        gateway_http_client: GatewayHTTPClient,
        # Ответ users-service, замоканный через мок
        mock_user_service_get_user: GetUserResponseSchema,
        # Ответ orders-service, замоканный через мок
        mock_orders_service_get_orders_summary: GetOrdersSummaryResponseSchema,
        # Ответ billing-service, замоканный через мок
        mock_billing_service_get_billing_summary: GetBillingSummaryResponseSchema,
):
    # Вызываем gateway так же, как это сделал бы реальный клиент
    response = await gateway_http_client.get_user_summary(
        user_id=mock_user_service_get_user.user.id
    )

    # Проверяем, что gateway корректно прокинул данные пользователя
    assert response.user == mock_user_service_get_user.user

    # Проверяем, что gateway корректно агрегировал данные по заказам
    assert response.orders_summary == mock_orders_service_get_orders_summary.summary

    # Проверяем, что gateway корректно агрегировал данные по биллингу
    assert response.billing_summary == mock_billing_service_get_billing_summary.summary

И на этом всё.

Здесь нет ни одной лишней строки. Тест:

• вызывает gateway по реальному HTTP,

• получает реальный HTTP-ответ,

• проверяет, что агрегированный результат соответствует тем контрактам, которые мы задали через моки.

Обратите внимание, что тест ничего не знает: о том, как устроены users, orders и billing, какие у них базы, как именно gateway внутри себя агрегирует данные.

Он проверяет ровно то, за что gateway отвечает по контракту. Ни больше, ни меньше.

Именно поэтому такие тесты:

• легко читаются,

• легко расширяются новыми сценариями,

• практически не флакают,

• и спокойно живут в CI/CD.

Дальше остаётся последний шаг — показать, как всё это запускается в CI/CD и сколько времени реально занимает такой прогон.

Запуск в CI/CD

Теперь посмотрим, как всё это запускается в CI/CD. И здесь тоже не будет никакой магии или сложных пайплайнов. Вся схема укладывается в стандартный docker-compose, один Dockerfile и простой workflow в GitHub Actions.

docker-compose.yaml

version: '3.9'

# Базовая конфигурация сервиса:
# сборка образа, рабочая директория и env-файл
x-base-service: &base-service
  build:
    context: .
    dockerfile: Dockerfile
  image: base-service
  volumes:
    - .:/app
  working_dir: /app
  environment:
    # ENV_FILE позволяет легко переключать конфигурации
    ENV_FILE: .env.ci

# Python-сервис с единым entrypoint
x-python-service: &python-service
  <<: *base-service
  entrypoint: [ "python", "-u", "-m" ]

services:
  # Мок-сервис
  mock:
    <<: *python-service
    ports: [ "8000:8000" ]
    # Запуск FastAPI-приложения мока
    command: "services.mock.server"
    container_name: "mock"

  # Gateway-сервис
  gateway:
    <<: *python-service
    ports: [ "9000:9000" ]
    # Запуск FastAPI-приложения gateway
    command: "services.gateway.server"
    container_name: "gateway"

Здесь важно, что:

• поднимаются только два сервиса — gateway и mock,

• никаких users / orders / billing нет вообще,

• вся изоляция достигается исключительно конфигурацией.

Dockerfile

FROM python:3.12-slim

WORKDIR /app

# Минимальный набор системных зависимостей
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Установка Python-зависимостей
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копируем исходный код проекта
COPY . .

Обычный, скучный Dockerfile. И это хорошо. Чем меньше магии — тем проще поддержка и воспроизводимость.

GitHub Actions

Финальный шаг — запуск в CI. Используем GitHub Actions.

./.github/workflows/test.yml

name: API mock tests

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  run-tests:
    runs-on: ubuntu-latest
    timeout-minutes: 30

    steps:
      - uses: actions/checkout@v6

      - uses: actions/setup-python@v6
        with:
          python-version: '3.12'

      - name: Start services
        # Поднимаем gateway и мок через docker-compose
        run: docker compose up -d --build

      - name: Install test dependencies
        # Устанавливаем зависимости для запуска pytest
        run: pip install -r requirements.txt

      - name: Run tests
        # Запускаем тесты с нужной конфигурацией
        run: pytest
        env:
          ENV_FILE: .env.ci

      - name: Stop services
        # Гарантированно останавливаем окружение
        if: always()
        run: docker compose down -v

Никаких сложных стадий, кэшей, кастомных runner’ов или танцев с бубном. Подняли сервисы → запустили тесты → прибрали за собой.

Результат

Результат выполнения можно посмотреть здесь: https://github.com/Nikita-Filonov/python-api-mock-tests/actions/runs/20349347572

И теперь самое интересное — время выполнения. Один тест проходит примерно за 0.28 секунды. Если прикинуть:

• 100 таких тестов — около 30 секунд,

• 1000 тестов — порядка 5 минут.

Для тестов, которые реально проверяют сетевое взаимодействие и интеграцию сервисов, это очень быстро.

Для сравнения: те же 1000 тестов на реальном окружении могут выполняться часами. И это в лучшем случае, который на практике почти никогда не встречается. Плюс флаки, таймауты, нестабильные зависимости и прочие «прелести».

Здесь же тесты:

• быстрые,

• стабильные,

• детерминированные,

• и прекрасно чувствуют себя в CI/CD.

Дальше можно спокойно масштабировать этот подход, не боясь, что пайплайн превратится в бутылочное горлышко.

А как же покрытие?

На этом месте обычно звучит возражение:

«Но у вас же плохое покрытие, вы ничего не тестируете».

И здесь важно честно ответить. Мы ничего не тестируем, когда тесты работают раз через раз, постоянно падают, флакают и в итоге выключаются из пайплайна. Такие тесты не дают покрытия — они дают иллюзию контроля.

А когда тесты быстрые, стабильные и детерминированные, мы как раз и тестируем всё, что действительно важно.

В текущей архитектуре мы тестируем gateway ровно в той зоне ответственности, за которую он отвечает: его HTTP-контракты и его логику агрегации. И при этом нам ничто не мешает зайти в репозитории users, orders и billing и написать там такие же изоляционные тесты. По тем же принципам, с тем же подходом.

В результате бизнес-логика покрыта полностью, просто:

• каждый сервис тестируется на своём уровне,

• в своей зоне ответственности,

• в изолированной и предсказуемой среде.

Это и есть нормальная, масштабируемая модель покрытия в микросервисной архитектуре.

И, конечно, никто не запрещает оставить несколько интеграционных happy path сценариев, чтобы убедиться, что всё базово работает в сборке. Но именно несколько — как контрольный слой, а не как основной способ тестирования.

Что дальше?

А дальше этот подход спокойно расширяется.

К таким тестам можно прикрутить всё, что угодно: отчёты, Allure, метки для left shift, подключение разработчиков, Kafka, более умные моки. Можно добавить трейсинг на мок-сервис и видеть, какие сервисы сколько раз вызывались, с какими параметрами и в каком порядке.

Возможности здесь не ограничены архитектурно. Всё зависит только от того, что вам действительно нужно.

Пример в статье показан на Python, но по факту он один в один переносится на любой другой язык и стек. Здесь нет ничего специфичного для FastAPI или httpx. Важна не реализация, важна концепция.

И, пожалуй, самое сильное во всём этом — результат. Мы написали тесты, которые получились:

• тонкими,

• стабильными,

• быстрыми.

При этом мы переиспользовали код разработчиков — те же клиенты, те же схемы, те же контракты. Мы не писали отдельный «тестовый мир», мы просто аккуратно подключились к уже существующей экосистеме.

В такой модели разработчикам не нужно разбираться в гигантском тестовом фреймворке. Они могут спокойно запускать эти тесты локально, дописывать новые сценарии и понимать, что именно проверяется. А при желании — расширять мок, добавлять трейсинг и получать прозрачность, до которой классические интеграционные тесты даже близко не доходят.

Именно поэтому такой подход работает. Не потому что он модный, а потому что он инженерно честный.

Заключение

Вся архитектура, код мок-сервиса, клиентов, фикстур и тестов, которые разобраны в этой статье, доступны в открытом виде на GitHub: https://github.com/Nikita-Filonov/python-api-mock-tests

---

В этой статье я показываю базовую идею изоляционных тестов: контракты → моки → локальный запуск сервиса → быстрые API-тесты.

Но в реальных системах инфраструктура обычно сложнее. Помимо HTTP-сервисов появляются базы данных, брокеры сообщений, кэш, асинхронные процессы. Соответственно, усложняется и изоляция.

Если тема показалась интересной и хочется разобрать её глубже на практике, я подробно рассматриваю этот подход в своём курсе «Автоматизация тестирования Backend с Python».

В курсе мы разбираем:

• построение полностью изолированного тестового окружения

• создание контрактных мок-сервисов

• тестирование HTTP и gRPC API

• изоляцию Kafka и асинхронных сценариев

• работу с базами данных в изолированных тестах

• запуск тестов в CI/CD

То есть по сути строим полноценную инженерную инфраструктуру для быстрых и стабильных автотестов.

Для читателей статьи действует промокод BACKBASE07QP1 со скидкой 30%.