forked from fgmacedo/python-statemachine
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_statechart_delayed.py
More file actions
100 lines (71 loc) · 3.18 KB
/
Copy pathtest_statechart_delayed.py
File metadata and controls
100 lines (71 loc) · 3.18 KB
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
"""Delayed event sends and cancellations.
Tests exercise queuing events with a delay (fires after elapsed time),
cancelling delayed events before they fire, zero-delay immediate firing,
and the Event(delay=...) definition syntax.
Theme: Beacons of Gondor — signal fires propagate with timing.
"""
import asyncio
import pytest
from statemachine.event import BoundEvent
from statemachine import Event
from statemachine import State
from statemachine import StateChart
@pytest.mark.timeout(10)
class TestDelayedEvents:
async def test_delayed_event_fires_after_delay(self, sm_runner):
"""Queuing a delayed event does not fire immediately; processing after delay does."""
class BeaconsOfGondor(StateChart):
dark = State(initial=True)
first_lit = State()
all_lit = State(final=True)
light_first = dark.to(first_lit)
light_all = first_lit.to(all_lit)
sm = await sm_runner.start(BeaconsOfGondor)
await sm_runner.send(sm, "light_first")
assert "first_lit" in sm.configuration_values
# Queue the event with delay without triggering the processing loop
event = BoundEvent(id="light_all", name="Light all", delay=50, _sm=sm)
event.put()
# Not yet processed
assert "first_lit" in sm.configuration_values
await asyncio.sleep(0.1)
await sm_runner.processing_loop(sm)
assert "all_lit" in sm.configuration_values
async def test_cancel_delayed_event(self, sm_runner):
"""Cancelled delayed events do not fire."""
class BeaconsOfGondor(StateChart):
dark = State(initial=True)
lit = State(final=True)
light = dark.to(lit)
sm = await sm_runner.start(BeaconsOfGondor)
# Queue delayed event
event = BoundEvent(id="light", name="Light", delay=500, _sm=sm)
event.put(send_id="beacon_signal")
sm.cancel_event("beacon_signal")
await asyncio.sleep(0.1)
await sm_runner.processing_loop(sm)
assert "dark" in sm.configuration_values
async def test_zero_delay_fires_immediately(self, sm_runner):
"""delay=0 fires immediately."""
class BeaconsOfGondor(StateChart):
dark = State(initial=True)
lit = State(final=True)
light = dark.to(lit)
sm = await sm_runner.start(BeaconsOfGondor)
await sm_runner.send(sm, "light", delay=0)
assert "lit" in sm.configuration_values
async def test_delayed_event_on_event_definition(self, sm_runner):
"""Event(transitions, delay=100) syntax queues with a delay."""
class BeaconsOfGondor(StateChart):
dark = State(initial=True)
lit = State(final=True)
light = Event(dark.to(lit), delay=50)
sm = await sm_runner.start(BeaconsOfGondor)
# Queue via BoundEvent.put() to avoid blocking in processing_loop
event = BoundEvent(id="light", name="Light", delay=50, _sm=sm)
event.put()
# Not yet processed
assert "dark" in sm.configuration_values
await asyncio.sleep(0.1)
await sm_runner.processing_loop(sm)
assert "lit" in sm.configuration_values