feat: run SCXML test suite with both sync and async engines

Add async SCXML test variant using a minimal AsyncListener to trigger
AsyncEngine selection, keeping SCXMLProcessor free of async concerns.
Fix __initial__ handling in AsyncEngine (break instead of continue in
phase 3) to ensure internal events are processed before external ones.

Author: fgmacedo

statemachine/engines/async_.py

@@ -324,13 +324,16 @@ async def processing_loop(self):  # noqa: C901
 
                     logger.debug("External event: %s", external_event.event)
 
-                    # Handle lazy initial state activation
+                    # Handle lazy initial state activation.
+                    # Break out of phase 3 so the outer loop restarts from phase 1
+                    # (eventless/internal), ensuring internal events queued during
+                    # initial entry are processed before any external events.
                     if external_event.event == "__initial__":
                         transitions = self._initial_transitions(external_event)
                         await self._enter_states(
                             transitions, external_event, OrderedSet(), OrderedSet()
                         )
-                        continue
+                        break
 
                     enabled_transitions = await self.select_transitions(external_event)
                     logger.debug("Enabled transitions: %s", enabled_transitions)

statemachine/io/scxml/processor.py

@@ -207,8 +207,7 @@ def _process_transitions(self, transitions: List[Transition]):
 
                 # Process actions
             if transition.on and not transition.on.is_empty:
-                callable = ExecuteBlock(transition.on)
-                transition_dict["on"] = callable
+                transition_dict["on"] = ExecuteBlock(transition.on)
 
             result.append(transition_dict)
         return result

tests/scxml/conftest.py

@@ -45,7 +45,7 @@ def pytest_generate_tests(metafunc):
                 id=str(testcase_path.relative_to(TESTCASES_DIR)),
                 marks=compute_testcase_marks(testcase_path),
             )
-            for testcase_path in TESTCASES_DIR.glob("**/*.scxml")
+            for testcase_path in sorted(TESTCASES_DIR.glob("**/*.scxml"))
             if "sub" not in testcase_path.name
         ],
     )

tests/scxml/test_scxml_cases.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import Any
 
+import pytest
 from statemachine.event import Event
 from statemachine.io.scxml.processor import SCXMLProcessor
 
@@ -60,6 +61,13 @@ def on_enter_state(self, event: Event, state: State, event_data):
         )
 
 
+class AsyncListener:
+    """No-op async listener to trigger AsyncEngine selection."""
+
+    async def on_enter_state(self, **kwargs):
+        pass
+
+
 @dataclass
 class FailedMark:
     reason: str
@@ -126,26 +134,38 @@ def write_fail_markdown(self, testcase_path: Path):
             fail_file.write(report)
 
 
-def test_scxml_usecase(
-    testcase_path: Path, update_fail_mark, should_generate_debug_diagram, caplog
-):
+def _run_scxml_testcase(
+    testcase_path: Path,
+    update_fail_mark,
+    should_generate_debug_diagram,
+    caplog,
+    *,
+    async_mode: bool = False,
+) -> StateChart:
+    """Shared logic for sync and async SCXML test variants.
+
+    Parses the SCXML file, starts the state machine, and asserts the final
+    configuration contains ``pass``.  Returns the SM instance.
+    """
     from statemachine.contrib.diagram import DotGraphMachine
 
     sm: "StateChart | None" = None
     try:
         debug = DebugListener()
+        listeners: list = [debug]
+        if async_mode:
+            listeners.append(AsyncListener())
         processor = SCXMLProcessor()
         processor.parse_scxml_file(testcase_path)
 
-        sm = processor.start(listeners=[debug])
+        sm = processor.start(listeners=listeners)
         if should_generate_debug_diagram:
             DotGraphMachine(sm).get_graph().write_png(
                 testcase_path.parent / f"{testcase_path.stem}.png"
             )
-        assert isinstance(sm, StateChart)
-        assert "pass" in {s.id for s in sm.configuration}, debug
+        assert sm is not None
+        return sm
     except Exception as e:
-        # Import necessary module
         if update_fail_mark:
             reason = f"{e.__class__.__name__}: {e.__class__.__doc__}"
             is_assertion_error = isinstance(e, AssertionError)
@@ -159,3 +179,38 @@ def test_scxml_usecase(
             )
             fail_mark.write_fail_markdown(testcase_path)
         raise
+
+
+def _assert_passed(sm: StateChart, debug: "DebugListener | None" = None):
+    assert isinstance(sm, StateChart)
+    assert "pass" in {s.id for s in sm.configuration}, debug
+
+
+def test_scxml_usecase_sync(
+    testcase_path: Path, update_fail_mark, should_generate_debug_diagram, caplog
+):
+    sm = _run_scxml_testcase(
+        testcase_path,
+        update_fail_mark,
+        should_generate_debug_diagram,
+        caplog,
+        async_mode=False,
+    )
+    _assert_passed(sm)
+
+
+@pytest.mark.asyncio()
+async def test_scxml_usecase_async(
+    testcase_path: Path, update_fail_mark, should_generate_debug_diagram, caplog
+):
+    sm = _run_scxml_testcase(
+        testcase_path,
+        update_fail_mark,
+        should_generate_debug_diagram,
+        caplog,
+        async_mode=True,
+    )
+    # In async context, the engine only queued __initial__ during __init__.
+    # Activate now within the running event loop.
+    await sm.activate_initial_state()
+    _assert_passed(sm)