✅ new tests for bpmn export

Author: Charizhardt

src/tests/test_bpmn.py

@@ -0,0 +1,145 @@
+import re
+import xml.etree.ElementTree as ET
+
+from processpiper import ProcessMap, EventType, ActivityType, GatewayType
+from util_test import get_test_file_path, get_solution_file_path, mock_uuid
+
+namespace = {'bpmn': 'http://www.omg.org/spec/BPMN/20100524/MODEL'}
+
+
+def test_case03(mock_uuid):
+    with ProcessMap(
+            "Sample Test Process", colour_theme="BLUEMOUNTAIN"
+    ) as my_process_map:
+        with my_process_map.add_lane("End User") as lane1:
+            start = lane1.add_element("Start", EventType.START)
+            enter_keyword = lane1.add_element("Enter Keyword", ActivityType.TASK)
+
+        with my_process_map.add_pool("System Search") as pool1:
+            with pool1.add_lane("Database System") as lane2:
+                login = lane2.add_element("Login", ActivityType.TASK)
+                search_records = lane2.add_element("Search Records", ActivityType.TASK)
+                result_found = lane2.add_element("Result Found?", GatewayType.EXCLUSIVE)
+                display_result = lane2.add_element("Display Result", ActivityType.TASK)
+                logout = lane2.add_element("Logout", ActivityType.TASK)
+                end = lane2.add_element("End", EventType.END)
+
+            with pool1.add_lane("Log System") as lane3:
+                log_error = lane3.add_element("Log Error", ActivityType.TASK)
+
+        start.connect(login, "User \nAuthenticates").connect(
+            enter_keyword, "Authenticated"
+        ).connect(search_records, "Search Criteria")
+        search_records.connect(result_found, "Result").connect(display_result, "Yes")
+        display_result.connect(logout).connect(end)
+        result_found.connect(log_error, "No").connect(display_result)
+
+        my_process_map.set_footer("Generated by ProcessPiper")
+        my_process_map.draw()
+
+        # Define paths for the generated BPMN file and the solution BPMN file
+        generated_bpmn_path = get_test_file_path("test_bpmn_01.bpmn")
+        solution_bpmn_path = get_solution_file_path("test_bpmn_01.bpmn")
+
+        # Save the BPMN file and the image
+        my_process_map.save(get_test_file_path("test_bpmn_01.png"))
+        my_process_map.export_to_bpmn(generated_bpmn_path)
+
+        # read bpmn file as xml and check if it is valid
+        with open(generated_bpmn_path, "r") as generated_bpmn:
+            generated_xml = generated_bpmn.read()
+
+            tree = ET.ElementTree(ET.fromstring(generated_xml))
+            root = tree.getroot()
+
+            # Does every start event have at least one outgoing element?
+            start_events = root.findall(".//bpmn:startEvent", namespace)
+            for start_event in start_events:
+                outgoing_count, incoming_count = _count_outgoing_incoming(start_event)
+                assert outgoing_count >= 1, "Start event does not have outgoing element"
+                assert incoming_count == 0, "Start event has incoming element"
+
+            # Does every end event have at least one incoming element?
+            end_events = root.findall(".//bpmn:endEvent", namespace)
+            for end_event in end_events:
+                outgoing_count, incoming_count = _count_outgoing_incoming(end_event)
+                assert outgoing_count == 0, "End event has outgoing element"
+                assert incoming_count >= 1, "End event does not have incoming element"
+
+            # Does intermediate event have at least one outgoing and incoming element?
+            intermediate_events = root.findall(".//bpmn:intermediateCatchEvent", namespace)
+            for intermediate_event in intermediate_events:
+                outgoing_count, incoming_count = _count_outgoing_incoming(intermediate_event)
+                assert outgoing_count >= 1, "Intermediate event does not have outgoing element"
+                assert incoming_count >= 1, "Intermediate event does not have incoming element"
+
+            # Does every task have at least one incoming and outgoing element?
+            tasks = root.findall(".//bpmn:task", namespace)
+            for task in tasks:
+                outgoing_count, incoming_count = _count_outgoing_incoming(task)
+                assert outgoing_count >= 1, "Task does not have outgoing element"
+                assert incoming_count >= 1, "Task does not have incoming element"
+
+            # Does every sub-process have at least one incoming and outgoing element?
+            sub_processes = root.findall(".//bpmn:subProcess", namespace)
+            for sub_process in sub_processes:
+                outgoing_count, incoming_count = _count_outgoing_incoming(sub_process)
+                assert outgoing_count >= 1, "Sub-process does not have outgoing element"
+                assert incoming_count >= 1, "Sub-process does not have incoming element"
+
+            # Does every service task have at least one incoming and outgoing element?
+            service_tasks = root.findall(".//bpmn:serviceTask", namespace)
+            for service_task in service_tasks:
+                outgoing_count, incoming_count = _count_outgoing_incoming(service_task)
+                assert outgoing_count >= 1, "Service task does not have outgoing element"
+                assert incoming_count >= 1, "Service task does not have incoming element"
+
+            # Does every exclusive gateway have at least one incoming and outgoing element?
+            exclusive_gateways = root.findall(".//bpmn:exclusiveGateway", namespace)
+            for exclusive_gateway in exclusive_gateways:
+                outgoing_count, incoming_count = _count_outgoing_incoming(exclusive_gateway)
+                assert outgoing_count >= 1, "Exclusive gateway does not have outgoing element"
+                assert incoming_count >= 1, "Exclusive gateway does not have incoming element"
+
+            # Does every inclusive gateway have at least one incoming and outgoing element?
+            inclusive_gateways = root.findall(".//bpmn:inclusiveGateway", namespace)
+            for inclusive_gateway in inclusive_gateways:
+                outgoing_count, incoming_count = _count_outgoing_incoming(inclusive_gateway)
+                assert outgoing_count >= 1, "Inclusive gateway does not have outgoing element"
+                assert incoming_count >= 1, "Inclusive gateway does not have incoming element"
+
+            # Does every parallel gateway have at least one incoming and outgoing element?
+            parallel_gateways = root.findall(".//bpmn:parallelGateway", namespace)
+            for parallel_gateway in parallel_gateways:
+                outgoing_count, incoming_count = _count_outgoing_incoming(parallel_gateway)
+                assert outgoing_count >= 1, "Parallel gateway does not have outgoing element"
+                assert incoming_count >= 1, "Parallel gateway does not have incoming element"
+
+            # Remove following attributes as they are always different:
+            # - sourceRef
+            # - targetRef
+            # - id
+            generated_xml = re.sub(r'sourceRef=".*?"', '', generated_xml)
+            generated_xml = re.sub(r'targetRef=".*?"', '', generated_xml)
+            generated_xml = re.sub(r'id=".*?"', '', generated_xml)
+            generated_bpmn.close()
+
+        # Remove following attributes as they are always different:
+        # - sourceRef
+        # - targetRef
+        # - id
+        with open(solution_bpmn_path, "r") as solution_bpmn:
+            solution_xml = solution_bpmn.read()
+            solution_xml = re.sub(r'sourceRef=".*?"', '', solution_xml)
+            solution_xml = re.sub(r'targetRef=".*?"', '', solution_xml)
+            solution_xml = re.sub(r'id=".*?"', '', solution_xml)
+            solution_bpmn.close()
+
+        # Compare the generated BPMN file with the solution BPMN file
+        assert generated_xml == solution_xml, "Generated BPMN file does not match the solution BPMN file"
+
+
+def _count_outgoing_incoming(event):
+    outgoing_count = len(event.findall("bpmn:outgoing", namespace))
+    incoming_count = len(event.findall("bpmn:incoming", namespace))
+    return outgoing_count, incoming_count