¶ Введение. Проектирование по контракту
В распределенной архитектуре с тысячами взаимодействующих друг с другом микросервисов обеспечение надежности и корректности этих взаимодействий становится первостепенной задачей. Сквозное тестирование (E2E) является медленным и дорогостоящим. Основой для построения надежной стратегии тестирования в таких системах является четкое и формализованное соглашение о том, как сервисы должны общаться — API контракт.
Термин "Контрактное программирование" (Design by Contract, DbC) введен Bertrand Meyer в контексте работы над языком программирования Eiffel. Предлагается рассматривать взаимодействие между программными компонентами (вызывающим кодом и методом класса) как формальный договор, имеющий обязательства и преимущества для обеих сторон. Такой подход предписывает разработчикам определять формальные, точные и проверяемые спецификации интерфейсов для компонентов ПО.
Составляющие контракта:
Предусловия (Preconditions)
Условия, которые клиент (вызывающая сторона) обязан гарантировать перед вызовом операции. Для поставщика (вызываемого метода) это является преимуществом, так как избавляет его от необходимости обрабатывать случаи, выходящие за рамки предусловий.
Нарушение предусловия указывает на ошибку в коде клиента
Постусловия (Postconditions)
Свойства, которые поставщик обязан гарантировать по завершении своей работы. Для клиента это является преимуществом — основной выгодой от вызова метода.
Нарушение постусловия указывает на ошибку в коде поставщика
Инварианты (Invariants)
Условия, которые должны сохраняться для компонента (например, класса) до и после выполнения любой его операции. Инварианты определяют стабильное состояние объекта и гарантируют, что его целостность не будет нарушена.
Пример для банковского счета:
- баланс никогда не должен быть отрицательным
- список всех транзакций должен быть всегда доступен (не
null)
Принципы
DbCбыли естественным образом адаптированы: "программный компонент" стал "микросервисом", а "интерфейс метода" — "конечной точкойAPI(endpoint)"
API и есть контракт между поставщиком программного обеспечения и его потребителем, описывающий, что система будет делать. Аналогии:
- предусловия соответствуют требованиям к входящему
API-запросу: обязательные параметры, заголовки, структура тела запроса - постусловия соответствуют гарантиям, предоставляемым
API-ответом: корректныйHTTP-статус код, структура тела ответа, определенные заголовки
Инварианты можно соотнести с ожидаемым состоянием ресурса или сервиса после выполнения операций.
Современный подход к разработке (
API-Firstилиcontract-first), является прямой реализацией принциповDbCв контекстеAPI
Суть в том, что проектирование и согласование API-контракта происходит до написания какого-либо кода реализации.
Это Code-First:
Это API-First:
Преимущества:
- команды фронтенда и бэкенда могут работать одновременно (фронтенд-команда может использовать mock-объекты, сгенерированные на основе контракта, не дожидаясь полной реализации бэкенда)
- контракт служит единым источником правды для всех участников процесса разработки
- большинство проблем проектирования могут быть выявлены и решены на этапе согласования контракта (снижение стоимости исправлений)
- наличие формального, машиночитаемого контракта позволяет автоматически генерировать документацию, клиентский и серверный код, а также тестовые сценарии
Часто
API-контракт становится не просто описанием, а центральным артефактом разработки
¶ Формализация контракта
¶ JSON Schema
Стандарт, предоставляющий словарь для аннотирования и валидации JSON-документов. Позволяет описать ожидаемую структуру JSON-объекта, включая типы данных полей, обязательные поля, форматы строк и многое другое.
Ключевые слова валидации включают:
type: тип данных ("object", "array", "string", "number")properties: поля объекта и их схемыrequired: список обязательных полей в объектеitems: схема для элементов массиваminItems / maxItems: минимальное / максимальное количество элементов в массивеadditionalProperties: разрешены ли в объекте поля, не описанные в properties
Пример:
pip install jsonschema
import pytest
from jsonschema import validate, ValidationError
# определение схемы
user_schema = {
"type": "object",
"properties": {
"id": {"type": "integer"},
"email": {"type": "string", "format": "email"},
"roles": {
"type": "array",
"items": {"type": "string"},
"minItems": 1
}
},
"required": ["id", "email", "roles"],
"additionalProperties": False
}
def test_valid_user_payload():
# проверка валидности
valid_user = {
"id": 101,
"email": "test@pmifi.ru",
"roles": ["admin", "editor"]
}
validate(instance=valid_user, schema=user_schema)
def test_invalid_user_payload_missing_required_field():
# здесь обязательное полк email отсутствует
invalid_user = {
"id": 102,
"roles": ["viewer"]
}
with pytest.raises(ValidationError) as exc_info:
validate(instance=invalid_user, schema=user_schema)
assert "'email' is a required property" in str(exc_info.value)
def test_invalid_user_payload_wrong_type():
# не тот тип
invalid_user = {
"id": "103",
"email": "user@pmifi.ru",
"roles": ["viewer"]
}
with pytest.raises(ValidationError) as exc_info:
validate(instance=invalid_user, schema=user_schema)
assert "'103' is not of type 'integer'" in str(exc_info.value)
def test_invalid_user_payload_additional_property():
# лишнее поле
invalid_user = {
"id": 104,
"email": "another@pmifi.ru",
"roles": ["viewer"],
"extra_field": "not allowed"
}
with pytest.raises(ValidationError) as exc_info:
validate(instance=invalid_user, schema=user_schema)
assert "Additional properties are not allowed ('extra_field' was unexpected)" in str(exc_info.value)
Тестирование схемы является первым уровнем контрактного тестирования
¶ OpenAPI Specification
Более широкий стандарт (ранее назывался Swagger) для описания REST API (YAML, JSON): конечные точки (paths), доступные на них операции (GET, POST и т.д.), параметры запросов, форматы ответов, схемы безопасности и другое.
Для описания структур данных (тел запросов и ответов)
OpenAPIиспользуетJSON Schema
Получить Swagger UI по спецификации
openapi: 3.0.3
info:
title: User API
version: 1.0.0
paths:
/users/{userId}:
get:
summary: Get user by ID
responses:
'200':
description: Successful response
content:
application/json:
schema:
$ref: '#/components/schemas/User'
components:
schemas:
User:
type: object
properties:
id:
type: integer
format: int64
username:
type: string
email:
type: string
format: email
required:
- id
- username
- email
pip install openapi-schema-validator pyyaml
Пример теста:
import yaml
from openapi_schema_validator import validate
def load_openapi_schema(component_name: str):
with open('openapi.yaml', 'r') as f:
spec = yaml.safe_load(f)
return spec['components']['schemas'][component_name]
def test_valid_api_response():
user_schema = load_openapi_schema('User')
api_response_body = {
"id": 1,
"username": "testuser",
"email": "testuser@pmifi.ru"
}
# валидация тела ответа
validate(instance=api_response_body, schema=user_schema)
def test_invalid_api_response_missing_field():
# отсутствует обязательное поле
user_schema = load_openapi_schema('User')
api_response_body = {
"id": 2,
"username": "anotheruser"
}
try:
validate(instance=api_response_body, schema=user_schema)
except Exception as e:
assert "'email' is a required property" in str(e)
¶ Тестирование поведения
После того как убедились в структурной корректности данных с помощью схем следующим шагом является проверка поведения сервисов.
¶ REST API
Тестирование REST API (HTTP + JSON) включает проверку нескольких ключевых аспектов:
- корректно ли API обрабатывает запросы к определенным эндпоинтам (/users/1) и поддерживает ли он правильные HTTP-методы (GET, POST, PUT, DELETE)
- возвращает ли API ожидаемые HTTP-статус-коды в различных ситуациях (200 OK, 201 Created, 404 Not Found, 400 Bad Request)
- присутствуют ли в ответе необходимые заголовки (Content-Type: application/json)
- соответствует ли тело ответа ожидаемой структуре и содержанию (валидация схемы)
При тестировании высокопроизводительных веб-сервисов, построенных на асинхронных фреймворках (
FastAPI), важно использовать асинхронныеHTTP-клиенты (httpx,APIсовместим сresuests)
Библиотека pytest-httpx предоставляет механизм мокирования HTTP-запросов прямо в тестах pytest (мок-сервер).
pip install pytest pytest-asyncio httpx pytest-httpx
Клиентский код:
import httpx
class UserClient:
def __init__(self, base_url: str):
self.base_url = base_url
async def get_user(self, user_id: int) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(f"{self.base_url}/users/{user_id}")
response.raise_for_status()
return response.json()
async def create_user(self, username: str, email: str) -> dict:
payload = {"username": username, "email": email}
async with httpx.AsyncClient() as client:
response = await client.post(f"{self.base_url}/users", json=payload)
response.raise_for_status()
return response.json()
Тест:
import pytest
from pytest_httpx import HTTPXMock
from user_client import UserClient
BASE_URL = "https://api.pmifi.ru"
@pytest.mark.asyncio
async def test_get_user_success(httpx_mock: HTTPXMock):
# все успешно
user_id = 123
expected_response_data = {"id": user_id, "username": "testuser", "email": "test@pmifi.ru"}
# настройка мок-сервера
httpx_mock.add_response(
method="GET",
url=f"{BASE_URL}/users/{user_id}",
json=expected_response_data,
status_code=200
)
client = UserClient(base_url=BASE_URL)
user_data = await client.get_user(user_id)
# проверка
assert user_data == expected_response_data
@pytest.mark.asyncio
async def test_create_user_success(httpx_mock: HTTPXMock):
request_payload = {"username": "newuser", "email": "new@pmifi.ru"}
expected_response_data = {"id": 456, **request_payload}
httpx_mock.add_response(
method="POST",
url=f"{BASE_URL}/users",
match_json=request_payload,
json=expected_response_data,
status_code=201
)
client = UserClient(base_url=BASE_URL)
created_user = await client.create_user(username="newuser", email="new@example.com")
assert created_user == expected_response_data
@pytest.mark.asyncio
async def test_get_user_not_found(httpx_mock: HTTPXMock):
# не найден
user_id = 999
httpx_mock.add_response(
method="GET",
url=f"{BASE_URL}/users/{user_id}",
status_code=404
)
client = UserClient(base_url=BASE_URL)
with pytest.raises(httpx.HTTPStatusError):
await client.get_user(user_id)
¶ gRPC-сервисы
Другой подход к построению API с рядом ключевых отличий от REST, которые напрямую влияют на стратегию тестирования.
| Характеристика | REST | gRPC |
|---|---|---|
| Протокол | HTTP/1.1 (чаще всего), HTTP/2 | HTTP/2 |
| Формат данных | JSON (типично), XML, текст и др. | Protocol Buffers (Protobuf) |
| Контракт | OpenAPI Specification (опционально) | Файл .proto (обязательно) |
| Коммуникация | Запрос-Ответ | Унарный, сервер-стриминг, клиент-стриминг, двунаправленный-стриминг |
| Генерация кода | Опционально (из OpenAPI) | Ключевая особенность (генерация заглушек, клиентов, серверов) |
| Производительность | Текстовый, обычно медленнее | Бинарный, компактный, обычно быстрее |
gRPC требует контрактизацию через .proto файл
Два подхода:
Тестирование с "in-process" сервером
Заключается в запуске реального, но легковесного gRPC-сервера в памяти для каждого теста или набора тестов. Клиент в тесте подключается к этому локальному серверу. Мини-интеграционный тест, но прост в реализации, хорошо отражает реальное поведение системы. Упрощается с использованием таких решений, как pytest-grpc.
Мокирование с grpc_testing
Официальная библиотека grpc_testing предлагает более юнит-тестовый подход, позволяя создавать тестовые двойники (test doubles) каналов и серверов. Полная изоляция.
Посмотрим на примере первый подход.
pip install grpcio grpcio-tools pytest pytest-grpc
Контракт:
syntax = "proto3";
package greeter;
service Greeter {
rpc SayHello (HelloRequest) returns (HelloReply);
rpc SayHelloStream (HelloRequest) returns (stream HelloReply);
}
message HelloRequest {
string name = 1;
}
message HelloReply {
string message = 1;
}
Генерация кода для сообщений, клиента и базового класса сервера:
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. greeter.proto
Реализация сервера:
import time
from concurrent import futures
import grpc
import greeter_pb2
import greeter_pb2_grpc
class GreeterServicer(greeter_pb2_grpc.GreeterServicer):
def SayHello(self, request, context):
return greeter_pb2.HelloReply(message=f"Hello, {request.name}!")
def SayHelloStream(self, request, context):
for i in range(3):
message = f"Hello, {request.name}! (message #{i+1})"
yield greeter_pb2.HelloReply(message=message)
time.sleep(0.1)
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
greeter_pb2_grpc.add_GreeterServicer_to_server(GreeterServicer(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()
if __name__ == '__main__':
serve()
Тесты:
import pytest
import greeter_pb2
from greeter_server import GreeterServicer
# фикстура сообщает, как добавить наш servicer к серверу
@pytest.fixture()
def grpc_add_to_server():
from greeter_pb2_grpc import add_GreeterServicer_to_server
return add_GreeterServicer_to_server
# фикстура предоставляет экземпляр servicer'а.
@pytest.fixture(scope='module')
def grpc_servicer():
return GreeterServicer()
def test_say_hello_unary(grpc_stub):
# унарный
request = greeter_pb2.HelloRequest(name="Gunenkov")
response = grpc_stub.SayHello(request)
assert response.message == "Hello, Gunenkov!"
def test_say_hello_stream(grpc_stub):
# сервер-стримминг
request = greeter_pb2.HelloRequest(name="Gunenkov")
# возвращается итератор
responses_iterator = grpc_stub.SayHelloStream(request)
responses = list(responses_iterator)
assert len(responses) == 3
assert responses.message == "Hello, Gunenkov! (message #1)"
assert responses.message == "Hello, Gunenkov! (message #2)"
assert responses.message == "Hello, Gunenkov! (message #3)"
¶ Consumer-Driven Contract Testing
В микросервисной архитектуре главная сложность — не в поведении отдельного сервиса, а в обеспечении их корректного взаимодействия
Изменения в одном сервисе не должны сломать другой
Проверка всех интеграций = запуск сквозных тестов. Недостатки:
- тесты могут падать по множеству причин, не связанных с тестируемым изменением (проблемы с сетью, недоступность одного из десятков сервисов, проблемы с тестовыми данными)
- полный прогон E2E-тестов может занимать часы, что замедляет цикл разработки.
- когда тест падает, то бывает сложно определить, какой именно сервис стал причиной сбоя
Вместо того чтобы тестировать все сервисы вместе тестируем их точки интеграции по отдельности, но с гарантией того, что они совместимы.
Ключевой принцип CDCT заключается в том, что именно потребитель (сервис, который инициирует запрос) определяет ожидания от провайдера (сервиса, который на него отвечает). Контракт генерируется на основе реальных потребностей потребителя
Pact — самый популярный инструмент для реализации CDCT. Он предоставляет библиотеки для множества языков и формализует процесс.
Pact — инструмент, который помогает убедиться, что два сервиса "понимают" друг друга, не запуская их одновременно
Consumer (Потребитель)
Сервис, который инициирует запрос к другому сервису.
Provider (Провайдер)
Сервис, который отвечает на запрос.
Pact-файл (Контракт)
JSON-файл, который генерируется тестами потребителя. Содержит список взаимодействий (interactions) — пар "запрос-ответ", которые потребитель ожидает от провайдера.
Pact Broker
Центральный сервис для хранения и обмена pact-файлами и результатами их верификации. Позволяет командам работать независимо друг от друга: потребители публикуют свои контракты, а провайдеры их забирают для проверки.
На стороне потребителя (генерация контракта):
Разработчик потребителя пишет модульный тест для своего кода, который обращается к провайдеру. Вместо реального провайдера используется мок-сервер Pact. В тесте описывается ожидаемое взаимодействие: "при таком-то состоянии провайдера (given) на такой-то запрос (with_request) ожидаем такой-то ответ (will_respond_with)".
Тест запускается. Код потребителя делает запрос к мок-серверу Pact. Мок-сервер проверяет, что запрос соответствует описанному в тесте, и возвращает заранее определенный ответ. Если все проверки в тесте потребителя проходят успешно, Pact генерирует pact-файл.
На стороне провайдера (верификация контракта):
Провайдер (обычно в рамках своего CI-пайплайна) скачивает pact-файл из Pact Broker. Фреймворк Pact запускает реальный сервис провайдера. Pact "проигрывает" запросы из pact-файла, отправляя их реальному провайдеру.
Перед каждым запросом Pact обращается к специальной конечной точке на провайдере для установки нужного состояния (provider state), описанного в given-секции контракта. Pact сравнивает реальные ответы от провайдера с ожидаемыми ответами из pact-файла. Если все реальные ответы соответствуют ожиданиям, верификация считается успешной.
Пример (UserConsumer получает данные о пользователе от UserProvider):
pip install pytest pact-python flask
Потребитель:
import requests
class UserClient:
def __init__(self, base_uri):
self.base_uri = base_uri
def get_user(self, user_id):
uri = f"{self.base_uri}/users/{user_id}"
response = requests.get(uri)
response.raise_for_status()
return response.json()
Контрактный тест для потребителя:
import atexit
import unittest
import requests
from pact import Consumer, Provider, Like, Term
from consumer_client import UserClient
pact = Consumer('UserConsumer').has_pact_with(
Provider('UserProvider'),
pact_dir='./pacts'
)
pact.start_service()
# регистрация функции, которая будет выполнена в любом случае при завершении работы программы
atexit.register(pact.stop_service)
class UserClientContractTest(unittest.TestCase):
def test_get_user(self):
# формат ответа
expected_body = {
'id': Like(1),
'username': Like('testuser'),
'last_login': Term(
r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}',
'2023-01-01T12:00:00'
)
}
(pact
.given('a user with id 1 exists') # состояние провайдера
.upon_receiving('a request for user 1')
.with_request(
method='GET',
path='/users/1'
)
.will_respond_with(200, body=expected_body))
with pact:
client = UserClient(pact.uri)
user = client.get_user(1)
self.assertEqual(user['id'], 1)
self.assertEqual(user['username'], 'testuser')
| Матчер | Описание | Пример использования в body |
|---|---|---|
Like(value) |
Проверяет соответствие типа данных value. Мок-сервер вернет value. |
{'id': Like(123), 'name': Like('a string')} |
Term(regex, example) |
Проверяет строку по регулярному выражению. Мок-сервер вернет example. |
{'uuid': Term('[0-9a-f]{8}-...', 'c5a3...-b4f5')} |
EachLike(matcher, min) |
Проверяет массив, где каждый элемент соответствует matcher (словарь, где для каждого ключа указан другой матчер (например, Like для проверки типа)). min задает минимальный размер. |
{'tags': EachLike('tag', min=1)} |
Format() |
Готовые матчеры для общих форматов (IP, UUID и т.д.). | {'ip_address': Format().ip_address} |
После запуска:
python -m unittest test_consumer.py
Получим Pact-файл:
{
"consumer": {
"name": "UserConsumer"
},
"provider": {
"name": "UserProvider"
},
"interactions": [
{
"description": "a request for user 1",
"providerState": "a user with id 1 exists",
"request": {
"method": "GET",
"path": "/users/1"
},
"response": {
"status": 200,
"headers": {
},
"body": {
"id": 1,
"username": "testuser",
"last_login": "2023-01-01T12:00:00"
},
"matchingRules": {
"$.body.id": {
"match": "type"
},
"$.body.username": {
"match": "type"
},
"$.body.last_login": {
"match": "regex",
"regex": "\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}"
}
}
}
}
],
"metadata": {
"pactSpecification": {
"version": "2.0.0"
}
}
}
Провайдер:
from flask import Flask, jsonify, request
import datetime
app = Flask(__name__)
DB = {}
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
if user_id in DB:
return jsonify(DB[user_id])
return jsonify({"error": "User not found"}), 404
# эндпоинт для установки состояния!
@app.route('/_provider_states', methods=['POST'])
def provider_states():
global DB
state_data = request.json
state = state_data.get("state")
if state == "a user with id 1 exists":
DB[1] = {
'id': 1,
'username': 'testuser',
'last_login': datetime.datetime.now().isoformat(timespec='seconds')
}
return jsonify({'status': 'ok'})
if __name__ == '__main__':
app.run(port=8000)
Верификация:
import pytest
from pact import Verifier
@pytest.fixture(name="pact_verifier", scope="session")
def setup_verifier():
verifier = Verifier(
provider='UserProvider',
provider_base_url='http://localhost:8000'
)
output, _ = verifier.verify_pacts(
'./pacts/userconsumer-userprovider.json',
provider_states_setup_url='http://localhost:8000/_provider_states'
)
# если верификация не удалась - здесь будет исключение
return output
def test_pact_verification(pact_verifier):
# запуск фикстуры
assert pact_verifier == 0 # код возврата успешный
Для запуска верификации нужно:
- запустить провайдер
python provider_app.py - запустить тесты:
pytest test_provider.py
¶ Выводы
Многоуровневая стратегия тестирования взаимодействия микросервисов
Уровень 1: валидация по схеме (проверка "существительных")
Проверка структуры и типов данных в телах запросов и ответов с использованием формальных описаний.
Гарантирует синтаксическую корректность обмена данными. Ответ на вопрос: "Соответствуют ли данные ("существительные") оговоренному формату?".
Уровень 2: функциональное тестирование API (проверка "глаголов")
Тестирование бизнес-логики и поведения отдельного сервиса в изоляции.
Проверяет, что сервис корректно выполняет свои функции. Ответ на вопрос: "Правильно ли работают действия ("глаголы"), которые выполняет API?".
Уровень 3: контрактное тестирование (проверка "диалога")
Проверка совместимости взаимодействий между сервисами (потребителем и провайдером) с использованием Consumer-Driven Contract Testing.
Самый высокий уровень юнит/интеграционного тестирования, который заменяет большую часть хрупких сквозных тестов. Гарантирует, что сервисы "понимают" друг друга. Ответ на вопрос: "Корректно ли построен диалог между сервисами?".
При тестировании взаимодействия микросервисов следует применяет все три уровня, чтобы получить максимально быструю и надежную обратную связь на каждом этапе жизненного цикла разработки
Ошибка в формате данных должна быть поймана на первом уровне, ошибка в логике — на втором, а проблема несовместимости версий API — на третьем.