Вступление
WebSocket API часто тестируют как что-то отдельное: подключились к сокету, отправили сообщение, получили ответ. HTTP API при этом живёт в другой части тестов: создали пользователя, проверили эндпоинт, сравнили JSON. В небольших примерах такой подход выглядит нормально, но в реальных сервисах HTTP и WebSocket обычно работают вместе.
Например, пользователь регистрируется через REST API, затем подключается к WebSocket-каналу, получает историю сообщений, отправляет новое сообщение, а остальные участники получают событие в реальном времени. Формально это два разных транспорта, но с точки зрения продукта — один сценарий. У него есть общие пользователи, общие идентификаторы, общий контракт данных и ожидаемый порядок событий.
В этой статье разберём, как писать автотесты для такого API на Python: не в формате «hello world с одним сокетом», а как полноценный тестовый слой вокруг небольшого FastAPI-приложения с REST и WebSocket. В качестве примера будет использоваться простой чат: HTTP-эндпоинты отвечают за пользователей, сброс состояния и вспомогательные операции, а WebSocket — за события чата.
Основной фокус будет не на продакшен-архитектуре сервиса, а на тестируемости. Поэтому демо-приложение намеренно простое: без базы данных, с in-memory-хранилищем и предсказуемыми эндпоинтами. Это позволяет сосредоточиться на том, как устроить сами автотесты: контракты, HTTP- и WebSocket-клиенты, фикстуры, строгие проверки событий, Allure-шаги и запуск в CI.
В итоге получится не набор разрозненных проверок, а единый подход: создать пользователя через HTTP, подключить его к WebSocket по тому же user_id, проверить порядок входящих событий и убедиться, что тесты читаются как сценарий, а не как набор случайных json.loads, send и receive.
Используемые технологии
Перед тем как переходить к структуре проекта и тестам, зафиксируем стек. В примере используется небольшой FastAPI-сервис и отдельный слой автотестов на Python. Все зависимости подобраны так, чтобы показать именно тестирование REST + WebSocket API, а не усложнять демо инфраструктурой, базой данных или внешними брокерами.
Библиотека | Зачем используется |
|---|---|
Основной фреймворк для демо-приложения. На нём реализованы HTTP-эндпоинты и WebSocket-ручка чата. | |
ASGI-сервер для локального запуска FastAPI-приложения и запуска сервиса в CI перед тестами. | |
Описание моделей данных: пользователей, сообщений, входящих и исходящих WebSocket-событий. В тестах используется для валидации контрактов. | |
Управление настройками тестов через переменные окружения: HTTP URL, WebSocket URL и другие параметры запуска. | |
Основной тестовый фреймворк. На нём строятся фикстуры, маркеры и сами сценарии. | |
Поддержка асинхронных тестов и фикстур. Нужна, потому что HTTP-клиенты и WebSocket-клиенты работают через | |
Асинхронный HTTP-клиент для REST API: создание пользователей, сброс состояния, health-check и другие HTTP-вызовы. | |
Клиентская библиотека для работы с WebSocket в тестах: подключение, отправка сообщений, чтение событий. | |
Интеграция с Allure: шаги в клиентах, названия тестов, отчётность и диагностика падений. | |
Генерация тестовых данных: имена пользователей, тексты сообщений и другие значения, которые не должны быть захардкожены в каждом тесте. |
Важный момент: тесты здесь не ходят напрямую в
FastAPI TestClientи не импортируют внутренности приложения. Сервис поднимается как обычное приложение, а тесты работают с ним снаружи — через HTTP и WebSocket. Это ближе к тому, как API проверяется в реальной среде: есть запущенный backend, есть публичный контракт, есть клиентский код, который этот контракт использует.
Такой стек позволяет показать полный путь: поднять приложение, подготовить данные через REST, подключиться к WebSocket, получить события, провалидировать их через Pydantic-модели и увидеть результат в Allure-отчёте.
Почему тесты будут асинхронными
Отдельно стоит проговорить момент с async/await, чтобы сразу закрыть типовое возражение: асинхронные тесты здесь используются не потому, что «так быстрее» или потому что асинхронность сама по себе делает тестовый проект лучше.
Причина проще: WebSocket-клиент работает в асинхронной модели. Подключение к сокету, отправка фрейма, ожидание входящего события — всё это естественно ложится на
async/await. Если внутри теста уже есть асинхронный WebSocket-клиент, то логично не смешивать разные подходы, а писать весь сценарий в одном асинхронном контексте.
Поэтому HTTP-клиенты в примере тоже будут асинхронными: используется httpx.AsyncClient, а сами тесты запускаются через pytest-asyncio. Это не попытка «ускорить REST-запросы», а способ сделать тестовый код однородным. Пользователь создаётся через HTTP, затем этот же user_id используется для WebSocket-подключения, после чего тест ждёт события из сокета. Вся цепочка получается асинхронной, без блокировки корутин и без лишних переходов между sync- и async-кодом.
В этой статье я не буду подробно разбирать саму тему асинхронного тестирования: как работает event loop, какие бывают проблемы с фикстурами, где async действительно полезен, а где только усложняет проект. Об этом я уже писал отдельно в статье «Асинхронные тесты для UI и API на Python: примеры, подводные камни и трезвый вывод».
Здесь важно другое: для WebSocket API асинхронный контекст — не декоративная деталь, а нормальный рабочий способ писать такие тесты. Поэтому дальше все клиенты, фикстуры и тестовые сценарии будут построены вокруг async/await.
Контракты: тестируем не сокет, а модель взаимодействия
Перед тем как писать клиенты и фикстуры, нужно понять, что именно мы проверяем. В случае с WebSocket легко скатиться к тестам вида: «отправили строку в сокет, получили какую-то строку обратно». Такие проверки быстро становятся хрупкими: в тестах появляются json.loads, ручной доступ к ключам, проверки отдельных полей и неочевидные ожидания по порядку событий.
В этой статье подход другой: тесты проверяют контракт API. Неважно, пришли данные по HTTP или через WebSocket, в тесте они должны быть представлены как понятная модель данных.
В демо-приложении есть простой чат. Пользователь создаётся через REST API, после этого он может подключиться к WebSocket-каналу и участвовать в обмене событиями.
Важно: сами автотесты UI не трогают. Интерфейс нужен только для наглядности. Тесты работают ниже уровнем — через HTTP и WebSocket API.
REST-контракт
REST-часть в этом проекте отвечает за базовые операции: создать пользователя, получить список пользователей, получить пользователя по id, посмотреть историю сообщений, отправить сообщение через HTTP и сбросить состояние приложения.
Например, пользователь создаётся через POST /api/users:
@router.post("", status_code=status.HTTP_201_CREATED, response_model=User) async def create_user(request: CreateUserRequest) -> User: try: return await storage.create_user(user_id=request.id, username=request.username) except ValueError as exc: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
Контракт здесь задаётся не самим хендлером, а моделями входа и выхода:
CreateUserRequest -> User
То есть тесту важно не то, как внутри устроен storage, и не то, какие функции вызывает хендлер. Тесту важно, что при корректном запросе API возвращает пользователя ожидаемой формы.
Упрощённо REST API в проекте выглядит так:
Метод | Путь | Что делает |
|---|---|---|
|
| Проверка, что приложение запущено |
|
| Сброс in-memory-состояния |
|
| Создание пользователя |
|
| Получение списка пользователей |
|
| Получение пользователя по |
|
| Получение истории сообщений |
|
| Создание сообщения и рассылка события в WebSocket |
Отдельно интересен POST /api/messages. Он не только создаёт сообщение, но и отправляет событие всем WebSocket-клиентам:
@router.post("", status_code=status.HTTP_201_CREATED, response_model=Message) async def create_message(request: CreateMessageRequest) -> Message: try: message = await storage.insert_message(request) except LookupError as error: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(error)) await manager.broadcast( WSOutgoingEvent(type=WSEventType.MESSAGE, message=message) ) return message
Это хороший пример того, почему HTTP и WebSocket нельзя рассматривать как полностью независимые части. HTTP-запрос меняет состояние системы, а результат этого изменения может прилететь другому клиенту уже через WebSocket.
WebSocket-контракт
WebSocket в проекте доступен по адресу:
ws://<host>:<port>/api/{user_id}
Здесь намеренно используется тот же префикс /api, что и для REST. Пользователь сначала регистрируется через HTTP, а затем подключается к WebSocket по своему user_id.
Сценарий выглядит так:
POST /api/users ↓ получили / зафиксировали user_id ↓ WS connect: /api/{user_id} ↓ получаем события чата
Если пользователя с таким user_id нет, WebSocket-подключение отклоняется:
user = await storage.get_user(user_id) if user is None: await websocket.close( code=status.WS_1008_POLICY_VIOLATION, reason=f"User with id '{user_id}' is not registered", ) return
После успешного подключения сервер отправляет пользователю историю сообщений:
history = await storage.list_messages() await manager.send( event=WSOutgoingEvent(type=WSEventType.HISTORY, history=history), user_id=user_id, )
А остальным пользователям сообщает, что в чат кто-то вошёл:
await manager.broadcast( event=WSOutgoingEvent(type=WSEventType.USER_JOINED, user=user), exclude=[user_id], )
Дальше сервер читает входящие сообщения из сокета, валидирует их и рассылает всем участникам событие message.
Типы WebSocket-событий
Все исходящие WebSocket-события имеют общую обёртку WSOutgoingEvent. Это сильно упрощает тестирование: тест не гадает, какая структура пришла из сокета, а валидирует её как конкретную модель.
class WSEventType(str, Enum): ERROR = "error" MESSAGE = "message" HISTORY = "history" USER_LEFT = "user_left" USER_JOINED = "user_joined" class WSOutgoingEvent(BaseModel): type: WSEventType detail: str | None = None user: User | None = None message: Message | None = None history: list[Message] | None = None
В зависимости от type, в событии заполняются разные поля:
Тип события | Когда приходит | Основное поле |
|---|---|---|
| После подключения пользователя |
|
| Когда пользователь отправил сообщение |
|
| Когда другой пользователь вошёл в чат |
|
| Когда пользователь вышел из чата |
|
| Когда клиент отправил некорректные данные |
|
Например, событие с историей может выглядеть так:
{ "type": "history", "detail": null, "user": null, "message": null, "history": [] }
А событие нового сообщения — так:
{ "type": "message", "detail": null, "user": null, "message": { "id": "c9d8b7b6-9b0d-4e9b-9c0d-0c2f5c2e8a7a", "user_id": "01bf73ec-6e9a-4d58-98d7-1d2b2d3f4a10", "username": "Test", "text": "Hello!", "created_at": "2026-04-24T22:58:07.123456Z" }, "history": null }
Для тестов это принципиально. Мы проверяем не «пришла строка с текстом Hello!», а полноценное событие:
type = message message.user_id = ожидаемый пользователь message.text = ожидаемый текст message.created_at = есть и соответствует формату лишних неожиданных расхождений в модели нет
Такой подход лучше масштабируется. Если завтра в событии изменится поле, тип, вложенная структура или порядок бизнес-событий, тест упадёт не где-то на случайном KeyError, а на уровне контрактной проверки.
Почему модели есть и в приложении, и в тестах
В проекте модели есть в app/, но тесты не обязаны импортировать внутренние модели приложения напрямую. В тестовом слое можно держать свои Pydantic-схемы, которые зеркалят публичный API-контракт.
На первый взгляд это выглядит как дублирование. Но у такого подхода есть смысл.
Тесты не должны быть слишком тесно связаны с внутренней реализацией сервиса. Если хендлеры переедут в другие модули, изменится структура
app/, появится другой слой сервисов или репозиториев, контракт API от этого не обязан меняться. И тесты тоже не должны падать только потому, что внутри приложения переименовали файл или переложили модель в другой пакет.
При этом важно не доводить идею до абсурда. В небольшом учебном проекте можно использовать и общие DTO, если это упрощает поддержку. Но для демонстрации контрактного подхода полезно разделить:
app/ — модели, которыми пользуется приложение tests/ — модели, которыми тесты описывают ожидания от публичного API
Тогда тесты становятся независимым потребителем API. Они не проверяют, как устроен код внутри сервиса. Они проверяют, что внешний контракт работает так, как ожидается.
Именно от этого дальше будут строиться клиенты, фикстуры и ассерты: HTTP-ответы и WebSocket-фреймы будут превращаться в Pydantic-модели, а тесты будут сравнивать уже не сырые JSON-строки, а нормальные объекты предметной области.
HTTP API клиенты
Начнём с HTTP-части. В тестах не хочется каждый раз писать httpx.AsyncClient, вручную собирать URL, передавать JSON, вызывать raise_for_status() и парсить ответ. Такой код быстро расползается по тестам, а сами сценарии начинают читатьcя как набор технических HTTP-вызовов.
Поэтому HTTP-слой в тестах вынесен в отдельные клиенты.
Идея простая:
тестовый сценарий ↓ предметный HTTP-клиент ↓ базовый HTTP-клиент ↓ httpx.AsyncClient ↓ FastAPI-приложение
Базовый клиент знает, как выполнить HTTP-запрос. Предметный клиент знает, какое бизнес-действие выполняется: создать пользователя, сбросить состояние, получить данные и так далее. В результате тесты работают не с «сырым POST», а с понятными методами уровня сценария.
Базовый HTTP-клиент
from typing import Any import allure from httpx import AsyncClient, QueryParams, URL, Response from tests.config import settings class BaseHTTPClient: def __init__(self, client: AsyncClient): # Внутри храним обычный httpx.AsyncClient. # Он уже знает base_url, поэтому в предметных клиентах # можно использовать относительные пути: /api/users, /api/reset и т.д. self.client = client async def get(self, url: URL | str, params: QueryParams | None = None) -> Response: # Все низкоуровневые GET-запросы проходят через один метод. # Так проще добавить логирование, Allure steps, headers, # обработку ошибок или другую общую механику. with allure.step(f"Make GET request to {url}"): return await self.client.get(url, params=params) async def post(self, url: URL | str, json: Any | None = None) -> Response: # Аналогично для POST. # Метод возвращает сырой httpx.Response, чтобы вызывающий код сам решил, # что с ним делать: проверить status_code, вызвать raise_for_status() # или распарсить тело ответа в Pydantic-модель. with allure.step(f"Make POST request to {url}"): return await self.client.post(url, json=json) def build_base_http_client() -> AsyncClient: # Клиент создаётся с base_url из настроек. # Благодаря этому тесты не хардкодят localhost, порт или адрес окружения. # # Локально это может быть http://localhost:8000, # в CI — такой же адрес или другой, переданный через SERVER_HTTP_URL. return AsyncClient(base_url=str(settings.server_http_url))
Здесь есть несколько важных решений.
Во-первых, все HTTP-запросы проходят через BaseHTTPClient. Сейчас внутри только get, post и Allure-шаги, но это уже единая точка расширения. Если позже понадобится добавить авторизацию, общий заголовок, correlation id, дополнительное логирование или обработку ошибок, это можно будет сделать в одном месте.
Во-вторых, клиент использует base_url из настроек. В предметных клиентах дальше не будет строк вида:
"http://localhost:8000/api/users"
Вместо этого используется относительный путь:
"/api/users"
Так тесты проще запускать в разных окружениях: локально, в CI, на другом порту или против удалённого стенда.
Отдельный момент — Allure. Для асинхронного кода здесь используется не декоратор @allure.step, а контекстный менеджер:
with allure.step(...): return await ...
Так шаги в отчёте лучше соответствуют реальному выполнению кода: шаг начался, внутри выполнился await, затем шаг завершился. Для клиентов это обычно более предсказуемый вариант.
Схемы пользователя
tests/clients/http/users/schema.py
from datetime import datetime from pydantic import BaseModel, Field from tests.fakers import fake class User(BaseModel): # Модель ответа API. # Именно такую структуру тесты ожидают получить от сервера # после создания или получения пользователя. id: str username: str created_at: datetime class CreateUserRequest(BaseModel): # Модель запроса на создание пользователя. # # id в этом проекте передаётся клиентом, а не генерируется сервером. # Это удобно для тестов: один и тот же user_id потом используется # и в HTTP-части, и при подключении к WebSocket. id: str = Field(default_factory=fake.uuid4) # username для сценариев не всегда принципиален, # поэтому его можно генерировать автоматически. username: str = Field(default_factory=fake.username)
Это тестовые схемы. Они находятся в tests/, а не импортируются из app/. Такой подход делает тесты отдельным потребителем API-контракта.
Да, это частично дублирует модели приложения. Но в автотестах такое дублирование часто оправдано: тесты проверяют внешний контракт, а не внутреннюю структуру кода сервиса. Если внутри приложения переедут модули, изменятся сервисы или слой хранения, тесты не должны падать только из-за рефакторинга. Они должны падать тогда, когда изменился публичный контракт.
CreateUserRequest также показывает, где удобно использовать Faker. Тесту не нужно каждый раз руками писать uuid4 и имя пользователя. Но структура запроса остаётся явной: пользователь создаётся из модели, а не из произвольного словаря.
Клиент пользователей
tests/clients/http/users/client.py
import allure from httpx import Response from tests.clients.http.base import BaseHTTPClient, build_base_http_client from tests.clients.http.users.schema import User, CreateUserRequest class UsersHTTPClient(BaseHTTPClient): async def create_user_api(self, request: CreateUserRequest) -> Response: # Низкоуровневый метод для вызова API. # # Он возвращает сырой Response и не делает raise_for_status(). # Это полезно для негативных тестов: например, когда нужно проверить, # что API вернул 409, 400 или другой ожидаемый статус. with allure.step("Create user"): return await self.post("/api/users", json=request.model_dump()) async def create_user(self) -> User: # Высокоуровневый метод для обычного сценария: # создать пользователя и сразу получить валидированную модель User. # # Такой метод удобно использовать в фикстурах и позитивных тестах, # где сам HTTP-статус не является предметом проверки. request = CreateUserRequest() response = await self.create_user_api(request) # Если сервер вернул ошибочный статус, тест упадёт здесь. # Для happy path это нормальное поведение: пользователь должен быть создан. response.raise_for_status() # Ответ API валидируется через Pydantic. # Тест дальше работает не с dict и не с response.json(), # а с типизированной моделью User. return User.model_validate_json(response.text) def build_users_http_client() -> UsersHTTPClient: # Фабрика предметного клиента. # Снаружи тестам и фикстурам не нужно знать, # как именно создаётся базовый httpx.AsyncClient. return UsersHTTPClient(client=build_base_http_client())
Здесь используется полезный паттерн: разделение методов на *_api и более предметные методы.
Метод
create_user_api()— это тонкая обёртка над HTTP-вызовом. Он принимает готовыйCreateUserRequestи возвращаетResponse. Такой метод удобен, когда тесту важно проверить сам HTTP-ответ: статус-код, тело ошибки, заголовки или поведение на невалидных данных.Метод
create_user()— это уже метод уровня сценария. Он сам создаёт запрос, вызывает API, проверяет успешность ответа и возвращаетUser.
В фикстурах чаще будет использоваться именно create_user():
sender_user = await users_http_client.create_user()
Так фикстура не знает деталей HTTP-запроса. Её задача — подготовить пользователя для теста. А то, что под капотом был POST /api/users, raise_for_status() и Pydantic-валидация ответа, остаётся внутри клиента.
Это делает тесты заметно чище. Вместо такого кода в каждом сценарии:
response = await http_client.post("/api/users", json={...}) response.raise_for_status() user = User.model_validate_json(response.text)
в тесте или фикстуре остаётся одна строка:
user = await users_http_client.create_user()
Системный HTTP-клиент
tests/clients/http/system/client.py
import allure from tests.clients.http.base import BaseHTTPClient, build_base_http_client class SystemHTTPClient(BaseHTTPClient): async def reset_storage_api(self): # Сервисный метод для сброса in-memory-состояния приложения. # # В этом демо нет базы данных, поэтому состояние очищается # через отдельный тестовый endpoint POST /api/reset. # # Этот вызов будет использоваться в autouse-фикстуре, # чтобы каждый тест начинался с чистого состояния. with allure.step("Reset storage"): return await self.post("/api/reset") def build_system_http_client() -> SystemHTTPClient: # Отдельная фабрика для системного клиента. # По аналогии с UsersHTTPClient скрываем создание AsyncClient # за функцией build_*. return SystemHTTPClient(client=build_base_http_client())
Системный клиент отвечает не за бизнес-сценарии пользователя, а за технические операции тестового стенда. В этом проекте главный пример — сброс in-memory-хранилища через POST /api/reset.
Это важная часть тестовой архитектуры. Если тесты используют общее состояние приложения, они начинают зависеть друг от друга: один тест создал пользователя, второй случайно увидел его в истории, третий упал из-за уже занятого
id. Поэтому состояние нужно сбрасывать до или после каждого сценария.
Сам тест при этом не должен знать, как именно очищается стенд. Он не должен напрямую вызывать /api/reset. Для этого есть SystemHTTPClient, а позже — фикстура, которая будет использовать этот клиент автоматически.
Что в итоге даёт слой HTTP-клиентов
После этих файлов HTTP-часть тестов становится достаточно простой:
user = await users_http_client.create_user()
или, если нужен низкоуровневый контроль ответа:
request = CreateUserRequest(id="existing-id", username="Bob") response = await users_http_client.create_user_api(request) assert response.status_code == 409
В этом и есть смысл клиентского слоя. Тесты остаются сценарными, но при необходимости можно опуститься на уровень HTTP-контракта и проверить конкретный статус, тело ответа или ошибку.
WebSocket-клиенты
Теперь перейдём к WebSocket-части. Здесь подход такой же, как и с HTTP: тесты не должны напрямую работать с низкоуровневым клиентом, вручную собирать URL, отправлять JSON-строки и каждый раз делать model_validate_json.
WebSocket в тестах тоже лучше завернуть в небольшой клиентский слой.
Общая схема получается такой:
тестовый сценарий ↓ предметный WebSocket-клиент ↓ базовый WebSocket-клиент ↓ websockets ClientConnection ↓ FastAPI WebSocket endpoint
На уровне теста мы хотим видеть не «отправить строку в сокет», а понятное действие: отправить сообщение в чат, получить следующее событие, закрыть соединение. А все технические детали должны остаться внутри клиента.
Базовый WebSocket-клиент
import allure from websockets.asyncio.client import ClientConnection, connect from websockets.typing import Data from tests.config import settings class BaseWSClient: def __init__(self, client: ClientConnection) -> None: # Внутри храним подключение websockets. # ClientConnection — это уже открытый WebSocket-клиент, # через который можно отправлять и получать сообщения. self.client = client async def send(self, message: Data) -> None: # Базовый метод отправки сообщения в WebSocket. # # На этом уровне мы ещё не знаем, что именно отправляем: # чат-сообщение, команду, событие или другой payload. # Поэтому метод принимает общий тип Data из websockets. with allure.step("WebSocket: send message"): await self.client.send(message) async def close(self) -> None: # Закрытие WebSocket-соединения. # # Этот метод будет использоваться в фикстурах после yield, # чтобы соединения не оставались открытыми после завершения теста. with allure.step("WebSocket: close connection"): await self.client.close() async def receive(self) -> Data: # Получение следующего сообщения из WebSocket. # # Метод возвращает сырые данные, потому что базовый клиент # не должен знать о конкретной схеме чата. # Парсинг в Pydantic-модель будет выше — в ChatWSClient. with allure.step("WebSocket: receive message"): return await self.client.recv() async def build_base_ws_client(route: str) -> ClientConnection: # Собираем полный WebSocket URL из настроек и относительного route. # # settings.server_ws_url может быть, например: # ws://localhost:8000 # # route для чата: # /api/{user_id} # # В итоге получится: # ws://localhost:8000/api/{user_id} # # rstrip("/") защищает от ситуации, когда базовый URL # случайно задан со слэшем на конце. return await connect(str(settings.server_ws_url).rstrip("/") + route)
Здесь важно, что WebSocket URL тоже не хардкодится в тестах. Как и в HTTP-клиентах, базовый адрес берётся из настроек:
SERVER_WS_URL=ws://localhost:8000
А конкретный путь передаётся отдельно:
/api/{user_id}
В этом проекте WebSocket endpoint намеренно расположен под тем же префиксом /api, что и REST API:
HTTP: POST /api/users WS: ws://localhost:8000/api/{user_id}
Это подчёркивает идею статьи: HTTP и WebSocket — не два независимых мира, а два транспорта в рамках одного API. Пользователь создаётся через REST, а затем тот же user_id используется для подключения к WebSocket.
Схемы WebSocket-событий
tests/clients/ws/chat/schema.py
from datetime import datetime from enum import Enum from pydantic import BaseModel, Field from tests.clients.http.users.schema import User from tests.fakers import fake class WSEventType(str, Enum): # Типы событий, которые сервер может отправить через WebSocket. # # Эти значения являются частью публичного контракта. # Если сервер начнёт присылать другой type, тесты должны это поймать. ERROR = "error" MESSAGE = "message" HISTORY = "history" USER_LEFT = "user_left" USER_JOINED = "user_joined" class Message(BaseModel): # Модель сообщения чата. # # Она используется внутри WebSocket-событий, # например в событии type="message" или в history. id: str text: str user_id: str username: str created_at: datetime class WSIncomingEvent(BaseModel): # Модель события, которое тестовый клиент отправляет в WebSocket. # # В текущем демо клиент отправляет только сообщения в чат, # поэтому type по умолчанию равен "message". type: WSEventType = WSEventType.MESSAGE # Текст сообщения генерируется автоматически, # чтобы в каждом тесте не писать руками строки вроде "hello". text: str = Field(default_factory=fake.sentence) class WSOutgoingEvent(BaseModel): # Единая модель всех событий, которые сервер отправляет клиенту. # # В зависимости от type будет заполнено одно из полей: # - history для истории сообщений; # - message для нового сообщения; # - user для user_joined / user_left; # - detail для error. type: WSEventType user: User | None = None detail: str | None = None message: Message | None = None history: list[Message] | None = None
Это центральная часть WebSocket-тестов. Мы не хотим, чтобы в каждом тесте был код такого вида:
raw = await websocket.recv() data = json.loads(raw) assert data["type"] == "message" assert data["message"]["text"] == "Hello" assert data["message"]["user_id"] == user.id
Такой код быстро становится шумным. Он размазывает знание о контракте по тестам и делает сценарии хуже читаемыми.
Вместо этого входящее сообщение из WebSocket превращается в Pydantic-модель:
event = WSOutgoingEvent.model_validate_json(raw)
И дальше тест работает уже с объектом:
assert event.type == WSEventType.MESSAGE assert event.message.user_id == sender_user.id assert event.message.text == sent_event.text
Это кажется небольшой разницей, но на практике она сильно влияет на качество тестов. Если сервер пришлёт невалидный JSON, неправильный тип события, некорректную дату или неожиданную структуру вложенного
message, тест упадёт уже на этапе валидации модели.
Клиент чата
tests/clients/ws/chat/client.py
import allure from websockets.typing import Data from tests.clients.ws.base import BaseWSClient, build_base_ws_client from tests.clients.ws.chat.schema import WSIncomingEvent, WSOutgoingEvent class ChatWSClient(BaseWSClient): async def send_chat_message_api(self, event: WSIncomingEvent) -> None: # Низкоуровневый метод отправки события в чат. # # На вход принимается готовая Pydantic-модель WSIncomingEvent. # Перед отправкой она сериализуется в JSON-строку. # # Метод не создаёт данные сам — это удобно для тестов, # где нужно отправить конкретный текст или проверить edge case. with allure.step("Send chat message"): await self.send(event.model_dump_json()) async def receive_chat_message_api(self) -> Data: # Низкоуровневый метод получения следующего сообщения из WebSocket. # # Возвращает сырые данные из сокета. # Это может быть полезно для отдельных проверок, # где важно увидеть именно исходный payload. with allure.step("Receive next chat message"): return await self.receive() async def send_chat_message(self) -> WSIncomingEvent: # Высокоуровневый метод для обычного сценария: # создать корректное входящее событие и отправить его в чат. # # Метод возвращает отправленное событие, # чтобы тест мог потом сравнить его с тем, # что пришло другим участникам через WebSocket. event = WSIncomingEvent() await self.send_chat_message_api(event) return event async def receive_chat_message(self) -> WSOutgoingEvent: # Высокоуровневый метод получения события от сервера. # # Сначала читаем сырой payload из WebSocket, # затем валидируем его через Pydantic. # # После этого тесты работают уже не с JSON-строкой, # а с типизированной моделью WSOutgoingEvent. raw = await self.receive_chat_message_api() return WSOutgoingEvent.model_validate_json(raw) async def build_chat_ws_client(user_id: str) -> ChatWSClient: # Фабрика клиента чата. # # Для подключения нужен user_id уже существующего пользователя. # Поэтому перед созданием ChatWSClient пользователь должен быть # зарегистрирован через HTTP API. # # Полный путь будет выглядеть так: # ws://localhost:8000/api/{user_id} client = await build_base_ws_client(f"/api/{user_id}") return ChatWSClient(client=client)
Здесь снова используется разделение на методы уровня *_api и более удобные методы уровня сценария.
Метод send_chat_message_api() отправляет конкретный WSIncomingEvent. Он полезен, когда в тесте нужно полностью контролировать входящие данные: текст сообщения, тип события или невалидный payload в негативных проверках.
Метод send_chat_message() — сценарный. Он сам создаёт корректное сообщение, отправляет его и возвращает модель отправленного события. Это удобно для позитивных тестов:
sent_event = await sender_chat_ws_client.send_chat_message()
Дальше другой WebSocket-клиент может получить событие от сервера:
actual_event = await receiver_chat_ws_client.receive_chat_message()
И тест уже сравнивает не байты сокета, а контрактные модели.
Что даёт слой WebSocket-клиентов
После введения WebSocket-клиентов тестовый код становится намного чище.
Вместо такого варианта:
websocket = await connect(f"ws://localhost:8000/api/{user_id}") await websocket.send(json.dumps({"type": "message", "text": "Hello"})) raw = await websocket.recv() data = json.loads(raw) assert data["type"] == "message" assert data["message"]["text"] == "Hello"
в тесте остаётся сценарный код:
sent_event = await sender_chat_ws_client.send_chat_message() actual_event = await receiver_chat_ws_client.receive_chat_message()
А дальше можно сравнивать Pydantic-модели или их отдельные поля.
Это и есть основная идея слоя WebSocket-клиентов: техническую работу с сокетом оставить внутри клиента, а тестам дать предметный интерфейс. Тест должен описывать сценарий чата, а не детали библиотеки
websockets.
В итоге HTTP- и WebSocket-клиенты работают в одной логике:
HTTP-клиент создаёт пользователя ↓ WebSocket-клиент подключается по user_id ↓ WebSocket-клиент отправляет событие ↓ другой WebSocket-клиент получает типизированное событие ↓ тест проверяет контракт и порядок событий
Дальше поверх этих клиентов можно строить фикстуры: автоматически создавать пользователей, открывать WebSocket-соединения, закрывать их после теста и сбрасывать состояние приложения между сценариями.
Фикстуры
Клиенты сами по себе ещё не дают удобных тестов. Да, у нас уже есть UsersHTTPClient, SystemHTTPClient и ChatWSClient, но в каждом тесте всё ещё не хочется вручную создавать пользователей, открывать WebSocket-соединения и не забывать закрывать их после выполнения сценария.
Для этого в проекте используются pytest-фикстуры.
Задача фикстур здесь вполне практическая:
создать HTTP-клиенты
создать пользователей
открыть WebSocket-подключения
сбросить состояние приложения до и после теста
закрыть сокеты после теста
Весь этот код лучше держать не в самих тестах, а в отдельных модулях tests/fixtures/..., разделённых по смыслу.
Фикстуры пользователей
import pytest from tests.clients.http.users.client import UsersHTTPClient, build_users_http_client from tests.clients.http.users.schema import User @pytest.fixture def users_http_client() -> UsersHTTPClient: # Фикстура возвращает предметный HTTP-клиент для работы с пользователями. # # Сам тест не должен знать, как создаётся httpx.AsyncClient, # какой base_url используется и где лежит фабрика клиента. return build_users_http_client() @pytest.fixture async def sender_user(users_http_client: UsersHTTPClient) -> User: # Пользователь-отправитель. # # Создаётся через REST API: POST /api/users. # Важно, что это не объект "из воздуха" и не мок. # Пользователь реально регистрируется в тестовом приложении, # а потом его id используется для WebSocket-подключения. return await users_http_client.create_user() @pytest.fixture async def receiver_user(users_http_client: UsersHTTPClient) -> User: # Пользователь-получатель. # # Он создаётся так же, как sender_user, но отдельной фикстурой, # чтобы в тестах было явно видно, кто отправляет сообщение, # а кто получает событие через WebSocket. return await users_http_client.create_user()
Здесь есть одна базовая клиентская фикстура и две фикстуры пользователей.
Можно было бы создать одну фикстуру user, а в тесте вызывать её несколько раз через фабрику. Но для сценариев чата часто полезнее иметь явно названные роли:
sender_user receiver_user
Так тест читается ближе к бизнес-сценарию: один пользователь отправляет сообщение, другой его получает.
Важно и то, что пользователи создаются именно через HTTP API. Это принципиальный момент: WebSocket-сценарий не стартует сам по себе. Сначала пользователь появляется в системе через REST, и только потом подключается к сокету по своему
user_id.
Системные фикстуры и сброс состояния
import pytest from tests.clients.http.system.client import SystemHTTPClient, build_system_http_client @pytest.fixture def system_http_client() -> SystemHTTPClient: # Системный HTTP-клиент. # # Через него тестовый слой выполняет технические операции: # например, сбрасывает in-memory-хранилище приложения. return build_system_http_client() @pytest.fixture(autouse=True) async def reset_storage(system_http_client: SystemHTTPClient) -> None: # autouse=True означает, что фикстура будет применяться # автоматически к каждому тесту. # # Тесту не нужно явно писать reset_storage в аргументах. # Состояние будет сброшено в любом случае. await system_http_client.reset_storage_api() # yield разделяет setup и teardown. # # Всё, что выше yield, выполняется до теста. # Сам тест выполняется в момент yield. # Всё, что ниже yield, выполняется после теста. yield # Повторный сброс после теста нужен, чтобы состояние не протекало # в следующие сценарии даже в том случае, если тест создал пользователей, # сообщения или открыл несколько соединений. await system_http_client.reset_storage_api()
В демо-приложении нет базы данных. Состояние хранится in-memory, поэтому его нужно явно очищать между тестами. Для этого есть технический endpoint:
POST /api/reset
Фикстура reset_storage вызывает его до и после каждого теста.
Почему до и после, а не только до?
Сброс до теста гарантирует, что сценарий стартует из чистого состояния. Сброс после теста помогает не оставлять мусор, если следующий тест по какой-то причине будет запускаться в том же процессе приложения. Это особенно полезно для локальных прогонов и CI, где хочется, чтобы тесты были максимально независимыми друг от друга.
autouse=True здесь уместен, потому что изоляция состояния — это не особенность конкретного теста, а общее правило всего набора автотестов. Каждый тест должен начинаться с чистого хранилища.
Фикстуры WebSocket-клиентов
from collections.abc import AsyncGenerator import pytest from tests.clients.http.users.schema import User from tests.clients.ws.chat.client import ChatWSClient, build_chat_ws_client @pytest.fixture async def sender_chat_ws_client(sender_user: User) -> AsyncGenerator[ChatWSClient, None]: # WebSocket-клиент для пользователя-отправителя. # # Фикстура зависит от sender_user, а значит сначала пользователь # будет создан через HTTP API, и только потом мы попробуем # подключиться к WebSocket по его id. client = await build_chat_ws_client(sender_user.id) try: # На время выполнения теста отдаём уже подключенный ChatWSClient. yield client finally: # После теста обязательно закрываем WebSocket-соединение. # # finally нужен, чтобы соединение закрывалось даже тогда, # когда тест упал на assert или получил исключение. await client.close() @pytest.fixture async def receiver_chat_ws_client(receiver_user: User) -> AsyncGenerator[ChatWSClient, None]: # WebSocket-клиент для пользователя-получателя. # # Логика такая же: # 1. receiver_user создаётся через REST; # 2. по receiver_user.id открывается WebSocket; # 3. тест получает готовый ChatWSClient; # 4. после теста соединение закрывается. client = await build_chat_ws_client(receiver_user.id) try: yield client finally: await client.close()
WebSocket-фикстуры асинхронные, потому что само подключение к сокету — асинхронная операция:
client = await build_chat_ws_client(sender_user.id)
Закрытие соединения тоже выполняется через await:
await client.close()
Поэтому здесь используется async def и AsyncGenerator. Фикстура открывает соединение до теста, отдаёт клиент через yield, а после завершения теста закрывает сокет в finally.
Это важная практика. Если не закрывать WebSocket-соединения, тесты могут начать влиять друг на друга: старые клиенты останутся подключёнными, будут получать события, сервер будет считать пользователя активным, а следующие проверки начнут падать неочевидным образом.
Схема жизненного цикла такой фикстуры выглядит так:
создать пользователя через HTTP ↓ открыть WebSocket по /api/{user_id} ↓ передать ChatWSClient в тест ↓ выполнить тестовый сценарий ↓ закрыть WebSocket в finally
Как фикстуры связаны между собой
Если разложить зависимости, получится такая цепочка:
Сначала создаётся HTTP-клиент пользователей. Через него создаются sender_user и receiver_user. После этого для каждого пользователя открывается свой WebSocket-клиент.
Параллельно для каждого теста автоматически срабатывает reset_storage: до теста и после теста. Сам тест в итоге получает уже готовые объекты и не занимается технической подготовкой.
В тесте это может выглядеть примерно так:
async def test_when_sender_sends_message_then_receiver_gets_it( sender_chat_ws_client: ChatWSClient, receiver_chat_ws_client: ChatWSClient, ): ...
Внутри теста уже не нужно думать о том, как создать пользователя, какой URL у WebSocket, как закрыть соединение и как очистить storage. Всё это описано в фикстурах.
Подключение фикстур через pytest_plugins
pytest_plugins = ( # Фикстуры для WebSocket-клиентов чата: # sender_chat_ws_client, receiver_chat_ws_client. "tests.fixtures.chat", # Фикстуры для пользователей и HTTP-клиента пользователей: # users_http_client, sender_user, receiver_user. "tests.fixtures.users", # Системные фикстуры: # system_http_client, reset_storage. "tests.fixtures.system", )
Вместо того чтобы складывать все фикстуры в один большой conftest.py, они разнесены по отдельным файлам:
tests/fixtures/users.py tests/fixtures/chat.py tests/fixtures/system.py
А в conftest.py остаётся только список подключаемых pytest-плагинов.
Так проще поддерживать проект. Когда нужно посмотреть, как создаются пользователи, открывается
tests/fixtures/users.py. Когда нужно разобраться с WebSocket-клиентами —tests/fixtures/chat.py. Когда интересует сброс состояния —tests/fixtures/system.py.На маленьком проекте это может казаться избыточным, но такой подход хорошо масштабируется. Через несколько месяцев в тестах могут появиться фикстуры для авторизации, сообщений, файлов, ролей, прав доступа, моков внешних сервисов. Если всё это заранее складывать в один
conftest.py, он быстро превращается в файл, который страшно открыть.
Настройки
Следующий небольшой, но важный слой — настройки тестов. Сейчас у проекта всего два параметра: HTTP URL и WebSocket URL. Но даже их лучше не хардкодить внутри клиентов.
Плохой вариант — когда адрес сервиса размазан по тестам и клиентам:
"http://localhost:8000/api/users" "ws://localhost:8000/api/{user_id}"
Пока проект запускается только локально и только на одном порту, это может не мешать. Но как только тесты нужно запустить в CI, на другом порту или против отдельного стенда, такие строки начинают мешать. Приходится искать их по проекту и менять руками.
В этом проекте настройки вынесены в один файл.
from pydantic import HttpUrl, WebsocketUrl from pydantic_settings import BaseSettings class Settings(BaseSettings): # Базовый WebSocket URL тестируемого приложения. # # По умолчанию используется локальный сервер: # ws://localhost:8000 # # В WebSocket-клиентах к этому адресу будет добавляться # конкретный route, например /api/{user_id}. # # Значение можно переопределить через переменную окружения: # SERVER_WS_URL=ws://127.0.0.1:8899 server_ws_url: WebsocketUrl = WebsocketUrl("ws://localhost:8000") # Базовый HTTP URL тестируемого приложения. # # По умолчанию используется тот же локальный сервер: # http://localhost:8000 # # HTTP-клиенты будут использовать его как base_url, # а дальше ходить по относительным путям: # /api/users, /api/reset, /api/messages и т.д. # # Значение можно переопределить через переменную окружения: # SERVER_HTTP_URL=http://127.0.0.1:8899 server_http_url: HttpUrl = HttpUrl("http://localhost:8000") # Единый объект настроек, который импортируется в клиентах. # # Например: # - build_base_http_client использует settings.server_http_url; # - build_base_ws_client использует settings.server_ws_url. settings = Settings()
Здесь используется pydantic-settings. Это удобно по двум причинам.
Первая — настройки имеют типы. Для HTTP-адреса используется
HttpUrl, для WebSocket-адреса —WebsocketUrl. Если случайно передать некорректное значение, проблема всплывёт сразу при созданииSettings, а не где-то глубоко в момент подключения клиента.Вторая — значения можно переопределять через переменные окружения. Поле
server_http_urlможно задать черезSERVER_HTTP_URL, аserver_ws_url— черезSERVER_WS_URL.
Как настройки используются в HTTP-клиенте
В базовом HTTP-клиенте настройки используются при создании httpx.AsyncClient:
def build_base_http_client() -> AsyncClient: return AsyncClient(base_url=str(settings.server_http_url))
После этого предметные HTTP-клиенты работают только с относительными путями:
return await self.post("/api/users", json=request.model_dump())
То есть клиент пользователей не знает, где именно запущено приложение. Он знает только путь /api/users. Базовый адрес приходит из settings.
Как настройки используются в WebSocket-клиенте
С WebSocket логика такая же:
async def build_base_ws_client(route: str) -> ClientConnection: return await connect(str(settings.server_ws_url).rstrip("/") + route)
А клиент чата передаёт только route:
client = await build_base_ws_client(f"/api/{user_id}")
В итоге полный адрес собирается из двух частей:
SERVER_WS_URL + route ws://localhost:8000 + /api/{user_id}
Получается:
ws://localhost:8000/api/{user_id}
Это особенно удобно, потому что HTTP и WebSocket URL можно менять независимо. Например, локально оба транспорта могут быть на localhost:8000, а в другом окружении WebSocket может быть доступен по отдельному адресу.
Проверки
Когда WebSocket-клиент уже возвращает не сырую строку, а WSOutgoingEvent, проверки тоже можно писать на уровне контракта. Не нужно в каждом тесте доставать поля из JSON, сравнивать словари и вручную собирать сообщения об ошибках.
Для этого в проекте есть отдельный модуль с assertion-хелперами.
import allure from tests.clients.ws.chat.schema import WSOutgoingEvent, WSEventType @allure.step("Assert WebSocket outgoing event") def assert_ws_outgoing_event(actual: WSOutgoingEvent, expected: WSOutgoingEvent) -> None: # Базовая проверка исходящего WebSocket-события. # # На вход принимаются две Pydantic-модели: # - actual: событие, которое реально пришло из WebSocket; # - expected: событие, которое тест ожидал получить. # # Сравниваем модели целиком, а не отдельные поля. # Это помогает не пропустить ситуацию, когда: # - изменился type; # - не заполнилось нужное поле; # - заполнилось не то поле; # - поменялась структура вложенного message/user/history. assert actual == expected, ( "WebSocket outgoing event does not match the expected payload.\n" f"Expected:\n{expected.model_dump_json(indent=2)}\n" f"Actual:\n{actual.model_dump_json(indent=2)}" ) def assert_empty_history_ws_outgoing_event(actual: WSOutgoingEvent) -> None: # Assertion для частого сценария: # пользователь подключился к пустому чату и должен получить HISTORY # с пустым списком сообщений. # # Сам тест в этом случае не собирает expected-объект вручную, # а использует готовую предметную проверку. assert_ws_outgoing_event( actual=actual, expected=WSOutgoingEvent(type=WSEventType.HISTORY, history=[]) )
Идея здесь простая: если в WebSocket прилетает событие, тест должен проверять его как целостный объект. Например, событие истории — это не просто
type == "history". Это событие типаhistory, у которого полеhistoryсодержит ожидаемый список сообщений, а остальные поля не заполнены.
То есть вместо проверки в стиле:
assert actual.type == WSEventType.HISTORY assert actual.history == []
можно написать:
assert_empty_history_ws_outgoing_event(actual)
А внутри уже будет полное сравнение с ожидаемой моделью:
WSOutgoingEvent(type=WSEventType.HISTORY, history=[])
Так тест становится короче, но не теряет строгость.
Faker и тестовые данные
В тестах постоянно нужны данные: user_id, username, текст сообщения. Можно каждый раз писать их руками, но тогда в коде быстро появляется шум:
user_id = "user-1" username = "test-user" text = "hello"
Для пары тестов это нормально. Но когда сценариев становится больше, копипаста начинает мешать: где-то забыли поменять
id, где-то два теста используют одинакового пользователя, где-то текст сообщения вообще не важен, но всё равно занимает место в тесте.
В этом проекте для генерации таких значений используется небольшой wrapper над Faker.
from faker import Faker class Fake: def __init__(self, faker: Faker): # Внутри храним объект Faker. # # Оборачиваем его в свой класс не потому, что Faker плохой, # а чтобы тестовый код зависел от нашего небольшого интерфейса, # а не от всей библиотеки напрямую. self.faker = faker def uuid4(self) -> str: # Генерация строкового UUID. # # В проекте id пользователя задаётся клиентом при регистрации, # поэтому тестам удобно уметь создавать новый user_id без копипасты. return self.faker.uuid4() def sentence(self) -> str: # Генерация текста сообщения. # # Для большинства позитивных сценариев нам не важно, # какой именно текст отправит пользователь. # Важно, что текст есть и что потом он пришёл в событии MESSAGE. return self.faker.sentence() def username(self) -> str: # Генерация имени пользователя. # # username в демо-приложении display-only: # он нужен для отображения и событий, # но уникальность пользователя определяется по id. return self.faker.user_name() # Единый объект fake, который используется в тестовых схемах. fake = Fake(faker=Faker())
На первый взгляд этот файл может показаться лишним: можно было бы импортировать
Fakerпрямо в схемах или тестах. Но отдельная обёртка даёт аккуратную точку управления тестовыми данными.Если завтра понадобится изменить формат
username, сделать более читаемые тексты сообщений или зафиксировать seed для воспроизводимости, это можно будет сделать в одном месте.
Как Faker связан с Pydantic-моделями
Главная польза появляется не там, где мы вызываем fake.uuid4() вручную, а там, где генерация данных встраивается в модели через default_factory.
Например, модель создания пользователя:
class CreateUserRequest(BaseModel): id: str = Field(default_factory=fake.uuid4) username: str = Field(default_factory=fake.username)
Теперь тесту или клиенту не нужно каждый раз собирать данные руками:
request = CreateUserRequest()
При создании модели автоматически появятся id и username.
То же самое используется для входящего WebSocket-события:
class WSIncomingEvent(BaseModel): type: WSEventType = WSEventType.MESSAGE text: str = Field(default_factory=fake.sentence)
В результате сценарный метод клиента может быть очень простым:
async def send_chat_message(self) -> WSIncomingEvent: event = WSIncomingEvent() await self.send_chat_message_api(event) return event
Тесты
Теперь всё, что было выше, складывается в сами тестовые сценарии.
К этому моменту у нас уже есть:
HTTP-клиенты для подготовки состояния
WebSocket-клиенты для работы с событиями
Pydantic-модели для контрактов
фикстуры для пользователей и сокетов
assertion-хелперы для читаемых проверок
настройки для локального запуска и CI
Поэтому сами тесты должны быть максимально сценарными. В идеале по телу теста должно быть понятно, что проверяется, без необходимости проваливаться в детали
httpx,websockets,json.loadsи ручной подготовки данных.
Настройки pytest
Сначала посмотрим на конфигурацию pytest.
[pytest] # Базовые опции запуска. # # -s не перехватывает stdout/stderr. # -v включает подробный вывод по тестам. addopts = -s -v # Включаем автоматический режим pytest-asyncio. # # Благодаря этому async-тесты и async-фикстуры # нормально обрабатываются pytest без ручной настройки event loop # в каждом тестовом модуле. asyncio_mode = auto # Правила поиска тестовых файлов. # # pytest будет искать: # - файлы вида *_tests.py # - файлы вида test_*.py python_files = *_tests.py test_*.py # Правила поиска тестовых классов. python_classes = Test* # Правила поиска тестовых функций. python_functions = test_* # Область жизни event loop для async-фикстур. # # В данном проекте function scope хорошо подходит: # каждый тест получает изолированный контекст выполнения. asyncio_default_fixture_loop_scope = function # Маркеры тестов. # # chat — тесты WebSocket-чата. # regression — тесты, входящие в регрессионный набор. markers = chat: Маркировка для тестов чата. regression: Маркировка для регрессионных тестов.
Для небольшого проекта это может казаться избыточным, но привычка маркировать тесты окупается быстро. Со временем появляются smoke-тесты, регрессия, интеграционные сценарии, негативные проверки, отдельные группы по фичам. Лучше заложить эту структуру сразу.
Тесты чата
Теперь сам тестовый модуль.
import allure import pytest from tests.assertions import assert_empty_history_ws_outgoing_event, assert_ws_outgoing_event from tests.clients.http.users.schema import User from tests.clients.ws.chat.client import ChatWSClient from tests.clients.ws.chat.schema import Message, WSEventType, WSOutgoingEvent # Все тесты внутри класса относятся к WebSocket-чату # и входят в регрессионный набор. # # Благодаря этому не нужно дублировать маркеры # над каждым отдельным тестом. @pytest.mark.chat @pytest.mark.regression class TestChat: @allure.title("On connect both peers receives empty history") async def test_on_connect_both_peers_receives_empty_history( self, sender_chat_ws_client: ChatWSClient, receiver_chat_ws_client: ChatWSClient, ) -> None: # Оба клиента уже созданы фикстурами: # - sender_user создан через HTTP; # - receiver_user создан через HTTP; # - для каждого пользователя открыт WebSocket. # # После подключения сервер должен отправить HISTORY. # Так как storage перед тестом очищен autouse-фикстурой, # история должна быть пустой. sender_history_event = await sender_chat_ws_client.receive_chat_message() receiver_history_event = await receiver_chat_ws_client.receive_chat_message() # Проверяем не отдельные поля, а контракт события целиком: # type=history и history=[]. assert_empty_history_ws_outgoing_event(sender_history_event) assert_empty_history_ws_outgoing_event(receiver_history_event) @allure.title("Message sent by one user is delivered to the other") async def test_message_sent_by_one_user_is_delivered_to_the_other( self, sender_user: User, sender_chat_ws_client: ChatWSClient, receiver_chat_ws_client: ChatWSClient, ) -> None: # Отправитель отправляет сообщение в чат. # # Метод send_chat_message сам создаёт WSIncomingEvent, # отправляет его в WebSocket и возвращает модель события, # чтобы потом можно было сравнить отправленный text # с тем, что получил другой пользователь. sender_message_event = await sender_chat_ws_client.send_chat_message() # Важно: при подключении receiver сначала получает HISTORY. # # Даже если сообщение уже отправлено, первое непрочитанное событие # у receiver — это событие истории, потому что сервер отправляет его # сразу после подключения. receiver_history_event = await receiver_chat_ws_client.receive_chat_message() # Следующее событие — MESSAGE, которое было разослано после отправки # сообщения sender-клиентом. receiver_message_event = await receiver_chat_ws_client.receive_chat_message() assert_empty_history_ws_outgoing_event(receiver_history_event) # Проверяем MESSAGE как полноценный WSOutgoingEvent. # # id и created_at генерируются сервером, поэтому берём их из actual. # Но бизнес-значимые поля проверяем строго: # - type события; # - text отправленного сообщения; # - user_id отправителя; # - username отправителя. assert_ws_outgoing_event( actual=receiver_message_event, expected=WSOutgoingEvent( type=WSEventType.MESSAGE, message=Message( id=receiver_message_event.message.id, text=sender_message_event.text, user_id=sender_user.id, username=sender_user.username, created_at=receiver_message_event.message.created_at ) ) ) @allure.title("When sender disconnects other peer receives user left") async def test_when_sender_disconnects_other_peer_receives_user_left( self, sender_user: User, sender_chat_ws_client: ChatWSClient, receiver_chat_ws_client: ChatWSClient, ) -> None: # Закрываем WebSocket-соединение отправителя. # # На стороне сервера это должно привести к disconnect # и рассылке события USER_LEFT остальным участникам. await sender_chat_ws_client.close() # Как и в предыдущем тесте, у receiver первым непрочитанным событием # остаётся HISTORY, полученный сразу после подключения. receiver_history_event = await receiver_chat_ws_client.receive_chat_message() # Следующее событие — USER_LEFT. receiver_user_left_event = await receiver_chat_ws_client.receive_chat_message() assert_empty_history_ws_outgoing_event(receiver_history_event) # Проверяем, что другой участник получил событие о выходе sender. assert_ws_outgoing_event( actual=receiver_user_left_event, expected=WSOutgoingEvent(type=WSEventType.USER_LEFT, user=sender_user) )
Первый сценарий: пустая история при подключении
Первый тест проверяет базовое поведение WebSocket API: после подключения клиент получает историю сообщений.
sender_history_event = await sender_chat_ws_client.receive_chat_message() receiver_history_event = await receiver_chat_ws_client.receive_chat_message()
Так как перед каждым тестом срабатывает reset_storage, история должна быть пустой. Это и проверяет helper:
assert_empty_history_ws_outgoing_event(sender_history_event) assert_empty_history_ws_outgoing_event(receiver_history_event)
Этот тест выглядит простым, но он важен. Он фиксирует контракт подключения: клиент после соединения не находится в неопределённом состоянии, а сразу получает событие history.
Второй сценарий: доставка сообщения другому участнику
Второй тест проверяет основной сценарий чата: один пользователь отправляет сообщение, другой получает событие message.
sender_message_event = await sender_chat_ws_client.send_chat_message()
Здесь отправляется входящее WebSocket-событие:
{ "type": "message", "text": "..." }
После этого receiver-клиент читает события из своего WebSocket-соединения:
receiver_history_event = await receiver_chat_ws_client.receive_chat_message() receiver_message_event = await receiver_chat_ws_client.receive_chat_message()
Порядок здесь важен. Сначала читается history, потому что это первое событие, которое сервер отправляет клиенту после подключения. Только затем читается message.
И уже после этого проверяется само сообщение:
assert_ws_outgoing_event( actual=receiver_message_event, expected=WSOutgoingEvent( type=WSEventType.MESSAGE, message=Message( id=receiver_message_event.message.id, text=sender_message_event.text, user_id=sender_user.id, username=sender_user.username, created_at=receiver_message_event.message.created_at ) ) )
Здесь хорошо видно, как работает паттерн с динамическими полями. Тест не знает заранее id сообщения и created_at, потому что их создаёт сервер. Поэтому они берутся из actual-события. Но всё остальное проверяется строго: тип события, текст, user_id и username.
Третий сценарий: событие о выходе пользователя
Третий тест проверяет, что при отключении одного пользователя остальные участники получают событие user_left.
await sender_chat_ws_client.close()
После закрытия соединения сервер должен выполнить disconnect и отправить событие другим клиентам.
Receiver снова сначала читает history:
receiver_history_event = await receiver_chat_ws_client.receive_chat_message()
А затем получает событие выхода пользователя:
receiver_user_left_event = await receiver_chat_ws_client.receive_chat_message()
Проверка выглядит так:
assert_ws_outgoing_event( actual=receiver_user_left_event, expected=WSOutgoingEvent(type=WSEventType.USER_LEFT, user=sender_user) )
То есть тест проверяет не просто факт, что «что-то пришло после disconnect», а конкретный контракт:
type = user_left user = sender_user
Что в итоге получилось
Финальные тесты не выглядят как работа с низкоуровневым WebSocket API. В них почти нет технического шума.
Тесты не создают пользователей вручную, не собирают URL, не вызывают
json.loads, не закрывают сокеты в теле сценария и не знают, как устроенhttpx.AsyncClient.
Они описывают поведение:
при подключении оба пользователя получают пустую историю
сообщение одного пользователя доставляется другому
при отключении пользователя другой участник получает
user_left
А вся инфраструктура спрятана в слоях ниже:
Именно это и было целью: WebSocket API тестируется не как набор случайных
send/recv, а как контрактный сценарий, связанный с HTTP API и общим состоянием приложения.
Запуск на CI/CD
Последний слой — запуск этих же тестов в CI. Здесь идея такая же, как и при локальном запуске: поднять приложение, дождаться, что оно готово принимать запросы, запустить pytest и сохранить Allure-результаты.
Важно, что в CI не появляется какой-то отдельный тестовый режим. Тесты остаются теми же самыми. Отличается только окружение: локально сервер запускается вручную в терминале, а в GitHub Actions он поднимается отдельным шагом workflow.
name: API WebSocket Tests on: # Запускаем workflow при push в main. push: branches: - main # И при pull request в main. pull_request: branches: - main jobs: run-tests: runs-on: ubuntu-latest steps: - name: Check out repository # Забираем код репозитория в runner. uses: actions/checkout@v6 - name: Set up Python # Устанавливаем нужную версию Python. uses: actions/setup-python@v6 with: python-version: '3.12' - name: Install dependencies # Устанавливаем зависимости приложения и тестов. # # В этом проекте один requirements.txt используется # и для FastAPI-приложения, и для тестового слоя. run: | python -m pip install --upgrade pip pip install -r requirements.txt - name: Start server # Запускаем FastAPI-приложение в фоне. # # Важно: в конце команды uvicorn стоит &. # Без него шаг зависнет на запущенном сервере, # и GitHub Actions просто не дойдёт до pytest. # # sleep 3 — самый простой способ дать приложению время стартовать. # Для демо этого достаточно, но в реальном проекте лучше заменить # sleep на ожидание /api/health. run: | uvicorn app.main:app --host 127.0.0.1 --port 8000 & sleep 3 - name: Run tests # Запускаем регрессионные тесты и сохраняем сырые Allure-результаты. # # --alluredir=allure-results создаёт директорию, # которую потом можно скачать как artifact # или использовать для генерации HTML-отчёта. run: | pytest -m regression --alluredir=allure-results - name: Upload Allure results # Загружаем allure-results как artifact. # # if: always() нужен, чтобы результаты сохранились даже тогда, # когда pytest завершился с ошибкой. if: always() uses: actions/upload-artifact@v7 with: name: allure-results path: allure-results publish-report: # Отчёт пытаемся опубликовать всегда: # и после успешного прогона, и после падения тестов. if: always() # Этот job зависит от run-tests, # потому что ему нужны allure-results из предыдущего job. needs: [ run-tests ] runs-on: ubuntu-latest steps: - name: Check out repository # Забираем ветку gh-pages. # # В ней будет храниться история Allure-отчётов # и собранная HTML-версия отчёта. uses: actions/checkout@v6 with: ref: gh-pages path: gh-pages - name: Download Allure results # Скачиваем artifact, который был загружен в run-tests. uses: actions/download-artifact@v8 with: name: allure-results path: allure-results - name: Build Allure report # Собираем HTML-отчёт из allure-results. # # allure_history позволяет сохранять историю запусков, # чтобы в отчёте были тренды и предыдущие результаты. uses: simple-elf/allure-report-action@v1.13 if: always() with: allure_results: allure-results allure_history: allure-history - name: Deploy Allure report # Публикуем собранный отчёт в ветку gh-pages. # # После настройки GitHub Pages содержимое этой ветки # будет доступно как статическая страница. if: always() uses: peaceiris/actions-gh-pages@v4 with: github_token: ${{ secrets.GITHUB_TOKEN }} publish_branch: gh-pages publish_dir: allure-history
После выполнения workflow artifact можно скачать прямо из GitHub Actions и локально открыть отчёт через Allure CLI. Но удобнее сразу публиковать HTML-версию на GitHub Pages.
Публикация отчёта на GitHub Pages
Второй job — publish-report — отвечает за публикацию Allure-отчёта.
Он делает три вещи:
скачивает allure-results из предыдущего job
собирает HTML-отчёт
публикует результат в ветку gh-pages
После этого GitHub Pages может загружать содержимое ветки gh-pages как статический сайт.
Но здесь есть несколько нюансов, о которых лучше сказать явно.
Во-первых, ветка gh-pages должна существовать или должна быть корректно создана при первой публикации.
В данном workflow есть checkout именно этой ветки:
with: ref: gh-pages path: gh-pages
Если ветки нет, этот шаг может упасть. Для первого запуска можно заранее создать ветку gh-pages вручную или адаптировать workflow так, чтобы он сам корректно обрабатывал первую публикацию.
Во-вторых, у GITHUB_TOKEN должны быть права на запись в репозиторий. Для публикации в gh-pages workflow должен иметь возможность пушить изменения в ветку.
Обычно это настраивается в настройках репозитория:
Settings → Actions → General → Workflow permissions
Там нужно разрешить workflow запись:
Read and write permissions
Дополнительно в некоторых репозиториях удобно явно указать permissions прямо в workflow:
permissions: contents: write
Например:
name: API WebSocket Tests permissions: contents: write on: push: branches: - main
В-третьих, нужно настроить GitHub Pages: указать, из какой ветки и какой директории публиковать сайт. В этом варианте отчёт публикуется в ветку gh-pages, поэтому в настройках Pages нужно выбрать публикацию из этой ветки.
Обычно логика такая:
Settings → Pages
Source: Deploy from a branch
Branch:
gh-pagesFolder:
/root
После этого GitHub выдаст URL, по которому будет доступен Allure-отчёт.
Заключение
Все ссылки на код, отчеты и запуски тестов в CI/CD можно найти на моем GitHub:
