This commit is contained in:
burnettk 2024-07-03 16:27:59 -04:00
parent d954a5b8ef
commit 477ad2134e
No known key found for this signature in database
1 changed files with 12 additions and 9 deletions

View File

@ -1,10 +1,13 @@
import glob
from lxml import etree
def detect_multiple_incoming_flows(bpmn_file):
# Parse the BPMN file
try:
tree = etree.parse(bpmn_file)
# actually use SpecFileService.get_etree_from_xml_bytes if we use this in future
tree = etree.parse(bpmn_file) # noqa: S320
except etree.XMLSyntaxError as e:
print(f"Error parsing {bpmn_file}: {e}")
return []
@ -12,17 +15,17 @@ def detect_multiple_incoming_flows(bpmn_file):
root = tree.getroot()
# Namespace dictionary for finding elements
ns = {'bpmn': 'http://www.omg.org/spec/BPMN/20100524/MODEL'}
ns = {"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL"}
# Find all sequence flows
sequence_flows = root.findall('.//bpmn:sequenceFlow', namespaces=ns)
sequence_flows = root.findall(".//bpmn:sequenceFlow", namespaces=ns)
# Dictionary to keep track of incoming flows for each target
incoming_flows = {}
# Iterate through sequence flows and count incoming flows for each target
for flow in sequence_flows:
target_ref = flow.get('targetRef')
target_ref = flow.get("targetRef")
if target_ref in incoming_flows:
incoming_flows[target_ref] += 1
else:
@ -36,19 +39,18 @@ def detect_multiple_incoming_flows(bpmn_file):
task_elements = root.findall(f".//*[@id='{element}']", namespaces=ns)
if task_elements:
task_element = task_elements[0]
if not any(tag in task_element.tag for tag in ['Gateway', 'Event']):
if not any(tag in task_element.tag for tag in ["Gateway", "Event"]):
tasks_with_multiple_incoming.append(element)
return tasks_with_multiple_incoming
def validate_bpmn_files():
# Hardcoded glob pattern for BPMN files
glob_pattern = '**/*.bpmn' # Adjust the path as needed
glob_pattern = "**/*.bpmn"
# Find all BPMN files matching the glob pattern
bpmn_files = glob.glob(glob_pattern, recursive=True)
for bpmn_file in bpmn_files:
# print(f"Validating {bpmn_file}...")
tasks_with_issues = detect_multiple_incoming_flows(bpmn_file)
if tasks_with_issues:
@ -56,5 +58,6 @@ def validate_bpmn_files():
for task in tasks_with_issues:
print(f" - {task}")
# Run the validation
validate_bpmn_files()