python-statemachine/tests/test_contrib_timeout.py at a993638b7737ddcb674cd1ab84dfcd19cbb87a7c · fgmacedo/python-statemachine · 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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
"""Tests for the timeout contrib module."""

import threading

import pytest
from statemachine.contrib.timeout import _Timeout
from statemachine.contrib.timeout import timeout

from statemachine import State
from statemachine import StateChart


class TestTimeoutValidation:
    def test_positive_duration(self):
        t = timeout(5)
        assert isinstance(t, _Timeout)
        assert t.duration == 5

    def test_zero_duration_raises(self):
        with pytest.raises(ValueError, match="must be positive"):
            timeout(0)

    def test_negative_duration_raises(self):
        with pytest.raises(ValueError, match="must be positive"):
            timeout(-1)

    def test_repr_without_on(self):
        assert repr(timeout(5)) == "timeout(5)"

    def test_repr_with_on(self):
        assert repr(timeout(3.5, on="expired")) == "timeout(3.5, on='expired')"


class TestTimeoutBasic:
    """Timeout fires done.invoke.<state> when no custom event is specified."""

    async def test_timeout_fires_done_invoke(self, sm_runner):
        class SM(StateChart):
            waiting = State(initial=True, invoke=timeout(0.05))
            done = State(final=True)
            done_invoke_waiting = waiting.to(done)

        sm = await sm_runner.start(SM)
        await sm_runner.sleep(0.15)
        await sm_runner.processing_loop(sm)

        assert "done" in sm.configuration_values

    async def test_timeout_cancelled_on_early_exit(self, sm_runner):
        """If the machine transitions out before the timeout, nothing fires."""

        class SM(StateChart):
            waiting = State(initial=True, invoke=timeout(10))
            other = State(final=True)
            go = waiting.to(other)
            # No done_invoke_waiting — would fail if timeout fired unexpectedly
            done_invoke_waiting = waiting.to(waiting)

        sm = await sm_runner.start(SM)
        await sm_runner.send(sm, "go")

        assert "other" in sm.configuration_values


class TestTimeoutCustomEvent:
    """Timeout fires a custom event via the `on` parameter."""

    async def test_custom_event_fires(self, sm_runner):
        class SM(StateChart):
            waiting = State(initial=True, invoke=timeout(0.05, on="expired"))
            timed_out = State(final=True)
            expired = waiting.to(timed_out)

        sm = await sm_runner.start(SM)
        await sm_runner.sleep(0.15)
        await sm_runner.processing_loop(sm)

        assert "timed_out" in sm.configuration_values

    async def test_custom_event_cancelled_on_early_exit(self, sm_runner):
        class SM(StateChart):
            waiting = State(initial=True, invoke=timeout(10, on="expired"))
            other = State(final=True)
            go = waiting.to(other)
            expired = waiting.to(waiting)

        sm = await sm_runner.start(SM)
        await sm_runner.send(sm, "go")

        assert "other" in sm.configuration_values


class TestTimeoutComposition:
    """Timeout combined with other invoke handlers — first to complete wins."""

    async def test_invoke_completes_before_timeout(self, sm_runner):
        """A fast invoke handler transitions out, cancelling the timeout."""

        def fast_handler():
            return "fast_result"

        class SM(StateChart):
            loading = State(initial=True, invoke=[fast_handler, timeout(10, on="too_slow")])
            ready = State(final=True)
            stuck = State(final=True)
            done_invoke_loading = loading.to(ready)
            too_slow = loading.to(stuck)

        sm = await sm_runner.start(SM)
        await sm_runner.sleep(0.15)
        await sm_runner.processing_loop(sm)

        assert "ready" in sm.configuration_values

    async def test_timeout_fires_before_slow_invoke(self, sm_runner):
        """Timeout fires while a slow invoke handler is still running."""
        handler_cancelled = threading.Event()

        class SlowHandler:
            def run(self, ctx):
                # Wait until cancelled (state exit) — simulates long-running work
                ctx.cancelled.wait()
                handler_cancelled.set()

        class SM(StateChart):
            loading = State(initial=True, invoke=[SlowHandler(), timeout(0.05, on="too_slow")])
            ready = State(final=True)
            stuck = State(final=True)
            done_invoke_loading = loading.to(ready)
            too_slow = loading.to(stuck)

        sm = await sm_runner.start(SM)
        await sm_runner.sleep(0.15)
        await sm_runner.processing_loop(sm)

        assert "stuck" in sm.configuration_values
        # The slow handler should have been cancelled when the state exited
        handler_cancelled.wait(timeout=2)
        assert handler_cancelled.is_set()