prosim/tests/conftest.py at main · alch3mistdev/prosim · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
"""Shared test fixtures."""

import pytest

from prosim.graph.models import (
    Edge,
    EdgeType,
    Node,
    NodeParams,
    NodeType,
    WorkflowGraph,
)


@pytest.fixture
def linear_workflow() -> WorkflowGraph:
    """Simple linear workflow: start -> validate -> process -> end."""
    return WorkflowGraph(
        name="Linear Invoice",
        nodes=[
            Node(id="start", name="Start", node_type=NodeType.START, params=NodeParams(exec_time_mean=0.0)),
            Node(
                id="validate",
                name="Validate Invoice",
                node_type=NodeType.API,
                params=NodeParams(
                    exec_time_mean=2.0,
                    exec_time_variance=0.5,
                    cost_per_transaction=0.01,
                    error_rate=0.02,
                    max_retries=2,
                    retry_delay=1.0,
                ),
            ),
            Node(
                id="process",
                name="Process Payment",
                node_type=NodeType.API,
                params=NodeParams(
                    exec_time_mean=5.0,
                    exec_time_variance=1.0,
                    cost_per_transaction=0.05,
                    error_rate=0.01,
                    queue_delay_mean=3.0,
                ),
            ),
            Node(id="end", name="End", node_type=NodeType.END, params=NodeParams(exec_time_mean=0.0)),
        ],
        edges=[
            Edge(source="start", target="validate"),
            Edge(source="validate", target="process"),
            Edge(source="process", target="end"),
        ],
    )


@pytest.fixture
def branching_workflow() -> WorkflowGraph:
    """Workflow with decision node and two branches."""
    return WorkflowGraph(
        name="Approval Flow",
        nodes=[
            Node(id="start", name="Start", node_type=NodeType.START, params=NodeParams(exec_time_mean=0.0)),
            Node(
                id="review",
                name="Manual Review",
                node_type=NodeType.HUMAN,
                params=NodeParams(
                    exec_time_mean=300.0,
                    exec_time_variance=100.0,
                    cost_per_transaction=5.0,
                    error_rate=0.03,
                ),
            ),
            Node(
                id="decide",
                name="Approval Decision",
                node_type=NodeType.DECISION,
                params=NodeParams(exec_time_mean=0.1),
            ),
            Node(
                id="approve",
                name="Process Approval",
                node_type=NodeType.API,
                params=NodeParams(
                    exec_time_mean=1.0,
                    cost_per_transaction=0.01,
                ),
            ),
            Node(
                id="reject",
                name="Send Rejection",
                node_type=NodeType.ASYNC,
                params=NodeParams(
                    exec_time_mean=0.5,
                    cost_per_transaction=0.005,
                ),
            ),
            Node(id="end", name="End", node_type=NodeType.END, params=NodeParams(exec_time_mean=0.0)),
        ],
        edges=[
            Edge(source="start", target="review"),
            Edge(source="review", target="decide"),
            Edge(source="decide", target="approve", probability=0.7, edge_type=EdgeType.CONDITIONAL, condition="Approved"),
            Edge(source="decide", target="reject", probability=0.3, edge_type=EdgeType.CONDITIONAL, condition="Rejected"),
            Edge(source="approve", target="end"),
            Edge(source="reject", target="end"),
        ],
    )