# 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.
"""Detailed pipeline stage tests for the paper example.

Verifies each stage of the Split Miner pipeline using the
running example from Section 3 of the SM 1.0 paper.
"""
from __future__ import annotations

import unittest

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 TestDFGAllEdges(unittest.TestCase):
    """Tests for all DFG edge frequencies (Table 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_all_edge_frequencies(self) -> None:
        """All 20 DFG edge frequencies match Table 1.

        Verifies every directly-follows frequency from the
        paper's example event log.
        """
        t: dict[str, Task] = self.__t
        expected: dict[tuple[Node, Node], int] = {
            (t["a"], t["b"]): 60,
            (t["a"], t["c"]): 20,
            (t["a"], t["d"]): 20,
            (t["b"], t["c"]): 20,
            (t["b"], t["d"]): 20,
            (t["b"], t["e"]): 40,
            (t["b"], t["f"]): 20,
            (t["c"], t["b"]): 20,
            (t["c"], t["f"]): 10,
            (t["c"], t["g"]): 20,
            (t["d"], t["b"]): 20,
            (t["d"], t["e"]): 10,
            (t["d"], t["g"]): 20,
            (t["e"], t["c"]): 10,
            (t["e"], t["d"]): 10,
            (t["e"], t["g"]): 30,
            (t["e"], t["h"]): 20,
            (t["f"], t["g"]): 30,
            (t["g"], t["e"]): 20,
            (t["g"], t["h"]): 80,
        }
        for (src, tgt), freq in expected.items():
            self.assertEqual(
                self.__dfg.df_frequency(src, tgt),
                freq,
                f"|{src.node_id} -> {tgt.node_id}|"
                f" should be {freq}"
            )

    def test_all_edges(self) -> None:
        """The DFG has exactly 20 edges."""
        t: dict[str, Task] = self.__t
        expected: set[tuple[Node, Node]] = {
            (t["a"], t["b"]), (t["a"], t["c"]),
            (t["a"], t["d"]),
            (t["b"], t["c"]), (t["b"], t["d"]),
            (t["b"], t["e"]), (t["b"], t["f"]),
            (t["c"], t["b"]), (t["c"], t["f"]),
            (t["c"], t["g"]),
            (t["d"], t["b"]), (t["d"], t["e"]),
            (t["d"], t["g"]),
            (t["e"], t["c"]), (t["e"], t["d"]),
            (t["e"], t["g"]), (t["e"], t["h"]),
            (t["f"], t["g"]),
            (t["g"], t["e"]), (t["g"], t["h"]),
        }
        self.assertEqual(self.__dfg.edges, expected)

    def test_edge_count(self) -> None:
        """The DFG has 20 edges."""
        self.assertEqual(len(self.__dfg.edges), 20)


class TestPrunedDFGEdges(unittest.TestCase):
    """Tests for PDFG edge set (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_pdfg_edges(self) -> None:
        """PDFG has 12 edges after concurrent pruning.

        Concurrent pairs b||c, b||d, d||e, e||g are
        removed along with their reverse edges.
        """
        t: dict[str, Task] = self.__t
        expected: set[tuple[Node, Node]] = {
            (t["a"], t["b"]), (t["a"], t["c"]),
            (t["a"], t["d"]),
            (t["b"], t["e"]), (t["b"], t["f"]),
            (t["c"], t["f"]), (t["c"], t["g"]),
            (t["d"], t["g"]),
            (t["e"], t["c"]), (t["e"], t["h"]),
            (t["f"], t["g"]),
            (t["g"], t["h"]),
        }
        self.assertEqual(self.__pdfg.edges, expected)

    def test_pdfg_edge_count(self) -> None:
        """PDFG has 12 edges."""
        self.assertEqual(len(self.__pdfg.edges), 12)

    def test_concurrent_pairs(self) -> None:
        """All four concurrent pairs are detected."""
        t: dict[str, Task] = self.__t
        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"])
        )

    def test_not_concurrent(self) -> None:
        """Non-concurrent pairs."""
        t: dict[str, Task] = self.__t
        self.assertFalse(
            self.__pdfg.is_concurrent(t["a"], t["b"])
        )
        self.assertFalse(
            self.__pdfg.is_concurrent(t["c"], t["d"])
        )
        self.assertFalse(
            self.__pdfg.is_concurrent(t["c"], t["f"])
        )


class TestFilteredDFGEdges(unittest.TestCase):
    """Tests for filtered PDFG edge set (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=0.4
        )

    def test_filtered_edges(self) -> None:
        """Filtered PDFG has 10 edges.

        Edges c->f and e->c are filtered out.
        """
        t: dict[str, Task] = self.__t
        expected: set[tuple[Node, Node]] = {
            (t["a"], t["b"]), (t["a"], t["c"]),
            (t["a"], t["d"]),
            (t["b"], t["e"]), (t["b"], t["f"]),
            (t["c"], t["g"]),
            (t["d"], t["g"]),
            (t["e"], t["h"]),
            (t["f"], t["g"]),
            (t["g"], t["h"]),
        }
        self.assertEqual(self.__fdfg.edges, expected)

    def test_filtered_edge_count(self) -> None:
        """Filtered PDFG has 10 edges."""
        self.assertEqual(len(self.__fdfg.edges), 10)

    def test_removed_edges(self) -> None:
        """Edges c->f and e->c are not in filtered PDFG."""
        t: dict[str, Task] = self.__t
        self.assertNotIn(
            (t["c"], t["f"]), self.__fdfg.edges
        )
        self.assertNotIn(
            (t["e"], t["c"]), self.__fdfg.edges
        )


class TestPaperExampleStructure(unittest.TestCase):
    """Tests for the final paper example structure."""

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

        :return: None.
        """
        traces: dict[tuple[str, ...], int] = (
            _make_paper_str_log()
        )
        self.__model: BPMNModel = split_miner(
            traces, epsilon=0.2, eta=0.4
        )

    def test_node_count(self) -> None:
        """The final model has 16 nodes."""
        self.assertEqual(
            len(self.__model.all_nodes), 16
        )

    def test_edge_count(self) -> None:
        """The final model has 18 edges."""
        self.assertEqual(len(self.__model.edges), 18)

    def test_gateway_counts(self) -> None:
        """6 gateways: AND=1, XOR=4, OR=1."""
        gw_types: list[GatewayType] = [
            gw.gateway_type
            for gw in self.__model.gateways.values()
        ]
        self.assertEqual(len(gw_types), 6)
        self.assertEqual(
            gw_types.count(GatewayType.AND), 1
        )
        self.assertEqual(
            gw_types.count(GatewayType.XOR), 4
        )
        self.assertEqual(
            gw_types.count(GatewayType.OR), 1
        )

    def test_and_split_after_a(self) -> None:
        """Task a leads to an AND split gateway."""
        a: Task = self.__model.get_task("a")
        a_succs: set[Node] = (
            self.__model.successors(a)
        )
        self.assertEqual(len(a_succs), 1)
        and_gw: Node = next(iter(a_succs))
        self.assertIsInstance(and_gw, Gateway)
        assert isinstance(and_gw, Gateway)
        self.assertEqual(
            and_gw.gateway_type, GatewayType.AND
        )

    def test_and_split_successors(self) -> None:
        """AND split has successors b and XOR split."""
        a: Task = self.__model.get_task("a")
        and_gw: Node = next(
            iter(self.__model.successors(a))
        )
        and_succs: set[Node] = (
            self.__model.successors(and_gw)
        )
        self.assertEqual(len(and_succs), 2)
        b: Task = self.__model.get_task("b")
        self.assertIn(b, and_succs)

    def test_xor_split_cd(self) -> None:
        """XOR split for c and d (successor of AND)."""
        a: Task = self.__model.get_task("a")
        and_gw: Node = next(
            iter(self.__model.successors(a))
        )
        and_succs: set[Node] = (
            self.__model.successors(and_gw)
        )
        b: Task = self.__model.get_task("b")
        xor1: Node = (and_succs - {b}).pop()
        self.assertIsInstance(xor1, Gateway)
        assert isinstance(xor1, Gateway)
        self.assertEqual(
            xor1.gateway_type, GatewayType.XOR
        )
        xor1_succs: set[Node] = (
            self.__model.successors(xor1)
        )
        c: Task = self.__model.get_task("c")
        d: Task = self.__model.get_task("d")
        self.assertEqual(xor1_succs, {c, d})

    def test_xor_split_ef(self) -> None:
        """XOR split for e and f (after b)."""
        b: Task = self.__model.get_task("b")
        b_succs: set[Node] = (
            self.__model.successors(b)
        )
        self.assertEqual(len(b_succs), 1)
        xor2: Node = next(iter(b_succs))
        self.assertIsInstance(xor2, Gateway)
        assert isinstance(xor2, Gateway)
        self.assertEqual(
            xor2.gateway_type, GatewayType.XOR
        )
        xor2_succs: set[Node] = (
            self.__model.successors(xor2)
        )
        e: Task = self.__model.get_task("e")
        f: Task = self.__model.get_task("f")
        self.assertEqual(xor2_succs, {e, f})

    def test_xor_join_cd(self) -> None:
        """XOR join for c and d."""
        c: Task = self.__model.get_task("c")
        d: Task = self.__model.get_task("d")
        c_succs: set[Node] = (
            self.__model.successors(c)
        )
        d_succs: set[Node] = (
            self.__model.successors(d)
        )
        self.assertEqual(len(c_succs), 1)
        self.assertEqual(len(d_succs), 1)
        self.assertEqual(c_succs, d_succs)
        join: Node = next(iter(c_succs))
        self.assertIsInstance(join, Gateway)
        assert isinstance(join, Gateway)
        self.assertEqual(
            join.gateway_type, GatewayType.XOR
        )

    def test_or_join_to_g(self) -> None:
        """OR join for {XOR-join, f} leading to g."""
        g: Task = self.__model.get_task("g")
        g_preds: set[Node] = (
            self.__model.predecessors(g)
        )
        self.assertEqual(len(g_preds), 1)
        or_gw: Node = next(iter(g_preds))
        self.assertIsInstance(or_gw, Gateway)
        assert isinstance(or_gw, Gateway)
        self.assertEqual(
            or_gw.gateway_type, GatewayType.OR
        )
        or_preds: set[Node] = (
            self.__model.predecessors(or_gw)
        )
        self.assertEqual(len(or_preds), 2)
        f: Task = self.__model.get_task("f")
        self.assertIn(f, or_preds)

    def test_xor_join_to_h(self) -> None:
        """XOR join for {e, g} leading to h.

        This was an OR-join that became XOR after
        OR-joins minimization (Algorithm 9).
        """
        h: Task = self.__model.get_task("h")
        h_preds: set[Node] = (
            self.__model.predecessors(h)
        )
        self.assertEqual(len(h_preds), 1)
        join: Node = next(iter(h_preds))
        self.assertIsInstance(join, Gateway)
        assert isinstance(join, Gateway)
        self.assertEqual(
            join.gateway_type, GatewayType.XOR
        )
        join_preds: set[Node] = (
            self.__model.predecessors(join)
        )
        self.assertEqual(len(join_preds), 2)
        e: Task = self.__model.get_task("e")
        g: Task = self.__model.get_task("g")
        self.assertIn(e, join_preds)
        self.assertIn(g, join_preds)


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