652 lines
17 KiB
Python
652 lines
17 KiB
Python
# Split Miner - BPMN process discovery from event logs.
|
|
# Authors:
|
|
# imacat@mail.imacat.idv.tw (imacat), 2026/3/11
|
|
# AI assistance: Claude Code (Anthropic)
|
|
|
|
# Copyright (c) 2026 imacat.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied. See the License for the specific language governing
|
|
# permissions and limitations under the License.
|
|
"""Tests for SPQR-tree and RPST integration.
|
|
|
|
Tests that the spqrtree library is correctly integrated via
|
|
build_rpst(), using known graph decompositions from:
|
|
- Wikimedia SPQR tree example
|
|
- RPST paper (Polyvyanyy et al., 2011) Fig. 3(a)
|
|
- SM 1.0 paper Fig. 5(b)
|
|
|
|
References:
|
|
* https://commons.wikimedia.org/wiki/File:SPQR_tree_2.svg
|
|
* Polyvyanyy, A., Vanhatalo, J., & Voelzer, H. (2011).
|
|
Simplified computation and generalization of the refined
|
|
process structure tree. Lecture Notes in Computer
|
|
Science, 25-41.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from spqrtree import MultiGraph, NodeType, SPQRTree
|
|
|
|
from split_miner.bpmn import (
|
|
BPMNModel,
|
|
EndEvent,
|
|
Gateway,
|
|
GatewayType,
|
|
Node,
|
|
StartEvent,
|
|
Task,
|
|
)
|
|
from split_miner.joins import SESEFragment, build_rpst
|
|
|
|
|
|
def _make_serial_model() -> tuple[
|
|
BPMNModel, StartEvent, EndEvent,
|
|
Task, Task, Task,
|
|
]:
|
|
"""Build a serial chain: start -> a -> b -> c -> end.
|
|
|
|
:return: The model and its nodes.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
a: Task = Task("a", "a")
|
|
b: Task = Task("b", "b")
|
|
c: Task = Task("c", "c")
|
|
for t in [a, b, c]:
|
|
model.add_task(t)
|
|
model.add_edge(start, a)
|
|
model.add_edge(a, b)
|
|
model.add_edge(b, c)
|
|
model.add_edge(c, end)
|
|
return model, start, end, a, b, c
|
|
|
|
|
|
def _make_diamond_model() -> tuple[
|
|
BPMNModel, StartEvent, EndEvent,
|
|
Task, Task,
|
|
]:
|
|
"""Build a diamond: start -> {a, b} -> end.
|
|
|
|
:return: The model and its nodes.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
a: Task = Task("a", "a")
|
|
b: Task = Task("b", "b")
|
|
model.add_task(a)
|
|
model.add_task(b)
|
|
model.add_edge(start, a)
|
|
model.add_edge(start, b)
|
|
model.add_edge(a, end)
|
|
model.add_edge(b, end)
|
|
return model, start, end, a, b
|
|
|
|
|
|
def _make_fig5b_model() -> tuple[
|
|
BPMNModel, dict[str, Node],
|
|
]:
|
|
"""Build the model from Fig. 5(b) of the SM 1.0 paper.
|
|
|
|
Graph structure (after splits, before joins):
|
|
- start -> gx1 (XOR split)
|
|
- gx1 -> {a, b}
|
|
- a -> gx2 (XOR split), b -> gx3 (XOR split)
|
|
- gx2 -> {j, c}, gx3 -> {j, d}
|
|
- j -> i, c -> i, d -> k, i -> k
|
|
- k -> end
|
|
|
|
:return: The model and its named nodes.
|
|
"""
|
|
start: StartEvent = StartEvent("start")
|
|
end: EndEvent = EndEvent("end")
|
|
model: BPMNModel = BPMNModel(start, end)
|
|
|
|
tasks: dict[str, Task] = {}
|
|
for label in ["a", "b", "c", "d", "i", "j", "k"]:
|
|
t: Task = Task(label, label)
|
|
model.add_task(t)
|
|
tasks[label] = t
|
|
|
|
gx1: Gateway = Gateway("gx1", GatewayType.XOR)
|
|
gx2: Gateway = Gateway("gx2", GatewayType.XOR)
|
|
gx3: Gateway = Gateway("gx3", GatewayType.XOR)
|
|
model.add_gateway(gx1)
|
|
model.add_gateway(gx2)
|
|
model.add_gateway(gx3)
|
|
|
|
nodes: dict[str, Node] = {
|
|
"start": start, "end": end,
|
|
"gx1": gx1, "gx2": gx2, "gx3": gx3,
|
|
}
|
|
nodes.update(tasks)
|
|
|
|
for src, tgt in [
|
|
("start", "gx1"),
|
|
("gx1", "a"), ("gx1", "b"),
|
|
("a", "gx2"), ("b", "gx3"),
|
|
("gx2", "j"), ("gx3", "j"),
|
|
("gx2", "c"), ("gx3", "d"),
|
|
("j", "i"), ("c", "i"),
|
|
("d", "k"), ("i", "k"),
|
|
("k", "end"),
|
|
]:
|
|
model.add_edge(nodes[src], nodes[tgt])
|
|
|
|
return model, nodes
|
|
|
|
|
|
class TestSpqrTreeWikimedia(unittest.TestCase):
|
|
"""Tests SPQR-tree on the Wikimedia Commons example.
|
|
|
|
Tests the set of all SPQR-tree nodes (type + vertices)
|
|
regardless of root choice, since the unrooted tree
|
|
structure is unique but the rooting may vary.
|
|
|
|
Reference:
|
|
https://commons.wikimedia.org/wiki/File:SPQR_tree_2.svg
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the Wikimedia example graph.
|
|
|
|
:return: None.
|
|
"""
|
|
mg: MultiGraph = MultiGraph()
|
|
for v in "abcdefghijklmnop":
|
|
mg.add_vertex(v)
|
|
for u, v in [
|
|
("a", "b"), ("a", "c"), ("a", "g"),
|
|
("b", "d"), ("b", "h"),
|
|
("c", "d"), ("c", "e"),
|
|
("d", "f"), ("e", "f"), ("e", "g"),
|
|
("f", "h"),
|
|
("h", "i"), ("h", "j"),
|
|
("i", "j"), ("i", "n"),
|
|
("j", "k"),
|
|
("k", "m"), ("k", "n"), ("m", "n"),
|
|
("l", "m"), ("l", "o"), ("l", "p"),
|
|
("m", "o"), ("m", "p"),
|
|
("o", "p"),
|
|
("g", "l"),
|
|
]:
|
|
mg.add_edge(u, v)
|
|
self.__tree: SPQRTree = SPQRTree(mg)
|
|
self.__all_nodes: list[
|
|
tuple[str, frozenset[str]]
|
|
] = []
|
|
_collect_all_nodes(
|
|
self.__tree.root, self.__all_nodes
|
|
)
|
|
|
|
def test_node_count(self) -> None:
|
|
"""The tree has 5 nodes (1 S, 1 P, 3 R).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__all_nodes), 5)
|
|
|
|
def test_node_types(self) -> None:
|
|
"""Node types are P, R, R, R, S (sorted).
|
|
|
|
:return: None.
|
|
"""
|
|
types: list[str] = sorted(
|
|
t for t, _ in self.__all_nodes
|
|
)
|
|
self.assertEqual(
|
|
types, ["P", "R", "R", "R", "S"]
|
|
)
|
|
|
|
def test_s_node(self) -> None:
|
|
"""S-node has vertices {g, h, l, m}.
|
|
|
|
:return: None.
|
|
"""
|
|
s_nodes: list[frozenset[str]] = [
|
|
v for t, v in self.__all_nodes
|
|
if t == "S"
|
|
]
|
|
self.assertEqual(len(s_nodes), 1)
|
|
self.assertEqual(
|
|
s_nodes[0],
|
|
frozenset({"g", "h", "l", "m"}),
|
|
)
|
|
|
|
def test_p_node(self) -> None:
|
|
"""P-node has vertices {l, m}.
|
|
|
|
:return: None.
|
|
"""
|
|
p_nodes: list[frozenset[str]] = [
|
|
v for t, v in self.__all_nodes
|
|
if t == "P"
|
|
]
|
|
self.assertEqual(len(p_nodes), 1)
|
|
self.assertEqual(
|
|
p_nodes[0], frozenset({"l", "m"})
|
|
)
|
|
|
|
def test_r_node_1(self) -> None:
|
|
"""R-node {a,b,c,d,e,f,g,h} exists.
|
|
|
|
:return: None.
|
|
"""
|
|
r_verts: list[frozenset[str]] = [
|
|
v for t, v in self.__all_nodes
|
|
if t == "R"
|
|
]
|
|
self.assertIn(
|
|
frozenset({
|
|
"a", "b", "c", "d",
|
|
"e", "f", "g", "h",
|
|
}),
|
|
r_verts,
|
|
)
|
|
|
|
def test_r_node_2(self) -> None:
|
|
"""R-node {h,i,j,k,m,n} exists.
|
|
|
|
:return: None.
|
|
"""
|
|
r_verts: list[frozenset[str]] = [
|
|
v for t, v in self.__all_nodes
|
|
if t == "R"
|
|
]
|
|
self.assertIn(
|
|
frozenset({
|
|
"h", "i", "j", "k", "m", "n",
|
|
}),
|
|
r_verts,
|
|
)
|
|
|
|
def test_r_node_3(self) -> None:
|
|
"""R-node {l,m,o,p} exists.
|
|
|
|
:return: None.
|
|
"""
|
|
r_verts: list[frozenset[str]] = [
|
|
v for t, v in self.__all_nodes
|
|
if t == "R"
|
|
]
|
|
self.assertIn(
|
|
frozenset({"l", "m", "o", "p"}),
|
|
r_verts,
|
|
)
|
|
|
|
|
|
class TestSpqrTreeRpstFig3a(unittest.TestCase):
|
|
"""Tests SPQR-tree on RPST paper Fig. 3(a).
|
|
|
|
Reference: Polyvyanyy et al. (2011), Fig. 3(a).
|
|
Graph: s->u, u->{v,w}, v->{w,x}, w->x, x->y,
|
|
y->z (x2), z->y, z->t, plus back-edge t->s.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the RPST Fig 3a graph.
|
|
|
|
:return: None.
|
|
"""
|
|
mg: MultiGraph = MultiGraph()
|
|
for v in [
|
|
"s", "u", "v", "w", "x",
|
|
"y", "z", "t",
|
|
]:
|
|
mg.add_vertex(v)
|
|
for u, v in [
|
|
("s", "u"), ("u", "v"), ("u", "w"),
|
|
("v", "w"), ("v", "x"), ("w", "x"),
|
|
("x", "y"),
|
|
("y", "z"), ("y", "z"), ("z", "y"),
|
|
("z", "t"), ("t", "s"),
|
|
]:
|
|
mg.add_edge(u, v)
|
|
self.__tree: SPQRTree = SPQRTree(mg)
|
|
|
|
def test_root_type(self) -> None:
|
|
"""The root is an S-node.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__tree.root.type, NodeType.S
|
|
)
|
|
|
|
def test_root_vertices(self) -> None:
|
|
"""Root S-node contains {s,t,u,x,y,z}.
|
|
|
|
:return: None.
|
|
"""
|
|
verts: set[str] = _skeleton_vertices(
|
|
self.__tree.root
|
|
)
|
|
self.assertEqual(
|
|
verts, {"s", "t", "u", "x", "y", "z"}
|
|
)
|
|
|
|
def test_child_count(self) -> None:
|
|
"""The root has 2 children: R and P.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
len(self.__tree.root.children), 2
|
|
)
|
|
|
|
def test_r_child(self) -> None:
|
|
"""R-node child has {u,v,w,x}.
|
|
|
|
:return: None.
|
|
"""
|
|
r1 = _find_child_by_vertices(
|
|
self.__tree.root, {"u", "v", "w", "x"}
|
|
)
|
|
self.assertIsNotNone(r1)
|
|
assert r1 is not None
|
|
self.assertEqual(r1.type, NodeType.R)
|
|
|
|
def test_p_child(self) -> None:
|
|
"""P-node child has {y,z}.
|
|
|
|
:return: None.
|
|
"""
|
|
p1 = _find_child_by_vertices(
|
|
self.__tree.root, {"y", "z"}
|
|
)
|
|
self.assertIsNotNone(p1)
|
|
assert p1 is not None
|
|
self.assertEqual(p1.type, NodeType.P)
|
|
|
|
def test_r_child_real_edges(self) -> None:
|
|
"""R-node has 5 real edges (the biconnected core).
|
|
|
|
:return: None.
|
|
"""
|
|
r1 = _find_child_by_vertices(
|
|
self.__tree.root, {"u", "v", "w", "x"}
|
|
)
|
|
assert r1 is not None
|
|
real: list[tuple[str, str]] = [
|
|
(e.u, e.v) for e in r1.skeleton.edges
|
|
if not e.virtual
|
|
]
|
|
self.assertEqual(len(real), 5)
|
|
|
|
def test_p_child_real_edges(self) -> None:
|
|
"""P-node has 3 real edges (y->z x2, z->y).
|
|
|
|
:return: None.
|
|
"""
|
|
p1 = _find_child_by_vertices(
|
|
self.__tree.root, {"y", "z"}
|
|
)
|
|
assert p1 is not None
|
|
real: list[tuple[str, str]] = [
|
|
(e.u, e.v) for e in p1.skeleton.edges
|
|
if not e.virtual
|
|
]
|
|
self.assertEqual(len(real), 3)
|
|
|
|
|
|
class TestBuildRpstSerial(unittest.TestCase):
|
|
"""Tests build_rpst on a serial chain."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up a serial model: start->a->b->c->end.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
model, _, _, _, _, _ = _make_serial_model()
|
|
self.__fragments: list[SESEFragment] = (
|
|
build_rpst(model)
|
|
)
|
|
|
|
def test_single_fragment(self) -> None:
|
|
"""A serial chain produces one S-type fragment.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__fragments), 1)
|
|
|
|
def test_fragment_type(self) -> None:
|
|
"""The fragment is S-type (serial).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__fragments[0].fragment_type,
|
|
NodeType.S,
|
|
)
|
|
|
|
def test_fragment_edges(self) -> None:
|
|
"""The fragment contains all 4 edges.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
len(self.__fragments[0].edges), 4
|
|
)
|
|
|
|
def test_fragment_nodes(self) -> None:
|
|
"""The fragment contains all 5 nodes.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
len(self.__fragments[0].nodes), 5
|
|
)
|
|
|
|
|
|
class TestBuildRpstDiamond(unittest.TestCase):
|
|
"""Tests build_rpst on a diamond graph."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up a diamond: start->{a,b}->end.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
model, _, _, _, _ = _make_diamond_model()
|
|
self.__fragments: list[SESEFragment] = (
|
|
build_rpst(model)
|
|
)
|
|
|
|
def test_fragment_count(self) -> None:
|
|
"""Diamond produces 3 fragments (2 S + 1 P).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__fragments), 3)
|
|
|
|
def test_has_p_fragment(self) -> None:
|
|
"""There is a P-type (parallel) fragment.
|
|
|
|
:return: None.
|
|
"""
|
|
p_frags: list[SESEFragment] = [
|
|
f for f in self.__fragments
|
|
if f.fragment_type == NodeType.P
|
|
]
|
|
self.assertEqual(len(p_frags), 1)
|
|
|
|
def test_p_fragment_covers_all(self) -> None:
|
|
"""The P-type fragment contains all 4 edges.
|
|
|
|
:return: None.
|
|
"""
|
|
p_frag: SESEFragment = [
|
|
f for f in self.__fragments
|
|
if f.fragment_type == NodeType.P
|
|
][0]
|
|
self.assertEqual(len(p_frag.edges), 4)
|
|
|
|
def test_p_fragment_nodes(self) -> None:
|
|
"""The P-type fragment contains all 4 nodes.
|
|
|
|
:return: None.
|
|
"""
|
|
p_frag: SESEFragment = [
|
|
f for f in self.__fragments
|
|
if f.fragment_type == NodeType.P
|
|
][0]
|
|
self.assertEqual(len(p_frag.nodes), 4)
|
|
|
|
def test_s_fragments(self) -> None:
|
|
"""Two S-type fragments (one per branch).
|
|
|
|
:return: None.
|
|
"""
|
|
s_frags: list[SESEFragment] = [
|
|
f for f in self.__fragments
|
|
if f.fragment_type == NodeType.S
|
|
]
|
|
self.assertEqual(len(s_frags), 2)
|
|
|
|
def test_s_fragment_edges(self) -> None:
|
|
"""Each S-type fragment has 2 edges.
|
|
|
|
:return: None.
|
|
"""
|
|
for f in self.__fragments:
|
|
if f.fragment_type == NodeType.S:
|
|
self.assertEqual(len(f.edges), 2)
|
|
|
|
def test_bottom_up_order(self) -> None:
|
|
"""Fragments are ordered bottom-up (small first).
|
|
|
|
:return: None.
|
|
"""
|
|
sizes: list[int] = [
|
|
len(f.edges) for f in self.__fragments
|
|
]
|
|
self.assertEqual(sizes, sorted(sizes))
|
|
|
|
|
|
class TestBuildRpstFig5b(unittest.TestCase):
|
|
"""Tests build_rpst on SM 1.0 paper Fig. 5(b)."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the Fig. 5(b) model.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
model, _ = _make_fig5b_model()
|
|
self.__fragments: list[SESEFragment] = (
|
|
build_rpst(model)
|
|
)
|
|
|
|
def test_has_fragments(self) -> None:
|
|
"""At least one fragment is produced.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertGreater(len(self.__fragments), 0)
|
|
|
|
def test_all_edges_covered(self) -> None:
|
|
"""Union of fragment edges covers all model edges.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
model, _ = _make_fig5b_model()
|
|
all_frag_edges: set[tuple[Node, Node]] = set()
|
|
for f in self.__fragments:
|
|
all_frag_edges |= f.edges
|
|
self.assertEqual(all_frag_edges, model.edges)
|
|
|
|
def test_has_r_fragment(self) -> None:
|
|
"""There is at least one R-type (rigid) fragment.
|
|
|
|
:return: None.
|
|
"""
|
|
r_frags: list[SESEFragment] = [
|
|
f for f in self.__fragments
|
|
if f.fragment_type == NodeType.R
|
|
]
|
|
self.assertGreater(len(r_frags), 0)
|
|
|
|
def test_bottom_up_order(self) -> None:
|
|
"""Fragments are ordered bottom-up (small first).
|
|
|
|
:return: None.
|
|
"""
|
|
sizes: list[int] = [
|
|
len(f.edges) for f in self.__fragments
|
|
]
|
|
self.assertEqual(sizes, sorted(sizes))
|
|
|
|
def test_entry_exit_are_nodes(self) -> None:
|
|
"""Entry and exit of each fragment are model nodes.
|
|
|
|
:return: None.
|
|
"""
|
|
model: BPMNModel
|
|
model, _ = _make_fig5b_model()
|
|
all_nodes: set[Node] = model.all_nodes
|
|
for f in self.__fragments:
|
|
self.assertIn(f.entry, all_nodes)
|
|
self.assertIn(f.exit_node, all_nodes)
|
|
|
|
|
|
def _collect_all_nodes(
|
|
spqr_node,
|
|
result: list[tuple[str, frozenset]],
|
|
) -> None:
|
|
"""Collect all SPQR-tree nodes as (type, vertices).
|
|
|
|
:param spqr_node: The SPQR-tree node.
|
|
:param result: The output list.
|
|
"""
|
|
verts: frozenset = frozenset(
|
|
_skeleton_vertices(spqr_node)
|
|
)
|
|
result.append((spqr_node.type.name, verts))
|
|
for child in spqr_node.children:
|
|
_collect_all_nodes(child, result)
|
|
|
|
|
|
def _skeleton_vertices(spqr_node) -> set:
|
|
"""Extract vertex set from an SPQR-tree node skeleton.
|
|
|
|
:param spqr_node: The SPQR-tree node.
|
|
:return: The set of vertices.
|
|
"""
|
|
verts: set = set()
|
|
for e in spqr_node.skeleton.edges:
|
|
verts.add(e.u)
|
|
verts.add(e.v)
|
|
return verts
|
|
|
|
|
|
def _find_child_by_vertices(
|
|
parent, target_verts: set
|
|
):
|
|
"""Find a child SPQR node by its vertex set.
|
|
|
|
:param parent: The parent SPQR-tree node.
|
|
:param target_verts: The expected vertex set.
|
|
:return: The matching child, or None.
|
|
"""
|
|
for child in parent.children:
|
|
if _skeleton_vertices(child) == target_verts:
|
|
return child
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|