636 lines
16 KiB
Python
636 lines
16 KiB
Python
# Split Miner - BPMN process discovery from event logs.
|
|
# Authors:
|
|
# imacat@mail.imacat.idv.tw (imacat), 2026/3/12
|
|
# 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 the refined DFG (Definition 6, SM 2.0).
|
|
|
|
The refined directly-follows relation uses activity lifecycle
|
|
(start/end) events: activity ay directly-follows ax iff
|
|
ay starts after ax ends with no other end events in between.
|
|
|
|
Reference:
|
|
Augusto, A., Dumas, M., & La Rosa, M. (2021).
|
|
Automated Discovery of Process Models with True
|
|
Concurrency and Inclusive Choices. Section 3.1,
|
|
Definition 6.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from split_miner.bpmn import Task
|
|
from split_miner.refined_dfg import RefinedDirectlyFollowsGraph
|
|
|
|
S: str = "start"
|
|
"""Lifecycle start constant."""
|
|
E: str = "end"
|
|
"""Lifecycle end constant."""
|
|
|
|
|
|
def _make_tasks(
|
|
*labels: str,
|
|
) -> dict[str, Task]:
|
|
"""Create Task objects from labels.
|
|
|
|
:param labels: The activity labels.
|
|
:return: A dict mapping label to Task.
|
|
"""
|
|
return {
|
|
label: Task(label, label)
|
|
for label in labels
|
|
}
|
|
|
|
|
|
def _make_paper_example() -> tuple[
|
|
dict[str, Task],
|
|
RefinedDirectlyFollowsGraph,
|
|
]:
|
|
"""Build the DFG from the paper's example Lrho_x.
|
|
|
|
Four traces with activities A-F, where B/C and
|
|
D/E have overlapping lifecycles:
|
|
|
|
Trace 1: As Ae Bs Cs Ce Be Es Ds De Ee Fs Fe
|
|
Trace 2: As Ae Bs Cs Be Ce Es Ds Ee De Fs Fe
|
|
Trace 3: As Ae Cs Bs Be Ce Ds Es De Ee Fs Fe
|
|
Trace 4: As Ae Cs Bs Ce Be Ds Es Ee De Fs Fe
|
|
|
|
Expected DFG (Figure 3c): A->B, A->C, B->D, B->E,
|
|
C->D, C->E, D->F, E->F.
|
|
|
|
:return: The tasks and the DFG.
|
|
"""
|
|
t: dict[str, Task] = _make_tasks(
|
|
"A", "B", "C", "D", "E", "F"
|
|
)
|
|
a: Task = t["A"]
|
|
b: Task = t["B"]
|
|
c: Task = t["C"]
|
|
d: Task = t["D"]
|
|
e: Task = t["E"]
|
|
f: Task = t["F"]
|
|
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
# Trace 1
|
|
((a, S), (a, E), (b, S), (c, S),
|
|
(c, E), (b, E), (e, S), (d, S),
|
|
(d, E), (e, E), (f, S), (f, E)): 1,
|
|
# Trace 2
|
|
((a, S), (a, E), (b, S), (c, S),
|
|
(b, E), (c, E), (e, S), (d, S),
|
|
(e, E), (d, E), (f, S), (f, E)): 1,
|
|
# Trace 3
|
|
((a, S), (a, E), (c, S), (b, S),
|
|
(b, E), (c, E), (d, S), (e, S),
|
|
(d, E), (e, E), (f, S), (f, E)): 1,
|
|
# Trace 4
|
|
((a, S), (a, E), (c, S), (b, S),
|
|
(c, E), (b, E), (d, S), (e, S),
|
|
(e, E), (d, E), (f, S), (f, E)): 1,
|
|
}
|
|
dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
return t, dfg
|
|
|
|
|
|
class TestSequentialTrace(unittest.TestCase):
|
|
"""Tests refined DFG with purely sequential traces.
|
|
|
|
When activities don't overlap, the refined DFG matches
|
|
the SM 1.0 DFG.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up a simple sequential trace.
|
|
|
|
Trace: As Ae Bs Be Cs Ce
|
|
Expected: A->B, B->C.
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task] = (
|
|
_make_tasks("A", "B", "C")
|
|
)
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (a, E),
|
|
(b, S), (b, E),
|
|
(c, S), (c, E)): 1,
|
|
}
|
|
self.__dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_nodes(self) -> None:
|
|
"""DFG has 3 nodes (A, B, C).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__dfg.nodes), 3)
|
|
|
|
def test_edges(self) -> None:
|
|
"""DFG has edges A->B and B->C.
|
|
|
|
:return: None.
|
|
"""
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
self.assertEqual(
|
|
self.__dfg.edges,
|
|
{(a, b), (b, c)},
|
|
)
|
|
|
|
def test_sources(self) -> None:
|
|
"""Source is A (first activity to start).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.sources,
|
|
{self.__tasks["A"]},
|
|
)
|
|
|
|
def test_sinks(self) -> None:
|
|
"""Sink is C (last activity to end).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.sinks,
|
|
{self.__tasks["C"]},
|
|
)
|
|
|
|
|
|
class TestOverlappingTrace(unittest.TestCase):
|
|
"""Tests refined DFG with overlapping lifecycles.
|
|
|
|
When activities overlap, no directly-follows relation
|
|
exists between them.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up overlapping and sequential activities.
|
|
|
|
Trace: As Bs Ae Be Cs Ce
|
|
A and B overlap. B ends last.
|
|
Expected: B->C only (not A->C, since A_end
|
|
precedes B_end which is an end event between
|
|
A_end and C_start).
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task] = (
|
|
_make_tasks("A", "B", "C")
|
|
)
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (b, S), (a, E),
|
|
(b, E), (c, S), (c, E)): 1,
|
|
}
|
|
self.__dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_no_edge_between_overlapping(self) -> None:
|
|
"""No A->B or B->A edge (overlapping lifecycles).
|
|
|
|
A starts before B, and A ends before B. But
|
|
A_end(3) > B_start(2), so A does not end before
|
|
B starts. Similarly B does not end before A
|
|
starts. So neither direction holds.
|
|
|
|
:return: None.
|
|
"""
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
self.assertNotIn((a, b), self.__dfg.edges)
|
|
self.assertNotIn((b, a), self.__dfg.edges)
|
|
|
|
def test_last_ender_follows_to_next(self) -> None:
|
|
"""Only B->C exists (B is the last to end).
|
|
|
|
B_end(4) precedes C_start(5) with no end events
|
|
between them. A_end(3) also precedes C_start(5)
|
|
but B_end(4) is between them, blocking A->C.
|
|
|
|
:return: None.
|
|
"""
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
self.assertIn((b, c), self.__dfg.edges)
|
|
|
|
def test_edge_count(self) -> None:
|
|
"""Only 1 edge total.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__dfg.edges), 1)
|
|
|
|
|
|
class TestPaperExample(unittest.TestCase):
|
|
"""Tests refined DFG on the paper's example Lrho_x.
|
|
|
|
Reference: SM 2.0 paper, Section 3.1, Figure 3(c).
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up the paper's example.
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task]
|
|
self.__dfg: RefinedDirectlyFollowsGraph
|
|
self.__tasks, self.__dfg = (
|
|
_make_paper_example()
|
|
)
|
|
|
|
def test_nodes(self) -> None:
|
|
"""DFG has 6 nodes (A through F).
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__dfg.nodes), 6)
|
|
|
|
def test_edges(self) -> None:
|
|
"""DFG has exactly 8 edges per Figure 3(c).
|
|
|
|
A->B, A->C, B->D, B->E, C->D, C->E, D->F,
|
|
E->F.
|
|
|
|
:return: None.
|
|
"""
|
|
t: dict[str, Task] = self.__tasks
|
|
expected: set[tuple[Task, Task]] = {
|
|
(t["A"], t["B"]), (t["A"], t["C"]),
|
|
(t["B"], t["D"]), (t["B"], t["E"]),
|
|
(t["C"], t["D"]), (t["C"], t["E"]),
|
|
(t["D"], t["F"]), (t["E"], t["F"]),
|
|
}
|
|
self.assertEqual(self.__dfg.edges, expected)
|
|
|
|
def test_edge_count(self) -> None:
|
|
"""DFG has 8 edges.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(len(self.__dfg.edges), 8)
|
|
|
|
def test_no_bidirectional_edges(self) -> None:
|
|
"""No bidirectional edges exist.
|
|
|
|
In the paper's example, overlapping lifecycles
|
|
prevent any bidirectional relations.
|
|
|
|
:return: None.
|
|
"""
|
|
for src, tgt in self.__dfg.edges:
|
|
self.assertNotIn(
|
|
(tgt, src), self.__dfg.edges,
|
|
f"Bidirectional edge {src}->{tgt} "
|
|
f"and {tgt}->{src}",
|
|
)
|
|
|
|
def test_sources(self) -> None:
|
|
"""Source is A.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.sources,
|
|
{self.__tasks["A"]},
|
|
)
|
|
|
|
def test_sinks(self) -> None:
|
|
"""Sink is F.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.sinks,
|
|
{self.__tasks["F"]},
|
|
)
|
|
|
|
def test_a_to_b_frequency(self) -> None:
|
|
"""A->B occurs in all 4 traces.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.df_frequency(
|
|
self.__tasks["A"],
|
|
self.__tasks["B"],
|
|
),
|
|
4,
|
|
)
|
|
|
|
def test_a_to_c_frequency(self) -> None:
|
|
"""A->C occurs in all 4 traces.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.df_frequency(
|
|
self.__tasks["A"],
|
|
self.__tasks["C"],
|
|
),
|
|
4,
|
|
)
|
|
|
|
def test_b_to_d_frequency(self) -> None:
|
|
"""B->D occurs in traces 1 and 4.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.df_frequency(
|
|
self.__tasks["B"],
|
|
self.__tasks["D"],
|
|
),
|
|
2,
|
|
)
|
|
|
|
def test_d_to_f_frequency(self) -> None:
|
|
"""D->F occurs in traces 2 and 4.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.df_frequency(
|
|
self.__tasks["D"],
|
|
self.__tasks["F"],
|
|
),
|
|
2,
|
|
)
|
|
|
|
def test_e_to_f_frequency(self) -> None:
|
|
"""E->F occurs in traces 1 and 3.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertEqual(
|
|
self.__dfg.df_frequency(
|
|
self.__tasks["E"],
|
|
self.__tasks["F"],
|
|
),
|
|
2,
|
|
)
|
|
|
|
|
|
class TestForkAfterEnd(unittest.TestCase):
|
|
"""Tests that multiple activities can follow one end.
|
|
|
|
When A ends and both B and C start (with no other end
|
|
events between), both A->B and A->C should exist.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up a fork pattern.
|
|
|
|
Trace: As Ae Bs Cs Be Ce
|
|
A ends, then B and C start (neither ends before
|
|
the other starts).
|
|
Expected: A->B, A->C.
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task] = (
|
|
_make_tasks("A", "B", "C")
|
|
)
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (a, E), (b, S), (c, S),
|
|
(b, E), (c, E)): 1,
|
|
}
|
|
self.__dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_fork_edges(self) -> None:
|
|
"""Both A->B and A->C exist.
|
|
|
|
A_end precedes both B_start and C_start with no
|
|
end events in between.
|
|
|
|
:return: None.
|
|
"""
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
self.assertEqual(
|
|
self.__dfg.edges,
|
|
{(a, b), (a, c)},
|
|
)
|
|
|
|
|
|
class TestSelfLoop(unittest.TestCase):
|
|
"""Tests self-loop detection from lifecycle traces.
|
|
|
|
An activity a is a self-loop iff a ->r a holds per
|
|
Definition 6, that is, an end event of a is followed by
|
|
a start event of a with no other end event in between.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up trace with self-loop: A repeats at once.
|
|
|
|
Trace: As Ae As Ae Bs Be Cs Ce
|
|
A ends and immediately starts again, so A ->r A.
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task] = (
|
|
_make_tasks("A", "B", "C")
|
|
)
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (a, E),
|
|
(a, S), (a, E),
|
|
(b, S), (b, E),
|
|
(c, S), (c, E)): 1,
|
|
}
|
|
self.__dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_a_is_self_loop(self) -> None:
|
|
"""A is detected as a self-loop.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertIn(
|
|
self.__tasks["A"],
|
|
self.__dfg.self_loops,
|
|
)
|
|
|
|
def test_b_not_self_loop(self) -> None:
|
|
"""B is not a self-loop.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertNotIn(
|
|
self.__tasks["B"],
|
|
self.__dfg.self_loops,
|
|
)
|
|
|
|
def test_no_self_edge(self) -> None:
|
|
"""The self-loop edge A->A is not in the edges.
|
|
|
|
:return: None.
|
|
"""
|
|
a: Task = self.__tasks["A"]
|
|
self.assertNotIn((a, a), self.__dfg.edges)
|
|
|
|
def test_other_edges(self) -> None:
|
|
"""The other edges are A->B and B->C.
|
|
|
|
:return: None.
|
|
"""
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
c: Task = self.__tasks["C"]
|
|
self.assertEqual(
|
|
self.__dfg.edges,
|
|
{(a, b), (b, c)},
|
|
)
|
|
|
|
def test_no_self_loop_in_paper(self) -> None:
|
|
"""Paper example has no self-loops.
|
|
|
|
:return: None.
|
|
"""
|
|
_, dfg = _make_paper_example()
|
|
self.assertEqual(dfg.self_loops, set())
|
|
|
|
|
|
class TestSelfLoopWithoutRepeatedEnd(unittest.TestCase):
|
|
"""Tests a self-loop with a single end event.
|
|
|
|
Definition 6 only requires an end event of a followed by
|
|
a start event of a, not a second end event of a.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up trace: As Ae As Bs Be.
|
|
|
|
A ends, then A starts again with no end event in
|
|
between, so A ->r A.
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task] = (
|
|
_make_tasks("A", "B")
|
|
)
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (a, E), (a, S),
|
|
(b, S), (b, E)): 1,
|
|
}
|
|
self.__dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_a_is_self_loop(self) -> None:
|
|
"""A is a self-loop even though it ends once.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertIn(
|
|
self.__tasks["A"],
|
|
self.__dfg.self_loops,
|
|
)
|
|
|
|
|
|
class TestTwoLoopIsNotSelfLoop(unittest.TestCase):
|
|
"""Tests that a 2-loop is not mistaken for a self-loop.
|
|
|
|
An activity repeating after another activity completes
|
|
forms a 2-loop, not a self-loop.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
"""Set up trace: As Ae Bs Be As Ae.
|
|
|
|
B ends between the end of A and the next start of
|
|
A, so A ->r A does not hold.
|
|
|
|
:return: None.
|
|
"""
|
|
self.__tasks: dict[str, Task] = (
|
|
_make_tasks("A", "B")
|
|
)
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
traces: dict[
|
|
tuple[tuple[Task, str], ...], int
|
|
] = {
|
|
((a, S), (a, E),
|
|
(b, S), (b, E),
|
|
(a, S), (a, E)): 1,
|
|
}
|
|
self.__dfg: RefinedDirectlyFollowsGraph = (
|
|
RefinedDirectlyFollowsGraph(traces)
|
|
)
|
|
|
|
def test_a_not_self_loop(self) -> None:
|
|
"""A is not a self-loop.
|
|
|
|
:return: None.
|
|
"""
|
|
self.assertNotIn(
|
|
self.__tasks["A"],
|
|
self.__dfg.self_loops,
|
|
)
|
|
|
|
def test_two_loop_edges(self) -> None:
|
|
"""The edges form the 2-loop A->B and B->A.
|
|
|
|
:return: None.
|
|
"""
|
|
a: Task = self.__tasks["A"]
|
|
b: Task = self.__tasks["B"]
|
|
self.assertEqual(
|
|
self.__dfg.edges,
|
|
{(a, b), (b, a)},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|