# Split Miner - BPMN process discovery from event logs.
# Authors:
#   imacat@mail.imacat.idv.tw (imacat), 2026/3/10
# 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 Split Miner algorithm.

Uses the running example from Section 3 of the SM 1.0 journal
paper (Augusto et al., 2018).
"""
from __future__ import annotations

import unittest
from collections import deque

from split_miner import (
    BPMNModel,
    Gateway,
    GatewayType,
    Node,
    Task,
    split_miner,
)
from split_miner.concurrency import PrunedDFG
from split_miner.dfg import DirectlyFollowsGraph
from split_miner.filtering import FilteredDFG


def _make_tasks(
    labels: str,
) -> dict[str, Task]:
    """Create a Task for each single-character label.

    :param labels: The labels as a string.
    :return: A dict mapping label to Task.
    """
    return {ch: Task(ch, ch) for ch in labels}


def _make_paper_node_log() -> tuple[
    dict[tuple[Node, ...], int], dict[str, Task]
]:
    """Build the paper example log with Node objects.

    :return: The Node-based traces and the task map.
    """
    t: dict[str, Task] = _make_tasks("abcdefgh")
    traces: dict[tuple[Node, ...], int] = {
        (t["a"], t["b"], t["c"], t["g"],
         t["e"], t["h"]): 10,
        (t["a"], t["b"], t["c"], t["f"],
         t["g"], t["h"]): 10,
        (t["a"], t["b"], t["d"], t["g"],
         t["e"], t["h"]): 10,
        (t["a"], t["b"], t["d"], t["e"],
         t["g"], t["h"]): 10,
        (t["a"], t["b"], t["e"], t["c"],
         t["g"], t["h"]): 10,
        (t["a"], t["b"], t["e"], t["d"],
         t["g"], t["h"]): 10,
        (t["a"], t["c"], t["b"], t["e"],
         t["g"], t["h"]): 10,
        (t["a"], t["c"], t["b"], t["f"],
         t["g"], t["h"]): 10,
        (t["a"], t["d"], t["b"], t["e"],
         t["g"], t["h"]): 10,
        (t["a"], t["d"], t["b"], t["f"],
         t["g"], t["h"]): 10,
    }
    return traces, t


def _make_paper_str_log() -> dict[
    tuple[str, ...], int
]:
    """Build the paper example log with string labels.

    :return: The string-based traces.
    """
    return {
        ("a", "b", "c", "g", "e", "h"): 10,
        ("a", "b", "c", "f", "g", "h"): 10,
        ("a", "b", "d", "g", "e", "h"): 10,
        ("a", "b", "d", "e", "g", "h"): 10,
        ("a", "b", "e", "c", "g", "h"): 10,
        ("a", "b", "e", "d", "g", "h"): 10,
        ("a", "c", "b", "e", "g", "h"): 10,
        ("a", "c", "b", "f", "g", "h"): 10,
        ("a", "d", "b", "e", "g", "h"): 10,
        ("a", "d", "b", "f", "g", "h"): 10,
    }


class TestDFGConstruction(unittest.TestCase):
    """Tests for DFG construction (Section 3.1)."""

    def setUp(self) -> None:
        """Set up the test.

        :return: None.
        """
        traces: dict[tuple[Node, ...], int]
        traces, self.__t = _make_paper_node_log()
        self.__dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )

    def test_nodes(self) -> None:
        """The DFG has the correct set of nodes."""
        self.assertEqual(
            self.__dfg.nodes,
            set(self.__t.values()),
        )

    def test_sources_and_sinks(self) -> None:
        """The DFG has correct sources and sinks."""
        self.assertIn(
            self.__t["a"], self.__dfg.sources
        )
        self.assertIn(
            self.__t["h"], self.__dfg.sinks
        )

    def test_df_frequencies(self) -> None:
        """Selected directly-follows frequencies match."""
        t: dict[str, Task] = self.__t
        # a -> b: appears in 6 trace types * 10 = 60
        self.assertEqual(
            self.__dfg.df_frequency(
                t["a"], t["b"]
            ), 60
        )
        # a -> c: 2 trace types * 10 = 20
        self.assertEqual(
            self.__dfg.df_frequency(
                t["a"], t["c"]
            ), 20
        )
        # a -> d: 2 trace types * 10 = 20
        self.assertEqual(
            self.__dfg.df_frequency(
                t["a"], t["d"]
            ), 20
        )

    def test_no_self_loops(self) -> None:
        """The paper example has no self-loops."""
        self.assertEqual(self.__dfg.self_loops, set())

    def test_no_short_loops(self) -> None:
        """The paper example has no short-loops."""
        self.assertEqual(
            self.__dfg.short_loops, set()
        )

    def test_self_loop_detection(self) -> None:
        """Self-loops are correctly detected."""
        t: dict[str, Task] = _make_tasks("abc")
        traces: dict[tuple[Node, ...], int] = {
            (t["a"], t["b"], t["b"], t["c"]): 10,
        }
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        self.assertIn(t["b"], dfg.self_loops)
        self.assertNotIn(t["a"], dfg.self_loops)

    def test_self_loop_edges_excluded(self) -> None:
        """Self-loop edges are excluded from edges set."""
        t: dict[str, Task] = _make_tasks("abc")
        traces: dict[tuple[Node, ...], int] = {
            (t["a"], t["b"], t["b"], t["c"]): 10,
        }
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        self.assertNotIn(
            (t["b"], t["b"]), dfg.edges
        )
        self.assertIn(
            (t["a"], t["b"]), dfg.edges
        )
        self.assertIn(
            (t["b"], t["c"]), dfg.edges
        )

    def test_short_loop_detection(self) -> None:
        """Short-loops are correctly detected."""
        t: dict[str, Task] = _make_tasks("abcd")
        traces: dict[tuple[Node, ...], int] = {
            (t["a"], t["b"], t["c"],
             t["b"], t["d"]): 10,
        }
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        self.assertIn(
            (t["b"], t["c"]), dfg.short_loops
        )
        self.assertIn(
            (t["c"], t["b"]), dfg.short_loops
        )


class TestConcurrencyDiscovery(unittest.TestCase):
    """Tests for concurrency discovery (Section 3.2)."""

    def setUp(self) -> None:
        """Set up the test.

        :return: None.
        """
        traces: dict[tuple[Node, ...], int]
        traces, self.__t = _make_paper_node_log()
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        self.__pdfg: PrunedDFG = PrunedDFG(
            dfg, epsilon=0.2
        )

    def test_concurrent_pairs(self) -> None:
        """Correct concurrency relations with epsilon=0.2.

        The paper identifies: b||c, b||d, d||e, e||g.
        """
        t: dict[str, Task] = self.__t
        # Check expected concurrent pairs
        self.assertTrue(
            self.__pdfg.is_concurrent(t["b"], t["c"])
        )
        self.assertTrue(
            self.__pdfg.is_concurrent(t["b"], t["d"])
        )
        self.assertTrue(
            self.__pdfg.is_concurrent(t["d"], t["e"])
        )
        self.assertTrue(
            self.__pdfg.is_concurrent(t["e"], t["g"])
        )
        # Non-concurrent pairs
        self.assertFalse(
            self.__pdfg.is_concurrent(t["a"], t["b"])
        )
        self.assertFalse(
            self.__pdfg.is_concurrent(t["c"], t["d"])
        )

    def test_self_loop_skipped_in_concurrency(
        self,
    ) -> None:
        """Self-loop nodes are never concurrent."""
        # b has a self-loop; a and b could look
        # concurrent but b should be skipped.
        t: dict[str, Task] = _make_tasks("abc")
        traces: dict[tuple[Node, ...], int] = {
            (t["a"], t["b"], t["b"],
             t["a"], t["c"]): 10,
            (t["a"], t["b"],
             t["a"], t["c"]): 10,
        }
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        pdfg: PrunedDFG = PrunedDFG(
            dfg, epsilon=1.0
        )
        self.assertFalse(
            pdfg.is_concurrent(t["a"], t["b"])
        )

    def test_short_loop_not_concurrent(self) -> None:
        """Short-loop pairs are not concurrent.

        Condition 4 prevents short-loop pairs from being
        declared concurrent.
        """
        t: dict[str, Task] = _make_tasks("abcd")
        traces: dict[tuple[Node, ...], int] = {
            (t["a"], t["b"], t["c"],
             t["b"], t["d"]): 10,
            (t["a"], t["c"],
             t["b"], t["d"]): 10,
        }
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        pdfg: PrunedDFG = PrunedDFG(
            dfg, epsilon=1.0
        )
        self.assertFalse(
            pdfg.is_concurrent(t["b"], t["c"])
        )


class TestFiltering(unittest.TestCase):
    """Tests for edge filtering (Section 3.3)."""

    def setUp(self) -> None:
        """Set up the test.

        :return: None.
        """
        traces: dict[tuple[Node, ...], int]
        traces, self.__t = _make_paper_node_log()
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        pdfg: PrunedDFG = PrunedDFG(
            dfg, epsilon=0.2
        )
        self.__fdfg: FilteredDFG = FilteredDFG(
            pdfg, eta=1.0
        )

    def test_filtered_edges_retain_best(self) -> None:
        """Filtered DFG retains best incoming/outgoing
        edges.

        Per Table 1 in the paper, edges (e,c) and (c,f)
        should be dropped.
        """
        t: dict[str, Task] = self.__t
        edges: set[tuple[Node, Node]] = (
            self.__fdfg.edges
        )
        # These should be retained (best edges)
        self.assertIn((t["a"], t["b"]), edges)
        self.assertIn((t["b"], t["e"]), edges)
        self.assertIn((t["f"], t["g"]), edges)
        self.assertIn((t["g"], t["h"]), edges)

    def test_sources_and_sinks_preserved(self) -> None:
        """Filtering preserves sources and sinks."""
        t: dict[str, Task] = self.__t
        self.assertIn(
            t["a"], self.__fdfg.sources
        )
        self.assertIn(
            t["h"], self.__fdfg.sinks
        )


class TestEndToEnd(unittest.TestCase):
    """End-to-end test for Split Miner."""

    def test_paper_example_basic(self) -> None:
        """Split Miner produces a valid BPMN model.

        The discovered model should have start/end events,
        all 8 tasks, gateways, and proper connectivity.
        """
        traces: dict[tuple[str, ...], int] = (
            _make_paper_str_log()
        )
        model: BPMNModel = split_miner(
            traces, epsilon=0.2, eta=0.4
        )
        # Has start and end
        self.assertIsNotNone(model.start)
        self.assertIsNotNone(model.end)
        # Has all 8 tasks
        task_labels: set[str] = {
            t.label for t in model.tasks.values()
            if t.label is not None
        }
        self.assertEqual(
            task_labels,
            {"a", "b", "c", "d", "e", "f", "g", "h"},
        )
        # Has edges
        self.assertGreater(len(model.edges), 0)
        # Start has outgoing edge
        self.assertGreater(
            len(model.outgoing_edges(model.start)), 0
        )
        # End has incoming edge
        self.assertGreater(
            len(model.incoming_edges(model.end)), 0
        )

    def test_paper_example_has_gateways(self) -> None:
        """The paper example produces split and join
        gateways.

        Per Fig. 3c, the model should have both XOR and
        AND gateways (or OR gateways that get minimized).
        """
        traces: dict[tuple[str, ...], int] = (
            _make_paper_str_log()
        )
        model: BPMNModel = split_miner(
            traces, epsilon=0.2, eta=0.4
        )
        self.assertGreater(len(model.gateways), 0)
        gw_types: set[GatewayType] = {
            gw.gateway_type
            for gw in model.gateways.values()
        }
        # Should have at least XOR or AND gateways
        self.assertTrue(
            GatewayType.XOR in gw_types
            or GatewayType.AND in gw_types,
            f"Expected XOR or AND gateways, "
            f"got {gw_types}"
        )

    def test_paper_example_all_tasks_connected(
        self,
    ) -> None:
        """Every task is reachable from start.

        Verifies syntactic correctness: all tasks on a
        path from start to end.
        """
        traces: dict[tuple[str, ...], int] = (
            _make_paper_str_log()
        )
        model: BPMNModel = split_miner(
            traces, epsilon=0.2, eta=0.4
        )
        # BFS from start
        reachable: set[Node] = set()
        queue: deque[Node] = deque([model.start])
        while queue:
            node: Node = queue.popleft()
            if node in reachable:
                continue
            reachable.add(node)
            for _, succ in model.outgoing_edges(node):
                queue.append(succ)
        # All tasks should be reachable
        for task in model.tasks.values():
            self.assertIn(
                task, reachable,
                f"Task {task.label!r} not reachable "
                f"from start"
            )
        # End should be reachable
        self.assertIn(model.end, reachable)

    def test_paper_example_all_tasks_reach_end(
        self,
    ) -> None:
        """Every task can reach the end event.

        Verifies syntactic correctness by backward BFS.
        """
        traces: dict[tuple[str, ...], int] = (
            _make_paper_str_log()
        )
        model: BPMNModel = split_miner(
            traces, epsilon=0.2, eta=0.4
        )
        # Backward BFS from end
        can_reach_end: set[Node] = set()
        queue: deque[Node] = deque([model.end])
        while queue:
            node: Node = queue.popleft()
            if node in can_reach_end:
                continue
            can_reach_end.add(node)
            for pred, _ in model.incoming_edges(node):
                queue.append(pred)
        # All tasks should reach end
        for task in model.tasks.values():
            self.assertIn(
                task, can_reach_end,
                f"Task {task.label!r} cannot reach end"
            )

    def test_simple_sequence(self) -> None:
        """A simple sequential log produces no gateways."""
        traces: dict[tuple[str, ...], int] = {
            ("a", "b", "c"): 10,
        }
        model: BPMNModel = split_miner(traces)
        self.assertEqual(len(model.gateways), 0)
        self.assertEqual(len(model.tasks), 3)

    def test_simple_xor_choice(self) -> None:
        """A log with exclusive choice produces XOR
        gateways.

        Log: {<a,b,d>^10, <a,c,d>^10}
        Expected: a -> XOR-split -> {b, c} ->
                  XOR-join -> d
        """
        traces: dict[tuple[str, ...], int] = {
            ("a", "b", "d"): 10,
            ("a", "c", "d"): 10,
        }
        model: BPMNModel = split_miner(
            traces, epsilon=0.1, eta=0.4
        )
        # Should have tasks a, b, c, d
        task_labels: set[str] = {
            t.label for t in model.tasks.values()
            if t.label is not None
        }
        self.assertEqual(
            task_labels, {"a", "b", "c", "d"}
        )
        # Should have gateways
        self.assertGreater(len(model.gateways), 0)
        # All tasks reachable from start
        reachable: set[Node] = set()
        queue: deque[Node] = deque([model.start])
        while queue:
            node: Node = queue.popleft()
            if node in reachable:
                continue
            reachable.add(node)
            for _, s in model.outgoing_edges(node):
                queue.append(s)
        for task in model.tasks.values():
            self.assertIn(task, reachable)

    def test_simple_concurrency(self) -> None:
        """A log with concurrency produces AND gateways.

        Log: {<a,b,c,d>^10, <a,c,b,d>^10}
        b and c are concurrent.
        """
        traces: dict[tuple[str, ...], int] = {
            ("a", "b", "c", "d"): 10,
            ("a", "c", "b", "d"): 10,
        }
        model: BPMNModel = split_miner(
            traces, epsilon=1.0, eta=0.4
        )
        task_labels: set[str] = {
            t.label for t in model.tasks.values()
            if t.label is not None
        }
        self.assertEqual(
            task_labels, {"a", "b", "c", "d"}
        )
        # Should have AND gateways for b||c
        and_gws: list[Gateway] = [
            gw for gw in model.gateways.values()
            if gw.gateway_type == GatewayType.AND
        ]
        self.assertGreater(
            len(and_gws), 0,
            "Expected AND gateways for concurrent "
            "b and c"
        )


class TestSelfLoopHandling(unittest.TestCase):
    """Tests for self-loop handling."""

    def test_self_loop_restored(self) -> None:
        """Self-loops are restored in the final BPMN model.

        A self-loop on task b should produce XOR-join and
        XOR-split gateways around b with a back-edge.
        """
        traces: dict[tuple[str, ...], int] = {
            ("a", "b", "c"): 10,
            ("a", "b", "b", "c"): 10,
            ("a", "b", "b", "b", "c"): 10,
        }
        model: BPMNModel = split_miner(traces)
        # Task b should have a gateway predecessor
        # and a gateway successor (the self-loop
        # XOR-join and XOR-split)
        b: Node = model.get_task("b")
        b_preds: set[Node] = model.predecessors(b)
        b_succs: set[Node] = model.successors(b)
        # b should have exactly 1 predecessor (XOR-join)
        # and 1 successor (XOR-split)
        self.assertEqual(len(b_preds), 1)
        self.assertEqual(len(b_succs), 1)
        join_node: Node = next(iter(b_preds))
        split_node: Node = next(iter(b_succs))
        self.assertIsInstance(join_node, Gateway)
        self.assertIsInstance(split_node, Gateway)
        assert isinstance(join_node, Gateway)
        assert isinstance(split_node, Gateway)
        self.assertEqual(
            join_node.gateway_type, GatewayType.XOR
        )
        self.assertEqual(
            split_node.gateway_type, GatewayType.XOR
        )
        # Back-edge: split -> join
        self.assertIn(
            split_node,
            model.predecessors(join_node)
        )


class TestShortLoopHandling(unittest.TestCase):
    """Tests for short-loop handling."""

    def test_short_loop_not_concurrent(self) -> None:
        """Short-loop pairs are excluded from concurrency.

        If a and b form a short-loop, they must not be
        declared concurrent even if they appear in both
        orders.
        """
        # a,b,a pattern = short-loop
        t: dict[str, Task] = _make_tasks("abxy")
        traces: dict[tuple[Node, ...], int] = {
            (t["x"], t["a"], t["b"],
             t["a"], t["y"]): 10,
            (t["x"], t["b"], t["a"],
             t["b"], t["y"]): 10,
            (t["x"], t["a"], t["y"]): 10,
            (t["x"], t["b"], t["y"]): 10,
        }
        dfg: DirectlyFollowsGraph = (
            DirectlyFollowsGraph(traces)
        )
        # Should detect short-loop
        self.assertIn(
            (t["a"], t["b"]), dfg.short_loops
        )
        # Should NOT be concurrent
        pdfg: PrunedDFG = PrunedDFG(
            dfg, epsilon=1.0
        )
        self.assertFalse(
            pdfg.is_concurrent(t["a"], t["b"])
        )


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