1- import traceback
2- from dataclasses import dataclass
3- from dataclasses import field
41from pathlib import Path
5- from typing import Any
62
73import pytest
8- from statemachine .event import Event
94from statemachine .io .scxml .processor import SCXMLProcessor
105
11- from statemachine import State
126from statemachine import StateChart
137
148"""
2216""" # noqa: E501
2317
2418
25- @dataclass (frozen = True , unsafe_hash = True )
26- class OnTransition :
27- source : str
28- event : str
29- data : str
30- target : str
31-
32-
33- @dataclass (frozen = True , unsafe_hash = True )
34- class OnEnterState :
35- state : str
36- event : str
37- data : str
38-
39-
40- @dataclass (frozen = True , unsafe_hash = True )
41- class DebugListener :
42- events : list [Any ] = field (default_factory = list )
43-
44- def on_transition (self , event : Event , source : State , target : State , event_data ):
45- self .events .append (
46- OnTransition (
47- source = f"{ source and source .id } " ,
48- event = f"{ event and event .id } " ,
49- data = f"{ event_data .trigger_data .kwargs } " ,
50- target = f"{ target and target .id } " ,
51- )
52- )
53-
54- def on_enter_state (self , event : Event , state : State , event_data ):
55- self .events .append (
56- OnEnterState (
57- state = f"{ state .id } " ,
58- event = f"{ event and event .id } " ,
59- data = f"{ event_data .trigger_data .kwargs } " ,
60- )
61- )
62-
63-
6419class AsyncListener :
6520 """No-op async listener to trigger AsyncEngine selection."""
6621
@@ -69,77 +24,9 @@ async def on_enter_state(
6924 ): ... # No-op: presence of async callback triggers AsyncEngine selection
7025
7126
72- @dataclass
73- class FailedMark :
74- reason : str
75- events : list [OnTransition ]
76- is_assertion_error : bool
77- exception : Exception
78- logs : str
79- configuration : list [str ] = field (default_factory = list )
80-
81- @staticmethod
82- def _get_header (report : str ) -> str :
83- header_end_index = report .find ("---" )
84- return report [:header_end_index ]
85-
86- def write_fail_markdown (self , testcase_path : Path ):
87- fail_file_path = testcase_path .with_suffix (".fail.md" )
88- if not self .is_assertion_error :
89- exception_traceback = "" .join (
90- traceback .format_exception (
91- type (self .exception ), self .exception , self .exception .__traceback__
92- )
93- )
94- else :
95- exception_traceback = "Assertion of the testcase failed."
96-
97- report = """# Testcase: {testcase_path.stem}
98-
99- {reason}
100-
101- Final configuration: `{configuration}`
102-
103- ---
104-
105- ## Logs
106- ```py
107- {logs}
108- ```
109-
110- ## "On transition" events
111- ```py
112- {events}
113- ```
114-
115- ## Traceback
116- ```py
117- {exception_traceback}
118- ```
119- """ .format (
120- testcase_path = testcase_path ,
121- reason = self .reason ,
122- configuration = self .configuration if self .configuration else "No configuration" ,
123- logs = self .logs if self .logs else "No logs" ,
124- events = "\n " .join (map (repr , self .events )) if self .events else "No events" ,
125- exception_traceback = exception_traceback ,
126- )
127-
128- if fail_file_path .exists ():
129- last_report = fail_file_path .read_text ()
130-
131- if self ._get_header (report ) == self ._get_header (last_report ):
132- return
133-
134- with fail_file_path .open ("w" ) as fail_file :
135- fail_file .write (report )
136-
137-
13827def _run_scxml_testcase (
13928 testcase_path : Path ,
140- update_fail_mark ,
14129 should_generate_debug_diagram ,
142- caplog ,
14330 * ,
14431 async_mode : bool = False ,
14532) -> StateChart :
@@ -150,65 +37,40 @@ def _run_scxml_testcase(
15037 """
15138 from statemachine .contrib .diagram import DotGraphMachine
15239
153- sm : "StateChart | None" = None
154- try :
155- debug = DebugListener ()
156- listeners : list = [debug ]
157- if async_mode :
158- listeners .append (AsyncListener ())
159- processor = SCXMLProcessor ()
160- processor .parse_scxml_file (testcase_path )
161-
162- sm = processor .start (listeners = listeners )
163- if should_generate_debug_diagram :
164- DotGraphMachine (sm ).get_graph ().write_png (
165- testcase_path .parent / f"{ testcase_path .stem } .png"
166- )
167- assert sm is not None
168- return sm
169- except Exception as e :
170- if update_fail_mark :
171- reason = f"{ e .__class__ .__name__ } : { e .__class__ .__doc__ } "
172- is_assertion_error = isinstance (e , AssertionError )
173- fail_mark = FailedMark (
174- reason = reason ,
175- is_assertion_error = is_assertion_error ,
176- events = debug .events ,
177- exception = e ,
178- logs = caplog .text ,
179- configuration = [s .id for s in sm .configuration ] if sm else [],
180- )
181- fail_mark .write_fail_markdown (testcase_path )
182- raise
183-
184-
185- def _assert_passed (sm : StateChart , debug : "DebugListener | None" = None ):
40+ listeners : list = []
41+ if async_mode :
42+ listeners .append (AsyncListener ())
43+ processor = SCXMLProcessor ()
44+ processor .parse_scxml_file (testcase_path )
45+
46+ sm = processor .start (listeners = listeners )
47+ if should_generate_debug_diagram :
48+ DotGraphMachine (sm ).get_graph ().write_png (
49+ testcase_path .parent / f"{ testcase_path .stem } .png"
50+ )
51+ assert isinstance (sm , StateChart )
52+ return sm
53+
54+
55+ def _assert_passed (sm : StateChart ):
18656 assert isinstance (sm , StateChart )
187- assert "pass" in {s .id for s in sm .configuration }, debug
57+ assert "pass" in {s .id for s in sm .configuration }
18858
18959
190- def test_scxml_usecase_sync (
191- testcase_path : Path , update_fail_mark , should_generate_debug_diagram , caplog
192- ):
60+ def test_scxml_usecase_sync (testcase_path : Path , should_generate_debug_diagram , caplog ):
19361 sm = _run_scxml_testcase (
19462 testcase_path ,
195- update_fail_mark ,
19663 should_generate_debug_diagram ,
197- caplog ,
19864 async_mode = False ,
19965 )
20066 _assert_passed (sm )
20167
20268
20369@pytest .mark .asyncio ()
204- async def test_scxml_usecase_async (
205- testcase_path : Path , update_fail_mark , should_generate_debug_diagram , caplog
206- ):
70+ async def test_scxml_usecase_async (testcase_path : Path , should_generate_debug_diagram , caplog ):
20771 sm = _run_scxml_testcase (
20872 testcase_path ,
209- update_fail_mark ,
21073 should_generate_debug_diagram ,
211- caplog ,
21274 async_mode = True ,
21375 )
21476 # In async context, the engine only queued __initial__ during __init__.
0 commit comments