Merge pull request #5 from strohganoff/feature/inject-command_sender-b

Feature/inject command sender in event handlers

Author: strohganoff

README.md

@@ -87,6 +87,18 @@ def handle_will_appear(event_data):
 
 !!!INFO Handlers for action-specific events are dispatched only if the event is triggered by the associated action, ensuring isolation and predictability. For other types of events that are not associated with a specific action, handlers are dispatched without such restrictions.
 
+Handlers can optionally include a `command_sender` parameter to access Stream Deck command sending capabilities.
+
+```python
+@my_action.on("willAppear")
+def handle_will_appear(event_data, command_sender: StreamDeckCommandSender):
+    # Use command_sender to interact with Stream Deck
+    command_sender.set_title(context=event_data.context, title="Hello!")
+    command_sender.set_state(context=event_data.context, state=0)
+```
+
+The `command_sender` parameter is optional. If included in the handler's signature, the SDK automatically injects a `StreamDeckCommandSender` instance that can be used to send commands back to the Stream Deck (like setting titles, states, or images).
+
 
 
 ### Writing Logs
@@ -234,7 +246,7 @@ Contributions are welcome! Please open an issue or submit a pull request on GitH
 The following upcoming improvements are in the works:
 
 - **Improved Documentation**: Expand and improve the documentation with more examples, guides, and use cases.
-- **Bind Command Sender**: Automatically bind `command_sender` and action instance context-holding objects to handler function arguments if they are included in the definition.
+- **Store & Bind Settings**: Automatically store and bind action instance context-holding objects to handler function arguments if they are included in the definition.
 - **Optional Event Pattern Matching on Hooks**: Add support for optional pattern-matching on event messages to further filter when hooks get called.
 - **Async Support**: Introduce asynchronous features to handle WebSocket communication more efficiently.
 

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/actions/test_action_registry.py

@@ -1,21 +1,14 @@
 from __future__ import annotations
 
 import pytest
-from polyfactory.factories.pydantic_factory import ModelFactory
 from streamdeck.actions import Action, ActionRegistry
 from streamdeck.models import events
 
-
-class DialUpEventFactory(ModelFactory[events.DialUp]):
-    """Polyfactory factory for creating a fake dialUp event message based on our Pydantic model."""
-
-class DialDownEventFactory(ModelFactory[events.DialDown]):
-    """Polyfactory factory for creating a fake dialDown event message based on our Pydantic model."""
-
-class KeyUpEventFactory(ModelFactory[events.KeyUp]):
-    """Polyfactory factory for creating a fake keyUp event message based on our Pydantic model."""
-
-
+from tests.test_utils.fake_event_factories import (
+    DialDownEventFactory,
+    DialUpEventFactory,
+    KeyUpEventFactory,
+)
 
 
 def test_register_action():

tests/actions/test_event_handler_filtering.py

@@ -4,14 +4,21 @@
 from unittest.mock import create_autospec
 
 import pytest
-from polyfactory.factories.pydantic_factory import ModelFactory
 from streamdeck.actions import Action, ActionRegistry, GlobalAction
 from streamdeck.models import events
 
+from tests.test_utils.fake_event_factories import (
+    ApplicationDidLaunchEventFactory,
+    DeviceDidConnectFactory,
+    KeyDownEventFactory,
+)
+
 
 if TYPE_CHECKING:
     from unittest.mock import Mock
 
+    from polyfactory.factories.pydantic_factory import ModelFactory
+
 
 
 @pytest.fixture
@@ -22,28 +29,6 @@ def dummy_handler(event: events.EventBase) -> None:
     return create_autospec(dummy_handler, spec_set=True)
 
 
-class ApplicationDidLaunchEventFactory(ModelFactory[events.ApplicationDidLaunch]):
-    """Polyfactory factory for creating fake applicationDidLaunch event message based on our Pydantic model.
-
-    ApplicationDidLaunchEvent's hold no unique identifier properties, besides the almost irrelevant `event` name property.
-    """
-
-class DeviceDidConnectFactory(ModelFactory[events.DeviceDidConnect]):
-    """Polyfactory factory for creating fake deviceDidConnect event message based on our Pydantic model.
-
-    DeviceDidConnectEvent's have `device` unique identifier property.
-    """
-
-class KeyDownEventFactory(ModelFactory[events.KeyDown]):
-    """Polyfactory factory for creating fake keyDown event message based on our Pydantic model.
-
-    KeyDownEvent's have the unique identifier properties:
-        `device`: Identifies the Stream Deck device that this event is associated with.
-        `action`: Identifies the action that caused the event.
-        `context`: Identifies the *instance* of an action that caused the event.
-    """
-
-
 @pytest.mark.parametrize(("event_name","event_factory"), [
     ("keyDown", KeyDownEventFactory),
     ("deviceDidConnect", DeviceDidConnectFactory),

tests/actions/test_global_action.py

@@ -3,14 +3,11 @@
 from typing import TYPE_CHECKING
 
 import pytest
-from polyfactory.factories.pydantic_factory import ModelFactory
-from streamdeck.actions import Action, GlobalAction, available_event_names
-from streamdeck.models import events
+from streamdeck.actions import GlobalAction, available_event_names
 
 
 if TYPE_CHECKING:
     from streamdeck.models.events import EventBase
-    from streamdeck.types import EventNameStr
 
 
 def test_global_action_register_event_handler():

tests/plugin_manager/__init__.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import uuid
+from unittest.mock import Mock, create_autospec
+
+import pytest
+from streamdeck.manager import PluginManager
+from streamdeck.websocket import WebSocketClient
+
+
+@pytest.fixture
+def plugin_manager(port_number: int) -> PluginManager:
+    """Fixture that provides a configured instance of PluginManager for testing.
+
+    Returns:
+        PluginManager: An instance of PluginManager with test parameters.
+    """
+    plugin_uuid = "test-plugin-uuid"
+    plugin_registration_uuid = str(uuid.uuid1())
+    register_event = "registerPlugin"
+    info = {"some": "info"}
+
+    return PluginManager(
+        port=port_number,
+        plugin_uuid=plugin_uuid,
+        plugin_registration_uuid=plugin_registration_uuid,
+        register_event=register_event,
+        info=info,
+    )
+
+
+@pytest.fixture
+def patch_websocket_client(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    """Fixture that uses pytest's MonkeyPatch to mock WebSocketClient for the PluginManager run method.
+
+    The mocked WebSocketClient can be given fake event messages to yield when listen_forever() is called:
+      ```patch_websocket_client.listen_forever.return_value = [fake_event_json1, fake_event_json2, ...]```
+
+    Args:
+        monkeypatch: pytest's monkeypatch fixture.
+
+    Returns:
+    """
+    mock_websocket_client: Mock = create_autospec(WebSocketClient, spec_set=True)
+    mock_websocket_client.__enter__.return_value = mock_websocket_client
+
+    monkeypatch.setattr("streamdeck.manager.WebSocketClient", (lambda *args, **kwargs: mock_websocket_client))
+
+    return mock_websocket_client
+
+
+@pytest.fixture
+def mock_command_sender(mocker: pytest_mock.MockerFixture) -> Mock:
+    """Fixture that patches the StreamDeckCommandSender.
+
+    Args:
+        mocker: pytest-mock's mocker fixture.
+
+    Returns:
+        Mock: Mocked instance of StreamDeckCommandSender
+    """
+    mock_cmd_sender = Mock()
+    mocker.patch("streamdeck.manager.StreamDeckCommandSender", return_value=mock_cmd_sender)
+    return mock_cmd_sender

tests/plugin_manager/conftest.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)