Add code and tests for new global actions feature

Author: strohganoff

streamdeck/actions.py

@@ -39,24 +39,17 @@
 }
 
 
-class Action:
-    """Represents an action that can be performed, with event handlers for specific event types."""
+class ActionBase:
+    """Base class for all actions."""
 
-    def __init__(self, uuid: str):
+    def __init__(self):
         """Initialize an Action instance.
 
         Args:
             uuid (str): The unique identifier for the action.
         """
-        self.uuid = uuid
-
         self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set)
 
-    @cached_property
-    def name(self) -> str:
-        """The name of the action, derived from the last part of the UUID."""
-        return self.uuid.split(".")[-1]
-
     def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc], EventHandlerFunc]:
         """Register an event handler for a specific event.
 
@@ -99,14 +92,36 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand
         yield from self._events[event_name]
 
 
+class GlobalAction(ActionBase):
+    """Represents an action that is performed at the plugin level, meaning it isn't associated with a specific device or action."""
+
+
+class Action(ActionBase):
+    """Represents an action that can be performed for a specific action, with event handlers for specific event types."""
+
+    def __init__(self, uuid: str):
+        """Initialize an Action instance.
+
+        Args:
+            uuid (str): The unique identifier for the action.
+        """
+        super().__init__()
+        self.uuid = uuid
+
+    @cached_property
+    def name(self) -> str:
+        """The name of the action, derived from the last part of the UUID."""
+        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[Action] = []
+        self._plugin_actions: list[ActionBase] = []
 
-    def register(self, action: Action) -> None:
+    def register(self, action: ActionBase) -> None:
         """Register an action with the registry.
 
         Args:
@@ -126,9 +141,10 @@ def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str |
             EventHandlerFunc: The event handler functions for the specified event.
         """
         for action in self._plugin_actions:
-            # If the event is action-specific, only get handlers for that action, as we don't want to trigger
+            # 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 event_action_uuid != action.uuid:
+            if event_action_uuid is not None and (hasattr(action, "uuid") and action.uuid != event_action_uuid):
                 continue
 
             yield from action.get_event_handlers(event_name)

streamdeck/models/events.py

@@ -21,6 +21,16 @@ def is_action_specific(cls) -> bool:
         """Check if the event is specific to an action instance (i.e. the event has an "action" field)."""
         return "action" in cls.model_fields
 
+    @classmethod
+    def is_device_specific(cls) -> bool:
+        """Check if the event is specific to a device instance (i.e. the event has a "device" field)."""
+        return "device" in cls.model_fields
+
+    @classmethod
+    def is_action_instance_specific(cls) -> bool:
+        """Check if the event is specific to an action instance (i.e. the event has a "context" field)."""
+        return "context" in cls.model_fields
+
 
 class ApplicationDidLaunchEvent(EventBase):
     event: Literal["applicationDidLaunch"]

tests/actions/__init__.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+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
+
+
+if TYPE_CHECKING:
+    from unittest.mock import Mock
+
+
+
+@pytest.fixture
+def mock_event_handler() -> Mock:
+    def dummy_handler(event: events.EventBase) -> None:
+        """Dummy event handler function that matches the EventHandlerFunc TypeAlias."""
+
+    return create_autospec(dummy_handler, spec_set=True)
+
+
+class ApplicationDidLaunchEventFactory(ModelFactory[events.ApplicationDidLaunchEvent]):
+    """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.DeviceDidConnectEvent]):
+    """Polyfactory factory for creating fake deviceDidConnect event message based on our Pydantic model.
+
+    DeviceDidConnectEvent's have `device` unique identifier property.
+    """
+
+class KeyDownEventFactory(ModelFactory[events.KeyDownEvent]):
+    """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),
+    ("applicationDidLaunch", ApplicationDidLaunchEventFactory)
+])
+def test_global_action_gets_triggered_by_event(
+    mock_event_handler: Mock,
+    event_name: str,
+    event_factory: ModelFactory[events.EventBase],
+):
+    """Test that a global action's event handlers are triggered by an event.
+
+    Global actions should be triggered by any event type that is registered with them,
+      regardless of the event's unique identifier properties (or whether they're even present).
+    """
+    fake_event_data = event_factory.build()
+
+    global_action = GlobalAction()
+
+    global_action.on(event_name)(mock_event_handler)
+
+    for handler in global_action.get_event_handlers(event_name):
+        handler(fake_event_data)
+
+    assert mock_event_handler.call_count == 1
+    assert fake_event_data in mock_event_handler.call_args.args
+
+
+@pytest.mark.parametrize(("event_name","event_factory"), [
+    ("keyDown", KeyDownEventFactory),
+    ("deviceDidConnect", DeviceDidConnectFactory),
+    ("applicationDidLaunch", ApplicationDidLaunchEventFactory)
+])
+def test_action_gets_triggered_by_event(mock_event_handler: Mock, event_name: str, event_factory: ModelFactory[events.EventBase]):
+    # Create a fake event model instance
+    fake_event_data: events.EventBase = event_factory.build()
+    # Extract the action UUID from the fake event data, or use a default value
+    action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else "my-fake-action-uuid"
+
+    action = Action(uuid=action_uuid)
+
+    # Register the mock event handler with the action
+    action.on(event_name)(mock_event_handler)
+
+    # Get the action's event handlers for the event and call them
+    for handler in action.get_event_handlers(event_name):
+        handler(fake_event_data)
+
+    # For some reason, assert_called_once() and assert_called_once_with() are returning False here...
+    # assert mock_event_handler.assert_called_once(fake_event_data)
+    assert mock_event_handler.call_count == 1
+    assert fake_event_data in mock_event_handler.call_args.args
+
+
+
+@pytest.mark.parametrize(("event_name","event_factory"), [
+    ("keyDown", KeyDownEventFactory),
+    ("deviceDidConnect", DeviceDidConnectFactory),
+    ("applicationDidLaunch", ApplicationDidLaunchEventFactory)
+])
+def test_global_action_registry_get_action_handlers_filtering(mock_event_handler: Mock, event_name: str, event_factory: ModelFactory[events.EventBase]):
+    # Create a fake event model instance
+    fake_event_data: events.EventBase = event_factory.build()
+    # Extract the action UUID from the fake event data, or use a default value
+    action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None
+
+    registry = ActionRegistry()
+    # Create an Action instance, without an action UUID as global actions aren't associated with a specific action
+    global_action = GlobalAction()
+
+    global_action.on(event_name)(mock_event_handler)
+
+    # Register the global action with the registry
+    registry.register(global_action)
+
+    for handler in registry.get_action_handlers(
+        event_name=fake_event_data.event,
+        event_action_uuid=action_uuid,
+    ):
+        handler(fake_event_data)
+
+    assert mock_event_handler.call_count == 1
+    assert fake_event_data in mock_event_handler.call_args.args
+
+
+
+@pytest.mark.parametrize(("event_name","event_factory"), [
+    ("keyDown", KeyDownEventFactory),
+    ("deviceDidConnect", DeviceDidConnectFactory),
+    ("applicationDidLaunch", ApplicationDidLaunchEventFactory)
+])
+def test_action_registry_get_action_handlers_filtering(mock_event_handler: Mock, event_name: str, event_factory: ModelFactory[events.EventBase]):
+    # Create a fake event model instance
+    fake_event_data: events.EventBase = event_factory.build()
+    # Extract the action UUID from the fake event data, or use a default value
+    action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None
+
+    registry = ActionRegistry()
+    # 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")
+
+    action.on(event_name)(mock_event_handler)
+
+    # Register the action with the registry
+    registry.register(action)
+
+    for handler in registry.get_action_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)
+    ):
+        handler(fake_event_data)
+
+    assert mock_event_handler.call_count == 1
+    assert fake_event_data in mock_event_handler.call_args.args
+
+
+
+def test_multiple_actions_filtering():
+    registry = ActionRegistry()
+    action = Action("my-fake-action-uuid-1")
+    global_action = GlobalAction()
+
+    global_action_event_handler_called = False
+    action_event_handler_called = False
+
+    @global_action.on("applicationDidLaunch")
+    def _global_app_did_launch_action_handler(event: events.EventBase):
+        nonlocal global_action_event_handler_called
+        global_action_event_handler_called = True
+
+    @action.on("keyDown")
+    def _action_key_down_event_handler(event: events.EventBase):
+        nonlocal action_event_handler_called
+        action_event_handler_called = True
+
+    # Register both actions with the registry
+    registry.register(global_action)
+    registry.register(action)
+
+    # Create a fake event model instances
+    fake_app_did_launch_event_data: events.ApplicationDidLaunchEvent = ApplicationDidLaunchEventFactory.build()
+    fake_key_down_event_data: events.KeyDownEvent = KeyDownEventFactory.build(action=action.uuid)
+
+    for handler in registry.get_action_handlers(event_name=fake_app_did_launch_event_data.event):
+        handler(fake_app_did_launch_event_data)
+
+    assert global_action_event_handler_called
+    assert not action_event_handler_called
+
+    # Reset the flag for global action event handler
+    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):
+        handler(fake_key_down_event_data)
+
+    assert action_event_handler_called
+    assert not global_action_event_handler_called

tests/test_action.py renamed to tests/actions/test_action.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+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
+
+
+if TYPE_CHECKING:
+    from streamdeck.models.events import EventBase
+    from streamdeck.types import EventNameStr
+
+
+def test_global_action_register_event_handler():
+    """Test that an event handler can be registered for each valid event name."""
+    global_action = GlobalAction()
+
+    for event_name in available_event_names:
+        @global_action.on(event_name)
+        def handler(event: EventBase) -> None:
+            pass
+
+        # Ensure the handler is registered for the correct event name
+        assert len(global_action._events[event_name]) == 1
+        assert handler in global_action._events[event_name]
+
+
+def test_global_action_register_invalid_event_handler():
+    """Test that attempting to register an invalid event handler raises an exception."""
+    global_action = GlobalAction()
+
+    with pytest.raises(Exception):
+        @global_action.on("InvalidEvent")
+        def handler(event: EventBase):
+            pass
+
+
+def test_global_action_get_event_handlers():
+    """Test that the correct event handlers are retrieved for each event name."""
+    global_action = GlobalAction()
+
+    for event_name in available_event_names:
+        # Register a handler for the event name
+        @global_action.on(event_name)
+        def handler(event: EventBase):
+            pass
+
+        # Retrieve the handlers using the generator
+        handlers = list(global_action.get_event_handlers(event_name))
+
+        # Ensure that the correct handler is retrieved
+        assert len(handlers) == 1
+        assert handlers[0] == handler
+
+
+def test_global_action_get_event_handlers_no_event_registered():
+    """Test that attempting to get handlers for an event with no registered handlers raises an exception."""
+    global_action = GlobalAction()
+
+    with pytest.raises(Exception):
+        list(global_action.get_event_handlers("InvalidEvent"))
+
+
+def test_global_action_register_multiple_handlers_for_event():
+    """Test that multiple handlers can be registered for an event."""
+    global_action = GlobalAction()
+
+    @global_action.on("keyDown")
+    def handler_one(event: EventBase):
+        pass
+
+    @global_action.on("keyDown")
+    def handler_two(event: EventBase):
+        pass
+
+    handlers = list(global_action.get_event_handlers("keyDown"))
+
+    # Ensure both handlers are registered for the event
+    assert len(handlers) == 2
+    assert handler_one in handlers
+    assert handler_two in handlers
+
+
+def test_global_action_get_event_handlers_invalid_event():
+    """Test that attempting to get handlers for an invalid event raises a KeyError."""
+    global_action = GlobalAction()
+
+    with pytest.raises(KeyError):
+        list(global_action.get_event_handlers("InvalidEvent"))