from __future__ import annotations
from collections import Counter
import unittest
from pathlib import Path
from govoplan_workflow_engine.backend.bpmn import (
BPMN_MODEL_NAMESPACE,
BpmnInspectionError,
inspect_bpmn_xml,
parse_bpmn_xml,
)
from govoplan_workflow_engine.backend.bpmn_adapters import (
BpmnAdapterError,
INTERCHANGE_ADAPTER_ID,
NATIVE_LINEAR_ADAPTER_ID,
bpmn_adapter_registry,
compile_bpmn_to_graph,
)
from govoplan_workflow_engine.backend.bpmn_graph import (
NATIVE_BPMN_ADAPTER_ID,
export_bpmn_graph,
import_bpmn_graph,
)
BPMN = """
"""
NATIVE_BPMN = """
"""
class BpmnInspectionTests(unittest.TestCase):
def test_notation_fixtures_are_safe_and_fully_inventoried(self) -> None:
fixture_directory = Path(__file__).parent / "fixtures" / "bpmn"
results = {
path.stem: inspect_bpmn_xml(path.read_text(encoding="utf-8"))
for path in sorted(fixture_directory.glob("*.bpmn"))
}
self.assertEqual(
{
"choreography",
"collaboration",
"control-flow",
"data",
"events-transaction-compensation",
"process",
},
set(results),
)
self.assertEqual(1, results["choreography"].choreography_count)
self.assertEqual(1, results["collaboration"].collaboration_count)
self.assertEqual(
1,
results["events-transaction-compensation"].element_counts[
"transaction"
],
)
self.assertEqual(
1,
results["data"].element_counts["dataStoreReference"],
)
self.assertEqual(
1,
results["control-flow"].element_counts["exclusiveGateway"],
)
self.assertEqual(
1,
results["control-flow"].element_counts["subProcess"],
)
self.assertEqual(
1,
results["control-flow"].element_counts["callActivity"],
)
self.assertEqual(
1,
results["control-flow"].element_counts[
"compensateEventDefinition"
],
)
self.assertEqual(
2,
results["control-flow"].element_counts["signalEventDefinition"],
)
def test_inventory_classifies_native_and_interchange_elements(self) -> None:
result = inspect_bpmn_xml(BPMN)
self.assertTrue(result.valid_xml)
self.assertEqual(1, result.process_count)
self.assertEqual(1, result.executable_process_count)
self.assertEqual(1, result.collaboration_count)
self.assertEqual(3, result.element_counts["sequenceFlow"])
review = next(
item for item in result.elements if item.element_id == "Review_1"
)
collaboration = next(
item
for item in result.elements
if item.element_id == "Collaboration_1"
)
self.assertEqual("native_execution", review.support_level)
self.assertEqual("native_mapping", collaboration.support_level)
def test_dangling_references_are_reported(self) -> None:
result = inspect_bpmn_xml(
BPMN.replace('targetRef="End_1"', 'targetRef="Missing_1"')
)
self.assertFalse(result.valid_xml)
self.assertTrue(
any(
item.code == "dangling_bpmn_reference"
for item in result.diagnostics
)
)
def test_entities_are_rejected(self) -> None:
unsafe = """]>
&xxe;"""
with self.assertRaisesRegex(
BpmnInspectionError,
"not safe and well formed",
):
inspect_bpmn_xml(unsafe)
def test_non_bpmn_root_is_rejected(self) -> None:
with self.assertRaisesRegex(
BpmnInspectionError,
"bpmn:definitions",
):
inspect_bpmn_xml("")
def test_profiles_are_versioned_and_native_compilation_is_stable(self) -> None:
profiles = {
item.id: item for item in bpmn_adapter_registry().profiles()
}
self.assertFalse(profiles[INTERCHANGE_ADAPTER_ID].executable)
self.assertTrue(profiles[NATIVE_LINEAR_ADAPTER_ID].executable)
self.assertTrue(profiles[NATIVE_BPMN_ADAPTER_ID].executable)
adapter, inspection, graph = compile_bpmn_to_graph(
NATIVE_BPMN,
adapter_id=NATIVE_LINEAR_ADAPTER_ID,
)
self.assertEqual("1.0.0", adapter.profile.version)
self.assertTrue(inspection.valid_xml)
self.assertIsNotNone(graph)
assert graph is not None
self.assertEqual(
[
"workflow.start.manual",
"workflow.activity",
"workflow.end.completed",
],
[item.type for item in graph.nodes],
)
self.assertEqual(220, graph.nodes[1].position.x)
self.assertEqual(
["Flow_start_review", "Flow_review_end"],
[item.id for item in graph.edges],
)
def test_native_profile_rejects_semantics_it_cannot_execute(self) -> None:
with self.assertRaisesRegex(
BpmnAdapterError,
"exclusiveGateway is not supported",
):
compile_bpmn_to_graph(
BPMN,
adapter_id=NATIVE_LINEAR_ADAPTER_ID,
)
def test_model_only_profile_remains_read_compatible(self) -> None:
_adapter, _inspection, graph = compile_bpmn_to_graph(
BPMN,
adapter_id=INTERCHANGE_ADAPTER_ID,
)
self.assertIsNone(graph)
def test_native_bpmn_graph_imports_full_notation_and_round_trips(self) -> None:
graph = import_bpmn_graph(BPMN)
self.assertEqual(
[
"bpmn.startEvent",
"bpmn.userTask",
"bpmn.exclusiveGateway",
"bpmn.endEvent",
"bpmn.participant",
],
[node.type for node in graph.nodes],
)
self.assertTrue(
all(edge.type == "bpmn.sequenceFlow" for edge in graph.edges)
)
rendered = export_bpmn_graph(graph, name="Round trip")
imported = import_bpmn_graph(rendered)
self.assertEqual(
[(node.id, node.type) for node in graph.nodes],
[(node.id, node.type) for node in imported.nodes],
)
self.assertEqual(
[(edge.id, edge.type, edge.source, edge.target) for edge in graph.edges],
[
(edge.id, edge.type, edge.source, edge.target)
for edge in imported.edges
],
)
def test_all_bpmn_fixtures_round_trip_through_the_native_graph(self) -> None:
fixture_directory = Path(__file__).parent / "fixtures" / "bpmn"
for path in sorted(fixture_directory.glob("*.bpmn")):
with self.subTest(path=path.name):
graph = import_bpmn_graph(path.read_text(encoding="utf-8"))
imported = import_bpmn_graph(
export_bpmn_graph(graph, name=path.stem)
)
self.assertEqual(
Counter(node.type for node in graph.nodes),
Counter(node.type for node in imported.nodes),
)
self.assertEqual(
Counter(edge.type for edge in graph.edges),
Counter(edge.type for edge in imported.edges),
)
self.assertEqual(len(graph.nodes), len(imported.nodes))
self.assertEqual(len(graph.edges), len(imported.edges))
def test_nested_flows_remain_in_their_bpmn_container(self) -> None:
fixture = (
Path(__file__).parent
/ "fixtures"
/ "bpmn"
/ "events-transaction-compensation.bpmn"
)
rendered = export_bpmn_graph(
import_bpmn_graph(fixture.read_text(encoding="utf-8"))
)
root = parse_bpmn_xml(rendered)
transaction = next(
item
for item in root.iter()
if item.tag == f"{{{BPMN_MODEL_NAMESPACE}}}transaction"
)
self.assertTrue(
any(
child.tag == f"{{{BPMN_MODEL_NAMESPACE}}}association"
and child.attrib.get("id") == "Compensation_Association"
for child in transaction
)
)
def test_default_flow_is_an_editable_edge_property(self) -> None:
source = BPMN.replace(
'',
'',
)
graph = import_bpmn_graph(source)
default_edge = next(edge for edge in graph.edges if edge.id == "Flow_3")
self.assertIs(default_edge.config.get("default"), True)
rendered = export_bpmn_graph(graph)
self.assertIn('default="Flow_3"', rendered)
graph.edges = [
edge.model_copy(update={"config": {**edge.config, "default": False}})
if edge.id == "Flow_3"
else edge
for edge in graph.edges
]
rendered_without_default = export_bpmn_graph(graph)
self.assertNotIn('default="Flow_3"', rendered_without_default)
def test_native_graph_import_is_separate_from_runtime_support(self) -> None:
_adapter, _inspection, graph = compile_bpmn_to_graph(
BPMN,
adapter_id=NATIVE_BPMN_ADAPTER_ID,
)
self.assertIsNotNone(graph)
with self.assertRaisesRegex(
BpmnAdapterError,
"exclusive gateway",
):
compile_bpmn_to_graph(
BPMN,
adapter_id=NATIVE_BPMN_ADAPTER_ID,
activation=True,
)
def test_adapter_versions_are_resolved_exactly_when_pinned(self) -> None:
with self.assertRaisesRegex(BpmnAdapterError, "is not installed"):
compile_bpmn_to_graph(
NATIVE_BPMN,
adapter_id=NATIVE_LINEAR_ADAPTER_ID,
adapter_version="9.0.0",
)
if __name__ == "__main__":
unittest.main()