Add unit tests for osism/services websocket_manager and event_bridge

Pipfile

@@ -44,6 +44,7 @@ watchdog = "==6.0.0"
 [dev-packages]
 httpx = "==0.28.1"
 pytest = "==9.1.1"
+pytest-asyncio = "==1.4.0"
 pytest-cov = "==7.1.0"
 pytest-mock = "==3.15.1"
 pytest-timeout = "==2.4.0"

osism/services/event_bridge.py

@@ -151,6 +151,15 @@ def _start_processor_thread(self):
         self._processor_thread.start()
         logger.info("Started event bridge processor thread")
 
+    def _close_subscriber(self):
+        """Close the current Redis subscriber and drop the reference."""
+        if self._redis_subscriber:
+            try:
+                self._redis_subscriber.close()
+            except Exception:
+                pass  # Ignore errors during cleanup
+            self._redis_subscriber = None
+
     def _redis_subscriber_loop(self):
         """Redis subscriber loop for receiving events from other containers with auto-reconnect."""
         retry_count = 0
@@ -202,14 +211,20 @@ def _redis_subscriber_loop(self):
 
                     except Exception as get_msg_error:
                         logger.error(f"Error getting Redis message: {get_msg_error}")
-                        break  # Break inner loop to trigger reconnect
+                        # Route through the bounded back-off path below so
+                        # the subscriber is recreated before resubscribing
+                        raise
 
             except Exception as e:
                 retry_count += 1
                 logger.error(
                     f"Redis subscriber error (attempt {retry_count}/{max_retries}): {e}"
                 )
 
+                # Close the failed subscriber before _init_redis() replaces it,
+                # so it does not leak and cannot be closed in its stead later
+                self._close_subscriber()
+
                 if retry_count < max_retries:
                     logger.info(
                         f"Retrying Redis subscription in {retry_delay} seconds..."
@@ -222,12 +237,7 @@ def _redis_subscriber_loop(self):
                     except Exception as init_error:
                         logger.error(f"Failed to reinitialize Redis: {init_error}")
 
-            finally:
-                if self._redis_subscriber:
-                    try:
-                        self._redis_subscriber.close()
-                    except Exception:
-                        pass  # Ignore errors during cleanup
+        self._close_subscriber()
 
         if retry_count >= max_retries:
             logger.error("Max Redis reconnection attempts reached, giving up")
@@ -243,18 +253,27 @@ def _process_single_event(self, event_data: Dict[str, Any]):
         try:
             import asyncio
 
-            # Create new event loop for this thread
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-
-            # Process the event
-            loop.run_until_complete(
-                self._websocket_manager.broadcast_event_from_notification(
-                    event_data["event_type"], event_data["payload"]
-                )
+            coro = self._websocket_manager.broadcast_event_from_notification(
+                event_data["event_type"], event_data["payload"]
             )
 
-            loop.close()
+            loop = getattr(self._websocket_manager, "loop", None)
+            if loop is not None and loop.is_running():
+                # The broadcaster awaits the manager's asyncio.Queue on this
+                # loop; waking its waiters from a worker thread is not
+                # thread-safe, so the coroutine must run on that loop
+                future = asyncio.run_coroutine_threadsafe(coro, loop)
+                future.result(timeout=10)
+            else:
+                # No broadcaster loop yet (no client has connected), so the
+                # queue has no waiters and a private loop is safe
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+                try:
+                    loop.run_until_complete(coro)
+                finally:
+                    loop.close()
+
             logger.debug(f"Processed event via bridge: {event_data['event_type']}")
 
         except Exception as e:

osism/services/websocket_manager.py

@@ -89,6 +89,9 @@ def __init__(self):
         self.event_queue: asyncio.Queue = asyncio.Queue()
         # Background task for event broadcasting
         self._broadcaster_task: Optional[asyncio.Task] = None
+        # Loop the broadcaster runs on; the event bridge marshals
+        # broadcasts from its worker threads onto this loop
+        self.loop: Optional[asyncio.AbstractEventLoop] = None
         # Lock for thread-safe operations
         self._lock = asyncio.Lock()
 
@@ -101,6 +104,7 @@ async def connect(self, websocket: WebSocket) -> None:
 
         # Start broadcaster if this is the first connection
         if not self._broadcaster_task or self._broadcaster_task.done():
+            self.loop = asyncio.get_running_loop()
             self._broadcaster_task = asyncio.create_task(self._broadcast_events())
 
     async def disconnect(self, websocket: WebSocket) -> None:

setup.cfg

@@ -182,5 +182,6 @@ python_files = test_*.py
 python_classes = Test*
 python_functions = test_*
 addopts = -ra --strict-markers
+asyncio_default_fixture_loop_scope = function
 markers =
     integration: integration tests requiring a reachable Redis and a Celery worker (run with: pytest tests/integration)