1. Введение
Всем привет! Меня зовут Ренат Дасаев и в прошлой статье Автоматизация Е2Е‑тестирования сквозных БП интеграционных проектов Операционного блока было рассказано о том, как устроено e2e‑автотестирование. Сегодня хочу рассказать о том, как используется camunda в автотестировании бизнес‑процессов (далее БП). На практических примерах рассмотрим, что и как мы делаем в своих тестах.
Если интересно узнать о движке camunda и о том, как он применяется в БП Московской Биржи на примере проекта ЦУП, — рекомендую ознакомиться с материалом нашего коллеги.
2. Проверки БП в camunda автотестами
Прежде чем погрузиться в техническую часть вопроса, определимся, какие проверки нам необходимо осуществить в ходе БП.
Стандартный набор проверок:
процесс начинается и завершается успешно;
gateway, boundary events и другие элементы двигают БП в нужном направлении;
на проверяемых шагах корректный контекст — присутствуют необходимые переменные процесса;
корректный результат вычислений в выражениях events/gateway;
успешно исполняются service‑таски и создаются user‑таски;
на шагах, где есть взаимодействие с внешними сервисами — успешно отрабатывают (в e2e нет заглушек, работаем с реальными интеграциями);
не только «success path», но и другие маршруты следования по схеме, в том числе, где обрабатываются ошибки в процессе;
процесс не впадает в инцидент.
Схемы могут вызывать подпроцессы со своим контекстом. В таком случае проверяются и порожденные процессы наравне с родительским.
3. Разработка модулей по работе с bpmn – процессами
В прошлой статье упоминалось, что совместно с другой командой разрабатываем python-модули для работы с микросервисами, а также модули (клиенты) по работе с сервисами/движками.
Для работы с bpmn-процессами разработано несколько модулей:
moex-pmh-bpmn (работа с bpmn-процессами);
moex-pmh-camunda-client (работа с camunda rest api);
moex-pmh-<название_БП> (работа с конкретным инстансом БП, на каждый бизнес-процесс по 1 такому модулю).
Иерархия модулей:
moex-pmh-<название_БП> → moex‑pmh‑bpmn → moex-pmh-camunda-client.
Модули разработаны, размещены в корпоративном pypi-репозитории, и любая из команд в нашей компании может их использовать в своем проекте. При этом у команд появляется также возможность предлагать изменения в кодовой базе (новые фичи, исправление ошибок) через merge-request.
На текущий момент модуль moex-pmh-camunda-client умеет работать со следующими контроллерами:
Decision Definition
Execution
External task
Job
Job Definition
Historic Process Instance
Historic Variable Instance
Incident
Process Definition
Process Instance
Task
Variable Instance
Контроллеры добавляются по мере необходимости. В каждом методе любого контроллера на выходе получаем объект датакласса со всеми полями, которые приходят в ответе. Ниже будет показан пример использования возможностей этого модуля при решении практических задач автотестирования БП.
4. Старт бизнес-процесса. Поиск запущенного процесса в camunda.
В любом БП движок camunda доступен через http — https://<тестовый_домен>/<camunda>. С этим endpoint автотест и работает в ходе проверок внутри БП.
Первое, с чего начинается автотестирование БП, — это его старт. В большинстве случаев запуск процесса происходит в ходе возникновения определенного рода событий:
отправки сообщения в шину, например, esb (enterprise service bus);
создания объекта в БД сервиса;
наступления определенной даты и/или времени;
получения электронных писем, файлов и прочее.
После того, как БП запустился, необходимо этот старт идентифицировать и найти запущенный процесс в списке активных. Его, как правило, можно найти по уникальным переменным процесса, таким как бизнес-ключ (business_key), торговый код инструмента (securityid) или другим переменным, присущим конкретным БП. В редких случаях идентификация процесса происходит по дате и времени запуска, если нет уникальных переменных процесса.
В большинстве случаев инженеру по автотестированию достаточно просто руками найти запущенный процесс в его camunda, основываясь на времени его запуска. Далее инженер анализирует переменные процесса и находит уникальные, по которым можно точно идентифицировать запускаемый экземпляр процесса.
Для наглядности возьмем реальный БП «Активация выкупа ценной бумаги» (входит в один большой процесс по Регистрации, активации и деактивации выкупа). Это одна из самых маленьких схем, что удалось найти. В 99% случаев схемы гораздо больше по объему и взаимодействию с внешними ресурсами. На рисунке 1 представлен скриншот данного БП из camunda:
В нашем примере поиск процесса будет производиться по переменной secuirtyId (торговый код инструмента). На рисунке 2 можно увидеть запущенный экземпляр процесса «Активация выкупа» со списком переменных процесса, среди которых есть необходимая переменная:
Ниже представлен код на языке Python, который реализует поиск среди активных процессов по переменной secuirtyId:
def wait_for_process_instance_by_security_id(
self,
security_id: str,
*,
max_time: int = 180,
sleep_time: float = 2.0,
) -> CamundaProcessInstance:
with allure.step(
f'Проверка того, что был запущен процесс "{self.pretty_name}" '
f'со значением переменной securityId="{security_id}"',
):
return poll_by_time_first_success(
fn=lambda: self.get_process_instance_by_security_id(security_id),
checker=lambda p: p is not None,
max_time=max_time,
sleep_time=sleep_time,
)Производятся попытки в течение max_time (120 секунд по умолчанию) с шагом в 2 секунды (по умолчанию) в цикле (поллинг) найти процесс, в котором обнаружилась необходимая переменная (get_process_variable_instances) и её значение:
def get_process_instance_by_security_id(
self,
security_id: str,
) -> Optional[CamundaProcessInstance]:
with allure.step(
f'Поиск процесса "{self.pretty_name}" '
f'со значением переменной securityId="{security_id}"',
):
processes = self.camunda.get_process_instances()
for process in processes:
variables = self.camunda.get_process_variable_instances(process.id) or {}
if 'securityId' not in variables:
continue
v_security_id = variables.get('securityId').value
if v_security_id == security_id:
return processЕсли процесс нашелся, то возвращается объект датакласса CamundaProcessInstance, либо None.
Пример CamundaProcessInstance объекта:
id='007f5c67-44c4-11ef-b669-b2d4f57bb43f'
rootProcessInstanceId='007f5c67-44c4-11ef-b669-b2d4f57bb43f'
superProcessInstanceId=None
superCaseInstanceId=None
caseInstanceId=None
processDefinitionName='Подтверждение активации режима выкупа'
processDefinitionKey='bp-offer-activation'
processDefinitionVersion=7
processDefinitionId='bp-offer-activation:7:b99a97ae-17e5-11ee-971d-7ef2aaea4619'
businessKey='bpms/offersreg/1/1/RUTEST48KTEP/5623'
startTime=datetime.datetime(2024, 7, 18, 8, 10, 8, 440000)
endTime=None
removalTime=None
durationInMillis=None
startUserId=None
startActivityId='StartEvent_1'
deleteReason=None
tenantId=None
state=<ProcessInstanceState.ACTIVE: 'ACTIVE'>Если в процессе поллинга процесс не обнаружился за промежуток времени max_time, то порождается исключение TimeoutError и автотест завершается на этом.
5. Мониторинг активностей по bpmn-схеме
Важной функцией является поиск активностей по camunda-схеме. Это могут быть разные шаги:
определенные вычисления в gateway/boundary/events;
сервис-таски (service-tasks);
пользовательские задачи (user-tasks);
взаимодействие с базами данных/шинами/почтовыми ящиками/сайтами и прочее.
На примере нашего БП по активации выкупа попробуем определить, что создалась пользовательская задача «Выполнить сверку ЕКБД с ASTS» (ЕКБД – Единая Клиентская База Данных, ASTS — Automated Securities Trading System).
Для начала нужно вычислить id данного шага на схеме в camunda. Загрузим bpmn-файл данного БП в camunda modeler и выделим необходимый шаг (см. рисунок 3):
В списке атрибутов находим id = compare-offers-manual. Это и будет тот идентификатор, который необходимо найти в списке активностей нашего тестируемого БП.
Для поиска активности используем функцию с поллингом (waitforprocess_activity):
def wait_for_process_activity(
self,
process_instance_id: str,
*,
activity_name: Optional[str] = None,
activity_id: Optional[str] = None,
max_time: int = 30,
sleep_time: int = 2,
) -> None:
moex_asserts.assert_true(
expr=(activity_name or activity_id) and not (activity_name and activity_id),
msg='Должен быть указан один из аргументов [activity_name, activity_id]',
)
kwargs = (
{'activity_name': activity_name}
if activity_name else {'activity_id': activity_id}
)
with allure.step(
f'Проверка того, что процесс "{self.pretty_name}" '
f'с id "{process_instance_id}" дошел до активности '
f'"{activity_name or activity_id}"',
):
poll_by_time_first_success(
fn=lambda: self.find_process_activity(
process_instance_id=process_instance_id,
**kwargs,
),
checker=lambda a: a is not None,
max_time=max_time,
sleep_time=sleep_time,
msg=(
f'Процесс "{self.pretty_name}" с id "{process_instance_id}" '
f'не дошел до активности "{activity_id or activity_name}"'
),
)В которой на каждой итерации получаем список активностей (find_process_activity):
def find_process_activity(
self,
process_instance_id: str,
*,
activity_name: Optional[str] = None,
activity_id: Optional[str] = None,
) -> Optional[CamundaActivity]:
moex_asserts.assert_true(
expr=(activity_name or activity_id) and not (activity_name and activity_id),
msg='Должен быть указан один из аргументов [activity_name, activity_id]',
)
with allure.step(
f'Поиск активности "{activity_name or activity_id}" '
f'процесса "{self.pretty_name}" с id "{process_instance_id}"',
):
for activity in self.get_process_activities(process_instance_id):
if activity_name:
if activity.activityName == activity_name:
return activity
elif activity_id:
if activity.activityId == activity_id:
return activityВнутри find_process_activity() вызывается get_process_activities() (внутри используется camunda api - /process-instance/{process_instance_id}/activity-instances) и ищем среди них наш активити по activity_id/activity_name:
def get_process_activities(self, process_instance_id: str) -> List[CamundaActivity]:
with allure.step(
f'Поиск активностей процесса "{self.pretty_name}" '
f'с id "{process_instance_id}"',
):
base_activity = self.camunda.get_process_instance_activities(
process_instance_id,
)
child_activities = self.__get_process_activities(base_activity)
return child_activitiesСоответственно, необходимо передать в аргументах wait_for_process_activity лишь:
process_instance_id=<id_процесса> (вычисляется из функции поиска старта процесса);
activity_id=’ compare-offers-manual’ (или activity_name).
Как только процесс дойдет до этого шага в указанные max_time (обычно 60 секунд), то поллинг это отловит, произведет проверку и, если все успешно, передаст управление следующей функции в тесте. Иначе получим TimeoutError.
После того, как тест нашел activity с activity_id=’ compare-offers-manual’ в camunda, необходимо вычислить id пользовательской задачи (user-tasks), чтобы потом с этим идентификатором найти задачу уже в Менеджере Задач платформы ЦУП.
Для вычисления id пользовательской задачи в camunda разработали функцию wait_for_process_user_task():
def wait_for_process_user_task(
self,
process_instance_id: str,
*,
activity_name: Optional[str] = None,
activity_id: Optional[str] = None,
max_time: int = 30,
sleep_time: int = 2,
post_await_time: int = 3,
) -> CamundaTask:
moex_asserts.assert_true(
expr=(activity_name or activity_id) and not (activity_name and activity_id),
msg='Должен быть указан один из аргументов [activity_name, activity_id]',
)
kwargs = (
{'activity_name': activity_name}
if activity_name else {'activity_id': activity_id}
)
with allure.step(
f'Проверка того, что по процессу "{self.pretty_name}" с id '
f'"{process_instance_id}" была создана пользовательская задача по джобе'
f'"{activity_id or activity_name}"',
):
task = poll_by_time_first_success(
fn=lambda: self.find_process_user_task(
process_instance_id=process_instance_id,
**kwargs,
),
checker=lambda a: a is not None,
max_time=max_time,
sleep_time=sleep_time,
msg=(
f'По процессу {self.pretty_name} с id "{process_instance_id}" '
'не была создана пользовательская задача по джобе '
f'"{activity_id or activity_name}"'
),
)
time.sleep(post_await_time)
return taskКак и в поиске необходимой активности, необходимо передать идентификаторы процесса и активити. Внутри поллинга используется функция find_process_user_task():
def find_process_user_task(
self,
process_instance_id: str,
*,
activity_name: Optional[str] = None,
activity_id: Optional[str] = None,
) -> Optional[CamundaTask]:
moex_asserts.assert_true(
expr=(activity_name or activity_id) and not (activity_name and activity_id),
msg='Должен быть указан один из аргументов [activity_name, activity_id]',
)
with allure.step(
'Поиск пользовательской задачи по джобе '
f'"{activity_id or activity_name}" процесса '
f'"{self.pretty_name}" с id "{process_instance_id}"',
):
if not activity_id:
activity_id = self.find_process_activity(
process_instance_id=process_instance_id,
activity_name=activity_name,
).activityId
tasks = self.camunda.get_process_instance_tasks(
process_instance_id=process_instance_id,
activity_id=activity_id,
)
return tasks[0] if tasks else NoneВнутри функции используется POST-метод /task. В случае успеха (нашлась таска в нужном процессе и с нужным активити) возвращается список объектов датакласса CamundaTask со всеми полями:
id='00977769-44c4-11ef-b669-b2d4f57bb43f'
name='Добавить "TEST-H9XQ8" (RUTEST48KTEP) на режим "Выкуп: Адресные заявки"'
assignee=None
created='2024-07-18T08:10:08.598+0300'
due='2024-07-18T23:59:59.647+0300'
followUp=None, delegationState=None
description='5623'
executionId='007f5c67-44c4-11ef-b669-b2d4f57bb43f'
owner=None
parentTaskId=None
priority=50
processDefinitionId='bp-offer-activation:7:b99a97ae-17e5-11ee-971d-7ef2aaea4619'
processInstanceId='007f5c67-44c4-11ef-b669-b2d4f57bb43f'
caseExecutionId=None
caseDefinitionId=None
caseInstanceId=None
taskDefinitionKey='compare-offers-manual'
suspended=False
formKey=None
camundaFormRef=None
tenantId=NoneИз данного объекта нас интересует id='00977769-44c4-11ef-b669-b2d4f57bb43f' это и есть идентификатор задачи, который будет такой же и в ЦУП. Далее необходимо будет отправить запрос на получение активных задач в ЦУП в Менеджер Задач, найти её в выдаче и далее выполнить над этой задачей необходимые действия («Выполнить» или «Отклонить»).
Если необходимо идентифицировать прохождение определенного шага в схеме, но этот шаг проскакивает очень быстро — достаточно в поллинге в sleep_time (задержка перед следующей итерации) задать низкое значение, например, 0.3 секунды.
Если же поллинг по активным процессам не позволяет вовремя его обнаружить, то стоит воспользоваться поиском в уже завершенных процессах.
Для получения информации об уже завершенном процессе достаточно передать process_instance_id в функцию get_historic_process_instance():
def get_historic_process_instance(
self,
process_instance_id: str,
) -> Optional[CamundaHistoricProcessInstance]:
url = f'{self.__url_prefix}/history/process-instance/{process_instance_id}'
resp = self.__get(url=url)
return CamundaHistoricProcessInstance(**resp) if resp else NoneЕсли же при этом нужно найти в уже завершенном процессе какой-то активити, то можно использовать get_process_instance_historic_activities():
def get_process_instance_historic_activities(
self,
process_instance_id: str,
) -> List[CamundaHistoricActivity]:
url = f'{self.__url_prefix}/history/activity-instance'
resp = self.__get(url=url, params={'processInstanceId': process_instance_id})
return [CamundaHistoricActivity(**a) for a in resp]и в списке всех активити найти нужный.
В случае если нужно найти переменную в завершенном процессе, то можно использовать get_historic_process_instance_variables():
def get_historic_process_instance_variables(
self,
process_instance_id: str,
) -> Optional[Dict[str, ProcessInstanceHistoricVariable]]:
url = f'{self.__url_prefix}/history/variable-instance'
params = {
'processInstanceId': process_instance_id,
'deserializeValues': 'false',
}
resp = self.__get(url=url, params=params)
return {
variable['name']: ProcessInstanceHistoricVariable(**variable)
for variable in resp
} if resp else None6. Взаимодействие с таймерами
Существуют БП, где в схеме запрограммированы таймеры. С теми процессами, что имели дело, таймер, как правило, срабатывал на следующие события:
наступление определенной даты и времени;
поступление сигнала от внешней системы.
Чтобы проверить, что БП встал на таймере также по camunda modeler, находим id этого шага и ищем этот идентификатор в списке активити по процессу (более детально в предыдущей главе). Часто бывает так, что необходимо не просто найти активный таймер, но и произвести взаимодействие с ним (симуляция наступления времени), если таймер очень долгий (> 30 сек).
Для взаимодействия с таймерами разработана функция pass_timer():
def pass_timer(
self,
*,
process_instance: Union[ProcessInstance, CamundaProcessInstance],
timer_id: Optional[str] = None,
job_type: Optional[str] = None,
) -> None:
if isinstance(process_instance, ProcessInstance):
process_instance_id = process_instance.processInstanceId
elif isinstance(process_instance, CamundaHistoricProcessInstance):
process_instance_id = process_instance.id
else:
process_instance_id = process_instance.id
timer_id = timer_id or self.TIMER_ID
job_type = job_type or self.TIMER_JOB_TYPE
with allure.step(
f'Проброс таймера с параметрами timer_id={timer_id} '
f'и job_type={job_type} для процесса "{self.pretty_name}" '
f'с id "{process_instance_id}"',
):
timer_job = self.get_timer(
process_instance=process_instance,
timer_id=timer_id,
job_type=job_type,
)
self.camunda.execute_job_by_id(timer_job.id)
time.sleep(self.TIMER_AWAIT_TIME)В качестве аргументов необходимо передать:
id процесса;
timer_id (id активити по схеме в camunda);
job_type (по дефолту используется ‘timer-intermediate-transition’).
Так как таймер является весьма специфичным активити, то была разработана отдельная функция get_timer(), который, помимо стандартных process_instance и timer_id (по сути activity_id), есть еще и job_type, который может варьироваться от типа реализации таймера
(timer-intermediate-transition или timer-transition):
def get_timer(
self,
*,
process_instance: Union[ProcessInstance, CamundaProcessInstance],
timer_id: Optional[str] = None,
job_type: Optional[str] = None,
) -> Optional[CamundaJob]:
if isinstance(process_instance, ProcessInstance):
process_instance_id = process_instance.processInstanceId
process_definition_id = process_instance.processDefinitionId
elif isinstance(process_instance, CamundaHistoricProcessInstance):
process_instance_id = process_instance.id
process_definition_id = process_instance.processDefinitionId
else:
process_instance_id = process_instance.id
process_definition_id = process_instance.definitionId
timer_id = timer_id or self.TIMER_ID
job_type = job_type or self.TIMER_JOB_TYPE
with allure.step(
f'Получение таймера с параметрами timer_id={timer_id} '
f'и job_type={job_type} для процесса "{self.pretty_name}" '
f'с id "{process_instance_id}"',
):
job_definition = self.camunda.get_job_definitions(
params={
'activityIdIn': timer_id,
'processDefinitionId': process_definition_id,
'jobType': job_type,
},
)[0]
timer_job = self.camunda.get_jobs(
params={
'jobDefinitionId': job_definition.id,
'processInstanceId': process_instance_id,
},
)
return timer_job[0] if timer_job else NoneКак нашли таймер, то исполняем его execute_job_by_id() — внутри зашит camunda-метод ‘/job/{job_id}/execute'.
7. Идентификация завершенного БП по camunda, удаление инстансов процессов до и после автотестирования
Важным шагом в тестировании БП является проверка, что процесс завершился без ошибок. Для этого разработана функция wait_completed_process():
def wait_completed_process(
self,
process_instance_id: str,
*,
max_time: int = 30,
sleep_time: int = 2,
) -> None:
with allure.step(
f'Проверка того, что процесс "{self.pretty_name}" '
f'с id "{process_instance_id}" завершен',
):
poll_by_time_first_success(
fn=lambda: self.__get_process_instance_with_incidents_check(
process_instance_id,
),
checker=lambda p: p is None,
max_time=max_time,
sleep_time=sleep_time,
msg=(
f'Процесс "{self.pretty_name}" с id "{process_instance_id}" '
f'не был завершен за max_time = {max_time} секунд'
),
)Используется поллинг по методу __get_process_instance_with_incidents_check() проверяется:
что в процессе с указанным process_instance_id нет инцидентов (внутри используется GET метод /incident);
что процесс с указанным process_instance_id исчез из списка активных процессов, что является показателем успешного завершения процесса в camunda.
Если определяется, что в процессе обнаружился инцидент, то выбрасывается RuntimeError. Найденный инцидент добавляется в отчет о тестировании, чтобы автоматизатор смог сразу увидеть причину падения теста. На рисунке 4 представлен фрагмент отчета с инцидентом процесса:
Иногда требуется проверить, что процесс в определенный промежуток времени НЕ завершился. Для этого используем обратный механизм, что при поллинге поиска активных процессов — проверяется, что процесс возвращается из camunda на протяжении необходимого времени.
Часто требуется перед тестом (setup) или после него (teardown) удалить активный процесс (классический пример — процесс выпал в инцидент и «висит» в camunda). Для удаления активных процессов разработана функция delete_process_instance_by_business_key() (удаление по business_key):
def delete_process_instance_by_business_key(
self,
business_key: str,
) -> None:
with allure.step(
f'Удаление процесса "{self.pretty_name}" '
f'с бизнес-ключом "{business_key}"',
):
self.camunda.delete_process_instance(
process_instance_id=self.get_process_instance_by_business_key(
business_key,
).id,
)В delete_process_instance() используется DELETE метод /process-instance/{process_instance_id}.
Есть аналогичные функции с параметрами, отличные от business_key. Также имеется в арсенале функция по удалению всех активных процессов внутри определенного process_definition.
8. Заключение
Все больше и больше команд в нашей компании подключаются к тестированию БП. Часть из них используют тот же самый подход, что и мы, в том числе через использование модулей, которые рассмотрели в данной статье. Надеюсь, что кому‑то материал поможет и даст отправную точку, чтобы начать тестирование БП через camunda. По крайней мере 5 лет назад, когда мы только начинали выстраивать автотестирование БП (не e2e), подобных материалов с подходами в тестировании camunda не встречали.
Конечно, это не полноценный how‑to или туториал от начала и до конца, как выстроить процесс автотестирования бизнес‑процессов, а лишь рассмотрение базовых возможностей в наших процессах. В наших модулях множество функций и описать их все в одной статье проблематично, да и смысла в этом не особо много. Очень много специфики в покрываемых нами автотестами БП.
Спасибо всем, кто дочитал статью до конца. Если остались вопросы, пишите их в комментариях — с радостью ответим! До новых встреч!