Автоматизация тестирования бизнес-процессов через camunda

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 None

6. Взаимодействие с таймерами

Существуют БП, где в схеме запрограммированы таймеры. С теми процессами, что имели дело, таймер, как правило, срабатывал на следующие события:

• наступление определенной даты и времени;

• поступление сигнала от внешней системы.

Чтобы проверить, что БП встал на таймере также по 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 или туториал от начала и до конца, как выстроить процесс автотестирования бизнес‑процессов, а лишь рассмотрение базовых возможностей в наших процессах. В наших модулях множество функций и описать их все в одной статье проблематично, да и смысла в этом не особо много. Очень много специфики в покрываемых нами автотестами БП.

Спасибо всем, кто дочитал статью до конца. Если остались вопросы, пишите их в комментариях — с радостью ответим! До новых встреч!