diff --git a/streamdeck/__init__.py b/streamdeck/__init__.py index d0f3bf8..906a1b5 100644 --- a/streamdeck/__init__.py +++ b/streamdeck/__init__.py @@ -1,11 +1,11 @@ from . import ( - actions, command_sender, manager, models, utils, websocket, ) +from .event_handlers import actions __all__ = [ diff --git a/tests/actions/__init__.py b/streamdeck/event_handlers/__init__.py similarity index 100% rename from tests/actions/__init__.py rename to streamdeck/event_handlers/__init__.py diff --git a/streamdeck/actions.py b/streamdeck/event_handlers/actions.py similarity index 56% rename from streamdeck/actions.py rename to streamdeck/event_handlers/actions.py index 757261d..06f8954 100644 --- a/streamdeck/actions.py +++ b/streamdeck/event_handlers/actions.py @@ -6,36 +6,32 @@ from logging import getLogger from typing import TYPE_CHECKING, cast +from streamdeck.event_handlers.protocol import ( + EventHandlerFunc, + EventModel_contra, + InjectableParams, + SupportsEventHandlers, +) + if TYPE_CHECKING: from collections.abc import Callable, Generator - from typing import Protocol - - from typing_extensions import ParamSpec, TypeVar - from streamdeck.models.events import EventBase from streamdeck.types import ActionUUIDStr, EventNameStr - EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True) - InjectableParams = ParamSpec("InjectableParams", default=...) - - class EventHandlerFunc(Protocol[EventModel_contra, InjectableParams]): - """Protocol for an event handler function that takes an event (of subtype of EventBase) and other parameters that are injectable.""" - def __call__(self, event_data: EventModel_contra, *args: InjectableParams.args, **kwargs: InjectableParams.kwargs) -> None: ... - - - logger = getLogger("streamdeck.actions") -class ActionBase(ABC): +class ActionBase(ABC, SupportsEventHandlers): """Base class for all actions.""" + _events: dict[EventNameStr, set[EventHandlerFunc]] + """Dictionary mapping event names to sets of event handler functions.""" def __init__(self) -> None: """Initialize an Action instance.""" - self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set) + self._events = defaultdict(set) def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[EventModel_contra, InjectableParams]], EventHandlerFunc[EventModel_contra, InjectableParams]]: """Register an event handler for a specific event. @@ -88,6 +84,8 @@ class GlobalAction(ActionBase): class Action(ActionBase): """Represents an action that can be performed for a specific action, with event handlers for specific event types.""" + uuid: ActionUUIDStr + """The unique identifier for the action.""" def __init__(self, uuid: ActionUUIDStr) -> None: """Initialize an Action instance. @@ -104,37 +102,3 @@ def name(self) -> str: return self.uuid.split(".")[-1] -class ActionRegistry: - """Manages the registration and retrieval of actions and their event handlers.""" - - def __init__(self) -> None: - """Initialize an ActionRegistry instance.""" - self._plugin_actions: list[ActionBase] = [] - - def register(self, action: ActionBase) -> None: - """Register an action with the registry. - - Args: - action (Action): The action to register. - """ - self._plugin_actions.append(action) - - def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: ActionUUIDStr | None = None) -> Generator[EventHandlerFunc, None, None]: - """Get all event handlers for a specific event from all registered actions. - - Args: - event_name (EventName): The name of the event to retrieve handlers for. - event_action_uuid (str | None): The action UUID to get handlers for. - If None (i.e., the event is not action-specific), get all handlers for the event. - - Yields: - EventHandlerFunc: The event handler functions for the specified event. - """ - for action in self._plugin_actions: - # If the event is action-specific (i.e is not a GlobalAction and has a UUID attribute), - # only get handlers for that action, as we don't want to trigger - # and pass this event to handlers for other actions. - if event_action_uuid is not None and (isinstance(action, Action) and action.uuid != event_action_uuid): - continue - - yield from action.get_event_handlers(event_name) diff --git a/streamdeck/event_handlers/protocol.py b/streamdeck/event_handlers/protocol.py new file mode 100644 index 0000000..89b520d --- /dev/null +++ b/streamdeck/event_handlers/protocol.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import inspect +from collections.abc import Iterable +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +from typing_extensions import ParamSpec, TypeGuard, TypeVar # noqa: UP035 + +from streamdeck.command_sender import StreamDeckCommandSender +from streamdeck.models.events import EventBase + + +if TYPE_CHECKING: + from collections.abc import Iterable + + from streamdeck.types import EventNameStr + + + +EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True) +InjectableParams = ParamSpec("InjectableParams", default=...) + + +class EventHandlerFunc(Protocol[EventModel_contra, InjectableParams]): + """Protocol for an event handler function that takes an event (of subtype of EventBase) and other parameters that are injectable.""" + def __call__(self, event_data: EventModel_contra, *args: InjectableParams.args, **kwargs: InjectableParams.kwargs) -> None: ... + + +BindableEventHandlerFunc = EventHandlerFunc[EventModel_contra, tuple[StreamDeckCommandSender]] # type: ignore[misc] +"""Type alias for a bindable event handler function that takes an event (of subtype of EventBase) and a command_sender parameter that is to be injected.""" +BoundEventHandlerFunc = EventHandlerFunc[EventModel_contra] +"""Type alias for a bound event handler function that takes an event (of subtype of EventBase) and no other parameters. + +Typically used for event handlers that have already had parameters injected. +""" + + +@runtime_checkable +class SupportsEventHandlers(Protocol): + """Protocol for a class that holds event-specific handler functions which can be pulled out by event name. + + Implementing classes should handle defining the registration of event handlers and ensuring + that they can be retrieved efficiently. + """ + + def get_event_handlers(self, event_name: EventNameStr, /) -> Iterable[EventHandlerFunc]: + """Get all event handlers for a specific event. + + Args: + event_name (str): The name of the event to get handlers for. + + Returns: + Iterable[EventHandlerFunc]: The event handler functions for the specified event. + """ + ... + + def get_registered_event_names(self) -> list[EventNameStr]: + """Get all event names for which handlers are registered. + + Returns: + list[str]: A list of event names for which handlers are registered. + """ + ... + + +# def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BindableEventHandlerFunc[EventModel_contra]]: +def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> bool: + """Check if the handler is prebound with the `command_sender` parameter.""" + # Check dynamically if the `command_sender`'s name is in the handler's arguments. + return "command_sender" in inspect.signature(handler).parameters + + +def is_not_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BoundEventHandlerFunc[EventModel_contra]]: + """Check if the handler only accepts the event_data parameter. + + If this function returns False after the is_bindable_handler check is True, then the function has invalid parameters, and will subsequently need to be handled in the calling code. + """ + handler_params = inspect.signature(handler).parameters + return len(handler_params) == 1 and "event_data" in handler_params + + + + + + + + diff --git a/streamdeck/event_handlers/registry.py b/streamdeck/event_handlers/registry.py new file mode 100644 index 0000000..14a325e --- /dev/null +++ b/streamdeck/event_handlers/registry.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from streamdeck.event_handlers.actions import Action + + +if TYPE_CHECKING: + from collections.abc import Generator + + from streamdeck.event_handlers.protocol import EventHandlerFunc, SupportsEventHandlers + from streamdeck.types import ActionUUIDStr, EventNameStr + + +class HandlersRegistry: + """Manages the registration and retrieval of event handler catalogs and their event handlers.""" + _plugin_event_handler_catalogs: list[SupportsEventHandlers] + """List of registered actions and other event handler catalogs.""" + + def __init__(self) -> None: + """Initialize a HandlersRegistry instance.""" + self._plugin_event_handler_catalogs = [] + + def register(self, catalog: SupportsEventHandlers) -> None: + """Register an event handler catalog with the registry. + + Args: + catalog (SupportsEventHandlers): The event handler catalog to register. + """ + self._plugin_event_handler_catalogs.append(catalog) + + def get_event_handlers(self, event_name: EventNameStr, event_action_uuid: ActionUUIDStr | None = None) -> Generator[EventHandlerFunc, None, None]: + """Get all event handlers for a specific event from all registered event handler catalogs. + + Args: + event_name (EventName): The name of the event to retrieve handlers for. + event_action_uuid (str | None): The action UUID to get handlers for. + If None (i.e., the event is not action-specific), get all handlers for the event. + + Yields: + EventHandlerFunc: The event handler functions for the specified event. + """ + for catalog in self._plugin_event_handler_catalogs: + # If the event is action-specific (i.e is not a GlobalAction and has a UUID attribute), + # only get handlers for that action, as we don't want to trigger + # and pass this event to handlers for other actions. + if event_action_uuid is not None and (isinstance(catalog, Action) and catalog.uuid != event_action_uuid): + continue + + yield from catalog.get_event_handlers(event_name) diff --git a/streamdeck/manager.py b/streamdeck/manager.py index a99b606..3e46d6c 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -1,15 +1,19 @@ from __future__ import annotations import functools -import inspect from logging import getLogger from typing import TYPE_CHECKING from pydantic import ValidationError -from typing_extensions import TypeGuard # noqa: UP035 -from streamdeck.actions import Action, ActionBase, ActionRegistry from streamdeck.command_sender import StreamDeckCommandSender +from streamdeck.event_handlers.actions import Action, ActionBase +from streamdeck.event_handlers.protocol import ( + SupportsEventHandlers, + is_bindable_handler, + is_not_bindable_handler, +) +from streamdeck.event_handlers.registry import HandlersRegistry from streamdeck.event_listener import EventListener, EventListenerManager from streamdeck.models.events.adapter import EventAdapter from streamdeck.models.events.common import ContextualEventMixin @@ -21,7 +25,9 @@ from collections.abc import Generator from typing import Any, Literal - from streamdeck.actions import ( + from streamdeck.event_handlers.protocol import ( + BindableEventHandlerFunc, + BoundEventHandlerFunc, EventHandlerFunc, EventModel_contra, InjectableParams, @@ -29,35 +35,11 @@ from streamdeck.models.events import EventBase - BindableEventHandlerFunc = EventHandlerFunc[EventModel_contra, [StreamDeckCommandSender]] - """Type alias for a bindable event handler function that takes an event (of subtype of EventBase) and a command_sender parameter that is to be injected.""" - BoundEventHandlerFunc = EventHandlerFunc[EventModel_contra, []] - """Type alias for a bound event handler function that takes an event (of subtype of EventBase) and no other parameters. - - Typically used for event handlers that have already had parameters injected. - """ - - # TODO: Fix this up to push to a log in the apropos directory and filename. logger = getLogger("streamdeck.manager") -def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BindableEventHandlerFunc[EventModel_contra]]: - """Check if the handler is prebound with the `command_sender` parameter.""" - # Check dynamically if the `command_sender`'s name is in the handler's arguments. - return "command_sender" in inspect.signature(handler).parameters - - -def is_not_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BoundEventHandlerFunc[EventModel_contra]]: - """Check if the handler only accepts the event_data parameter. - - If this function returns False after the is_bindable_handler check is True, then the function has invalid parameters, and will subsequently need to be handled in the calling code. - """ - handler_params = inspect.signature(handler).parameters - return len(handler_params) == 1 and "event_data" in handler_params - - class PluginManager: """Manages plugin actions and communicates with a WebSocket server to handle events.""" @@ -88,12 +70,12 @@ def __init__( self._register_event = register_event self._info = info - self._action_registry = ActionRegistry() + self._handlers_registry = HandlersRegistry() self._event_listener_manager = EventListenerManager() self._event_adapter = EventAdapter() - def _ensure_action_has_valid_events(self, action: ActionBase) -> None: - """Ensure that the action's registered events are valid. + def _ensure_catalog_has_valid_events(self, action: SupportsEventHandlers) -> None: + """Ensure that the event handler catalog's registered events are valid. Args: action (Action): The action to validate. @@ -111,13 +93,13 @@ def register_action(self, action: ActionBase) -> None: action (Action): The action to register. """ # First, validate that the action's registered events are valid. - self._ensure_action_has_valid_events(action) + self._ensure_catalog_has_valid_events(action) # Next, configure a logger for the action, giving it the last part of its uuid as name (if it has one). action_component_name = action.uuid.split(".")[-1] if isinstance(action, Action) else "global" configure_streamdeck_logger(name=action_component_name, plugin_uuid=self.uuid) - self._action_registry.register(action) + self._handlers_registry.register(action) def register_event_listener(self, listener: EventListener) -> None: """Register an event listener with the PluginManager, and add its event models to the event adapter. @@ -130,6 +112,31 @@ def register_event_listener(self, listener: EventListener) -> None: for event_model in listener.event_models: self._event_adapter.add_model(event_model) + def _stream_event_data(self) -> Generator[EventBase, None, None]: + """Stream event data from the event listeners. + + Validate and model the incoming event data before yielding it. + + Yields: + EventBase: The event data received from the event listeners. + """ + for message in self._event_listener_manager.event_stream(): + try: + data: EventBase = self._event_adapter.validate_json(message) + except ValidationError: + logger.exception("Error modeling event data.") + continue + + logger.debug("Event received: %s", data.event) + + # TODO: is this necessary? Or would this be covered by the event adapter validation? + if not self._event_adapter.event_name_exists(data.event): + logger.error("Invalid event received: %s", data.event) + continue + + yield data + + # def _inject_command_sender(self, handler: EventHandlerFunc[EventModel_contra, InjectableParams], device: DeviceUUIDStr | None, action: ActionUUIDStr | None, action_instance: ActionInstanceUUIDStr | None) -> BoundEventHandlerFunc[EventModel_contra]: def _inject_command_sender(self, handler: EventHandlerFunc[EventModel_contra, InjectableParams], command_sender: StreamDeckCommandSender) -> BoundEventHandlerFunc[EventModel_contra]: """Inject command_sender into handler if it accepts it as a parameter. @@ -146,33 +153,27 @@ def _inject_command_sender(self, handler: EventHandlerFunc[EventModel_contra, In if not is_not_bindable_handler(handler): # If the handler is neither bindable nor not bindable, raise an error. - raise TypeError(f"Invalid event handler function signature: {handler}") # noqa: TRY003, EM102 + raise TypeError(f"Invalid event handler function signature: {handler}") return handler - def _stream_event_data(self) -> Generator[EventBase, None, None]: - """Stream event data from the event listeners. + # return self._injector.bind_injectable(handler, device, action, action_instance) - Validate and model the incoming event data before yielding it. + # TODO: rather than an explicit 'command_sender' arg, this will eventually be handled by dynamically binded args. + def _dispatch_event(self, event: EventBase, command_sender: StreamDeckCommandSender) -> None: + """Dispatch an event to the appropriate action handlers. - Yields: - EventBase: The event data received from the event listeners. + Args: + event (EventBase): The event data model for the event to dispatch. """ - for message in self._event_listener_manager.event_stream(): - try: - data: EventBase = self._event_adapter.validate_json(message) - except ValidationError: - logger.exception("Error modeling event data.") - continue + # If the event is action-specific, we'll pass the action's uuid to the handler to ensure only the correct action is triggered. + event_action_uuid = event.action if isinstance(event, ContextualEventMixin) else None - logger.debug("Event received: %s", data.event) + for event_handler in self._handlers_registry.get_event_handlers(event_name=event.event, event_action_uuid=event_action_uuid): + processed_handler = self._inject_command_sender(event_handler, command_sender) + # TODO: from contextual event occurrences, save metadata to the action's properties. - # TODO: is this necessary? Or would this be covered by the event adapter validation? - if not self._event_adapter.event_name_exists(data.event): - logger.error("Invalid event received: %s", data.event) - continue - - yield data + processed_handler(event) def run(self) -> None: """Run the PluginManager by connecting to the WebSocket server and processing incoming events. @@ -188,13 +189,6 @@ def run(self) -> None: command_sender.send_action_registration(register_event=self._register_event) for data in self._stream_event_data(): - # If the event is action-specific, we'll pass the action's uuid to the handler to ensure only the correct action is triggered. - event_action_uuid = data.action if isinstance(data, ContextualEventMixin) else None - - for event_handler in self._action_registry.get_action_handlers(event_name=data.event, event_action_uuid=event_action_uuid): - processed_handler = self._inject_command_sender(event_handler, command_sender) - # TODO: from contextual event occurences, save metadata to the action's properties. - - processed_handler(data) + self._dispatch_event(data, command_sender) logger.info("PluginManager has stopped processing events.") diff --git a/tests/data/test_action1.py b/tests/data/test_action1.py index f2d366b..3360876 100644 --- a/tests/data/test_action1.py +++ b/tests/data/test_action1.py @@ -1,5 +1,5 @@ -from streamdeck.actions import Action from streamdeck.command_sender import StreamDeckCommandSender +from streamdeck.event_handlers.actions import Action from streamdeck.models.events import KeyDown, WillAppear diff --git a/tests/data/test_action2.py b/tests/data/test_action2.py index d676184..2357629 100644 --- a/tests/data/test_action2.py +++ b/tests/data/test_action2.py @@ -1,4 +1,4 @@ -from streamdeck.actions import Action, GlobalAction +from streamdeck.event_handlers.actions import Action, GlobalAction from streamdeck.models.events import ApplicationDidLaunch diff --git a/tests/event_handlers/__init__.py b/tests/event_handlers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/actions/test_action_event_handler_filtering.py b/tests/event_handlers/test_action_event_handler_filtering.py similarity index 89% rename from tests/actions/test_action_event_handler_filtering.py rename to tests/event_handlers/test_action_event_handler_filtering.py index efd7d17..87eec2b 100644 --- a/tests/actions/test_action_event_handler_filtering.py +++ b/tests/event_handlers/test_action_event_handler_filtering.py @@ -4,7 +4,8 @@ from unittest.mock import create_autospec import pytest -from streamdeck.actions import Action, ActionRegistry, GlobalAction +from streamdeck.event_handlers.actions import Action, GlobalAction +from streamdeck.event_handlers.registry import HandlersRegistry from streamdeck.models.events.common import ContextualEventMixin from tests.test_utils.fake_event_factories import ( @@ -86,14 +87,14 @@ def test_action_gets_triggered_by_event( -def test_global_action_registry_get_action_handlers_filtering( +def test_global_handlers_registry_get_event_handlers_filtering( mock_event_handler: Mock, fake_event_data: events.EventBase, ) -> None: # Extract the action UUID from the fake event data, or use a default value action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else None - registry = ActionRegistry() + registry = HandlersRegistry() # Create an Action instance, without an action UUID as global actions aren't associated with a specific action global_action = GlobalAction() @@ -102,7 +103,7 @@ def test_global_action_registry_get_action_handlers_filtering( # Register the global action with the registry registry.register(global_action) - for handler in registry.get_action_handlers( + for handler in registry.get_event_handlers( event_name=fake_event_data.event, event_action_uuid=action_uuid, ): @@ -113,14 +114,14 @@ def test_global_action_registry_get_action_handlers_filtering( -def test_action_registry_get_action_handlers_filtering( +def test_handlers_registry_get_event_handlers_filtering( mock_event_handler: Mock, fake_event_data: events.EventBase, ) -> None: # Extract the action UUID from the fake event data, or use a default value action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else None - registry = ActionRegistry() + registry = HandlersRegistry() # Create an Action instance, using either the fake event's action UUID or a default value action = Action(uuid=action_uuid or "my-fake-action-uuid") @@ -129,7 +130,7 @@ def test_action_registry_get_action_handlers_filtering( # Register the action with the registry registry.register(action) - for handler in registry.get_action_handlers( + for handler in registry.get_event_handlers( event_name=fake_event_data.event, event_action_uuid=action_uuid, # This will be None if the event is not action-specific (i.e. doesn't have an action UUID property) ): @@ -141,7 +142,7 @@ def test_action_registry_get_action_handlers_filtering( def test_multiple_actions_filtering() -> None: - registry = ActionRegistry() + registry = HandlersRegistry() action = Action("my-fake-action-uuid-1") global_action = GlobalAction() @@ -166,7 +167,7 @@ def _action_key_down_event_handler(event: events.EventBase) -> None: fake_app_did_launch_event_data: events.ApplicationDidLaunch = ApplicationDidLaunchEventFactory.build() fake_key_down_event_data: events.KeyDown = KeyDownEventFactory.build(action=action.uuid) - for handler in registry.get_action_handlers(event_name=fake_app_did_launch_event_data.event): + for handler in registry.get_event_handlers(event_name=fake_app_did_launch_event_data.event): handler(fake_app_did_launch_event_data) assert global_action_event_handler_called @@ -176,7 +177,7 @@ def _action_key_down_event_handler(event: events.EventBase) -> None: global_action_event_handler_called = False # Get the action handlers for the event and call them - for handler in registry.get_action_handlers(event_name=fake_key_down_event_data.event, event_action_uuid=fake_key_down_event_data.action): + for handler in registry.get_event_handlers(event_name=fake_key_down_event_data.event, event_action_uuid=fake_key_down_event_data.action): handler(fake_key_down_event_data) assert action_event_handler_called diff --git a/tests/actions/test_action_registry.py b/tests/event_handlers/test_action_registry.py similarity index 64% rename from tests/actions/test_action_registry.py rename to tests/event_handlers/test_action_registry.py index 34cf396..f5128f6 100644 --- a/tests/actions/test_action_registry.py +++ b/tests/event_handlers/test_action_registry.py @@ -3,7 +3,8 @@ from typing import TYPE_CHECKING import pytest -from streamdeck.actions import Action, ActionRegistry +from streamdeck.event_handlers.actions import Action +from streamdeck.event_handlers.registry import HandlersRegistry from tests.test_utils.fake_event_factories import ( DialDownEventFactory, @@ -18,31 +19,31 @@ def test_register_action() -> None: """Test that an action can be registered.""" - registry = ActionRegistry() + registry = HandlersRegistry() action = Action("my-fake-action-uuid") - assert len(registry._plugin_actions) == 0 + assert len(registry._plugin_event_handler_catalogs) == 0 registry.register(action) - assert len(registry._plugin_actions) == 1 - assert registry._plugin_actions[0] == action + assert len(registry._plugin_event_handler_catalogs) == 1 + assert registry._plugin_event_handler_catalogs[0] == action -def test_get_action_handlers_no_handlers() -> None: +def test_get_event_handlers_no_handlers() -> None: """Test that getting action handlers when there are no handlers yields nothing.""" - registry = ActionRegistry() + registry = HandlersRegistry() action = Action("my-fake-action-uuid") registry.register(action) fake_event_data: events.DialUp = DialUpEventFactory.build() - handlers = list(registry.get_action_handlers(event_name=fake_event_data.event, event_action_uuid=fake_event_data.action)) + handlers = list(registry.get_event_handlers(event_name=fake_event_data.event, event_action_uuid=fake_event_data.action)) assert len(handlers) == 0 -def test_get_action_handlers_with_handlers() -> None: +def test_get_event_handlers_with_handlers() -> None: """Test that registered event handlers can be retrieved correctly.""" - registry = ActionRegistry() + registry = HandlersRegistry() action = Action("my-fake-action-uuid") @action.on("dialDown") @@ -52,15 +53,15 @@ def dial_down_handler(event_data: events.EventBase) -> None: registry.register(action) fake_event_data: events.DialDown = DialDownEventFactory.build(action=action.uuid) - handlers = list(registry.get_action_handlers(event_name=fake_event_data.event, event_action_uuid=fake_event_data.action)) - # handlers = list(registry.get_action_handlers("dialDown")) + handlers = list(registry.get_event_handlers(event_name=fake_event_data.event, event_action_uuid=fake_event_data.action)) + # handlers = list(registry.get_event_handlers("dialDown")) assert len(handlers) == 1 assert handlers[0] == dial_down_handler -def test_get_action_handlers_multiple_actions() -> None: +def test_get_event_handlers_multiple_actions() -> None: """Test that multiple actions with registered handlers return all handlers.""" - registry = ActionRegistry() + registry = HandlersRegistry() action1 = Action("fake-action-uuid-1") action2 = Action("fake-action-uuid-2") @@ -78,7 +79,7 @@ def key_up_handler2(event_data: events.EventBase) -> None: fake_event_data: events.KeyUp = KeyUpEventFactory.build(action=action1.uuid) # Notice no action uuid is passed in here, so we should get all handlers for the event. - handlers = list(registry.get_action_handlers(event_name=fake_event_data.event)) + handlers = list(registry.get_event_handlers(event_name=fake_event_data.event)) assert len(handlers) == 2 assert key_up_handler1 in handlers diff --git a/tests/actions/test_actions.py b/tests/event_handlers/test_actions.py similarity index 97% rename from tests/actions/test_actions.py rename to tests/event_handlers/test_actions.py index 12f77c8..0f6b49e 100644 --- a/tests/actions/test_actions.py +++ b/tests/event_handlers/test_actions.py @@ -3,7 +3,7 @@ from typing import TYPE_CHECKING, Any, cast import pytest -from streamdeck.actions import Action, ActionBase, GlobalAction +from streamdeck.event_handlers.actions import Action, ActionBase, GlobalAction from streamdeck.models.events import DEFAULT_EVENT_NAMES diff --git a/tests/event_listener/test_event_listener.py b/tests/event_listener/test_event_listener.py index ea49587..55fc222 100644 --- a/tests/event_listener/test_event_listener.py +++ b/tests/event_listener/test_event_listener.py @@ -1,7 +1,8 @@ +from __future__ import annotations + import threading import time -from collections.abc import Generator -from typing import Any, ClassVar +from typing import TYPE_CHECKING from unittest.mock import Mock, patch import pytest @@ -9,6 +10,13 @@ from streamdeck.models.events import ApplicationDidLaunch, EventBase +if TYPE_CHECKING: + from collections.abc import Generator + from typing import Any, ClassVar + + from streamdeck.types import EventNameStr + + class MockEventListener(EventListener): """Mock implementation of EventListener for testing.""" event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] diff --git a/tests/plugin_manager/test_command_sender_binding.py b/tests/plugin_manager/test_command_sender_binding.py index 98a25f2..bfd0fce 100644 --- a/tests/plugin_manager/test_command_sender_binding.py +++ b/tests/plugin_manager/test_command_sender_binding.py @@ -5,14 +5,14 @@ from unittest.mock import Mock, create_autospec import pytest -from streamdeck.actions import Action +from streamdeck.event_handlers.actions import Action if TYPE_CHECKING: from functools import partial - from streamdeck.actions import EventHandlerFunc from streamdeck.command_sender import StreamDeckCommandSender + from streamdeck.event_handlers.protocol import EventHandlerFunc from streamdeck.manager import PluginManager from streamdeck.models import events diff --git a/tests/plugin_manager/test_plugin_manager.py b/tests/plugin_manager/test_plugin_manager.py index b471539..95ae3c6 100644 --- a/tests/plugin_manager/test_plugin_manager.py +++ b/tests/plugin_manager/test_plugin_manager.py @@ -3,7 +3,7 @@ import pytest import pytest_mock -from streamdeck.actions import Action +from streamdeck.event_handlers.actions import Action from streamdeck.manager import EventAdapter, PluginManager from streamdeck.models.events import ( #, event_adapter DEFAULT_EVENT_MODELS, @@ -13,10 +13,10 @@ @pytest.fixture -def _spy_action_registry_get_action_handlers( +def _spy_handlers_registry_get_event_handlers( mocker: pytest_mock.MockerFixture, plugin_manager: PluginManager ) -> None: - """Fixture that wraps and spies on the get_action_handlers method of the action_registry. + """Fixture that wraps and spies on the get_event_handlers method of the handlers_registry. Args: mocker: pytest-mock's mocker fixture. @@ -25,7 +25,7 @@ def _spy_action_registry_get_action_handlers( Returns: None """ - mocker.spy(plugin_manager._action_registry, "get_action_handlers") + mocker.spy(plugin_manager._handlers_registry, "get_event_handlers") @pytest.fixture @@ -43,13 +43,13 @@ def _spy_event_adapter_validate_json(mocker: pytest_mock.MockerFixture) -> None: def test_plugin_manager_register_action(plugin_manager: PluginManager) -> None: """Test that an action can be registered in the PluginManager.""" - assert len(plugin_manager._action_registry._plugin_actions) == 0 + assert len(plugin_manager._handlers_registry._plugin_event_handler_catalogs) == 0 action = Action("my-fake-action-uuid") plugin_manager.register_action(action) - assert len(plugin_manager._action_registry._plugin_actions) == 1 - assert plugin_manager._action_registry._plugin_actions[0] == action + assert len(plugin_manager._handlers_registry._plugin_event_handler_catalogs) == 1 + assert plugin_manager._handlers_registry._plugin_event_handler_catalogs[0] == action def test_plugin_manager_register_event_listener(plugin_manager: PluginManager) -> None: @@ -116,11 +116,11 @@ def test_plugin_manager_adds_websocket_event_listener( @pytest.mark.usefixtures("patch_websocket_client") def test_plugin_manager_process_event( mock_event_listener_manager_with_fake_events: tuple[Mock, list[EventBase]], - _spy_action_registry_get_action_handlers: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. + _spy_handlers_registry_get_event_handlers: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. _spy_event_adapter_validate_json: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. plugin_manager: PluginManager, # This fixture must come after patch_event_listener_manager and spy-fixtures to ensure things are mocked and spied correctly. ) -> None: - """Test that PluginManager processes events correctly, calling event_adapter.validate_json and action_registry.get_action_handlers.""" + """Test that PluginManager processes events correctly, calling event_adapter.validate_json and handlers_registry.get_event_handlers.""" mock_event_listener_mgr, fake_event_messages = mock_event_listener_manager_with_fake_events fake_event_message = fake_event_messages[0] @@ -137,9 +137,9 @@ def test_plugin_manager_process_event( # Check that the validate_json method returns the same event type model as the fake_event_message. assert spied_event_adapter__validate_json.spy_return == fake_event_message - # Check that the action_registry.get_action_handlers method was called with the event name and action uuid. - spied_action_registry__get_action_handlers = cast("Mock", plugin_manager._action_registry.get_action_handlers) - spied_action_registry__get_action_handlers.assert_called_once_with( + # Check that the handlers_registry.get_event_handlers method was called with the event name and action uuid. + spied_handlers_registry__get_event_handlers = cast("Mock", plugin_manager._handlers_registry.get_event_handlers) + spied_handlers_registry__get_event_handlers.assert_called_once_with( event_name=fake_event_message.event, event_action_uuid=fake_event_message.action ) diff --git a/tests/test_utils/fake_event_factories.py b/tests/test_utils/fake_event_factories.py index 47ce2ec..62816e6 100644 --- a/tests/test_utils/fake_event_factories.py +++ b/tests/test_utils/fake_event_factories.py @@ -1,5 +1,8 @@ +from polyfactory import Use from polyfactory.factories.pydantic_factory import ModelFactory from streamdeck.models import events +from streamdeck.models.events.settings import SingleActionSettingsPayload +from streamdeck.models.events.visibility import SingleActionVisibilityPayload class DialDownEventFactory(ModelFactory[events.DialDown]): @@ -40,3 +43,29 @@ class DeviceDidConnectFactory(ModelFactory[events.DeviceDidConnect]): DeviceDidConnectEvent's have `device` unique identifier property. """ + + +ControllerTypeFactory = Use(ModelFactory.__random__.choice, ["Keypad", "Encoder"]) + + +class DidReceiveSettingsFactory(ModelFactory[events.DidReceiveSettings]): + + class SingleActionSettingsPayloadFactory(ModelFactory[SingleActionSettingsPayload]): + __set_as_default_factory_for_type__ = True + + controller = ControllerTypeFactory + + +class WillAppearFactory(ModelFactory[events.WillAppear]): + """Polyfactory factory for creating fake willAppear event message based on our Pydantic model. + + WillAppearEvent'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 SingleActionVisibilityPayloadFactory(ModelFactory[SingleActionVisibilityPayload]): + __set_as_default_factory_for_type__ = True + + controller = ControllerTypeFactory