Automation of business process testing via camunda
1. Introduction
Hello everyone! My name is Renat Dasayev and in the previous article Automation of E2E testing of end-to-end BP of integration projects of the Operational Block it was told about how e2e-auto testing is arranged. Today I want to tell about how it is used camunda in automated testing of business processes (hereinafter BP). Using practical examples, we will consider what and how we do in our tests.
If you are interested in learning about the camunda engine and how it is used in the Moscow Exchange BP using the example of the TsUP project, I recommend that you read our colleague’s material.
2 Checking the power supply in camunda with autotests
Before we dive into the technical part of the issue, let's determine what checks we need to perform during the BP.
Standard set of checks:
the process starts and completes successfully;
gateway, boundary events and other elements move the BP in the right direction;
the context is correct at the steps being checked – the necessary process variables are present;
correct result of calculations in events/gateway expressions;
service tasks are successfully executed and user tasks are created;
at the steps where there is interaction with external services, they work successfully (in e2e there are no stubs, we work with real integrations);
not only the “success path”, but also other routes along the scheme, including where errors are handled in the process;
the process does not fall into an incident.
Schemes can call subprocesses with their own context. In this case, the child processes are checked along with the parent.
3 Development of modules for working with bpmn processes
In the previous article, it was mentioned that together with another team we are developing python modules for working with microservices, as well as modules (clients) for working with services/engines.
Several modules have been developed for working with bpmn processes:
moex-pmh-bpmn (working with bpmn processes);
moex-pmh-camunda-client (working with camunda rest api);
moex-pmh-
(work with a specific BP instance, 1 such module for each business process).
Module hierarchy:
moex-pmh-
The modules are developed, placed in the corporate pypi repository, and any team in our company can use them in their project. At the same time, teams also have the opportunity to propose changes to the code base (new features, bug fixes) via merge request.
At the moment the module moex-pmh-camunda-client can work with the following controllers:
Controllers are added as needed. In each method of any controller, we get a dataclass object with all the fields that come in the response. Below is an example of using the capabilities of this module in solving practical problems of BP autotesting.
4 Starting a business process. Finding a running process in camunda.
In any BP, the camunda engine is accessible via http – https://
The first thing that begins the automatic testing of the BP is its start. In most cases, the process is started when a certain type of event occurs:
sending a message to a bus, such as esb (enterprise service bus);
creating an object in the service database;
the onset of a specific date and/or time;
receiving emails, files, etc.
After the BP has started, it is necessary to identify this start and find the running process in the list of active ones. As a rule, it can be found by unique process variables, such as a business key (business_key), a trading instrument code (securityid), or other variables inherent in specific BPs. In rare cases, the process is identified by the date and time of launch, if there are no unique process variables.
In most cases, it is enough for an automated testing engineer to simply manually find the running process in its camunda, based on its launch time. Then the engineer analyzes the process variables and finds unique ones, by which it is possible to accurately identify the running instance of the process.
For clarity, let's take a real BP “Activation of security redemption” (part of one large process for Registration, activation and deactivation of redemption). This is one of the smallest schemes that we managed to find. In 99% of cases, the schemes are much larger in volume and interaction with external resources. Figure 1 shows a screenshot of this BP from camunda:
In our example, the process search will be performed by variable securityId (trading code of the instrument). In Figure 2 you can see a running instance of the “Buyout Activation” process with a list of process variables, among which is the required variable:
Below is a Python code that implements searching among active processes by variable securityId:
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,
)Attempts are being made during max_time (120 seconds by default) with a step of 2 seconds (by default) in a loop (polling) find the process in which the required variable was found (get_process_variable_instances) and its meaning:
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 processIf the process is found, then the dataclass object is returned. CamundaProcessInstance, or None.
Example CamundaProcessInstance object:
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'>If during the polling process the process is not detected within the time period max_timethen an exception is thrown TimeoutError and the autotest ends at this point.
5 Monitoring activities according to bpmn-scheme
An important function is to search for activities by camunda scheme. These can be different steps:
certain calculations in gateway/boundary/events;
service-tasks;
user tasks;
interaction with databases/buses/mailboxes/sites, etc.
Using our BP for activating the buyback as an example, we will try to determine that a user task “Complete reconciliation of the EKBD with ASTS” has been created (EKBD – Unified Client Database, ASTS – Automated Securities Trading System).
First, we need to calculate the id of this step on the diagram in camunda. Let's load the bpmn file of this BP into Camunda Modeler and select the required step (see Figure 3):
In the list of attributes we find id = compare-offers-manual. This will be the identifier that needs to be found in the list of activities of our tested BP.
To search for activity, we use a function with polling (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}"'
),
)In which at each iteration we get a list of activities (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 activityInside find_process_activity() is called get_process_activities() (camunda api is used internally – /process-instance/{process_instance_id}/activity-instances) and we are looking for our activity among them 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_activitiesAccordingly, it is necessary to pass in arguments wait_for_process_activity only:
process_instance_id=
(calculated from the process start search function); activity_id=' compare-offers-manual' (or activity_name).
Once the process reaches this step at the specified time max_time (usually 60 seconds), then polling will catch it, perform a check and, if everything is successful, pass control to the next function in the test. Otherwise, we get TimeoutError.
After the test has found activity with activity_id='compare-offers-manual' in camunda, it is necessary to calculate the user task id (user-tasks), so that later with this identifier you can find the task in the Task Manager of the TsUP platform.
To calculate the id of a user task, a function was developed in 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 taskAs in searching for the required activity, it is necessary to pass the process and activity identifiers. The function is used inside polling 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 NoneThe POST method is used inside the function. /task. If successful (task was found in the required process and with the required activity), a list of dataclass objects is returned. CamundaTask with all fields:
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=NoneWhat interests us from this object is id='00977769-44c4-11ef-b669-b2d4f57bb43f' this is the task identifier, which will be the same in the MCC. Next, you will need to send a request to receive active tasks in the MCC in the Task Manager, find it in the output and then perform the necessary actions on this task (“Execute” or “Reject”).
If you need to identify the passage of a certain step in the scheme, but this step skips very quickly, it is enough to set a low value in the polling in sleep_time (delay before the next iteration), for example, 0.3 seconds.
If polling by active processes does not allow you to detect it in time, then you should use a search in already completed processes.
To obtain information about a process that has already been completed, it is sufficient to transmit process_instance_id into function 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 NoneIf you need to find some activity in an already completed process, you can use 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]and find the one you need in the list of all activities.
If you need to find a variable in a completed process, you can use 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 Interacting with timers.
There are power supplies where timers are programmed into the circuit. With the processes that we dealt with, the timer, as a rule, was triggered by the following events:
To check that the BP has stopped on the timer also according to camunda modeler, we find the id of this step and look for this identifier in the list of activities by process (more details in the previous chapter). It often happens that it is necessary not only to find an active timer, but also to interact with it (simulate the onset of time), if the timer is very long (> 30 sec).
A function has been developed to interact with timers 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)The following must be passed as arguments:
Since the timer is a very specific activity, a separate function was developed get_timer(), which, in addition to the standard process_instance And timer_id (in fact activity_id), there is also job_typewhich may vary depending on the type of timer implementation
(timer-intermediate-transition or 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 NoneOnce we find the timer, we execute it. execute_job_by_id() – the camunda method is built inside '/job/{job_id}/execute'.
7 Identification of completed BP by camunda, deletion of process instances before and after autotesting
An important step in testing a BP is to check that the process has completed without errors. The function 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} секунд'
),
)Polling is used according to the __ methodget_process_instance_with_incidents_check() is checked:
what is in the process with the specified process_instance_id no incidents (GET method is used internally) /incident);
that the process with the specified process_instance_id disappeared from the list of active processes, which indicates that the process has successfully completed in camunda.
If it is determined that an incident has been detected in the process, then it is thrown out RuntimeError. The detected incident is added to the test report so that the automation engineer can immediately see the reason for the test failure. Figure 4 shows a fragment of the report with a process incident:
Sometimes you need to check that the process is a certain interval time NOT completed. For this we use the reverse mechanism, that when polling the search for active processes – it is checked that the process returns from camunda during the required time.
It is often necessary to remove an active process before a test (setup) or after it (teardown) (a classic example is when a process has fallen into an incident and is “hanging” in camunda). To remove active processes, a function has been developed delete_process_instance_by_business_key() (delete by 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,
)IN delete_process_instance() DELETE method is used /process-instance/{process_instance_id}.
There are similar functions with parameters different from business_key. There is also a function in the arsenal to delete all active processes within a certain process_definition.
8 Conclusion
More and more teams in our company are joining the BP testing. Some of them use the same approach as we do, including using the modules that we have discussed in this article. I hope that this material will help someone and give them a starting point to start testing BP via camunda. At least 5 years ago, when we were just starting to build automated BP testing (not e2e), we had not come across similar materials with approaches to testing camunda.
Of course, this is not a full-fledged how-to or tutorial from start to finish on how to build a process of automated testing of business processes, but only a review of the basic capabilities in our processes. Our modules have many functions and describing them all in one article is problematic, and there is not much point in this. There are a lot of specifics in the BPs covered by our automated tests.
Thanks to everyone who read the article to the end. If you have any questions, write them in the comments – we will be happy to answer! Until next time!
