Добрый день. Я занимаюсь автоматизацией тестирования. Как и у всех автоматизаторов, у меня есть набор библиотек и инструментов, которые я обычно выбираю для написания тестов. Но периодически возникают ситуации, когда ни одна из знакомых библиотек может решить задачу с риском сделать автотесты нестабильными или хрупкими. В этой статье я хотел бы рассказать, как вроде бы стандартная задача использования mock’ов привела меня к написанию своего модуля. Также хотел бы поделиться своим решением и услышать обратную связь.

Приложение

Одной из необходимых отраслей в финансовом секторе является аудит. Данные необходимо регулярно сверять (reconciliation). В связи с этим и появилось то приложение, которое я тестировал. Чтобы не говорить о чём-то абстрактном, представим, что наша команда разрабатывает приложение для обработки заявок из мессенджеров. Для каждой заявки должно быть создано соответствующее событие в elasticsearch’е. Сверяющее же приложение будет нашим мониторингом, что заявки не пропускаются.

Итак, представим, что у нас есть система, имеющая следующие компоненты:

1. Сервер конфигурации. Для пользователя это единая точка входа, где он настраивает не только приложение для сверки, но и другие компоненты системы.
2. Сверяющее приложение.
3. Данные от приложения, обрабатывающего заявки, которые хранятся в elasticsearch’е.
4. Эталонные данные. Формат данных зависит от мессенджера, с которым приложение интегрируется.

Задача

Автоматизация тестирования в данном случае выглядит достаточно прямолинейно:

1. Подготовка окружения:
   
   • Устанавливается elasticsearch с минимальной конфигурацией (с помощью msi и командной строки).
   • Устанавливается сверяющее приложение.
2. Выполнение тестов:
   
   • Конфигурируется сверяющее приложение.
   • Elasticsearch наполняется тестовыми данными для соответствующего теста (сколько заявок было обработано).
   • Приложению подаются "эталонные" данные от мессенджера (сколько заявок якобы было на самом деле).
   • Проверяется вердикт, который выдало приложение: количество успешно сверенных заявок, количество недостающих и т.п.
3. Очистка окружения.

Проблема заключается в том, что мы тестируем мониторинг, но чтобы его настроить, нужны данные от сервера конфигурации. Во-первых, устанавливать и настраивать сервер для каждого прогона — трудоёмкая операция (он имеет свою базу, например). Во-вторых, хочется изолировать приложения, чтобы упростить локализацию проблем при нахождении дефекта. В итоге, было решено использовать mock.

Тут может возникнуть вопрос: "Если мы всё равно будем mock’ать сервер, может и на установку и наполнение elasticsearch’а время не тратить, а заменить mock’ом?". Но всё-таки всегда нужно помнить, что использование mock’ов предоставляет гибкость, но добавляет обязательства следить за актуальностью поведения mock’а. Поэтому от замены elasticsearch’а я отказался: его и устанавливать, и наполнять достаточно просто.

Первый mock

Сервер отдаёт конфигурацию на GET-запросы по нескольким путям в /configuration. Нас интересуют два пути. Первый — это /configuration/data_cluster с конфигурацией кластера

{     "host": "127.0.0.1",     "port": 443,     "credentials": {         "username": "user",         "password": "pass"     } }

Второй — это /configuration/reconciliation с конфигурацией сверяющего приложения

{     "reconciliation_interval": 3600,     "configuration_update_interval": 60,     "source": {         "address": "file:///c:/path",         "credentials": {             "username": "user",             "password": "pass"         }     } }

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

Итак, статические mock’и и средства для mock’ов в юниттестах (mock, monkeypatch из pytest’а и т.п.) нам не подойдут. Я нашёл отличную библиотеку pretenders, которая, как я думал, мне подходит. Pretenders предоставляет возможность создавать HTTP сервер с правилами, которые определяют, как сервер будет отвечать на запросы. Правила хранятся в preset’ах, что позволяет изолировать mock’и для разных наборов тестов. Preset’ы можно очищать и заполнять заново, что позволяет обновлять ответы по необходимости. Сам сервер достаточно поднять один раз во время подготовки окружения:

python -m pretenders.server.server --host 127.0.0.1 --port 8000

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

import json  import pytest from pretenders.client.http import HTTPMock from pretenders.common.constants import FOREVER  @pytest.fixture def configuration_server_mock(request):     mock = HTTPMock(host="127.0.0.1", port=8000, name="server")     request.addfinalizer(mock.reset)     return mock  def test_something(configuration_server_mock):     configuration_server_mock.when("GET /configuration/data_cluster").reply(         headers={"Content-Type": "application/json"},         body=json.dumps({             "host": "127.0.0.1",             "port": 443,             "credentials": {                 "username": "user",                 "password": "pass",             },           }),          status=200,         times=FOREVER,     )        configuration_server_mock.when("GET /configuration/reconciliation").reply(         headers={"Content-Type": "application/json"},         body=json.dumps({             "reconciliation_interval": 3600,             "configuration_update_interval": 60,              "source": {                 "address": "file:///c:/path",                 "credentials": {                     "username": "user",                     "password": "pass",                 },               },           }),          status=200,         times=FOREVER,     )       # test application

Но это ещё не всё. При своей гибкости, pretenders имеет два ограничения, о которых надо помнить и необходимо решить в нашем случае:

1. Правила нельзя удалять по одному. Для того, чтобы изменить ответ, необходимо удалить весь preset и пересоздать все правила заново.
2. Все пути, используемые в правилах, являются относительными. Preset’ы имеют уникальный путь вида /mockhttp/<preset_name>, и этот путь является общим префиксом для всех создаваемых путей в правилах. Тестируемое же приложение получает только имя хоста, и о префиксе оно знать не может.

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

configuration.data_cluster.port = 443

или (чтобы делать запросы на обновление реже)

data_cluster_config = get_default_data_cluster_config() data_cluster_config.port = 443 configuration.update_data_cluster(data_cluster_config)

Такая инкапсуляция позволяет нам практически безболезненно обновлять все пути. Также можно создать индивидуальный preset для каждого отдельного endpoint’а и общий (основной) preset, перенаправляющий (через 307 или 308) на индивидуальные. Тогда можно очищать только один preset для обновления правила.

Для того, чтобы избавиться от префиксов, можно воспользоваться библиотекой mitmproxy. Это мощный инструмент, который позволяет, помимо прочего, перенаправлять запросы. Мы уберём префиксы следующим образом:

mitmdump --mode reverse:http://127.0.0.1:8000 --replacements :~http:^/:/mockhttp/server/ --listen-host 127.0.01 --listen-port 80

Параметры этой команды делают следующее:

1. --listen-host 127.0.0.1 и --listen-port 80 очевидные. Mitmproxy поднимает свой сервер, и этими параметрами мы определяем интерфейс и порт, которые будет слушать этот сервер.
2. --mode reverse:http://127.0.0.1:8000 означает, что запросы на сервер mitproxy будут перенаправлены на http://127.0.0.1:8000. Подробнее можно почитать тут.
3. --replacements :~http:^/:/mockhttp/server/ определяет шаблон, с помощью которого запросы будут изменены. Он состоит из трёх частей: фильтра запросов (~http для HTTP запросов), шаблона для изменения (^/, чтобы заменить начало пути) и собственно замены (/mockhttp/server). Подробнее можно почитать тут.

В нашем случае, мы добавляем mockhttp/server ко всем HTTP-запросам и перенаправляем их на http://127.0.0.1:8000, т.е. на наш сервер pretenders. В итоге мы добились того, что теперь конфигурацию можно получить GET-запросом к http://127.0.0.1/configuration/data_cluster.

В целом меня устраивала конструкция с pretenders и mitmproxy. При кажущейся сложности — всё-таки 2 сервера вместо одного настоящего — подготовка заключается в установке 2 пакетов и выполнении 2 команд в командной строке для их запуска. В управлении mock’ом не всё так однозначно, но вся сложность заключена только в одном месте (управлении preset’ами) и решается достаточно просто и надёжно. Однако в задаче появились новые обстоятельства, которые заставили меня задуматься о новом решении.

Второй mock

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

GET /application-history?page=2&size=5&start=1569148012&end=1569148446

Тут настораживают 2 момента. Во-первых, несколько параметров. Дело в том, что параметры могут быть указаны в любом порядке, что сильно усложняет регулярное выражение для правила pretenders. Ещё необходимо помнить, что параметры необязательные, но это не такая проблема, как произвольный порядок. Во-вторых, последние параметры (start и end) задают временной промежуток для фильтрации заявки. И проблема в данном случае в том, что мы не может заранее предугадать, какой промежуток (не величину, а время старта) будет использован приложением, чтобы сформировать ответ mock’а. Проще говоря, нам необходимо знать и использовать значения параметров, чтобы сформировать "разумный" ответ. "Разумность" в данном случае важна, например, чтобы мы могли протестировать, что приложение проходит по всем страницам пагинации: если на все запросы мы будем отвечать одинаково, то мы не сможем найти дефектов, связанных с тем, что запрашивается только одна страница из пяти.

Я попробовал поискать альтернативные варианты решения, но в итоге решил попробовать написать своё. Так появился loose server. Это Flask-приложение, у которого пути и ответы могут быть настроены после его старта. "Из коробки" он умеет работать с правилами для типа запроса (GET, POST и т.п.) и для пути. Это позволяет заменить pretenders и mitmproxy в изначальной задаче. Также я покажу, как его можно использовать, чтобы создать mock для API вендора.

Приложению необходимы 2 основных пути:

1. Base endpoint. Это тот самый префикс, который будет использован для всех сконфигурированных правил.
2. Configuration endpoint. Это префикс тех запросов, с помощью которых можно конфигурировать сам mock-сервер.

python -m looseserver.default.server.run --host 127.0.0.1 --port 80 --base-endpoint / --configuration-endpoint /_mock_configuration/

В общем случае, лучше не настраивать base endpoint и configuration endpoint так, чтобы один был родителем другого. Иначе есть риск, что пути для конфигурации и для тестирования будут конфликтовать. Приоритет будет у configuration endpoint, так как Flask-правила для конфигурации добавляются раньше, чем для динамических путей. В нашем случае можно было бы использовать --base-endpoint /configuration/, если бы мы не собирались включать API вендора в этот mock.

Простейший вариант тестов не сильно меняется

import json  import pytest from looseserver.default.client.http import HTTPClient from looseserver.default.client.rule import PathRule from looseserver.default.client.response import FixedResponse  @pytest.fixture def configuration_server_mock(request):     class MockFactory:         def __init__(self):             self._client = HTTPClient(configuration_url="http://127.0.0.1/_mock_configuration/")             self._rule_ids = []          def create_rule(self, path, json_response):             rule = self._client.create_rule(PathRule(path=path))             self._rule_ids.append(rule.rule_id)              response = FixedResponse(                 headers={"Content-Type": "application/json"},                 status=200,                 body=json.dumps(json_response),             )             self._client.set_response(rule_id=rule.rule_id, response=response)          def _delete_rules(self):             for rule_id in self._rule_ids:                 self._client.remove_rule(rule_id=rule_id)      mock = MockFactory()     request.addfinalizer(mock._delete_rules)     return mock  def test_something(configuration_server_mock):     configuration_server_mock.create_rule(         path="configuration/data_cluster",         json_response={             "host": "127.0.0.1",             "port": 443,             "credentials": {                 "username": "user",                 "password": "pass",             },         }     )      configuration_server_mock.create_rule(         path="configuration/reconciliation",         json_response={             "reconciliation_interval": 3600,             "configuration_update_interval": 60,             "source": {                 "address": "file:///applications",                 "credentials": {                     "username": "user",                     "password": "pass",                 },             },         }     )

Фикстура стала сложнее, но правила теперь можно удалять по одному, что упрощает работу с ними. Использование mitmproxy уже не нужно.

Вернёмся теперь к API вендора. Мы создадим новый тип правил для loose server’а, который в зависимости от значения параметра будет отдавать различные ответы. Далее мы воспользуемся этим правилом для параметра page.

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

from looseserver.server.rule import ServerRule  class ServerParameterRule(ServerRule):     def __init__(self, parameter_name, parameter_value=None, rule_type="PARAMETER"):         super(ServerParameterRule, self).__init__(rule_type=rule_type)         self._parameter_name = parameter_name         self._parameter_value = parameter_value      def is_match_found(self, request):         if self._parameter_value is None:             return self._parameter_name not in request.args          return request.args.get(self._parameter_name) == self._parameter_value

Каждое правило должно определить метод is_match_found, которое определяет, должно ли оно сработать для данного запроса или нет. Входным параметром для него является объект запроса. После того, как новое правило создано, надо "научить" сервер принимать его от клиента. Для этого используется RuleFactory:

from looseserver.default.server.rule import create_rule_factory from looseserver.default.server.application import configure_application  def _create_application(base_endpoint, configuration_endpoint):     server_rule_factory = create_rule_factory(base_endpoint)      def _parse_param_rule(rule_type, parameters):         return ServerParameterRule(             rule_type=rule_type,             parameter_name=parameters["parameter_name"],             parameter_value=parameters["parameter_value"],         )      server_rule_factory.register_rule(         rule_type="PARAMETER",         parser=_parse_param_rule,         serializer=lambda rule_type, rule: None,     )      return configure_application(         rule_factory=server_rule_factory,         base_endpoint=base_endpoint,         configuration_endpoint=configuration_endpoint,     )  if __name__ == "__main__":     application = _create_application(base_endpoint="/", configuration_endpoint="/_mock_configuration")     application.run(host="127.0.0.1", port=80)

Здесь мы создаём фабрику для правил по умолчанию, чтобы она содержала те правила, которые мы использовали раньше, и регистрируем новый тип. Клиенту в данном случае информация о правиле не нужна, поэтому serializer ничего фактически не делает. Далее эта фабрика передаётся в приложение. И его уже можно запускать, как обычное Flask-приложение.

С клиентом ситуация аналогичная: создаём правило и фабрику. Но для клиента, во-первых, не нужно определять метод is_match_found, а, во-вторых, serializer в данном случае необходим, чтобы отправить правило серверу.

from looseserver.client.rule import ClientRule from looseserver.default.client.rule import create_rule_factory  class ClientParameterRule(ClientRule):     def __init__(self, parameter_name, parameter_value=None, rule_type="PARAMETER", rule_id=None):         super(ClientParameterRule, self).__init__(rule_type=rule_type, rule_id=rule_id)         self.parameter_name = parameter_name         self.parameter_value = parameter_value  def _create_client(configuration_url):     def _serialize_param_rule(rule):         return {             "parameter_name": rule.parameter_name,             "parameter_value": rule.parameter_value,         }         client_rule_factory = create_rule_factory()     client_rule_factory.register_rule(         rule_type="PARAMETER",         parser=lambda rule_type, parameters: ClientParameterRule(rule_type=rule_type, parameter_name=None),         serializer=_serialize_param_rule,     )         return HTTPClient(configuration_url=configuration_url, rule_factory=client_rule_factory)

Осталось воспользоваться _create_client для создания клиента, и правила можно использовать в тестах. В примере ниже я добавил использование ещё одного правила по умолчанию: CompositeRule. Оно позволяет объединять несколько правил в одно так, что они срабатывают, только если каждое из них возвращает True для вызова is_match_found.

@pytest.fixture def configuration_server_mock(request):     class MockFactory:         def __init__(self):             self._client = _create_client("http://127.0.0.1/_mock_configuration/")             self._rule_ids = []          def create_paged_rule(self, path, page, json_response):             rule_prototype = CompositeRule(                 children=[                     PathRule(path=path),                     ClientParameterRule(parameter_name="page", parameter_value=page),                 ]             )             rule = self._client.create_rule(rule_prototype)             self._rule_ids.append(rule.rule_id)              response = FixedResponse(                 headers={"Content-Type": "application/json"},                 status=200,                 body=json.dumps(json_response),             )             self._client.set_response(rule_id=rule.rule_id, response=response)           ...      mock = MockFactory()     request.addfinalizer(mock._delete_rules)     return mock  def test_something(configuration_server_mock):     ...      configuration_server_mock.create_paged_rule(         path="application-history",         page=None,         json_response=["1", "2", "3"],     )      configuration_server_mock.create_paged_rule(         path="application-history",         page="1",         json_response=["1", "2", "3"],     )      configuration_server_mock.create_paged_rule(         path="application-history",         page="2",         json_response=["4", "5"],     )

Заключение

Связка библиотек pretenders и mitmproxy предоставляет мощный и достаточно гибкий инструмент для создания mock’ов. Его плюсы:

1. Лёгкая настройка.
2. Возможность изолировать наборы запросов с помощью preset’ов.
3. Удаление сразу всего изолированного набора.

К минусам можно отнести:

1. Необходимость создания регулярных выражений для правил.
2. Отсутствие возможности изменять правила индивидуально.
3. Наличие префикса для всех созданных путей либо использование перенаправления с помощью mitmproxy.

Ссылки на документацию:
Pretenders
Mitmproxy
Loose server