split-miner/tests/test_node_splitting.py

# 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 node-splitting normalization in build_rpst().

When aggressive edge filtering produces a graph with cut
vertices, the completed version C(G) is not biconnected.
Polyvyanyy et al. (2011), Section 4 describes node-splitting
as the correct fix: split each node with >1 incoming AND >1
outgoing edges into two nodes, making C(G) biconnected.

These tests verify that build_rpst() correctly applies
node-splitting normalization instead of falling back to a
single fragment.

Reference:
    Polyvyanyy, A., Vanhatalo, J., & Volzer, H. (2011).
    Simplified Computation and Generalization of the
    Refined Process Structure Tree. Section 4.
"""
from __future__ import annotations

import unittest

from split_miner.bpmn import (
    BPMNModel,
    EndEvent,
    Gateway,
    GatewayType,
    Node,
    StartEvent,
    Task,
)
from split_miner.joins import SESEFragment, build_rpst
from split_miner.joins import discover_joins


def _make_cut_vertex_model() -> tuple[
    BPMNModel, dict[str, Node],
]:
    """Build a model with a cut vertex at node c.

    Graph structure:
    - start -> a -> c -> d -> end
    - c -> e -> f -> c  (loop)

    Node c has 2 incoming edges (from a and f) and
    2 outgoing edges (to d and e).  In C(G), removing c
    disconnects {e, f} from the rest, making c a cut
    vertex (separation point).

    :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", "c", "d", "e", "f"]:
        t: Task = Task(label, label)
        model.add_task(t)
        tasks[label] = t

    nodes: dict[str, Node] = {
        "start": start, "end": end,
    }
    nodes.update(tasks)

    for src, tgt in [
        ("start", "a"), ("a", "c"),
        ("c", "d"), ("d", "end"),
        ("c", "e"), ("e", "f"), ("f", "c"),
    ]:
        model.add_edge(nodes[src], nodes[tgt])

    return model, nodes


class TestBuildRpstCutVertex(unittest.TestCase):
    """Tests build_rpst with a cut vertex graph.

    Verifies that node-splitting normalization produces
    proper RPST fragments instead of a single fallback
    fragment.
    """

    def setUp(self) -> None:
        """Set up the cut vertex model.

        :return: None.
        """
        self.__model: BPMNModel
        self.__nodes: dict[str, Node]
        self.__model, self.__nodes = (
            _make_cut_vertex_model()
        )
        self.__fragments: list[SESEFragment] = (
            build_rpst(self.__model)
        )

    def test_multiple_fragments(self) -> None:
        """Produces multiple fragments, not single fallback.

        With node-splitting normalization, the SPQR-tree
        should decompose the graph into multiple SESE
        fragments instead of falling back to a single
        R-type fragment.

        :return: None.
        """
        self.assertGreater(len(self.__fragments), 1)

    def test_all_edges_covered(self) -> None:
        """Union of fragment edges covers all model edges.

        :return: None.
        """
        all_frag_edges: set[tuple[Node, Node]] = set()
        for f in self.__fragments:
            all_frag_edges |= f.edges
        self.assertEqual(
            all_frag_edges, self.__model.edges
        )

    def test_entry_exit_are_model_nodes(self) -> None:
        """Entry and exit are original model nodes.

        No split proxy nodes should appear as fragment
        entry or exit.

        :return: None.
        """
        all_nodes: set[Node] = self.__model.all_nodes
        for f in self.__fragments:
            self.assertIn(f.entry, all_nodes)
            self.assertIn(f.exit_node, all_nodes)

    def test_fragment_nodes_are_model_nodes(self) -> None:
        """All fragment nodes are original model nodes.

        No split proxy nodes should leak into fragment
        node sets.

        :return: None.
        """
        all_nodes: set[Node] = self.__model.all_nodes
        for f in self.__fragments:
            for node in f.nodes:
                self.assertIn(
                    node, all_nodes,
                    f"Proxy node {node!r} leaked "
                    f"into fragment",
                )

    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 TestDiscoverJoinsCutVertex(unittest.TestCase):
    """Tests discover_joins on a graph with a cut vertex.

    Verifies that join gateway discovery works correctly
    when the graph requires node-splitting normalization.
    """

    def setUp(self) -> None:
        """Set up and run discover_joins.

        :return: None.
        """
        self.__model: BPMNModel
        self.__nodes: dict[str, Node]
        self.__model, self.__nodes = (
            _make_cut_vertex_model()
        )
        discover_joins(self.__model)

    def test_completes_without_raising(self) -> None:
        """discover_joins completes without exception.

        :return: None.
        """
        # If we get here, it didn't raise.
        self.assertTrue(True)

    def test_join_for_c(self) -> None:
        """Task c gets a join gateway predecessor.

        Task c has 2 incoming edges (from a and f),
        so it should get a join gateway.

        :return: None.
        """
        c: Task = self.__model.get_task("c")
        preds: set[Node] = (
            self.__model.predecessors(c)
        )
        self.assertEqual(len(preds), 1)
        join: Node = next(iter(preds))
        self.assertIsInstance(join, Gateway)

    def test_join_for_c_is_xor(self) -> None:
        """Task c's join is XOR (loop-join).

        The f -> c edge creates a cycle (c -> e -> f
        -> c), making this a loop-join which should be
        XOR per Definition 12 of the SM 1.0 paper.

        :return: None.
        """
        c: Task = self.__model.get_task("c")
        preds: set[Node] = (
            self.__model.predecessors(c)
        )
        join: Node = next(iter(preds))
        assert isinstance(join, Gateway)
        self.assertEqual(
            join.gateway_type, GatewayType.XOR,
            "Loop-join for c should be XOR",
        )


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