Skip to content

Commit 6dd9ab8

Browse files
committed
test: add dedicated backward-compat tests for StateMachine (v2 API)
Cover all four flag defaults, TransitionNotAllowed behavior (sync and async), error_on_execution=False propagation, self-transition entries, current_state deprecated property, and basic smoke tests.
1 parent 531a5f2 commit 6dd9ab8

1 file changed

Lines changed: 374 additions & 0 deletions

File tree

tests/test_statemachine_compat.py

Lines changed: 374 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,374 @@
1+
"""Backward-compatibility tests for the StateMachine (v2) API.
2+
3+
These tests verify that ``StateMachine`` (which inherits from ``StateChart``
4+
with different defaults) continues to work as expected. Tests here exercise
5+
behaviour that differs from ``StateChart`` defaults:
6+
7+
- ``allow_event_without_transition = False`` → ``TransitionNotAllowed``
8+
- ``enable_self_transition_entries = False``
9+
- ``atomic_configuration_update = True``
10+
- ``error_on_execution = False`` → exceptions propagate directly
11+
- ``current_state`` deprecated property
12+
"""
13+
14+
import warnings
15+
16+
import pytest
17+
18+
from statemachine import State
19+
from statemachine import StateMachine
20+
from statemachine import exceptions
21+
22+
# ---------------------------------------------------------------------------
23+
# Flag defaults
24+
# ---------------------------------------------------------------------------
25+
26+
27+
class TestStateMachineDefaults:
28+
"""Verify the four class-level flag defaults on StateMachine."""
29+
30+
def test_allow_event_without_transition(self):
31+
assert StateMachine.allow_event_without_transition is False
32+
33+
def test_enable_self_transition_entries(self):
34+
assert StateMachine.enable_self_transition_entries is False
35+
36+
def test_atomic_configuration_update(self):
37+
assert StateMachine.atomic_configuration_update is True
38+
39+
def test_error_on_execution(self):
40+
assert StateMachine.error_on_execution is False
41+
42+
43+
# ---------------------------------------------------------------------------
44+
# Smoke test
45+
# ---------------------------------------------------------------------------
46+
47+
48+
class TestStateMachineSmoke:
49+
"""StateMachine as a subclass works for basic operations."""
50+
51+
def test_create_send_and_check_state(self):
52+
class TrafficLight(StateMachine):
53+
green = State(initial=True)
54+
yellow = State()
55+
red = State()
56+
57+
cycle = green.to(yellow) | yellow.to(red) | red.to(green)
58+
59+
sm = TrafficLight()
60+
assert sm.green.is_active
61+
62+
sm.send("cycle")
63+
assert sm.yellow.is_active
64+
65+
sm.send("cycle")
66+
assert sm.red.is_active
67+
68+
def test_final_state_terminates(self):
69+
class Simple(StateMachine):
70+
s1 = State(initial=True)
71+
s2 = State(final=True)
72+
73+
go = s1.to(s2)
74+
75+
sm = Simple()
76+
sm.send("go")
77+
assert sm.is_terminated
78+
79+
80+
# ---------------------------------------------------------------------------
81+
# TransitionNotAllowed (allow_event_without_transition = False)
82+
# ---------------------------------------------------------------------------
83+
84+
85+
class TestTransitionNotAllowed:
86+
"""StateMachine raises TransitionNotAllowed for invalid events."""
87+
88+
@pytest.fixture()
89+
def sm(self):
90+
class Workflow(StateMachine):
91+
draft = State(initial=True)
92+
published = State(final=True)
93+
94+
publish = draft.to(published)
95+
96+
return Workflow()
97+
98+
def test_invalid_event_raises(self, sm):
99+
with pytest.raises(exceptions.TransitionNotAllowed):
100+
sm.send("nonexistent")
101+
102+
def test_event_not_available_in_current_state(self, sm):
103+
sm.send("publish")
104+
with pytest.raises(exceptions.TransitionNotAllowed):
105+
sm.send("publish")
106+
107+
def test_condition_blocks_transition(self):
108+
class Gated(StateMachine):
109+
s1 = State(initial=True)
110+
s2 = State(final=True)
111+
112+
go = s1.to(s2, cond="allowed")
113+
114+
allowed: bool = False
115+
116+
sm = Gated()
117+
with pytest.raises(sm.TransitionNotAllowed):
118+
sm.go()
119+
120+
def test_multiple_destinations_all_blocked(self):
121+
def never(event_data):
122+
return False
123+
124+
class Multi(StateMachine):
125+
requested = State(initial=True)
126+
accepted = State(final=True)
127+
rejected = State(final=True)
128+
129+
validate = requested.to(accepted, cond=never) | requested.to(
130+
rejected, cond="also_never"
131+
)
132+
133+
@property
134+
def also_never(self):
135+
return False
136+
137+
sm = Multi()
138+
with pytest.raises(exceptions.TransitionNotAllowed):
139+
sm.validate()
140+
assert sm.requested.is_active
141+
142+
def test_from_any_with_cond_blocked(self):
143+
class Account(StateMachine):
144+
active = State(initial=True)
145+
closed = State(final=True)
146+
147+
close = closed.from_.any(cond="can_close")
148+
149+
can_close: bool = False
150+
151+
sm = Account()
152+
with pytest.raises(sm.TransitionNotAllowed):
153+
sm.close()
154+
155+
def test_condition_algebra_any_false(self):
156+
class CondAlgebra(StateMachine):
157+
start = State(initial=True)
158+
end = State(final=True)
159+
160+
submit = start.to(end, cond="used_money or used_credit")
161+
162+
used_money: bool = False
163+
used_credit: bool = False
164+
165+
sm = CondAlgebra()
166+
with pytest.raises(sm.TransitionNotAllowed):
167+
sm.submit()
168+
169+
170+
# ---------------------------------------------------------------------------
171+
# TransitionNotAllowed — async
172+
# ---------------------------------------------------------------------------
173+
174+
175+
class TestTransitionNotAllowedAsync:
176+
"""TransitionNotAllowed in async machines."""
177+
178+
@pytest.fixture()
179+
def async_sm_cls(self):
180+
class AsyncWorkflow(StateMachine):
181+
s1 = State(initial=True)
182+
s2 = State()
183+
s3 = State(final=True)
184+
185+
go = s1.to(s2, cond="is_ready")
186+
finish = s2.to(s3)
187+
188+
is_ready: bool = False
189+
190+
async def on_go(self): ...
191+
192+
return AsyncWorkflow
193+
194+
async def test_async_transition_not_allowed(self, async_sm_cls):
195+
sm = async_sm_cls()
196+
await sm.activate_initial_state()
197+
with pytest.raises(sm.TransitionNotAllowed):
198+
await sm.send("go")
199+
200+
def test_sync_context_transition_not_allowed(self, async_sm_cls):
201+
sm = async_sm_cls()
202+
with pytest.raises(sm.TransitionNotAllowed):
203+
sm.send("go")
204+
205+
async def test_async_condition_blocks(self):
206+
class AsyncCond(StateMachine):
207+
s1 = State(initial=True)
208+
s2 = State(final=True)
209+
210+
go = s1.to(s2, cond="check")
211+
212+
async def check(self):
213+
return False
214+
215+
sm = AsyncCond()
216+
await sm.activate_initial_state()
217+
with pytest.raises(sm.TransitionNotAllowed):
218+
await sm.go()
219+
220+
221+
# ---------------------------------------------------------------------------
222+
# error_on_execution = False (exceptions propagate directly)
223+
# ---------------------------------------------------------------------------
224+
225+
226+
class TestErrorOnExecutionFalse:
227+
"""With error_on_execution=False, exceptions propagate without being caught."""
228+
229+
def test_runtime_error_in_action_propagates(self):
230+
class SM(StateMachine):
231+
s1 = State(initial=True)
232+
s2 = State(final=True)
233+
234+
go = s1.to(s2)
235+
236+
def on_go(self):
237+
raise RuntimeError("boom")
238+
239+
sm = SM()
240+
with pytest.raises(RuntimeError, match="boom"):
241+
sm.send("go")
242+
243+
def test_runtime_error_in_after_propagates(self):
244+
class SM(StateMachine):
245+
s1 = State(initial=True)
246+
s2 = State(final=True)
247+
248+
go = s1.to(s2)
249+
250+
def after_go(self):
251+
raise RuntimeError("after boom")
252+
253+
sm = SM()
254+
with pytest.raises(RuntimeError, match="after boom"):
255+
sm.send("go")
256+
257+
@pytest.mark.timeout(5)
258+
async def test_async_runtime_error_in_after_propagates(self):
259+
class SM(StateMachine):
260+
s1 = State(initial=True)
261+
s2 = State(final=True)
262+
263+
go = s1.to(s2)
264+
265+
async def after_go(self, **kwargs):
266+
raise RuntimeError("async after boom")
267+
268+
sm = SM()
269+
await sm.activate_initial_state()
270+
with pytest.raises(RuntimeError, match="async after boom"):
271+
await sm.send("go")
272+
273+
274+
# ---------------------------------------------------------------------------
275+
# enable_self_transition_entries = False
276+
# ---------------------------------------------------------------------------
277+
278+
279+
class TestSelfTransitionNoEntries:
280+
"""With enable_self_transition_entries=False, internal self-transitions do NOT fire entry/exit.
281+
282+
Note: ``enable_self_transition_entries`` only applies to *internal* self-transitions
283+
(``internal=True``). External self-transitions always fire entry/exit regardless.
284+
"""
285+
286+
def test_internal_self_transition_does_not_fire_enter_exit(self):
287+
log = []
288+
289+
class SM(StateMachine):
290+
s1 = State(initial=True)
291+
292+
loop = s1.to.itself(internal=True)
293+
294+
def on_enter_s1(self):
295+
log.append("enter_s1")
296+
297+
def on_exit_s1(self):
298+
log.append("exit_s1")
299+
300+
sm = SM()
301+
log.clear() # clear initial enter
302+
sm.send("loop")
303+
assert "enter_s1" not in log
304+
assert "exit_s1" not in log
305+
306+
def test_external_self_transition_fires_enter_exit(self):
307+
"""External self-transitions always fire, regardless of the flag."""
308+
log = []
309+
310+
class SM(StateMachine):
311+
s1 = State(initial=True)
312+
313+
loop = s1.to.itself()
314+
315+
def on_enter_s1(self):
316+
log.append("enter_s1")
317+
318+
def on_exit_s1(self):
319+
log.append("exit_s1")
320+
321+
sm = SM()
322+
log.clear()
323+
sm.send("loop")
324+
assert "enter_s1" in log
325+
assert "exit_s1" in log
326+
327+
328+
# ---------------------------------------------------------------------------
329+
# current_state deprecated property
330+
# ---------------------------------------------------------------------------
331+
332+
333+
class TestCurrentStateDeprecated:
334+
"""The current_state property emits DeprecationWarning but still works."""
335+
336+
def test_current_state_returns_state(self):
337+
class SM(StateMachine):
338+
s1 = State(initial=True)
339+
s2 = State(final=True)
340+
341+
go = s1.to(s2)
342+
343+
sm = SM()
344+
with warnings.catch_warnings():
345+
warnings.simplefilter("ignore", DeprecationWarning)
346+
cs = sm.current_state
347+
assert cs == sm.s1
348+
349+
def test_current_state_emits_warning(self):
350+
class SM(StateMachine):
351+
s1 = State(initial=True)
352+
s2 = State(final=True)
353+
354+
go = s1.to(s2)
355+
356+
sm = SM()
357+
with pytest.warns(DeprecationWarning, match="current_state"):
358+
_ = sm.current_state # noqa: F841
359+
360+
def test_current_state_with_list_value(self):
361+
"""current_state handles list current_state_value (backward compat)."""
362+
363+
class SM(StateMachine):
364+
s1 = State(initial=True)
365+
s2 = State(final=True)
366+
367+
go = s1.to(s2)
368+
369+
sm = SM()
370+
setattr(sm.model, sm.state_field, [sm.s1.value])
371+
with warnings.catch_warnings():
372+
warnings.simplefilter("ignore", DeprecationWarning)
373+
config = sm.current_state
374+
assert sm.s1 in config

0 commit comments

Comments
 (0)