"""Is a document valid BPMN 2.0? Against the OMG's own schema (tests/fixtures/bpmn20: BPMN20.xsd and its imports,
fetched once from https://www.omg.org/spec/BPMN/20100501/ — no network at test time), plus what a schema cannot
say: every sequence flow joins two flow nodes, every lane names flow nodes, every flow node has exactly one
diagram shape and every flow one edge, and every node is reachable from the start event."""
import pathlib
import xml.etree.ElementTree as ET
import xmlschema
from brick_hr.domain.bpmn import NS
XSD = pathlib.Path(__file__).parent / "fixtures" / "bpmn20" / "BPMN20.xsd"
_SCHEMA = None
FLOW_NODES = ("startEvent", "endEvent", "userTask", "serviceTask", "exclusiveGateway", "parallelGateway",
"intermediateCatchEvent")
def schema() -> xmlschema.XMLSchema:
global _SCHEMA
if _SCHEMA is None:
_SCHEMA = xmlschema.XMLSchema(str(XSD))
return _SCHEMA
def q(tag: str) -> str:
prefix, local = tag.split(":")
return f"{{{NS[prefix]}}}{local}"
def check(xml: str) -> ET.Element:
"""Raise AssertionError naming the first problem; return the parsed root."""
errors = [str(e).splitlines()[0] for e in schema().iter_errors(xml)]
assert not errors, errors[:3]
root = ET.fromstring(xml)
assert root.tag == q("bpmn:definitions")
ids = [e.get("id") for e in root.iter() if e.get("id")]
assert len(ids) == len(set(ids)), "duplicate ids"
proc = root.find(q("bpmn:process"))
nodes = {e.get("id"): e for e in proc if e.tag in {q(f"bpmn:{t}") for t in FLOW_NODES}}
flows = proc.findall(q("bpmn:sequenceFlow"))
for f in flows:
assert f.get("sourceRef") in nodes and f.get("targetRef") in nodes, f.get("id")
for lane in proc.iter(q("bpmn:lane")):
for ref in lane.findall(q("bpmn:flowNodeRef")):
assert ref.text in nodes, ref.text
in_lanes = [r.text for r in proc.iter(q("bpmn:flowNodeRef"))]
assert sorted(in_lanes) == sorted(nodes), "every flow node sits in exactly one lane"
plane = root.find(f"{q('bpmndi:BPMNDiagram')}/{q('bpmndi:BPMNPlane')}")
shapes = [s.get("bpmnElement") for s in plane.findall(q("bpmndi:BPMNShape"))]
edges = [s.get("bpmnElement") for s in plane.findall(q("bpmndi:BPMNEdge"))]
for n in nodes:
assert shapes.count(n) == 1, f"{n}: {shapes.count(n)} shapes"
for f in flows:
assert edges.count(f.get("id")) == 1, f.get("id")
for mf in root.iter(q("bpmn:messageFlow")):
assert edges.count(mf.get("id")) == 1, mf.get("id")
starts = [i for i, e in nodes.items() if e.tag == q("bpmn:startEvent")]
assert len(starts) == 1
succ: dict = {}
for f in flows:
succ.setdefault(f.get("sourceRef"), []).append(f.get("targetRef"))
seen, todo = set(), list(starts)
while todo:
n = todo.pop()
if n not in seen:
seen.add(n)
todo += succ.get(n, [])
assert seen == set(nodes), f"not reachable from the start: {sorted(set(nodes) - seen)}"
return root
def upstream_steps(root: ET.Element) -> dict:
"""For each step's own task (the element carrying dodil:step = its key), the keys of the steps whose tasks lie on
a path from the start to it."""
proc = root.find(q("bpmn:process"))
pred: dict = {}
for f in proc.findall(q("bpmn:sequenceFlow")):
pred.setdefault(f.get("targetRef"), []).append(f.get("sourceRef"))
attr = f"{{{NS['dodil']}}}step"
key = {e.get("id"): e.get(attr) for e in proc if e.get(attr)}
out = {}
for i, k in key.items():
seen, todo = set(), list(pred.get(i, []))
while todo:
x = todo.pop()
if x not in seen:
seen.add(x)
todo += pred.get(x, [])
out[k] = {key[x] for x in seen if x in key}
return out