split-miner/tests/test_split_miner_2.py

# 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.
"""Integration tests for Split Miner 2.0 pipeline.

Verifies that split_miner_2() produces valid BPMN models
from lifecycle-aware event logs.

Reference:
    Augusto, A., Dumas, M., & La Rosa, M. (2021).
    Automated Discovery of Process Models with True
    Concurrency and Inclusive Choices.
"""
from __future__ import annotations

import unittest

from split_miner.bpmn import (
    BPMNModel,
    Gateway,
    GatewayType,
    Node,
    StartEvent,
    Task,
)
from split_miner.miner import split_miner_2

S: str = "start"
"""Lifecycle start constant."""
E: str = "end"
"""Lifecycle end constant."""


class TestSequentialPipeline(unittest.TestCase):
    """Tests SM 2.0 with a simple sequential log.

    A -> B -> C with no concurrency.
    """

    def setUp(self) -> None:
        """Set up sequential lifecycle traces.

        :return: None.
        """
        traces: dict[
            tuple[tuple[str, str], ...], int
        ] = {
            (("A", S), ("A", E),
             ("B", S), ("B", E),
             ("C", S), ("C", E)): 5,
        }
        self.__model: BPMNModel = split_miner_2(
            traces
        )

    def test_has_start_and_end(self) -> None:
        """Model has start and end events.

        :return: None.
        """
        self.assertIsNotNone(self.__model.start)
        self.assertIsNotNone(self.__model.end)

    def test_has_three_tasks(self) -> None:
        """Model has 3 tasks (A, B, C).

        :return: None.
        """
        self.assertEqual(
            len(self.__model.tasks), 3
        )

    def test_no_gateways(self) -> None:
        """Sequential model has no gateways.

        :return: None.
        """
        self.assertEqual(
            len(self.__model.gateways), 0
        )


class TestConcurrentPipeline(unittest.TestCase):
    """Tests SM 2.0 with concurrent activities.

    A -> (B || C) -> D where B and C always overlap.
    """

    def setUp(self) -> None:
        """Set up concurrent lifecycle traces.

        :return: None.
        """
        traces: dict[
            tuple[tuple[str, str], ...], int
        ] = {
            (("A", S), ("A", E),
             ("B", S), ("C", S),
             ("B", E), ("C", E),
             ("D", S), ("D", E)): 5,
        }
        self.__model: BPMNModel = split_miner_2(
            traces
        )

    def test_has_four_tasks(self) -> None:
        """Model has 4 tasks (A, B, C, D).

        :return: None.
        """
        self.assertEqual(
            len(self.__model.tasks), 4
        )

    def test_has_gateways(self) -> None:
        """Model has gateway(s) for concurrency.

        :return: None.
        """
        self.assertGreater(
            len(self.__model.gateways), 0
        )

    def test_has_and_split(self) -> None:
        """Model has an AND-split for B || C.

        :return: None.
        """
        and_splits: list[Gateway] = [
            gw for gw
            in self.__model.gateways.values()
            if gw.gateway_type == GatewayType.AND
            and len(
                self.__model.outgoing_edges(gw)
            ) > 1
        ]
        self.assertGreater(len(and_splits), 0)


class TestPaperExample(unittest.TestCase):
    """Tests SM 2.0 on the paper's Lrho_x example.

    Should produce a valid BPMN model with A-F.
    """

    def setUp(self) -> None:
        """Set up the paper's example.

        :return: None.
        """
        traces: dict[
            tuple[tuple[str, str], ...], int
        ] = {
            (("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,
            (("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,
            (("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,
            (("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,
        }
        self.__model: BPMNModel = split_miner_2(
            traces
        )

    def test_has_six_tasks(self) -> None:
        """Model has 6 tasks (A through F).

        :return: None.
        """
        self.assertEqual(
            len(self.__model.tasks), 6
        )

    def test_completes_without_error(self) -> None:
        """Pipeline completes without error.

        :return: None.
        """
        self.assertIsNotNone(self.__model)

    def test_start_connects_to_a(self) -> None:
        """Start event connects to task A.

        :return: None.
        """
        start_succs: set[Node] = (
            self.__model.successors(
                self.__model.start
            )
        )
        # Either directly to A or through a gateway
        reachable: set[str] = set()
        queue: list[Node] = list(start_succs)
        visited: set[Node] = set()
        while queue:
            n: Node = queue.pop(0)
            if n in visited:
                continue
            visited.add(n)
            if isinstance(n, Task):
                reachable.add(n.node_id)
            elif isinstance(n, Gateway):
                queue.extend(
                    self.__model.successors(n)
                )
        self.assertIn("A", reachable)


class TestSelfLoopPipeline(unittest.TestCase):
    """Tests SM 2.0 with a self-loop activity.

    A -> B (self-loop) -> C.  B repeats in the same
    trace without going through other activities.
    The self-loop XOR back-edge is the only source of
    gateways around B.
    """

    def setUp(self) -> None:
        """Set up lifecycle traces with self-loop on B.

        Trace: As Ae Bs Be Bs Be Cs Ce
        B completes twice.  No other loops exist,
        so only self-loop restoration adds gateways
        around B.

        :return: None.
        """
        traces: dict[
            tuple[tuple[str, str], ...], int
        ] = {
            (("A", S), ("A", E),
             ("B", S), ("B", E),
             ("B", S), ("B", E),
             ("C", S), ("C", E)): 5,
        }
        self.__model: BPMNModel = split_miner_2(
            traces
        )

    def test_self_loop_xor_join_before_b(
        self,
    ) -> None:
        """Task B has an XOR-join predecessor.

        The self-loop restoration inserts an XOR-join
        before B.  Without restoration, B connects
        directly to its predecessor (A or start).

        :return: None.
        """
        b: Task = self.__model.get_task("B")
        preds: set[Node] = (
            self.__model.predecessors(b)
        )
        xor_joins: list[Gateway] = [
            p for p in preds
            if isinstance(p, Gateway)
            and p.gateway_type == GatewayType.XOR
            and len(
                self.__model.incoming_edges(p)
            ) > 1
        ]
        self.assertGreater(
            len(xor_joins), 0,
            "Self-loop task B should have an "
            "XOR-join predecessor",
        )

    def test_self_loop_xor_split_after_b(
        self,
    ) -> None:
        """Task B has an XOR-split successor.

        The self-loop restoration inserts an XOR-split
        after B with a back-edge to the XOR-join.

        :return: None.
        """
        b: Task = self.__model.get_task("B")
        succs: set[Node] = (
            self.__model.successors(b)
        )
        xor_splits: list[Gateway] = [
            s for s in succs
            if isinstance(s, Gateway)
            and s.gateway_type == GatewayType.XOR
            and len(
                self.__model.outgoing_edges(s)
            ) > 1
        ]
        self.assertGreater(
            len(xor_splits), 0,
            "Self-loop task B should have an "
            "XOR-split successor",
        )


if __name__ == "__main__":
    unittest.main()