test: cover remaining branch partials in invoke, callbacks, and statemachine

Move visit condition tests to test_callbacks.py and invalid state value
test to test_statemachine.py. Add tests for async cancelled-during-execution,
sync error-after-cancel, and spawn_pending_async empty paths.

Add pragma: no branch to IInvoke Protocol method body (coverage.py
limitation with Protocol classes).

Author: fgmacedo

statemachine/invoke.py

@@ -44,7 +44,7 @@ class IInvoke(Protocol):
     Optionally implement ``on_cancel()`` for cleanup when the state is exited.
     """
 
-    def run(self, ctx: "InvokeContext") -> Any: ...
+    def run(self, ctx: "InvokeContext") -> Any: ...  # pragma: no branch
 
 
 class _InvokeCallableWrapper:

tests/test_callbacks.py

@@ -2,6 +2,7 @@
 
 import pytest
 from statemachine.callbacks import CallbackGroup
+from statemachine.callbacks import CallbacksExecutor
 from statemachine.callbacks import CallbackSpec
 from statemachine.callbacks import CallbackSpecList
 from statemachine.callbacks import CallbacksRegistry
@@ -351,3 +352,35 @@ class ExampleStateMachine(StateChart):
             match="Error on transition start from Created to Started when resolving callbacks",
         ):
             ExampleStateMachine()
+
+
+class TestVisitConditionFalse:
+    """visit/async_visit skip callbacks whose condition returns False."""
+
+    def test_visit_skips_when_condition_is_false(self):
+        visited = []
+        spec = CallbackSpec(
+            "never_called",
+            group=CallbackGroup.INVOKE,
+            is_convention=True,
+            cond=lambda *a, **kw: False,
+        )
+        executor = CallbacksExecutor()
+        executor.add("test_key", spec, lambda: lambda **kw: True)
+
+        executor.visit(lambda cb, *a, **kw: visited.append(str(cb)))
+        assert visited == []
+
+    async def test_async_visit_skips_when_condition_is_false(self):
+        visited = []
+        spec = CallbackSpec(
+            "never_called",
+            group=CallbackGroup.INVOKE,
+            is_convention=True,
+            cond=lambda *a, **kw: False,
+        )
+        executor = CallbacksExecutor()
+        executor.add("test_key", spec, lambda: lambda **kw: True)
+
+        await executor.async_visit(lambda cb, *a, **kw: visited.append(str(cb)))
+        assert visited == []

tests/test_invoke.py

@@ -880,3 +880,110 @@ async def async_visitor(callback, **kwargs):
 
         await executor.async_visit(async_visitor)
         assert visited == ["dummy"]
+
+
+class TestIInvokeProtocolRun:
+    """IInvoke.run() protocol method can be called on a concrete implementation."""
+
+    def test_protocol_run_is_callable(self):
+        """Verify that calling run() on a concrete IInvoke instance works."""
+
+        class ConcreteInvoker:
+            def run(self, ctx):
+                return "concrete_result"
+
+        invoker: IInvoke = ConcreteInvoker()
+        result = invoker.run(None)
+        assert result == "concrete_result"
+
+
+class TestSpawnPendingAsyncEmpty:
+    """spawn_pending_async with nothing pending is a no-op."""
+
+    async def test_spawn_pending_async_no_pending(self, sm_runner):
+        class SM(StateChart):
+            idle = State(initial=True)
+            active = State(final=True)
+            go = idle.to(active)
+
+        sm = await sm_runner.start(SM)
+        # Directly call spawn_pending_async with empty pending list
+        await sm._engine._invoke_manager.spawn_pending_async()
+
+
+class TestInvokeAsyncCancelledDuringExecution:
+    """Async handler completes or errors after state was already exited."""
+
+    async def test_success_after_cancel(self):
+        """Handler returns successfully but ctx.cancelled is already set."""
+        from tests.conftest import SMRunner
+
+        class SM(StateChart):
+            loading = State(initial=True)
+            stopped = State(final=True)
+            cancel = loading.to(stopped)
+
+            def on_invoke_loading(self, ctx=None, **kwargs):
+                if ctx is None:
+                    return
+                # Simulate: cancelled is set during execution but we still return
+                ctx.cancelled.set()
+                return "should_be_ignored"
+
+        sm_runner = SMRunner(is_async=True)
+        sm = await sm_runner.start(SM)
+        await sm_runner.sleep(0.2)
+        await sm_runner.processing_loop(sm)
+
+        # The done.invoke event should NOT have been sent (cancelled)
+        assert "loading" in sm.configuration_values
+
+    async def test_error_after_cancel(self):
+        """Handler raises but ctx.cancelled is already set — error is swallowed."""
+        from tests.conftest import SMRunner
+
+        class SM(StateChart):
+            loading = State(initial=True)
+            error_state = State(final=True)
+            error_execution = loading.to(error_state)
+
+            def on_invoke_loading(self, ctx=None, **kwargs):
+                if ctx is None:
+                    return
+                # Simulate: cancelled during execution, then error
+                ctx.cancelled.set()
+                raise ValueError("should be ignored")
+
+        sm_runner = SMRunner(is_async=True)
+        sm = await sm_runner.start(SM)
+        await sm_runner.sleep(0.2)
+        await sm_runner.processing_loop(sm)
+
+        # The error.execution event should NOT have been sent (cancelled)
+        assert "loading" in sm.configuration_values
+
+
+class TestSyncInvokeErrorAfterCancel:
+    """Sync handler errors after state was already exited."""
+
+    async def test_sync_error_after_cancel(self):
+        """Sync handler raises but ctx.cancelled is set — error.execution not sent."""
+        from tests.conftest import SMRunner
+
+        class SM(StateChart):
+            loading = State(initial=True)
+            error_state = State(final=True)
+            error_execution = loading.to(error_state)
+
+            def on_invoke_loading(self, ctx=None, **kwargs):
+                if ctx is None:
+                    return
+                ctx.cancelled.set()
+                raise ValueError("should be ignored")
+
+        sm_runner = SMRunner(is_async=False)
+        sm = await sm_runner.start(SM)
+        await sm_runner.sleep(0.2)
+        await sm_runner.processing_loop(sm)
+
+        assert "loading" in sm.configuration_values

tests/test_statemachine.py

@@ -709,3 +709,23 @@ def is_blocked(self):
 
         sm = MyMachine()
         assert [e.id for e in sm.enabled_events()] == ["go"]
+
+
+class TestInvalidStateValueNonNone:
+    """current_state raises InvalidStateValue when state value is non-None but invalid."""
+
+    def test_invalid_non_none_state_value(self):
+        import warnings
+
+        class SM(StateChart):
+            idle = State(initial=True)
+            active = State(final=True)
+            go = idle.to(active)
+
+        sm = SM()
+        # Bypass setter validation by writing directly to the model attribute
+        setattr(sm.model, sm.state_field, "nonexistent_state")
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            with pytest.raises(exceptions.InvalidStateValue):
+                _ = sm.current_state