Update PluginManager class to bind the command_sender object to handler args when present in their signature. Add tests as well.

Author: strohganoff

streamdeck/manager.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import functools
+import inspect
 import logging
 from logging import getLogger
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING
 
 from streamdeck.actions import ActionRegistry
 from streamdeck.command_sender import StreamDeckCommandSender
@@ -12,6 +14,7 @@
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from typing import Any, Literal
 
     from streamdeck.actions import Action
@@ -67,6 +70,24 @@ def register_action(self, action: Action) -> None:
 
         self._registry.register(action)
 
+    def _inject_command_sender(self, handler: Callable[..., None], command_sender: StreamDeckCommandSender) -> Callable[..., None]:
+        """Inject command_sender into handler if it accepts it as a parameter.
+
+        Args:
+            handler: The event handler function
+            command_sender: The StreamDeckCommandSender instance
+
+        Returns:
+            The handler with command_sender injected if needed
+        """
+        args: dict[str, inspect.Parameter] = inspect.signature(handler).parameters
+
+        # Check dynamically if the `command_sender`'s name is in the handler's arguments.
+        if "command_sender" in args:
+            return functools.partial(handler, command_sender=command_sender)
+
+        return handler
+
     def run(self) -> None:
         """Run the PluginManager by connecting to the WebSocket server and processing incoming events.
 
@@ -86,5 +107,7 @@ def run(self) -> None:
                 event_action_uuid = data.action if isinstance(data, ContextualEventMixin) else None
 
                 for handler in self._registry.get_action_handlers(event_name=data.event, event_action_uuid=event_action_uuid):
+                    handler = self._inject_command_sender(handler, command_sender)
                     # TODO: from contextual event occurences, save metadata to the action's properties.
+
                     handler(data)

tests/plugin_manager/test_command_sender_binding.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import inspect
+from functools import partial
+from typing import TYPE_CHECKING, Any, cast
+from unittest.mock import Mock, create_autospec
+
+import pytest
+from pprintpp import pprint
+from streamdeck.actions import Action
+from streamdeck.command_sender import StreamDeckCommandSender
+from streamdeck.websocket import WebSocketClient
+
+from tests.test_utils.fake_event_factories import KeyDownEventFactory
+
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from streamdeck.manager import PluginManager
+    from streamdeck.models import events
+    from typing_extensions import TypeAlias  # noqa: UP035
+
+    EventHandlerFunc: TypeAlias = Callable[[events.EventBase], None] | Callable[[events.EventBase, StreamDeckCommandSender], None]
+
+
+
+def create_event_handler(include_command_sender_param: bool = False) -> EventHandlerFunc:
+    """Create a dummy event handler function that matches the EventHandlerFunc TypeAlias.
+
+    Args:
+        include_command_sender_param (bool, optional): Whether to include the `command_sender` parameter in the handler. Defaults to False.
+
+    Returns:
+        Callable[[events.EventBase], None] | Callable[[events.EventBase, StreamDeckCommandSender], None]: A dummy event handler function.
+    """
+    if not include_command_sender_param:
+        def dummy_handler_without_cmd_sender(event: events.EventBase) -> None:
+            """Dummy event handler function that matches the EventHandlerFunc TypeAlias without `command_sender` param."""
+
+        return dummy_handler_without_cmd_sender
+
+    def dummy_handler_with_cmd_sender(event: events.EventBase, command_sender: StreamDeckCommandSender) -> None:
+        """Dummy event handler function that matches the EventHandlerFunc TypeAlias with `command_sender` param."""
+
+    return dummy_handler_with_cmd_sender
+
+
+@pytest.fixture(params=[True, False])
+def mock_event_handler(request: pytest.FixtureRequest) -> Mock:
+    include_command_sender_param: bool = request.param
+    dummy_handler: EventHandlerFunc = create_event_handler(include_command_sender_param)
+
+    return create_autospec(dummy_handler, spec_set=True)
+
+
+@pytest.fixture
+def mock_websocket_client_with_fake_events(patch_websocket_client: Mock) -> tuple[Mock, list[events.EventBase]]:
+    """Fixture that mocks the WebSocketClient and provides a list of fake event messages yielded by the mock client.
+
+    Args:
+        patch_websocket_client: Mocked instance of the patched WebSocketClient.
+
+    Returns:
+        tuple: Mocked instance of WebSocketClient, and a list of fake event messages.
+    """
+    # Create a list of fake event messages, and convert them to json strings to be passed back by the client.listen_forever() method.
+    fake_event_messages: list[events.EventBase] = [
+        KeyDownEventFactory.build(action="my-fake-action-uuid"),
+    ]
+
+    patch_websocket_client.listen_forever.return_value = [event.model_dump_json() for event in fake_event_messages]
+
+    return patch_websocket_client, fake_event_messages
+
+
+
+def test_inject_command_sender_func(
+    plugin_manager: PluginManager,
+    mock_event_handler: Mock,
+):
+    """Test that the command_sender is injected into the handler."""
+    mock_command_sender = Mock()
+    result_handler = plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender)
+
+    resulting_handler_params = inspect.signature(result_handler).parameters
+
+    # If this condition is true, then the `result_handler` is a partial function.
+    if "command_sender" in resulting_handler_params:
+
+        # Check that the `result_handler` is not the same as the original `mock_event_handler`.
+        assert result_handler != mock_event_handler
+
+        # Check that the `command_sender` parameter is bound to the correct value.
+        resulting_handler_bound_kwargs: dict[str, Any] = cast(partial[Any], result_handler).keywords
+        assert resulting_handler_bound_kwargs["command_sender"] == mock_command_sender
+
+    # If there isn't a `command_sender` parameter, then the `result_handler` is the original handler unaltered.
+    else:
+        assert result_handler == mock_event_handler
+
+
+def test_run_manager_events_handled_with_correct_params(
+    mock_websocket_client_with_fake_events: tuple[Mock, list[events.EventBase]],
+    plugin_manager: PluginManager,
+    mock_command_sender: Mock,
+):
+    """Test that the PluginManager runs and triggers event handlers with the correct parameter binding.
+
+    This test will:
+      - Register an action with the PluginManager.
+      - Create and register mock event handlers with and without the `command_sender` parameter.
+      - Run the PluginManager and let it process the fake event messages generated by the mocked WebSocketClient.
+      - Ensure that mocked event handlers were called with the correct params,
+          binding the `command_sender` parameter if defined in the handler's signature.
+
+    Args:
+        mock_websocket_client_with_fake_events (tuple[Mock, list[events.EventBase]]): Mocked instance of WebSocketClient, and a list of fake event messages it will yield.
+        plugin_manager (PluginManager): Instance of PluginManager with test parameters.
+        mock_command_sender (Mock): Patched instance of StreamDeckCommandSender. Used here to ensure that the `command_sender` parameter is bound correctly.
+    """
+    # As of now, fake_event_messages is a list of one KeyDown event. If this changes, I'll need to update this test.
+    fake_event_message: events.KeyDown = mock_websocket_client_with_fake_events[1][0]
+
+    action = Action(fake_event_message.action)
+
+    # Create a mock event handler with the `command_sender` parameter and register it with the action for an event type.
+    mock_event_handler_with_cmd_sender: Mock = create_autospec(create_event_handler(include_command_sender_param=True), spec_set=True)
+    action.on("keyDown")(mock_event_handler_with_cmd_sender)
+
+    # Create a mock event handler without the `command_sender` parameter and register it with the action for an event type.
+    mock_event_handler_without_cmd_sender: Mock = create_autospec(create_event_handler(include_command_sender_param=False), spec_set=True)
+    action.on("keyDown")(mock_event_handler_without_cmd_sender)
+
+    plugin_manager.register_action(action)
+
+    # Run the PluginManager and let it process the fake event messages generated by the mocked WebSocketClient.
+    plugin_manager.run()
+
+    # Ensure that mocked event handlers were called with the correct params, binding the `command_sender` parameter if defined in the handler's signature.
+    mock_event_handler_without_cmd_sender.assert_called_once_with(fake_event_message)
+    mock_event_handler_with_cmd_sender.assert_called_once_with(fake_event_message, mock_command_sender)