Как правильно писать gRPC API автотесты на Python

Вступление

В этой статье будем писать автотесты для gRPC API на Python.

В качестве тестируемого сервиса возьмём небольшой CoursesService. Это gRPC-сервис для работы с курсами: он умеет создавать курс, получать курс по id, получать список курсов с фильтрацией, обновлять курс и удалять его.

Контракт сервиса описан через .proto-файлы. Из этих контрактов генерируется Python-код: protobuf-модели, request/response-классы и gRPC stub. Именно через этот сгенерированный код автотесты будут обращаться к сервису.

Тесты будем писать не через прямой импорт handler-ов и не через вызов внутренних функций приложения. Сервис будет запускаться как отдельное приложение, а тесты будут ходить в него снаружи — через настоящий gRPC-канал. То есть так же, как с ним работал бы реальный клиент или другой сервис.

В статье разберём полный тестовый слой вокруг gRPC API:

• генерацию Python-кода из .proto-контрактов;
• структуру проекта с отдельным слоем автотестов;
• базовый gRPC-клиент для тестов;
• клиент-обёртку для CoursesService;
• pytest-фикстуры для подготовки данных;
• генерацию тестовых request-объектов через Faker;
• проверки успешных protobuf-ответов;
• проверки gRPC-ошибок и status codes;
• Allure-шаги и запуск тестов.

Демо-сервис намеренно простой: Python, gRPC, SQLite и понятные CRUD-операции. Здесь задача не в том, чтобы показать сложную backend-архитектуру, а в том, чтобы на небольшом примере разобрать, как удобно и поддерживаемо тестировать gRPC API.

В итоге получится не набор случайных вызовов stub.CreateCourse(...), а нормальная структура API-автотестов: с контрактами, сгенерированными классами, тестовым клиентом, фикстурами, assertion-хелперами и читаемыми сценариями.

Используемые технологии

Перед тем как переходить к структуре проекта и автотестам, зафиксируем стек. В примере используется небольшой gRPC-сервис на Python и отдельный слой API-автотестов. Все зависимости подобраны так, чтобы показать именно тестирование gRPC API, а не перегружать пример лишней инфраструктурой.

Библиотека	Зачем используется
grpcio	Основная библиотека для работы с gRPC в Python. Используется и на стороне сервера, и на стороне тестового клиента.
grpcio-tools	Нужна для генерации Python-кода из .proto-контрактов: сообщений, enum-ов и gRPC stub-классов.
grpcio-reflection	Включает server reflection. Благодаря этому сервис можно изучать через grpcurl или GUI-клиенты без ручной передачи всех proto-файлов.
mypy-protobuf	Генерирует .pyi-файлы с типами для protobuf-классов. Это улучшает подсказки в IDE и делает работу с generated-кодом более удобной.
pytest	Основной тестовый фреймворк. На нём строятся тестовые сценарии, фикстуры, маркировки и запуск regression-набора.
allure-pytest	Интеграция с Allure. Используется для названий тестов, шагов в клиентах и более понятной диагностики падений.
Faker	Генерация тестовых данных: названий курсов, описаний, авторов, категорий и других значений, которые не хочется хардкодить в каждом тесте.
pydantic-settings	Управление настройками через переменные окружения: адрес gRPC-сервера, порт и другие параметры тестового запуска.
SQLAlchemy	Работа с базой данных на стороне демо-сервиса. Для тестов это не главный объект проверки, но сервису нужно где-то хранить курсы.
aiosqlite	SQLite-драйвер для асинхронной работы с базой. Используется в демо-приложении как простое локальное хранилище.

Автотесты в этой статье проверяют внешний gRPC-контракт сервиса, а не его внутреннюю реализацию.

Тестам не важно, как сервис хранит данные: в SQLite, PostgreSQL, Redis или in-memory-хранилище. Важно поведение на уровне API:

• CreateCourseRequest должен создавать курс и возвращать корректный CreateCourseResponse;
• GetCourseRequest с существующим id должен возвращать нужный Course;
• невалидные данные должны приводить к ожидаемому gRPC status code.

Поэтому тесты работают через gRPC-клиент. Они не импортируют handler-ы, repository или другие внутренние модули приложения. Сервис запускается отдельно, а тесты обращаются к нему через публичный gRPC-интерфейс.

gRPC-контракты и protobuf

В REST API контракт чаще всего описывается через HTTP-метод, URL, JSON-тело запроса и JSON-тело ответа. Например: POST /courses, передаём JSON с названием курса, получаем JSON с созданной сущностью.

В gRPC всё устроено иначе. Контракт описывается в .proto-файлах. В них фиксируются сервисы, RPC-методы, структуры запросов, структуры ответов, enum-ы и типы полей. Эти файлы являются источником правды для клиента и сервера.

В нашем проекте контракт лежит в директории: protos/contracts/services/courses/. После изменения .proto-файлов из них генерируется Python-код: protos/gen/

И уже этот сгенерированный код используется в тестах. То есть тесты не пишут произвольные словари и не собирают JSON руками. Они создают нормальные protobuf-объекты: CreateCourseRequest, GetCourseRequest, UpdateCourseRequest и так далее.

Что такое protobuf

Protocol Buffers, или protobuf, — это формат описания и сериализации данных. Если сильно упростить, .proto-файл говорит:

У нас есть такое-то сообщение, в нём есть такие-то поля, у каждого поля есть тип и номер.

Например, базовая модель курса описана так:

message Course {
  string id = 1;
  string title = 2;
  string description = 3;
  string author = 4;
  string category = 5;
  double price = 6;
  CourseLevel level = 7;
  string created_at = 8;
}

Это означает, что Course — не просто произвольный словарь. У него есть заранее определённая структура: id, title, description, author, category, price, level и created_at.

Числа справа — это номера полей в protobuf-контракте. Они нужны не для красоты и не как порядок отображения в документации. Эти номера используются при бинарной сериализации данных. Поэтому в protobuf важно аккуратно относиться к изменению существующих полей: переименование поля обычно менее опасно, чем изменение его номера или типа.

Для уровня курса используется enum:

enum CourseLevel {
  COURSE_LEVEL_UNSPECIFIED = 0;
  COURSE_LEVEL_BEGINNER = 1;
  COURSE_LEVEL_INTERMEDIATE = 2;
  COURSE_LEVEL_ADVANCED = 3;
}

Enum помогает не передавать уровень курса произвольной строкой вроде "easy", "junior" или "advanced-level". Вместо этого клиент выбирает одно из заранее известных значений.

При этом значение 0 в protobuf часто используется как значение по умолчанию. В нашем контракте это COURSE_LEVEL_UNSPECIFIED. Для демо-сервиса оно считается невалидным при создании и обновлении курса. Это хороший пример для негативных тестов: можно проверить, что сервис не принимает курс без явно указанного нормального уровня.

Сервис и RPC-методы

Главный gRPC-сервис описан в файле courses_service.proto:

service CoursesService {
  rpc GetCourse (GetCourseRequest) returns (GetCourseResponse);
  rpc GetCourses (GetCoursesRequest) returns (GetCoursesResponse);
  rpc CreateCourse (CreateCourseRequest) returns (CreateCourseResponse);
  rpc UpdateCourse (UpdateCourseRequest) returns (UpdateCourseResponse);
  rpc DeleteCourse (DeleteCourseRequest) returns (DeleteCourseResponse);
}

Здесь CoursesService — это публичный gRPC-сервис, а строки внутри него — RPC-методы.

Каждый метод явно описывает, какой request он принимает и какой response возвращает. Например:

rpc CreateCourse (CreateCourseRequest) returns (CreateCourseResponse);

Это значит: метод CreateCourse принимает сообщение CreateCourseRequest, а возвращает CreateCourseResponse.

Запрос на создание курса выглядит так:

message CreateCourseRequest {
  string title = 1;
  string description = 2;
  string author = 3;
  string category = 4;
  double price = 5;
  CourseLevel level = 6;
}

message CreateCourseResponse {
  Course course = 1;
}

То есть клиент передаёт данные для создания курса, а в ответ получает полноценный объект Course, уже с серверными полями. Например, с id и created_at.

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

Получение одного курса и списка курсов

Для получения одного курса используется простой request с идентификатором:

message GetCourseRequest {
  string id = 1;
}

message GetCourseResponse {
  Course course = 1;
}

Такой метод удобно тестировать через сценарий: сначала создать курс через CreateCourse, затем запросить его через GetCourse и сравнить данные.

Для списка курсов контракт чуть интереснее:

message GetCoursesRequest {
  optional string category = 1;
  optional string author = 2;
}

message GetCoursesResponse {
  repeated Course courses = 1;
}

Здесь появляются два важных protobuf-механизма.

• Первый — optional. Поля category и author можно передать, а можно не передавать. Это удобно для фильтрации: можно получить все курсы, можно отфильтровать по категории, можно отфильтровать по автору.
• Второй — repeated. Поле courses содержит не один объект Course, а список курсов. В Python после генерации proto-кода с таким полем можно работать как с коллекцией protobuf-сообщений.

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

Обновление и удаление курса

Обновление курса описано отдельными request/response-сообщениями:

message UpdateCourseRequest {
  string id = 1;
  string title = 2;
  string description = 3;
  string author = 4;
  string category = 5;
  double price = 6;
  CourseLevel level = 7;
}

message UpdateCourseResponse {
  Course course = 1;
}

Здесь в запросе обязательно передаётся id курса и новый набор полей. В тестах такой метод обычно проверяется через цепочку: создать курс, обновить его, затем убедиться, что в ответе пришли новые данные.

Удаление выглядит минималистично:

message DeleteCourseRequest {
  string id = 1;
}

message DeleteCourseResponse {}

Пустой response в gRPC — это нормальная практика. Если удаление прошло успешно, сервису не всегда нужно возвращать тело ответа. Для теста в таком случае важны сам факт успешного RPC-вызова и дальнейшая проверка состояния: например, после удаления попытка получить курс по тому же id должна завершиться ожидаемой gRPC-ошибкой.

Тестовые утилиты

Перед клиентами, фикстурами и самими тестами подготовим несколько утилит:

• конфигурацию gRPC-клиента;
• очистку хранилища курсов;
• генератор тестовых данных;
• тестовый логгер.

Эти модули будут лежать в tests/tools/.

Конфигурация gRPC-клиента

Файл tests/tools/config.py описывает настройки подключения к gRPC-сервису.

from pydantic import BaseModel, IPvAnyAddress


class GRPCClientTestConfig(BaseModel):
    # Порт gRPC-сервиса, к которому будут подключаться автотесты.
    # Например: 9000.
    port: int

    # Адрес gRPC-сервиса.
    # В локальном запуске это обычно 127.0.0.1.
    address: IPvAnyAddress

    @property
    def url(self) -> str:
        # Итоговый адрес подключения для grpc.insecure_channel.
        # gRPC-клиент ожидает строку в формате "host:port".
        return f'{self.address}:{self.port}'

Эта модель нужна, чтобы не хардкодить адрес сервиса внутри клиента. Позже настройки будут загружаться из tests/.env через pydantic-settings, а клиент будет получать готовую строку подключения через config.url.

Очистка базы перед тестами

Файл tests/tools/database.py содержит утилиту для очистки таблицы с курсами.

import asyncio

from app.repository import CoursesRepository


def clear_courses_db() -> None:
    # Очищает хранилище курсов перед тестом и после теста.
    #
    # Это нужно, чтобы один тест не зависел от данных,
    # которые были созданы в другом тесте.
    #
    # Сами проверки при этом остаются на уровне gRPC API:
    # тесты создают, получают, обновляют и удаляют курсы через gRPC-клиент.
    asyncio.run(CoursesRepository.clear())

Здесь есть прямой импорт CoursesRepository, но он используется только для подготовки чистого состояния. В тестовых проверках база напрямую не читается: результат проверяется через ответы gRPC-сервиса.

Генерация тестовых данных

Файл tests/tools/fakers.py отвечает за генерацию данных для request-объектов.

import uuid

from faker import Faker
from google.protobuf.internal.enum_type_wrapper import EnumTypeWrapper


class Fake:
    def __init__(self, faker: Faker):
        # Оборачиваем Faker в свой класс,
        # чтобы хранить все генераторы тестовых данных в одном месте.
        self.faker = faker

    def uuid(self) -> uuid.UUID:
        # Генерирует случайный UUID.
        #
        # Используется, например, в негативных тестах,
        # когда нужно запросить несуществующий курс.
        return self.faker.uuid4(cast_to=None)

    def title(self) -> str:
        # Генерирует название курса.
        #
        # Значение используется в CreateCourseRequest
        # и UpdateCourseRequest.
        return self.faker.catch_phrase()

    def price(self, min_value: float = 9.99, max_value: float = 499.99) -> float:
        # Генерирует цену курса в заданном диапазоне.
        #
        # right_digits=2 ограничивает цену двумя знаками после запятой,
        # чтобы значение было похоже на реальную стоимость курса.
        return self.faker.pyfloat(
            min_value=min_value,
            max_value=max_value,
            right_digits=2,
        )

    def author(self) -> str:
        # Генерирует имя автора курса.
        return self.faker.name()

    def category(self) -> str:
        # Возвращает одну из допустимых категорий курса.
        #
        # Фиксированный список удобен для тестов фильтрации:
        # можно создать курс с категорией, а затем искать курсы по ней.
        return self.faker.random_element([
            "cloud",
            "design",
            "devops",
            "mobile",
            "business",
            "security",
            "marketing",
            "analytics",
            "programming",
            "data-science",
        ])

    def proto_enum(self, value: EnumTypeWrapper) -> int:
        # Выбирает случайное значение из protobuf enum.
        #
        # Значение 0 пропускается, потому что в protobuf оно обычно означает
        # значение по умолчанию. В нашем контракте это COURSE_LEVEL_UNSPECIFIED,
        # который не должен использоваться как валидный уровень курса.
        return self.faker.random_element([
            item for item in value.values() if item != 0
        ])

    def description(self) -> str:
        # Генерирует описание курса из нескольких предложений.
        return self.faker.paragraph(nb_sentences=3)


# Общий объект для тестов.
# Его можно импортировать в клиентах, фикстурах и тестовых сценариях.
fake = Fake(faker=Faker())

Такая обёртка удобнее, чем прямой вызов Faker() в каждом тесте. Все правила генерации данных находятся в одном месте: формат цены, список категорий, генерация валидных enum-значений и структура текстовых полей.

Тестовый логгер

Файл tests/tools/logger.py создаёт логгер для тестового gRPC-клиента.

import logging
from functools import lru_cache


@lru_cache(maxsize=None)
def get_test_logger(name: str) -> logging.Logger:
    # Создаёт и кэширует логгер по имени.
    #
    # Кэш нужен, чтобы при повторном вызове с тем же name
    # не создавать новый logger и не добавлять обработчики повторно.
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # Логи будут выводиться в консоль при запуске тестов.
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)

    # Формат логов:
    # время события, имя логгера, уровень логирования и сообщение.
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    console_handler.setFormatter(formatter)

    # Подключаем обработчик к логгеру.
    logger.addHandler(console_handler)

    return logger

Логгер понадобится при создании gRPC-канала. Через него можно выводить информацию о gRPC-вызовах: какой метод был вызван, какой статус вернулся и где произошла ошибка. Это упрощает диагностику падений в локальном запуске и CI.

Конфигурация тестового запуска

Теперь добавим настройки для автотестов. Адрес и порт gRPC-сервиса не будем хардкодить в клиенте. Вынесем их в .env, а в коде опишем через pydantic-settings.

Структура будет такой:

tests/
├── .env
├── settings.py
└── tools/
    └── config.py

Модель GRPCClientTestConfig уже описана в tests/tools/config.py. Теперь подключим её к общим настройкам тестов.

Файл tests/settings.py

from pydantic_settings import BaseSettings, SettingsConfigDict

from tests.tools.config import GRPCClientTestConfig


class TestSettings(BaseSettings):
    # Конфигурация загрузки настроек для автотестов.
    #
    # Настройки читаются из файла tests/.env.
    # Это позволяет менять адрес и порт gRPC-сервиса
    # без изменения кода тестового клиента.
    model_config = SettingsConfigDict(
        # Разрешаем дополнительные переменные окружения.
        # Это удобно, если позже в tests/.env появятся настройки
        # для других клиентов или сервисов.
        extra='allow',

        # Файл с переменными окружения для тестового запуска.
        env_file='./tests/.env',

        # Кодировка .env-файла.
        env_file_encoding='utf-8',

        # Разделитель для вложенных настроек.
        #
        # Например:
        # COURSES_GRPC_CLIENT.PORT=9000
        #
        # будет преобразовано в:
        # courses_grpc_client.port
        env_nested_delimiter='.'
    )

    # Настройки подключения к gRPC-сервису курсов.
    #
    # Внутри лежат:
    # - address: IP-адрес сервиса;
    # - port: порт сервиса;
    # - url: строка подключения в формате "address:port".
    courses_grpc_client: GRPCClientTestConfig


# Глобальный объект настроек для тестов.
#
# Его можно импортировать в клиентах и фикстурах,
# чтобы получать параметры подключения к сервису.
test_settings = TestSettings()

Файл tests/.env

# Порт gRPC-сервиса CoursesService.
# На этот порт тестовый клиент будет отправлять RPC-вызовы.
COURSES_GRPC_CLIENT.PORT=9000

# IP-адрес gRPC-сервиса CoursesService.
# Для локального запуска используется 127.0.0.1.
COURSES_GRPC_CLIENT.ADDRESS=127.0.0.1

После загрузки настроек объект test_settings.courses_grpc_client будет содержать готовую конфигурацию клиента.

Например:

test_settings.courses_grpc_client.address
# 127.0.0.1

test_settings.courses_grpc_client.port
# 9000

test_settings.courses_grpc_client.url
# 127.0.0.1:9000

Эта конфигурация позже будет использоваться при создании gRPC-канала:

channel = grpc.insecure_channel(test_settings.courses_grpc_client.url)

Так тестовый клиент остаётся независимым от конкретного окружения. Локально можно использовать 127.0.0.1:9000, а в CI или Docker-окружении поменять значения через переменные окружения или другой .env-файл.

gRPC-клиенты для автотестов

Теперь подготовим клиентский слой.

Задача клиента — скрыть технические детали gRPC-вызовов от тестов. В тестах не должно быть постоянного создания channel, stub, request-объектов и логирования. Тест должен читаться как сценарий: создать курс, получить курс, обновить курс, удалить курс.

Клиентский слой разобьём на три части:

tests/clients/
├── client.py              # Базовый gRPC-клиент и сборка channel
├── logger_interceptor.py  # Интерсептор для логирования gRPC-вызовов
└── courses/
    └── client.py          # Клиент для CoursesService

Базовый gRPC-клиент

Файл tests/clients/client.py отвечает за базовую инфраструктуру клиента: хранение channel и создание gRPC-канала.

from logging import Logger

import grpc
import grpc.experimental.gevent as grpc_gevent

from tests.clients.logger_interceptor import GRPCLoggerInterceptor
from tests.tools.config import GRPCClientTestConfig


# Включаем gevent-совместимость для gRPC.
#
# Это полезно, если тестовая инфраструктура или другие инструменты проекта
# используют gevent-модель конкурентности.
grpc_gevent.init_gevent()


class GRPCTestClient:
    def __init__(self, channel: grpc.Channel):
        # gRPC channel — это соединение с gRPC-сервером.
        #
        # Через этот channel дальше создаются service stubs,
        # которые вызывают конкретные RPC-методы сервиса.
        self.channel = channel


def build_grpc_test_channel(logger: Logger, config: GRPCClientTestConfig) -> grpc.Channel:
    # Создаём insecure channel для тестового окружения.
    #
    # В демо-проекте сервис запускается локально без TLS,
    # поэтому используется grpc.insecure_channel.
    channel = grpc.insecure_channel(config.url)

    # Оборачиваем channel интерсептором.
    #
    # Так логирование RPC-вызовов подключается один раз на уровне канала,
    # а не размазывается по каждому методу каждого клиента.
    return grpc.intercept_channel(
        channel,
        GRPCLoggerInterceptor(logger=logger),
    )

Здесь важно два момента.

• Первый — GRPCTestClient ничего не знает про конкретный сервис. Это базовый класс, который просто хранит channel.
• Второй — логирование подключается через GRPCLoggerInterceptor. За счёт этого не нужно писать logger.info(...) внутри каждого метода клиента.

Интерсептор для логирования

Файл tests/clients/logger_interceptor.py содержит интерсептор для unary-unary RPC-вызовов.

from logging import Logger

from grpc import UnaryUnaryClientInterceptor


class GRPCLoggerInterceptor(UnaryUnaryClientInterceptor):
    def __init__(self, logger: Logger):
        # Логгер передаётся снаружи, чтобы разные клиенты
        # могли иметь разные имена логирования.
        self.logger = logger

    def intercept_unary_unary(self, continuation, client_call_details, request):
        # Логируем начало RPC-вызова.
        #
        # client_call_details.method содержит полный путь метода,
        # например:
        # /contracts.services.courses.CoursesService/CreateCourse
        self.logger.info(f"REQUEST: {client_call_details.method}")

        # Передаём выполнение дальше в настоящий gRPC-вызов.
        response = continuation(client_call_details, request)

        # Логируем завершение RPC-вызова.
        self.logger.info(f"RESPONSE: {client_call_details.method}")

        return response

В нашем сервисе все методы обычные unary-unary:

rpc GetCourse (GetCourseRequest) returns (GetCourseResponse);
rpc CreateCourse (CreateCourseRequest) returns (CreateCourseResponse);

То есть клиент отправляет один request и получает один response. Поэтому здесь используется UnaryUnaryClientInterceptor.

Интерсептор решает простую задачу: любой gRPC-вызов через этот channel автоматически попадает в логи. Клиентам не нужно думать о логировании.

Клиент для CoursesService

Теперь добавим клиент для конкретного сервиса CoursesService.

Файл tests/clients/courses/client.py

import uuid

import allure
import grpc

from contracts.services.courses.course_pb2 import CourseLevel
from contracts.services.courses.courses_service_pb2_grpc import CoursesServiceStub
from contracts.services.courses.rpc_create_course_pb2 import (
    CreateCourseRequest,
    CreateCourseResponse,
)
from contracts.services.courses.rpc_delete_course_pb2 import (
    DeleteCourseRequest,
    DeleteCourseResponse,
)
from contracts.services.courses.rpc_get_course_pb2 import (
    GetCourseRequest,
    GetCourseResponse,
)
from contracts.services.courses.rpc_get_courses_pb2 import (
    GetCoursesRequest,
    GetCoursesResponse,
)
from contracts.services.courses.rpc_update_course_pb2 import (
    UpdateCourseRequest,
    UpdateCourseResponse,
)
from tests.clients.client import GRPCTestClient, build_grpc_test_channel
from tests.settings import test_settings
from tests.tools.fakers import fake
from tests.tools.logger import get_test_logger


class CoursesGRPCTestClient(GRPCTestClient):
    def __init__(self, channel: grpc.Channel):
        super().__init__(channel)

        # Stub — это сгенерированный gRPC-клиент для CoursesService.
        #
        # Он создаётся из proto-контракта и содержит методы:
        # GetCourse, GetCourses, CreateCourse, UpdateCourse, DeleteCourse.
        self.stub = CoursesServiceStub(channel)

    @allure.step("Get course")
    def get_course_api(self, request: GetCourseRequest) -> GetCourseResponse:
        # Низкоуровневый метод клиента.
        #
        # Он принимает готовый protobuf request.
        # Это удобно для тестов, где важно явно контролировать тело запроса.
        return self.stub.GetCourse(request)

    @allure.step("Get courses")
    def get_courses_api(self, request: GetCoursesRequest) -> GetCoursesResponse:
        # Получение списка курсов.
        #
        # Request может быть пустым или содержать фильтры:
        # author и category.
        return self.stub.GetCourses(request)

    @allure.step("Create course")
    def create_course_api(self, request: CreateCourseRequest) -> CreateCourseResponse:
        # Создание курса через gRPC API.
        #
        # В Allure-отчёт попадёт шаг Create course
        # и protobuf-модель запроса.
        # Это удобно при разборе падений:
        # сразу видно, с какими данными был вызван RPC-метод.
        return self.stub.CreateCourse(request)

    @allure.step("Update course")
    def update_course_api(self, request: UpdateCourseRequest) -> UpdateCourseResponse:
        # Обновление курса через gRPC API.
        #
        # Метод принимает готовый UpdateCourseRequest,
        # чтобы тест мог явно задать id и новые значения полей.
        return self.stub.UpdateCourse(request)

    @allure.step("Delete course")
    def delete_course_api(self, request: DeleteCourseRequest) -> DeleteCourseResponse:
        # Удаление курса через gRPC API.
        #
        # Успешный ответ пустой, но сам RPC-вызов важен для сценария:
        # после удаления курс больше не должен находиться по id.
        return self.stub.DeleteCourse(request)

Методы с суффиксом _api — это прямые обёртки над RPC-методами. Они принимают уже собранный protobuf request и возвращают protobuf response.

Такой слой нужен, чтобы тест мог сам собрать request и потом сравнить его с response:

request = CreateCourseRequest(...)
response = courses_grpc_test_client.create_course_api(request)

assert_create_course_response(request, response)

Декораторы @allure.step(...) стоят именно на этих методах намеренно. В отчёте будет видно, какой RPC-вызов выполнялся и с каким request-объектом. Для gRPC это особенно полезно, потому что при падении теста нужно быстро понять, какие поля ушли в protobuf-запросе.

Высокоуровневые методы клиента

Ниже добавим методы, которые сами собирают request-объекты.

def get_course(self, course_id: uuid.UUID) -> GetCourseResponse:
        # Удобный метод для получения курса по UUID.
        #
        # Тесту не нужно каждый раз вручную создавать GetCourseRequest.
        request = GetCourseRequest(id=str(course_id))
        return self.get_course_api(request)

    def get_courses(
            self,
            author: str | None = None,
            category: str | None = None,
    ) -> GetCoursesResponse:
        # Удобный метод для получения списка курсов.
        #
        # author и category — фильтры из proto-контракта.
        # Если параметры не переданы, сервис вернёт список без фильтрации.
        request = GetCoursesRequest(
            author=author,
            category=category,
        )
        return self.get_courses_api(request)

    def create_course(self) -> CreateCourseResponse:
        # Создаёт курс со случайными валидными данными.
        #
        # Такой метод удобно использовать в фикстурах,
        # когда тесту нужен уже существующий курс.
        request = CreateCourseRequest(
            price=fake.price(),
            title=fake.title(),
            level=fake.proto_enum(CourseLevel),
            author=fake.author(),
            category=fake.category(),
            description=fake.description(),
        )
        return self.create_course_api(request)

    def update_course(self, course_id: uuid.UUID) -> UpdateCourseResponse:
        # Обновляет курс случайными валидными данными.
        #
        # course_id передаётся явно, потому что обновление всегда
        # относится к уже существующему курсу.
        request = UpdateCourseRequest(
            id=str(course_id),
            price=fake.price(),
            title=fake.title(),
            level=fake.proto_enum(CourseLevel),
            author=fake.author(),
            category=fake.category(),
            description=fake.description(),
        )
        return self.update_course_api(request)

    def delete_course(self, course_id: uuid.UUID) -> DeleteCourseResponse:
        # Удаляет курс по UUID.
        #
        # Тесту не нужно вручную создавать DeleteCourseRequest.
        request = DeleteCourseRequest(id=str(course_id))
        return self.delete_course_api(request)

Получается два уровня методов:

create_course_api(request)  # тест сам собирает request
create_course()             # клиент сам генерирует валидный request

Это удобно разделяет сценарии.

• Когда тест проверяет конкретные поля запроса и ответа, он использует _api-метод и сам создаёт request.
• Когда тесту просто нужно подготовить данные, он использует короткий метод вроде create_course().

Билдер клиента

В конце файла добавим функцию сборки клиента.

def build_courses_grpc_test_client() -> CoursesGRPCTestClient:
    # Собираем gRPC channel для CoursesService.
    #
    # Здесь подключаются:
    # - настройки подключения из tests/.env;
    # - тестовый логгер;
    # - gRPC-интерсептор для логирования.
    channel = build_grpc_test_channel(
        logger=get_test_logger("COURSES_GRPC_TEST_CLIENT"),
        config=test_settings.courses_grpc_client,
    )

    # Возвращаем готовый клиент для тестов и фикстур.
    return CoursesGRPCTestClient(channel=channel)

Билдер нужен, чтобы сборка клиента оставалась внутри клиентского слоя.

Фикстуры и тесты не должны знать:

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

Они должны получить готовый CoursesGRPCTestClient и использовать его методы.

Дальше в фикстуре это будет выглядеть коротко:

@pytest.fixture
def courses_grpc_test_client() -> CoursesGRPCTestClient:
    return build_courses_grpc_test_client()

Вся техническая сборка остаётся в tests/clients/courses/client.py, а тестовый код работает уже с готовым клиентом.

Фикстуры для gRPC-тестов

Теперь подготовим фикстуры. Они будут отвечать за три вещи:

• создание готового gRPC-клиента для тестов;
• подготовку тестового курса;
• очистку хранилища перед каждым тестом и после него.

Фикстуры вынесем в отдельный файл tests/fixtures/courses.py

import pytest

from contracts.services.courses.course_pb2 import Course
from tests.clients.courses.client import (
    CoursesGRPCTestClient,
    build_courses_grpc_test_client,
)
from tests.tools.database import clear_courses_db


@pytest.fixture
def courses_grpc_test_client() -> CoursesGRPCTestClient:
    # Возвращает готовый gRPC-клиент для CoursesService.
    #
    # Фикстуры и тесты не знают, как именно собирается клиент:
    # где берутся настройки, как создаётся channel,
    # какой interceptor подключается и какой stub используется.
    #
    # Всё это остаётся внутри build_courses_grpc_test_client().
    return build_courses_grpc_test_client()


@pytest.fixture
def course(courses_grpc_test_client: CoursesGRPCTestClient) -> Course:
    # Создаёт курс через публичный gRPC API.
    #
    # Эта фикстура нужна тестам, которым для проверки
    # требуется уже существующий курс.
    #
    # Например:
    # - получить курс по id;
    # - найти курс по автору;
    # - найти курс по категории;
    # - обновить курс;
    # - удалить курс.
    response = courses_grpc_test_client.create_course()

    # Возвращаем сам protobuf-объект Course,
    # чтобы тест мог использовать его id, author, category
    # и другие поля как ожидаемые данные.
    return response.course


@pytest.fixture(autouse=True)
def clear_courses_storage():
    # Очищаем хранилище курсов перед каждым тестом.
    #
    # Это защищает тесты от зависимости друг от друга:
    # данные, созданные в одном тесте, не попадут в следующий.
    clear_courses_db()

    yield

    # Очищаем хранилище ещё раз после теста.
    #
    # Это полезно, если тест упал посередине сценария
    # и оставил после себя созданные данные.
    clear_courses_db()

Зачем нужна фикстура courses_grpc_test_client

Фикстура courses_grpc_test_client отдаёт тестам готовый клиент:

@pytest.fixture
def courses_grpc_test_client() -> CoursesGRPCTestClient:
    return build_courses_grpc_test_client()

За счёт этого тесты не создают grpc.Channel, не подключают interceptor, не читают настройки и не создают CoursesServiceStub.

В тесте клиент будет использоваться напрямую:

def test_get_course(courses_grpc_test_client: CoursesGRPCTestClient):
    response = courses_grpc_test_client.get_course(...)

Сборка клиента остаётся в клиентском слое, а тест работает только с бизнес-действиями.

Зачем нужна фикстура course

Во многих тестах нужен уже существующий курс. Например, чтобы проверить получение курса по id, обновление или удаление.

Можно было бы в каждом тесте писать:

create_response = courses_grpc_test_client.create_course()
course = create_response.course

Но это быстро начнёт повторяться. Поэтому выносим подготовку курса в фикстуру:

@pytest.fixture
def course(courses_grpc_test_client: CoursesGRPCTestClient) -> Course:
    response = courses_grpc_test_client.create_course()
    return response.course

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

Зачем нужна autouse-фикстура очистки

Фикстура clear_courses_storage помечена как autouse=True:

@pytest.fixture(autouse=True)
def clear_courses_storage():
    clear_courses_db()
    yield
    clear_courses_db()

Это значит, что pytest будет автоматически применять её к каждому тесту. Подключать её вручную в аргументы тестовой функции не нужно.

Она делает две очистки:

до теста     → тест получает чистое состояние
после теста  → данные теста не протекают дальше

Прямой вызов clear_courses_db() здесь используется только для управления состоянием тестового окружения. Проверки всё равно выполняются через gRPC API.

Подключение фикстур через pytest plugins

Теперь нужно сделать так, чтобы pytest увидел фикстуры из tests/fixtures/courses.py.

Для этого в корневом tests/conftest.py подключим файл с фикстурами как pytest-плагин.

pytest_plugins = (
    # Подключаем фикстуры для тестов CoursesService.
    #
    # После этого фикстуры courses_grpc_test_client,
    # course и clear_courses_storage будут доступны в тестах
    # без дополнительных импортов.
    "tests.fixtures.courses",
)

Такой подход удобнее, чем складывать все фикстуры в один большой conftest.py.

В маленьком проекте один conftest.py ещё выглядит нормально. Но когда появляются десятки клиентов, сотни фикстур и тысячи тестов, файл быстро превращается в свалку: фикстуры для разных сервисов лежат рядом, зависимости смешиваются, искать нужный setup становится сложнее.

Лучше сразу разносить фикстуры по доменам:

tests/fixtures/
├── courses.py
├── users.py
├── auth.py
├── payments.py
└── documents.py

А в tests/conftest.py оставить только регистрацию плагинов:

pytest_plugins = (
    "tests.fixtures.courses",
    "tests.fixtures.users",
    "tests.fixtures.auth",
    "tests.fixtures.payments",
    "tests.fixtures.documents",
)

В итоге conftest.py остаётся коротким, а фикстуры лежат рядом с той областью, к которой относятся. Это проще поддерживать и масштабировать.

Проверки ответов

Теперь вынесем проверки в отдельный слой tests/assertions/.

Тесты не должны быть забиты десятками assert response.course.title == request.title. Такой код быстро разрастается, плохо читается и даёт слабую диагностику при падении.

Поэтому сделаем два файла:

tests/assertions/
├── base.py     # базовые проверки
└── courses.py  # проверки для CoursesService

Базовые проверки

Файл tests/assertions/base.py содержит универсальные assertion-хелперы.

from typing import Any, Sized

import allure
import grpc

from tests.tools.logger import get_test_logger


logger = get_test_logger("BASE_ASSERTIONS")


@allure.step("Check that {name} equals to {expected}")
def assert_equal(actual: Any, expected: Any, name: str):
    # Проверяет равенство двух значений.
    #
    # Используется для полей protobuf-ответов:
    # title, description, author, category, price, level и других.
    #
    # name нужен для понятной ошибки:
    # сразу видно, какое именно поле не совпало.
    logger.info(f'Check that "{name}" equals to {expected}')

    assert actual == expected, (
        f'Incorrect value: "{name}". '
        f'Expected value: {expected}. '
        f'Actual value: {actual}'
    )


@allure.step("Check that length of {name} equals to {expected_length}")
def assert_length(actual: Sized, expected_length: int, name: str):
    # Проверяет длину коллекции.
    #
    # В этом проекте используется для списка courses
    # в GetCoursesResponse.
    logger.info(f'Check that length of "{name}" equals to {expected_length}')

    assert len(actual) == expected_length, (
        f'Incorrect length: "{name}". '
        f'Expected length: {expected_length}. '
        f'Actual length: {len(actual)}'
    )


@allure.step("Check that {name} is not empty")
def assert_not_empty(actual: Any, name: str):
    # Проверяет, что значение заполнено.
    #
    # Используется для серверных полей,
    # которые не приходят из request, а создаются сервисом.
    #
    # Например:
    # - id курса;
    # - created_at.
    logger.info(f'Check that "{name}" is not empty')

    assert actual, (
        f'Incorrect value: "{name}". '
        f'Expected non-empty value but got: {actual}'
    )


@allure.step("Check gRPC status code equals to {expected}")
def assert_grpc_status(error: grpc.RpcError, expected: grpc.StatusCode):
    # Проверяет gRPC status code у ошибки.
    #
    # В негативных тестах важно проверять не просто факт ошибки,
    # а конкретный статус: NOT_FOUND, INVALID_ARGUMENT и т.д.
    logger.info(f"Check gRPC status code equals to {expected}")

    assert error.code() == expected, (
        f"Incorrect gRPC status code. "
        f"Expected: {expected}. "
        f"Actual: {error.code()}"
    )

Здесь у каждой проверки есть:

• @allure.step(...);
• запись в лог;
• понятное сообщение об ошибке.

Это нужно для диагностики. Если тест упал, в Allure будет видно не просто AssertionError, а конкретный шаг: какое поле проверялось, какое значение ожидалось и что пришло фактически.

Проверки для CoursesService

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

Файл tests/assertions/courses.py

import allure

from contracts.services.courses.course_pb2 import Course
from contracts.services.courses.rpc_create_course_pb2 import (
    CreateCourseRequest,
    CreateCourseResponse,
)
from contracts.services.courses.rpc_get_courses_pb2 import GetCoursesResponse
from contracts.services.courses.rpc_update_course_pb2 import (
    UpdateCourseRequest,
    UpdateCourseResponse,
)
from tests.assertions.base import assert_equal, assert_length, assert_not_empty
from tests.tools.logger import get_test_logger


logger = get_test_logger("COURSES_ASSERTIONS")


@allure.step("Check course")
def assert_course(actual: Course, expected: Course):
    # Проверяет, что два protobuf-объекта Course совпадают по полям.
    #
    # Используется, когда нужно сравнить курс,
    # полученный из API, с заранее ожидаемым курсом.
    #
    # Например:
    # - создали курс;
    # - получили его по id;
    # - сравнили полученный Course с исходным Course.
    logger.info("Check course")

    assert_equal(actual.id, expected.id, "id")
    assert_equal(actual.title, expected.title, "title")
    assert_equal(actual.description, expected.description, "description")
    assert_equal(actual.author, expected.author, "author")
    assert_equal(actual.category, expected.category, "category")
    assert_equal(actual.price, expected.price, "price")
    assert_equal(actual.level, expected.level, "level")
    assert_equal(actual.created_at, expected.created_at, "created_at")


@allure.step("Check create course response")
def assert_create_course_response(
        request: CreateCourseRequest,
        response: CreateCourseResponse,
):
    # Проверяет ответ на создание курса.
    #
    # Часть полей должна совпасть с request:
    # title, description, author, category, price, level.
    #
    # Часть полей создаёт сам сервис:
    # id и created_at.
    # Их не сравниваем с request, а проверяем, что они заполнены.
    logger.info("Check create course response")

    assert_not_empty(response.course.id, "id")
    assert_not_empty(response.course.created_at, "created_at")

    assert_equal(response.course.price, request.price, "price")
    assert_equal(response.course.title, request.title, "title")
    assert_equal(response.course.level, request.level, "level")
    assert_equal(response.course.author, request.author, "author")
    assert_equal(response.course.category, request.category, "category")
    assert_equal(response.course.description, request.description, "description")


@allure.step("Check update course response")
def assert_update_course_response(
        request: UpdateCourseRequest,
        response: UpdateCourseResponse,
):
    # Проверяет ответ на обновление курса.
    #
    # После UpdateCourse сервис должен вернуть Course
    # с тем же id и новыми значениями полей из request.
    logger.info("Check update course response")

    assert_equal(response.course.id, request.id, "id")
    assert_equal(response.course.title, request.title, "title")
    assert_equal(response.course.description, request.description, "description")
    assert_equal(response.course.author, request.author, "author")
    assert_equal(response.course.category, request.category, "category")
    assert_equal(response.course.price, request.price, "price")
    assert_equal(response.course.level, request.level, "level")


@allure.step("Check get courses response")
def assert_get_courses_response(
        response: GetCoursesResponse,
        expected_courses: list[Course],
):
    # Проверяет ответ на получение списка курсов.
    #
    # Сначала проверяем количество курсов.
    # Потом сравниваем каждый Course по полям.
    #
    # Это удобно для тестов фильтрации:
    # например, получить курсы по author или category
    # и проверить, что в ответе только ожидаемые записи.
    logger.info("Check get courses response")

    assert_length(response.courses, len(expected_courses), "courses")

    for index, expected_course in enumerate(expected_courses):
        assert_course(response.courses[index], expected_course)

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

Так тесты остаются короткими и читаемыми.

Вместо такого кода:

assert response.course.title == request.title
assert response.course.description == request.description
assert response.course.author == request.author
assert response.course.category == request.category
assert response.course.price == request.price
assert response.course.level == request.level

в тесте будет одна строка:

assert_create_course_response(request, response)

При этом диагностика не теряется. Наоборот, она становится понятнее:

• в Allure видно отдельные шаги проверок;
• в логах видно, какое поле проверялось;
• в ошибке видно expected и actual;
• одинаковые проверки не копируются между тестами.

Для gRPC это особенно удобно, потому что ответы — это protobuf-объекты с полями, enum-ами и вложенными структурами. Лучше один раз описать проверку контракта в tests/assertions/, чем размазывать одинаковые assert по всем тестовым сценариям.

Настройка pytest

Перед запуском тестов добавим pytest.ini в корень проекта.

Файл нужен не только для маркеров и параметров запуска. В этом проекте через него также настраивается pythonpath, чтобы тесты могли импортировать:

• модули проекта из корня репозитория;
• сгенерированные protobuf-модули из protos/gen.

Файл pytest.ini

[pytest]

# Базовые параметры запуска pytest.
#
# -s отключает перехват stdout/stderr, поэтому в консоли будут видны логи тестового клиента.
# -v включает подробный вывод: pytest покажет каждый тест отдельной строкой.
addopts = -s -v

# Пути, которые pytest добавит в Python import path перед запуском тестов.
#
# . нужен, чтобы тесты могли импортировать модули проекта:
# app, tests, config и другие пакеты из корня репозитория.
#
# protos/gen нужен, чтобы резолвились сгенерированные protobuf-модули:
# contracts.services.courses.course_pb2
# contracts.services.courses.courses_service_pb2_grpc
# contracts.services.courses.rpc_create_course_pb2
# и остальные файлы, созданные из .proto-контрактов.
pythonpath = . protos/gen

# Шаблон имён файлов с тестами.
#
# pytest будет искать:
# - файлы вида test_*.py
# - файлы вида *_tests.py
python_files = *_tests.py test_*.py

# Шаблон имён тестовых классов.
#
# Например:
# class TestCourses:
python_classes = Test*

# Шаблон имён тестовых функций и методов.
#
# Например:
# def test_create_course(...):
python_functions = test_*

# Описание пользовательских маркеров.
#
# Это убирает предупреждения PytestUnknownMarkWarning
# и документирует назначение маркеров в проекте.
markers =
    courses: Маркировка для тестов сервиса курсов.
    regression: Маркировка для регрессионных тестов.

Зачем здесь нужен pythonpath

После генерации proto-файлов Python-код лежит в директории: protos/gen/

В тестах мы импортируем сгенерированные классы так:

from contracts.services.courses.course_pb2 import CourseLevel, Course
from contracts.services.courses.courses_service_pb2_grpc import CoursesServiceStub
from contracts.services.courses.rpc_create_course_pb2 import CreateCourseRequest

Но пакет contracts физически находится не в корне проекта, а внутри protos/gen.

То есть структура такая:

protos/
└── gen/
    └── contracts/
        └── services/
            └── courses/
                ├── course_pb2.py
                ├── courses_service_pb2_grpc.py
                ├── rpc_create_course_pb2.py
                └── ...

Без настройки pythonpath = protos/gen Python не найдёт пакет contracts.

В таком случае при запуске тестов можно получить ошибку:

ModuleNotFoundError: No module named 'contracts'

Поэтому в pytest.ini добавляем:

pythonpath = . protos/gen

Точка . нужна для импортов из корня проекта:

from tests.clients.courses.client import CoursesGRPCTestClient
from tests.tools.fakers import fake
from app.repository import CoursesRepository

А protos/gen нужен для импортов generated-кода:

from contracts.services.courses.course_pb2 import Course

Альтернатива через переменную окружения

То же самое можно сделать через переменную окружения:

PYTHONPATH=./:./protos/gen pytest tests/ -m regression

Но для учебного проекта удобнее зафиксировать это в pytest.ini.

Тогда команда запуска остаётся простой:

pytest tests/ -m regression

И локально, и в CI pytest сам добавит нужные пути для импортов.

Автотесты для CoursesService

Теперь соберём сами тесты.

Файл tests/suites/test_courses.py. В этом файле проверим основные сценарии CoursesService:

• создание курса;
• получение курса по id;
• получение списка курсов с фильтром по автору;
• получение списка курсов с фильтром по категории;
• обновление курса;
• удаление курса;
• получение несуществующего курса.

import uuid

import allure
import grpc
import pytest

from contracts.services.courses.course_pb2 import CourseLevel, Course
from contracts.services.courses.rpc_create_course_pb2 import CreateCourseRequest
from contracts.services.courses.rpc_update_course_pb2 import UpdateCourseRequest
from tests.assertions.base import assert_grpc_status
from tests.assertions.courses import (
    assert_course,
    assert_get_courses_response,
    assert_create_course_response,
    assert_update_course_response,
)
from tests.clients.courses.client import CoursesGRPCTestClient
from tests.tools.fakers import fake


@pytest.mark.courses
@pytest.mark.regression
class TestCourses:
    # Маркируем весь класс как courses и regression.
    #
    # courses — тесты относятся к CoursesService.
    # regression — тесты входят в основной регрессионный набор.

    @allure.title("[gRPC] Create course")
    def test_create_course(self, courses_grpc_test_client: CoursesGRPCTestClient):
        # Собираем request явно внутри теста.
        #
        # Здесь важно контролировать входные данные,
        # потому что дальше response будет сравниваться именно с этим request.
        request = CreateCourseRequest(
            price=fake.price(),
            title=fake.title(),
            level=fake.proto_enum(CourseLevel),
            author=fake.author(),
            category=fake.category(),
            description=fake.description(),
        )

        # Отправляем CreateCourseRequest в gRPC API.
        response = courses_grpc_test_client.create_course_api(request)

        # Проверяем, что сервис вернул созданный курс:
        # - серверные поля id и created_at заполнены;
        # - остальные поля совпадают с request.
        assert_create_course_response(request, response)

    @allure.title("[gRPC] Get course")
    def test_get_course(
            self,
            course: Course,
            courses_grpc_test_client: CoursesGRPCTestClient,
    ):
        # Фикстура course заранее создаёт курс через gRPC API.
        #
        # В тесте используем его id, чтобы запросить этот же курс
        # через GetCourse.
        response = courses_grpc_test_client.get_course(uuid.UUID(course.id))

        # Проверяем, что полученный курс полностью совпадает
        # с курсом, созданным в фикстуре.
        assert_course(response.course, course)

    @allure.title("[gRPC] Get courses by author")
    def test_get_courses_by_author(
            self,
            course: Course,
            courses_grpc_test_client: CoursesGRPCTestClient,
    ):
        # Созданный курс содержит author.
        #
        # Используем его как фильтр для GetCourses.
        response = courses_grpc_test_client.get_courses(author=course.author)

        # Проверяем, что в ответе вернулся список
        # с ожидаемым курсом.
        assert_get_courses_response(response, [course])

    @allure.title("[gRPC] Get courses by category")
    def test_get_courses_by_category(
            self,
            course: Course,
            courses_grpc_test_client: CoursesGRPCTestClient,
    ):
        # Созданный курс содержит category.
        #
        # Используем её как фильтр для GetCourses.
        response = courses_grpc_test_client.get_courses(category=course.category)

        # Проверяем, что фильтрация по категории вернула
        # ожидаемый курс.
        assert_get_courses_response(response, [course])

    @allure.title("[gRPC] Update course")
    def test_update_course(
            self,
            course: Course,
            courses_grpc_test_client: CoursesGRPCTestClient,
    ):
        # Фикстура course создаёт курс, который будем обновлять.
        #
        # В UpdateCourseRequest передаём существующий id
        # и новый набор значений для полей курса.
        update_request = UpdateCourseRequest(
            id=course.id,
            price=fake.price(),
            title=fake.title(),
            level=fake.proto_enum(CourseLevel),
            author=fake.author(),
            category=fake.category(),
            description=fake.description(),
        )

        # Отправляем запрос на обновление курса.
        update_response = courses_grpc_test_client.update_course_api(update_request)

        # Проверяем, что ответ на UpdateCourse содержит
        # новые значения из update_request.
        assert_update_course_response(update_request, update_response)

        # Дополнительно запрашиваем курс по id.
        #
        # Это проверяет, что обновление не просто вернуло корректный response,
        # а действительно изменило состояние сервиса.
        get_course_response = courses_grpc_test_client.get_course(uuid.UUID(course.id))

        # Проверяем, что после повторного чтения курс совпадает
        # с результатом обновления.
        assert_course(get_course_response.course, update_response.course)

    @allure.title("[gRPC] Delete course")
    def test_delete_course(
            self,
            course: Course,
            courses_grpc_test_client: CoursesGRPCTestClient,
    ):
        # Фикстура course создаёт курс, который будем удалять.
        courses_grpc_test_client.delete_course(uuid.UUID(course.id))

        # После удаления курс больше не должен находиться по id.
        #
        # gRPC-клиент в таком случае выбросит grpc.RpcError.
        with pytest.raises(grpc.RpcError) as error:
            courses_grpc_test_client.get_course(uuid.UUID(course.id))

        # Проверяем конкретный gRPC status code.
        #
        # Для удалённого курса ожидаем NOT_FOUND.
        assert_grpc_status(error.value, grpc.StatusCode.NOT_FOUND)

    @allure.title("[gRPC] Get course. Not found")
    def test_get_course_not_found(self, courses_grpc_test_client):
        # Генерируем случайный UUID.
        #
        # Курс с таким id не создавался, поэтому сервис должен вернуть ошибку.
        with pytest.raises(grpc.RpcError) as error:
            courses_grpc_test_client.get_course(fake.uuid())

        # Проверяем, что сервис вернул именно NOT_FOUND,
        # а не любую другую gRPC-ошибку.
        assert_grpc_status(error.value, grpc.StatusCode.NOT_FOUND)

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

Тесты работают только через CoursesGRPCTestClient. В них нет создания grpc.Channel, CoursesServiceStub, ручного подключения интерсепторов или чтения .env. Всё это уже собрано в клиентском слое.

Позитивные сценарии проверяют не только факт успешного вызова RPC-метода, но и содержимое protobuf-ответа:

assert_create_course_response(request, response)
assert_course(response.course, course)
assert_update_course_response(update_request, update_response)
assert_get_courses_response(response, [course])

Негативные сценарии проверяют grpc.RpcError и конкретный gRPC status code:

with pytest.raises(grpc.RpcError) as error:
    courses_grpc_test_client.get_course(fake.uuid())

assert_grpc_status(error.value, grpc.StatusCode.NOT_FOUND)

Это важнее, чем просто проверить, что «какая-то ошибка произошла». Для клиента сервиса имеет значение конкретный статус: NOT_FOUND, INVALID_ARGUMENT, ALREADY_EXISTS, UNAUTHENTICATED и так далее.

Почему request иногда создаётся прямо в тесте

В test_create_course и test_update_course request собирается внутри теста:

request = CreateCourseRequest(...)

Так проще проверить связь между входными данными и ответом сервиса.

Например, в тесте создания курса нужно убедиться, что сервис вернул именно те значения, которые были переданы в CreateCourseRequest:

assert_create_course_response(request, response)

А в фикстурах или подготовительных шагах можно использовать короткие методы клиента:

course = courses_grpc_test_client.create_course().course

Так в проекте есть два режима работы:

create_course_api(request)  # явный request для точной проверки
create_course()             # быстрый способ подготовить валидный курс

Запуск тестов

Запустить весь regression-набор можно так:

pytest tests/ -m regression -v

Запустить только тесты CoursesService:

pytest tests/ -m courses -v

С Allure-результатами:

pytest tests/ -m regression --alluredir=allure-results
allure serve allure-results

В Allure-отчёте будут видны:

• названия тестов из @allure.title;
• шаги gRPC-клиента из @allure.step;
• шаги assertion-хелперов;
• данные protobuf request-объектов;
• место падения и конкретное несовпавшее поле.

Такой набор тестов уже покрывает базовый контракт CoursesService: успешные CRUD-сценарии, фильтрацию списка и ошибку получения несуществующего курса.

Запуск на CI/CD

Последний слой — запуск этих же тестов в CI. Здесь идея такая же, как и при локальном запуске: поднять приложение, дождаться, что оно готово принимать запросы, запустить pytest и сохранить Allure-результаты.

Важно, что в CI не появляется какой-то отдельный тестовый режим. Тесты остаются теми же самыми. Отличается только окружение: локально сервер запускается вручную в терминале, а в GitHub Actions он поднимается отдельным шагом workflow.

Файл .github/workflows/tests.yml

name: API gRPC Tests

on:
  # Запускаем workflow при push в main.
  push:
    branches:
      - main

  # И при pull request в main.
  pull_request:
    branches:
      - main

jobs:
  run-tests:
    runs-on: ubuntu-latest

    env:
      # Добавляем корень проекта и директорию со сгенерированными protobuf-модулями
      # в PYTHONPATH.
      #
      # . нужен для импортов app, tests и других модулей из корня проекта.
      #
      # ./protos/gen нужен для импортов сгенерированных gRPC-контрактов:
      # contracts.services.courses.course_pb2
      # contracts.services.courses.courses_service_pb2_grpc
      # contracts.services.courses.rpc_create_course_pb2
      PYTHONPATH: ./:./protos/gen

    steps:
      - name: Check out repository
        # Забираем код репозитория в runner.
        uses: actions/checkout@v6

      - name: Set up Python
        # Устанавливаем нужную версию Python.
        #
        # В проекте используется Python 3.12.
        uses: actions/setup-python@v6
        with:
          python-version: '3.12'

      - name: Install dependencies
        # Устанавливаем зависимости приложения и тестов.
        #
        # В этом проекте один requirements.txt используется
        # и для gRPC-сервиса, и для тестового слоя.
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Start server
        # Запускаем gRPC-сервис в фоне.
        #
        # Важно: в конце команды стоит &.
        # Без него шаг зависнет на запущенном сервере,
        # и GitHub Actions не перейдёт к запуску pytest.
        #
        # sleep 3 даёт сервису время подняться и начать слушать порт.
        # Для демо-проекта этого достаточно.
        # В реальном проекте лучше заменить sleep на health-check
        # или проверку доступности gRPC-сервиса.
        run: |
          python -m app.server &
          sleep 3

      - name: Run tests
        # Запускаем регрессионные gRPC API-тесты.
        #
        # Маркер regression описан в pytest.ini.
        #
        # --alluredir=allure-results сохраняет сырые результаты Allure,
        # из которых дальше будет собираться HTML-отчёт.
        run: |
          pytest -m regression --alluredir=allure-results

      - name: Upload Allure results
        # Загружаем allure-results как artifact.
        #
        # if: always() нужен, чтобы результаты сохранились даже тогда,
        # когда тесты завершились с ошибкой.
        #
        # Это важно для диагностики:
        # можно открыть Allure-результаты и посмотреть,
        # на каком gRPC-вызове или assertion-шаге произошёл сбой.
        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 allure-results,
        # который был загружен в job run-tests.
        uses: actions/download-artifact@v8
        with:
          name: allure-results
          path: allure-results

      - name: Build Allure report
        # Собираем HTML-отчёт из allure-results.
        #
        # allure_results — директория с сырыми результатами тестов.
        # allure_history — директория, куда будет собран отчёт
        # с сохранением истории запусков.
        #
        # История нужна, чтобы в Allure были тренды,
        # предыдущие результаты и статистика по тестам.
        uses: simple-elf/[email protected]
        if: always()
        with:
          allure_results: allure-results
          allure_history: allure-history

      - name: Deploy Allure report
        # Публикуем собранный Allure-отчёт в ветку gh-pages.
        #
        # После настройки GitHub Pages отчёт будет доступен
        # как статическая HTML-страница.
        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 gRPC Tests

permissions:
  contents: write

on:
  push:
    branches:
      - main

В-третьих, нужно настроить GitHub Pages: указать, из какой ветки и какой директории публиковать сайт. В этом варианте отчёт публикуется в ветку gh-pages, поэтому в настройках Pages нужно выбрать публикацию из этой ветки.

Обычно логика такая:

• Settings → Pages
• Source: Deploy from a branch
• Branch: gh-pages
• Folder: /root

После этого GitHub выдаст URL, по которому будет доступен Allure-отчёт.

Заключение

Все ссылки на код, отчеты и запуски тестов в CI/CD можно найти на моем GitHub:

• Проект с API gRPC автотестами
• Allure-отчет с результатами API gRPC автотестов
• CI/CD запуски API gRPC автотестов