added tests

Author: Andrei Neagu

services/dynamic-sidecar/tests/unit/test_modules_outputs_event_handler.py

@@ -2,8 +2,9 @@
 # pylint: disable=protected-access
 
 import asyncio
+from collections.abc import AsyncIterable
 from pathlib import Path
-from typing import AsyncIterable
+from typing import Any, Final
 from unittest.mock import Mock
 
 import aioprocessing
@@ -17,8 +18,16 @@
 from simcore_service_dynamic_sidecar.modules.outputs._event_handler import (
     EventHandlerObserver,
     _EventHandlerProcess,
+    _PortKeysEventHandler,
 )
 from simcore_service_dynamic_sidecar.modules.outputs._manager import OutputsManager
+from watchdog.events import (
+    DirModifiedEvent,
+    FileClosedEvent,
+    FileCreatedEvent,
+    FileMovedEvent,
+    FileSystemEvent,
+)
 
 
 @pytest.fixture
@@ -124,3 +133,62 @@ async def test_event_handler_observer_health_degraded(
     await asyncio.sleep(observer_monitor.wait_for_heart_beat_interval_s * 3)
     await observer_monitor.stop()
     assert outputs_manager.set_all_ports_for_upload.call_count >= 1
+
+
+_STATE_PATH: Final[Path] = Path("/some/random/fake/path/for/state/")
+
+
+@pytest.fixture
+def mock_state_path() -> Path:
+    return _STATE_PATH
+
+
+class _MockAioQueue:
+    def __init__(self) -> None:
+        self.items: list[Any] = []
+
+    def put(self, item: Any) -> None:
+        self.items.append(item)
+
+    def get(self) -> Any | None:
+        try:
+            return self.items.pop()
+        except IndexError:
+            return None
+
+
+@pytest.mark.parametrize(
+    "event, expected_port_key",
+    [
+        (FileCreatedEvent(src_path=f"{_STATE_PATH}/untitled.txt", dest_path=""), None),
+        (DirModifiedEvent(src_path=f"{_STATE_PATH}", dest_path=""), None),
+        (FileClosedEvent(src_path=f"{_STATE_PATH}/untitled.txt", dest_path=""), None),
+        (
+            FileMovedEvent(
+                src_path=f"{_STATE_PATH}/untitled.txt",
+                dest_path=f"{_STATE_PATH}/asdsadsasad.txt",
+            ),
+            None,
+        ),
+        (
+            FileMovedEvent(
+                src_path=f"{_STATE_PATH}/asdsadsasad.txt",
+                dest_path=f"{_STATE_PATH}/output_1/asdsadsasad.txt",
+            ),
+            "output_1",
+        ),
+        (DirModifiedEvent(src_path=f"{_STATE_PATH}/output_1", dest_path=""), None),
+    ],
+)
+def test_port_keys_event_handler_triggers_for_events(
+    mock_state_path: Path, event: FileSystemEvent, expected_port_key: str | None
+) -> None:
+
+    queue = _MockAioQueue()
+
+    event_handler = _PortKeysEventHandler(mock_state_path, queue)
+    event_handler.handle_set_outputs_port_keys(outputs_port_keys={"output_1"})
+    event_handler.handle_toggle_event_propagation(is_enabled=True)
+
+    event_handler.event_handler(event)
+    assert queue.get() == expected_port_key