From 01fed4758d746b909b22d801339dca29f9648d49 Mon Sep 17 00:00:00 2001 From: Elizabeth Esswein Date: Mon, 19 Dec 2022 11:33:48 -0500 Subject: [PATCH] add integration tests for process nav endpoints --- .../process_navigation.bpmn | 137 +++++++++++++++++ .../integration/test_process_api.py | 140 ++++++++++++++++++ 2 files changed, 277 insertions(+) create mode 100644 spiffworkflow-backend/tests/data/process_navigation/process_navigation.bpmn diff --git a/spiffworkflow-backend/tests/data/process_navigation/process_navigation.bpmn b/spiffworkflow-backend/tests/data/process_navigation/process_navigation.bpmn new file mode 100644 index 00000000..9f2f26bf --- /dev/null +++ b/spiffworkflow-backend/tests/data/process_navigation/process_navigation.bpmn @@ -0,0 +1,137 @@ + + + + + + + + Flow_1l15rbh + + + + Flow_1l15rbh + Flow_0d35i06 + Flow_0tzaigt + Flow_1vld4r2 + + + + Flow_0d35i06 + Flow_1w3n49n + + + + Flow_0tzaigt + Flow_1q47ol8 + + + + + + + Flow_1q47ol8 + + + + + Flow_1w3n49n + + + + Flow_1vld4r2 + Flow_13ai5vv + + timedelta(hours=1) + + + + + Click the button. + + Flow_13ai5vv + Flow_1vwnf3n + + + Flow_1vwnf3n + + + + + result + + + + + result + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_process_api.py b/spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_process_api.py index 3bc21456..14508285 100644 --- a/spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_process_api.py +++ b/spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_process_api.py @@ -2497,6 +2497,146 @@ class TestProcessApi(BaseTest): print("test_script_unit_test_run") + def test_send_event( + self, + app: Flask, + client: FlaskClient, + with_db_and_bpmn_file_cleanup: None, + with_super_admin_user: UserModel, + ) -> None: + """Test_script_unit_test_run.""" + process_group_id = "test_group" + process_model_id = "process_navigation" + bpmn_file_name = "process_navigation.bpmn" + bpmn_file_location = "process_navigation" + process_model_identifier = self.create_group_and_model_with_bpmn( + client=client, + user=with_super_admin_user, + process_group_id=process_group_id, + process_model_id=process_model_id, + bpmn_file_name=bpmn_file_name, + bpmn_file_location=bpmn_file_location, + ) + + bpmn_file_data_bytes = self.get_test_data_file_contents( + bpmn_file_name, bpmn_file_location + ) + self.create_spec_file( + client=client, + process_model_id=process_model_identifier, + process_model_location=process_model_identifier, + file_name=bpmn_file_name, + file_data=bpmn_file_data_bytes, + user=with_super_admin_user, + ) + + headers = self.logged_in_headers(with_super_admin_user) + response = self.create_process_instance_from_process_model_id( + client, process_model_identifier, headers + ) + process_instance_id = response.json["id"] + + client.post( + f"/v1.0/process-instances/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}/run", + headers=self.logged_in_headers(with_super_admin_user), + ) + + data = { + "correlation_properties": [], + "expression": None, + "external": True, + "internal": False, + "payload": {"message": "message 1"}, + "name": "Message 1", + "typename": "MessageEventDefinition", + } + response = client.post( + f"/v1.0/send-event/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}", + headers=self.logged_in_headers(with_super_admin_user), + content_type="application/json", + data=json.dumps(data), + ) + assert response.json["status"] == "complete" + + response = client.get( + f"/v1.0/task-data/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}?all_tasks=true", + headers=self.logged_in_headers(with_super_admin_user), + ) + end = next(task for task in response.json if task["name"] == "End") + assert end["data"]["result"] == {"message": "message 1"} + + def test_manual_complete_task( + self, + app: Flask, + client: FlaskClient, + with_db_and_bpmn_file_cleanup: None, + with_super_admin_user: UserModel, + ) -> None: + """Test_script_unit_test_run.""" + process_group_id = "test_group" + process_model_id = "process_navigation" + bpmn_file_name = "process_navigation.bpmn" + bpmn_file_location = "process_navigation" + process_model_identifier = self.create_group_and_model_with_bpmn( + client=client, + user=with_super_admin_user, + process_group_id=process_group_id, + process_model_id=process_model_id, + bpmn_file_name=bpmn_file_name, + bpmn_file_location=bpmn_file_location, + ) + + bpmn_file_data_bytes = self.get_test_data_file_contents( + bpmn_file_name, bpmn_file_location + ) + self.create_spec_file( + client=client, + process_model_id=process_model_identifier, + process_model_location=process_model_identifier, + file_name=bpmn_file_name, + file_data=bpmn_file_data_bytes, + user=with_super_admin_user, + ) + + headers = self.logged_in_headers(with_super_admin_user) + response = self.create_process_instance_from_process_model_id( + client, process_model_identifier, headers + ) + process_instance_id = response.json["id"] + + client.post( + f"/v1.0/process-instances/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}/run", + headers=self.logged_in_headers(with_super_admin_user), + ) + + data = { + "dateTime": "timedelta(hours=1)", + "external": True, + "internal": True, + "label": "Event_0e4owa3", + "typename": "TimerEventDefinition", + } + response = client.post( + f"/v1.0/send-event/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}", + headers=self.logged_in_headers(with_super_admin_user), + content_type="application/json", + data=json.dumps(data), + ) + + response = client.get( + f"/v1.0/task-data/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}", + headers=self.logged_in_headers(with_super_admin_user), + ) + assert len(response.json) == 1 + task = response.json[0] + + response = client.post( + f"/v1.0/task-complete/{self.modify_process_identifier_for_path_param(process_model_identifier)}/{process_instance_id}/{task['id']}", + headers=self.logged_in_headers(with_super_admin_user), + content_type="application/json", + ) + assert response.json["status"] == "complete" + def setup_initial_groups_for_move_tests( self, client: FlaskClient, with_super_admin_user: UserModel ) -> None: