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

Большинство API-тестов бесполезны: они флакают и тормозят CI. Показываю альтернативу — изоляционные тесты без стендов и боли.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Endpoint

GET /api/v1/users/{user_id}

Описание

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

Ответ

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

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

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 не считает эти данные сам и не применяет к ним бизнес-логику — он работает строго с тем, что пришло по контракту.

Endpoint

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

Описание

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

Ответ

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

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

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-параметров.

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

./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, что вместо реальных сервисов он должен ходить в него. Делается это исключительно через конфигурацию, без единого изменения в коде приложения.

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

# 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-клиенты для взаимодействия с 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,
обрабатывает ошибки,
валидирует ответы через схемы.

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

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

./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 здесь ничего не ускоряет — я подробно писал об этом отдельно в статье: «Асинхронные тесты для 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. И здесь тоже не будет никакой магии или сложных пайплайнов. Вся схема укладывается в стандартный 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. И это хорошо. Чем меньше магии — тем проще поддержка и воспроизводимость.

Финальный шаг — запуск в 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-автотесты без флаков, стендов и боли: изоляционный подход в CI/CD

Вступление

Контекст

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

Users service

Orders service

Billing service

Gateway service

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

Делаем мок

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

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

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

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

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

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

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

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

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

API-клиенты

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

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

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

Фикстуры

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

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

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

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

Тесты

Запуск в CI/CD

GitHub Actions

Результат

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

Что дальше?

Заключение