diff --git a/README.md b/README.md index 9b5512c..5c781d2 100644 --- a/README.md +++ b/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. diff --git a/streamdeck/manager.py b/streamdeck/manager.py index a9f89b8..7bdd930 100644 --- a/streamdeck/manager.py +++ b/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) diff --git a/tests/actions/test_action_registry.py b/tests/actions/test_action_registry.py index aac8797..7efecd8 100644 --- a/tests/actions/test_action_registry.py +++ b/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(): diff --git a/tests/actions/test_event_handler_filtering.py b/tests/actions/test_event_handler_filtering.py index 9bd887c..0b075aa 100644 --- a/tests/actions/test_event_handler_filtering.py +++ b/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), diff --git a/tests/actions/test_global_action.py b/tests/actions/test_global_action.py index 5c96169..39c4bc0 100644 --- a/tests/actions/test_global_action.py +++ b/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(): diff --git a/tests/plugin_manager/__init__.py b/tests/plugin_manager/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/plugin_manager/conftest.py b/tests/plugin_manager/conftest.py new file mode 100644 index 0000000..9b1a11a --- /dev/null +++ b/tests/plugin_manager/conftest.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 diff --git a/tests/plugin_manager/test_command_sender_binding.py b/tests/plugin_manager/test_command_sender_binding.py new file mode 100644 index 0000000..a388664 --- /dev/null +++ b/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) diff --git a/tests/test_plugin_manager.py b/tests/plugin_manager/test_plugin_manager.py similarity index 61% rename from tests/test_plugin_manager.py rename to tests/plugin_manager/test_plugin_manager.py index 4d96209..0062396 100644 --- a/tests/test_plugin_manager.py +++ b/tests/plugin_manager/test_plugin_manager.py @@ -1,77 +1,31 @@ -import uuid from typing import cast -from unittest.mock import MagicMock, Mock +from unittest.mock import Mock import pytest import pytest_mock -from polyfactory.factories.pydantic_factory import ModelFactory from streamdeck.actions import Action from streamdeck.manager import PluginManager from streamdeck.models.events import DialRotate, EventBase, event_adapter -from streamdeck.websocket import WebSocketClient - -class DialRotateEventFactory(ModelFactory[DialRotate]): - """Polyfactory factory for creating a fake event message based on our Pydantic model.""" - - -@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, - ) +from tests.test_utils.fake_event_factories import DialRotateEventFactory @pytest.fixture -def patch_websocket_client(monkeypatch: pytest.MonkeyPatch) -> tuple[MagicMock, EventBase]: - """Fixture that uses pytest's MonkeyPatch to mock WebSocketClient and StreamDeckCommandSender for the PluginManager run method. +def mock_websocket_client_with_event(patch_websocket_client: Mock) -> tuple[Mock, EventBase]: + """Fixture that mocks the WebSocketClient and provides a fake DialRotateEvent message. Args: - monkeypatch: pytest's monkeypatch fixture. + patch_websocket_client: Mocked instance of the patched WebSocketClient. Returns: tuple: Mocked instance of WebSocketClient, and a fake DialRotateEvent. """ - mock_websocket_client = MagicMock(spec=WebSocketClient) - - mock_websocket_client.__enter__.return_value = mock_websocket_client - # Create a fake event message, and convert it to a json string to be passed back by the client.listen_forever() method. fake_event_message: DialRotate = DialRotateEventFactory.build() - mock_websocket_client.listen_forever.return_value = [fake_event_message.model_dump_json()] + patch_websocket_client.listen_forever.return_value = [fake_event_message.model_dump_json()] - monkeypatch.setattr("streamdeck.manager.WebSocketClient", lambda port: mock_websocket_client) + return patch_websocket_client, fake_event_message - return mock_websocket_client, fake_event_message - - -@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 @pytest.fixture @@ -114,7 +68,7 @@ def test_plugin_manager_register_action(plugin_manager: PluginManager): assert plugin_manager._registry._plugin_actions[0] == action -@pytest.mark.usefixtures("patch_websocket_client") +@pytest.mark.usefixtures("mock_websocket_client_with_event") def test_plugin_manager_sends_registration_event( mock_command_sender: Mock, plugin_manager: PluginManager ): @@ -130,10 +84,10 @@ def test_plugin_manager_sends_registration_event( @pytest.mark.usefixtures("_spy_action_registry_get_action_handlers") @pytest.mark.usefixtures("_spy_event_adapter_validate_json") def test_plugin_manager_process_event( - patch_websocket_client: tuple[MagicMock, EventBase], plugin_manager: PluginManager + mock_websocket_client_with_event: tuple[Mock, EventBase], plugin_manager: PluginManager ): """Test that PluginManager processes events correctly, calling event_adapter.validate_json and action_registry.get_action_handlers.""" - mock_websocket_client, fake_event_message = patch_websocket_client + mock_websocket_client, fake_event_message = mock_websocket_client_with_event plugin_manager.run() diff --git a/tests/test_utils/__init__.py b/tests/test_utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_utils/fake_event_factories.py b/tests/test_utils/fake_event_factories.py new file mode 100644 index 0000000..47ce2ec --- /dev/null +++ b/tests/test_utils/fake_event_factories.py @@ -0,0 +1,42 @@ +from polyfactory.factories.pydantic_factory import ModelFactory +from streamdeck.models import events + + +class DialDownEventFactory(ModelFactory[events.DialDown]): + """Polyfactory factory for creating a fake dialDown event message based on our Pydantic model.""" + + +class DialUpEventFactory(ModelFactory[events.DialUp]): + """Polyfactory factory for creating a fake dialUp event message based on our Pydantic model.""" + + +class DialRotateEventFactory(ModelFactory[events.DialRotate]): + """Polyfactory factory for creating a fake event message based on our Pydantic model.""" + + +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. + """ + + +class KeyUpEventFactory(ModelFactory[events.KeyUp]): + """Polyfactory factory for creating a fake keyUp event message based on our Pydantic model.""" + + +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. + """