diff --git a/Pipfile b/Pipfile index 89b5b4a91..1011a7f59 100644 --- a/Pipfile +++ b/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" diff --git a/Pipfile.lock b/Pipfile.lock index 263328c6d..a31083ad8 100644 --- a/Pipfile.lock +++ b/Pipfile.lock @@ -1,7 +1,7 @@ { "_meta": { "hash": { - "sha256": "5994dd3a1ce1cfb3f75011a0868b5dcd9ac78ce5fba0ad7df5cc713ccaebce8e" + "sha256": "6d1a2ba30cc14de4b48384c97f6dd066f4a1d4b07c6d17011736ed6f27f4af7b" }, "pipfile-spec": 6, "requires": {}, @@ -3148,6 +3148,15 @@ "markers": "python_version >= '3.10'", "version": "==9.1.1" }, + "pytest-asyncio": { + "hashes": [ + "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", + "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42" + ], + "index": "pypi", + "markers": "python_version >= '3.10'", + "version": "==1.4.0" + }, "pytest-cov": { "hashes": [ "sha256:30674f2b5f6351aa09702a9c8c364f6a01c27aae0c1366ae8016160d1efc56b2", @@ -3189,6 +3198,14 @@ "sha256:a163dcaede0f1c021485e957a39245190e74249897e2ae4b2aa38595db237ee0" ], "version": "==2.4.0" + }, + "typing-extensions": { + "hashes": [ + "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", + "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5" + ], + "markers": "python_version >= '3.9'", + "version": "==4.16.0" } } } diff --git a/osism/services/event_bridge.py b/osism/services/event_bridge.py index a0c051856..bd7bdf478 100644 --- a/osism/services/event_bridge.py +++ b/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,7 +211,9 @@ 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 @@ -210,6 +221,10 @@ def _redis_subscriber_loop(self): 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: diff --git a/osism/services/websocket_manager.py b/osism/services/websocket_manager.py index c8713df05..a5997bcb8 100644 --- a/osism/services/websocket_manager.py +++ b/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: diff --git a/setup.cfg b/setup.cfg index 51d6707ba..dceae6b90 100644 --- a/setup.cfg +++ b/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) diff --git a/tests/unit/services/test_event_bridge.py b/tests/unit/services/test_event_bridge.py new file mode 100644 index 000000000..a05bef12e --- /dev/null +++ b/tests/unit/services/test_event_bridge.py @@ -0,0 +1,590 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for :mod:`osism.services.event_bridge`. + +Covers Redis initialization, the publish path with reconnect and local-queue +fallback in ``add_event``, the subscriber and processor loops, single-event +processing and shutdown. + +Every test constructs a fresh ``EventBridge`` with +``osism.services.event_bridge.redis.Redis`` patched (the ``bridge`` fixture) +instead of using the module-level ``event_bridge`` singleton, which attempts +a real Redis connection at import time. + +The thread targets ``_redis_subscriber_loop`` and ``_process_events`` are +called synchronously; termination is driven through ``_shutdown_event`` from +``get_message`` / ``get`` side effects, so no real threads or sleeps are +involved. Failures of ``subscribe()`` and ``get_message()`` both run through +the same bounded back-off path: the failed subscriber is closed, the retry +waits, and ``_init_redis()`` provides a fresh subscriber. + +The module logs via the stdlib ``logging`` module (``osism.event_bridge``), +so the plain ``caplog`` fixture is used for log assertions. +""" + +import asyncio +import json +import logging +import queue +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from osism.services.event_bridge import EventBridge +from osism.services.websocket_manager import WebSocketManager + + +@pytest.fixture +def redis_cls(mocker): + return mocker.patch("osism.services.event_bridge.redis.Redis") + + +@pytest.fixture +def bridge(redis_cls): + return EventBridge() + + +class TestInitRedis: + def test_defaults(self, redis_cls, monkeypatch): + for name in ("REDIS_HOST", "REDIS_PORT", "REDIS_DB"): + monkeypatch.delenv(name, raising=False) + EventBridge() + redis_cls.assert_called_once_with( + host="redis", + port=6379, + db=0, + decode_responses=True, + socket_connect_timeout=10, + socket_timeout=None, + health_check_interval=30, + ) + + def test_environment_variables_override_defaults(self, redis_cls, monkeypatch): + monkeypatch.setenv("REDIS_HOST", "redis.example.com") + monkeypatch.setenv("REDIS_PORT", "16379") + monkeypatch.setenv("REDIS_DB", "2") + EventBridge() + assert redis_cls.call_args.kwargs["host"] == "redis.example.com" + assert redis_cls.call_args.kwargs["port"] == 16379 + assert redis_cls.call_args.kwargs["db"] == 2 + + def test_successful_ping_sets_client_and_subscriber(self, redis_cls): + bridge = EventBridge() + client = redis_cls.return_value + client.ping.assert_called_once_with() + assert bridge._redis_client is client + client.pubsub.assert_called_once_with() + assert bridge._redis_subscriber is client.pubsub.return_value + + def test_connection_failure_leaves_client_unset(self, redis_cls, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + redis_cls.return_value.ping.side_effect = ConnectionError("no redis") + bridge = EventBridge() + assert bridge._redis_client is None + assert bridge._redis_subscriber is None + assert "Failed to connect to Redis" in caplog.text + + def test_redis_not_available_uses_local_queue_only(self, redis_cls, mocker, caplog): + caplog.set_level(logging.WARNING, logger="osism.event_bridge") + mocker.patch("osism.services.event_bridge.REDIS_AVAILABLE", False) + bridge = EventBridge() + redis_cls.assert_not_called() + assert bridge._redis_client is None + assert "event bridge will use local queue only" in caplog.text + + +class TestSetWebsocketManager: + def test_stores_manager_and_starts_threads(self, bridge, mocker): + start_subscriber = mocker.patch.object(bridge, "_start_redis_subscriber") + start_processor = mocker.patch.object(bridge, "_start_processor_thread") + manager = MagicMock() + bridge.set_websocket_manager(manager) + assert bridge._websocket_manager is manager + start_subscriber.assert_called_once_with() + start_processor.assert_called_once_with() + + def test_subscriber_not_started_without_redis_client(self, bridge, mocker): + bridge._redis_client = None + start_subscriber = mocker.patch.object(bridge, "_start_redis_subscriber") + mocker.patch.object(bridge, "_start_processor_thread") + bridge.set_websocket_manager(MagicMock()) + start_subscriber.assert_not_called() + + def test_subscriber_not_started_twice(self, bridge, mocker): + bridge._subscriber_thread = MagicMock() + start_subscriber = mocker.patch.object(bridge, "_start_redis_subscriber") + mocker.patch.object(bridge, "_start_processor_thread") + bridge.set_websocket_manager(MagicMock()) + start_subscriber.assert_not_called() + + def test_processor_not_started_when_alive(self, bridge, mocker): + thread = MagicMock() + thread.is_alive.return_value = True + bridge._processor_thread = thread + mocker.patch.object(bridge, "_start_redis_subscriber") + start_processor = mocker.patch.object(bridge, "_start_processor_thread") + bridge.set_websocket_manager(MagicMock()) + start_processor.assert_not_called() + + def test_processor_restarted_when_dead(self, bridge, mocker): + thread = MagicMock() + thread.is_alive.return_value = False + bridge._processor_thread = thread + mocker.patch.object(bridge, "_start_redis_subscriber") + start_processor = mocker.patch.object(bridge, "_start_processor_thread") + bridge.set_websocket_manager(MagicMock()) + start_processor.assert_called_once_with() + + +class TestAddEvent: + def test_publishes_event_to_redis(self, bridge, caplog): + caplog.set_level(logging.INFO, logger="osism.event_bridge") + client = bridge._redis_client + client.publish.return_value = 3 + payload = {"a": 1} + bridge.add_event("baremetal.node.power_set", payload) + client.publish.assert_called_once_with( + "osism:events", + json.dumps({"event_type": "baremetal.node.power_set", "payload": payload}), + ) + assert bridge._event_queue.qsize() == 0 + assert "Published event to Redis" in caplog.text + assert "No Redis subscribers" not in caplog.text + + def test_warns_when_no_subscribers(self, bridge, caplog): + caplog.set_level(logging.WARNING, logger="osism.event_bridge") + bridge._redis_client.publish.return_value = 0 + bridge.add_event("a.b", {}) + assert "No Redis subscribers for event: a.b" in caplog.text + + def test_publish_retried_after_successful_reconnect(self, bridge, mocker, caplog): + caplog.set_level(logging.INFO, logger="osism.event_bridge") + client = bridge._redis_client + client.publish.side_effect = [ConnectionError("gone"), 2] + init_redis = mocker.patch.object(bridge, "_init_redis") + bridge.add_event("a.b", {"x": 1}) + init_redis.assert_called_once_with() + assert client.publish.call_count == 2 + assert bridge._event_queue.qsize() == 0 + assert "Published event to Redis after reconnect" in caplog.text + + def test_falls_back_to_local_queue_when_reconnect_fails( + self, bridge, mocker, caplog + ): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + client = bridge._redis_client + client.publish.side_effect = ConnectionError("gone") + + def drop_client(): + bridge._redis_client = None + + mocker.patch.object(bridge, "_init_redis", side_effect=drop_client) + bridge.add_event("a.b", {"x": 1}) + assert client.publish.call_count == 1 + assert bridge._event_queue.get_nowait() == { + "event_type": "a.b", + "payload": {"x": 1}, + } + assert "Redis reconnection failed" in caplog.text + + def test_uses_local_queue_without_redis_client(self, bridge): + bridge._redis_client = None + bridge.add_event("a.b", {"x": 1}) + assert bridge._event_queue.get_nowait() == { + "event_type": "a.b", + "payload": {"x": 1}, + } + + def test_full_queue_drops_event_with_warning(self, bridge, mocker, caplog): + """Defensive branch only: ``_event_queue`` is an unbounded + ``queue.Queue``, so ``put_nowait`` never raises ``queue.Full`` in + production. The handler is reachable solely by patching + ``put_nowait``; this does not document real drop behavior. + """ + caplog.set_level(logging.WARNING, logger="osism.event_bridge") + bridge._redis_client = None + mocker.patch.object(bridge._event_queue, "put_nowait", side_effect=queue.Full) + bridge.add_event("a.b", {}) + assert "Event bridge queue is full, dropping event" in caplog.text + + def test_generic_error_is_swallowed(self, bridge, mocker, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + bridge._redis_client = None + mocker.patch.object( + bridge._event_queue, "put_nowait", side_effect=ValueError("boom") + ) + bridge.add_event("a.b", {}) + assert "Error adding event to bridge: boom" in caplog.text + + +class TestRedisSubscriberLoop: + @pytest.mark.timeout(10) + def test_returns_immediately_without_subscriber(self, bridge, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + bridge._redis_subscriber = None + bridge._redis_subscriber_loop() + assert "Redis subscriber not available" in caplog.text + + @pytest.mark.timeout(10) + def test_subscribes_and_stops_on_shutdown(self, bridge, caplog): + caplog.set_level(logging.INFO, logger="osism.event_bridge") + subscriber = MagicMock() + bridge._redis_subscriber = subscriber + + def get_message(timeout=None): + bridge._shutdown_event.set() + return None + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + subscriber.subscribe.assert_called_once_with("osism:events") + subscriber.get_message.assert_called_once_with(timeout=10.0) + subscriber.close.assert_called_once_with() + assert "Redis subscriber stopped" in caplog.text + + @pytest.mark.timeout(10) + def test_valid_message_is_processed_with_manager(self, bridge, mocker): + bridge._websocket_manager = MagicMock() + process = mocker.patch.object(bridge, "_process_single_event") + subscriber = MagicMock() + bridge._redis_subscriber = subscriber + event_data = {"event_type": "a.b", "payload": {"x": 1}} + + def get_message(timeout=None): + bridge._shutdown_event.set() + return {"type": "message", "data": json.dumps(event_data)} + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + process.assert_called_once_with(event_data) + + @pytest.mark.timeout(10) + def test_valid_message_is_queued_without_manager(self, bridge): + subscriber = MagicMock() + bridge._redis_subscriber = subscriber + event_data = {"event_type": "a.b", "payload": {"x": 1}} + + def get_message(timeout=None): + bridge._shutdown_event.set() + return {"type": "message", "data": json.dumps(event_data)} + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + assert bridge._event_queue.get_nowait() == event_data + + @pytest.mark.timeout(10) + def test_invalid_json_logs_and_continues(self, bridge, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + subscriber = MagicMock() + bridge._redis_subscriber = subscriber + calls = {"count": 0} + + def get_message(timeout=None): + calls["count"] += 1 + if calls["count"] == 1: + return {"type": "message", "data": "not-json"} + bridge._shutdown_event.set() + return None + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + assert subscriber.get_message.call_count == 2 + assert "Failed to decode Redis event message" in caplog.text + + @pytest.mark.timeout(10) + def test_processing_error_logs_and_continues(self, bridge, mocker, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + bridge._websocket_manager = MagicMock() + mocker.patch.object( + bridge, "_process_single_event", side_effect=ValueError("boom") + ) + subscriber = MagicMock() + bridge._redis_subscriber = subscriber + calls = {"count": 0} + + def get_message(timeout=None): + calls["count"] += 1 + if calls["count"] == 1: + return {"type": "message", "data": json.dumps({"event_type": "a.b"})} + bridge._shutdown_event.set() + return None + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + assert subscriber.get_message.call_count == 2 + assert "Error processing Redis event: boom" in caplog.text + + @pytest.mark.timeout(10) + def test_get_message_error_reconnects_with_backoff(self, bridge, mocker, caplog): + """A failing ``get_message`` runs through the same bounded back-off + as a failing ``subscribe``: the failed subscriber is closed, the + retry waits, and the resubscribe happens on the newly created + instance.""" + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + failed = MagicMock() + failed.get_message.side_effect = ConnectionError("lost") + bridge._redis_subscriber = failed + fresh = MagicMock() + + def get_message(timeout=None): + bridge._shutdown_event.set() + return None + + fresh.get_message.side_effect = get_message + + def install_fresh_subscriber(): + bridge._redis_subscriber = fresh + + init_redis = mocker.patch.object( + bridge, "_init_redis", side_effect=install_fresh_subscriber + ) + wait_mock = mocker.patch.object( + bridge._shutdown_event, "wait", return_value=False + ) + bridge._redis_subscriber_loop() + failed.subscribe.assert_called_once_with("osism:events") + failed.close.assert_called_once_with() + fresh.subscribe.assert_called_once_with("osism:events") + fresh.close.assert_called_once_with() + wait_mock.assert_called_once_with(5) + init_redis.assert_called_once_with() + assert "Error getting Redis message: lost" in caplog.text + assert "Redis subscriber error (attempt 1/5): lost" in caplog.text + + @pytest.mark.timeout(10) + def test_subscribe_error_waits_and_reinitializes_redis( + self, bridge, mocker, caplog + ): + """A failed subscriber is closed before ``_init_redis`` replaces it, + and the retry subscribes on the newly created instance.""" + caplog.set_level(logging.INFO, logger="osism.event_bridge") + failed = MagicMock() + failed.subscribe.side_effect = ConnectionError("down") + bridge._redis_subscriber = failed + fresh = MagicMock() + + def get_message(timeout=None): + bridge._shutdown_event.set() + return None + + fresh.get_message.side_effect = get_message + + def install_fresh_subscriber(): + bridge._redis_subscriber = fresh + + init_redis = mocker.patch.object( + bridge, "_init_redis", side_effect=install_fresh_subscriber + ) + wait_mock = mocker.patch.object( + bridge._shutdown_event, "wait", return_value=False + ) + bridge._redis_subscriber_loop() + wait_mock.assert_called_once_with(5) + init_redis.assert_called_once_with() + failed.subscribe.assert_called_once_with("osism:events") + failed.close.assert_called_once_with() + fresh.subscribe.assert_called_once_with("osism:events") + fresh.close.assert_called_once_with() + assert "Redis subscriber error (attempt 1/5)" in caplog.text + assert "Retrying Redis subscription in 5 seconds" in caplog.text + + @pytest.mark.timeout(10) + def test_gives_up_after_max_retries(self, bridge, mocker, caplog): + """Every attempt subscribes on a distinct freshly created subscriber + and closes it after its failure.""" + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + subscribers = [MagicMock() for _ in range(5)] + for subscriber in subscribers: + subscriber.subscribe.side_effect = ConnectionError("down") + replacements = iter(subscribers[1:]) + + def install_fresh_subscriber(): + bridge._redis_subscriber = next(replacements) + + bridge._redis_subscriber = subscribers[0] + init_redis = mocker.patch.object( + bridge, "_init_redis", side_effect=install_fresh_subscriber + ) + wait_mock = mocker.patch.object( + bridge._shutdown_event, "wait", return_value=False + ) + bridge._redis_subscriber_loop() + for subscriber in subscribers: + subscriber.subscribe.assert_called_once_with("osism:events") + subscriber.close.assert_called_once_with() + # No back-off after the fifth and final failure. + assert wait_mock.call_count == 4 + assert init_redis.call_count == 4 + assert "Max Redis reconnection attempts reached, giving up" in caplog.text + + @pytest.mark.timeout(10) + def test_close_error_in_cleanup_is_ignored(self, bridge): + subscriber = MagicMock() + subscriber.close.side_effect = ConnectionError("close boom") + bridge._redis_subscriber = subscriber + + def get_message(timeout=None): + bridge._shutdown_event.set() + return None + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + subscriber.close.assert_called_once_with() + + +class TestProcessSingleEvent: + def test_without_manager_warns_and_returns(self, bridge, caplog): + caplog.set_level(logging.WARNING, logger="osism.event_bridge") + bridge._process_single_event({"event_type": "a.b", "payload": {}}) + assert "No WebSocket manager available, dropping event" in caplog.text + + def test_broadcasts_event_via_manager(self, bridge): + manager = MagicMock() + manager.broadcast_event_from_notification = AsyncMock() + # No broadcaster loop: the coroutine is driven on a private loop + manager.loop = None + bridge._websocket_manager = manager + bridge._process_single_event({"event_type": "a.b", "payload": {"x": 1}}) + manager.broadcast_event_from_notification.assert_awaited_once_with( + "a.b", {"x": 1} + ) + + def test_coroutine_error_is_swallowed(self, bridge, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + manager = MagicMock() + manager.broadcast_event_from_notification = AsyncMock( + side_effect=ValueError("boom") + ) + manager.loop = None + bridge._websocket_manager = manager + bridge._process_single_event({"event_type": "a.b", "payload": {}}) + assert "Error processing event via bridge: boom" in caplog.text + + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_worker_thread_event_reaches_broadcaster_loop(self, bridge): + """End-to-end: an event handed to ``_process_single_event`` in a + worker thread is marshalled onto the loop the broadcaster of a real + ``WebSocketManager`` runs on and delivered to a connected client.""" + manager = WebSocketManager() + websocket = MagicMock() + websocket.accept = AsyncMock() + delivered = asyncio.Event() + websocket.send_text = AsyncMock(side_effect=lambda message: delivered.set()) + await manager.connect(websocket) + assert manager.loop is asyncio.get_running_loop() + bridge._websocket_manager = manager + try: + await asyncio.to_thread( + bridge._process_single_event, + {"event_type": "baremetal.node.power_set", "payload": {"x": 1}}, + ) + await asyncio.wait_for(delivered.wait(), timeout=5.0) + finally: + manager._broadcaster_task.cancel() + try: + await manager._broadcaster_task + except asyncio.CancelledError: + pass + message = json.loads(websocket.send_text.await_args.args[0]) + assert message["event_type"] == "baremetal.node.power_set" + assert message["data"]["x"] == 1 + assert message["data"]["service_type"] == "baremetal" + + +class TestProcessEvents: + @pytest.mark.timeout(10) + def test_processes_queued_event_and_marks_task_done(self, bridge, mocker): + event_data = {"event_type": "a.b", "payload": {}} + bridge._event_queue.put(event_data) + + def process(data): + bridge._shutdown_event.set() + + process_mock = mocker.patch.object( + bridge, "_process_single_event", side_effect=process + ) + bridge._process_events() + process_mock.assert_called_once_with(event_data) + assert bridge._event_queue.unfinished_tasks == 0 + + @pytest.mark.timeout(10) + def test_exits_immediately_when_shutdown_is_preset(self, bridge, mocker, caplog): + caplog.set_level(logging.INFO, logger="osism.event_bridge") + bridge._shutdown_event.set() + get_mock = mocker.patch.object(bridge._event_queue, "get") + bridge._process_events() + get_mock.assert_not_called() + assert "Event bridge processor stopped" in caplog.text + + @pytest.mark.timeout(10) + def test_empty_queue_continues_until_shutdown(self, bridge, mocker): + def get(timeout=None): + bridge._shutdown_event.set() + raise queue.Empty + + get_mock = mocker.patch.object(bridge._event_queue, "get", side_effect=get) + process_mock = mocker.patch.object(bridge, "_process_single_event") + bridge._process_events() + get_mock.assert_called_once_with(timeout=1.0) + process_mock.assert_not_called() + + @pytest.mark.timeout(10) + def test_processing_error_logs_and_continues(self, bridge, mocker, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + first = {"event_type": "a.b", "payload": {}} + second = {"event_type": "c.d", "payload": {}} + bridge._event_queue.put(first) + bridge._event_queue.put(second) + processed = [] + + def process(data): + processed.append(data) + if len(processed) == 1: + raise ValueError("boom") + bridge._shutdown_event.set() + + mocker.patch.object(bridge, "_process_single_event", side_effect=process) + bridge._process_events() + assert processed == [first, second] + assert "Unexpected error in event bridge processor: boom" in caplog.text + + +class TestShutdown: + def test_sets_shutdown_event_and_closes_subscriber(self, bridge): + subscriber = MagicMock() + bridge._redis_subscriber = subscriber + bridge.shutdown() + assert bridge._shutdown_event.is_set() + subscriber.close.assert_called_once_with() + + def test_close_error_is_logged_and_shutdown_continues(self, bridge, caplog): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + subscriber = MagicMock() + subscriber.close.side_effect = ConnectionError("boom") + bridge._redis_subscriber = subscriber + processor = MagicMock() + processor.is_alive.return_value = True + bridge._processor_thread = processor + bridge.shutdown() + assert "Error closing Redis subscriber: boom" in caplog.text + processor.join.assert_called_once_with(timeout=5.0) + + def test_joins_alive_threads(self, bridge): + processor = MagicMock() + processor.is_alive.return_value = True + subscriber_thread = MagicMock() + subscriber_thread.is_alive.return_value = True + bridge._processor_thread = processor + bridge._subscriber_thread = subscriber_thread + bridge.shutdown() + processor.join.assert_called_once_with(timeout=5.0) + subscriber_thread.join.assert_called_once_with(timeout=5.0) + + def test_skips_dead_or_missing_threads(self, bridge): + processor = MagicMock() + processor.is_alive.return_value = False + bridge._processor_thread = processor + bridge._subscriber_thread = None + bridge.shutdown() + processor.join.assert_not_called() diff --git a/tests/unit/services/test_websocket_manager.py b/tests/unit/services/test_websocket_manager.py new file mode 100644 index 000000000..56551f625 --- /dev/null +++ b/tests/unit/services/test_websocket_manager.py @@ -0,0 +1,593 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for :mod:`osism.services.websocket_manager`. + +Covers the ``EventMessage`` value object, the per-connection filter logic in +``WebSocketConnection.matches_filters``, and the ``WebSocketManager`` +connection registry, event queue and broadcaster loop. + +Every test creates a fresh ``WebSocketManager`` instead of using the +module-level ``websocket_manager`` singleton so queues and locks bind to the +event loop of the running test and no state leaks between tests. Fake +websockets are plain ``MagicMock`` objects with ``AsyncMock`` methods; the +``WebSocket`` instance is only used as a dictionary key, so no spec is +required. + +The modules under test log via the stdlib ``logging`` module +(``osism.websocket``), so the plain ``caplog`` fixture is used for log +assertions. +""" + +import asyncio +import json +import logging +import uuid +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import WebSocketDisconnect + +from osism.services.websocket_manager import ( + EventMessage, + WebSocketConnection, + WebSocketManager, +) + + +def make_websocket(): + """Return a fake websocket usable as a connection key.""" + websocket = MagicMock() + websocket.accept = AsyncMock() + websocket.send_text = AsyncMock() + return websocket + + +async def wait_until(predicate, *, timeout=1.0): + """Drive the event loop until ``predicate()`` is truthy or ``timeout`` elapses. + + Returns as soon as the condition holds, so the broadcaster tests assert on + real state (a message was sent, a connection was dropped, the queue + drained) instead of sleeping for a fixed, timing-sensitive duration. + """ + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while not predicate(): + if loop.time() >= deadline: + raise AssertionError("condition not met within timeout") + await asyncio.sleep(0) + + +class TestEventMessage: + def test_constructor_sets_attributes(self): + event = EventMessage( + event_type="baremetal.node.power_set", + source="openstack", + data={"key": "value"}, + node_name="node1", + ) + assert event.event_type == "baremetal.node.power_set" + assert event.source == "openstack" + assert event.data == {"key": "value"} + assert event.node_name == "node1" + + def test_node_name_defaults_to_none(self): + event = EventMessage(event_type="a.b", source="test", data={}) + assert event.node_name is None + + def test_id_is_uuid4_and_unique(self): + first = EventMessage(event_type="a.b", source="test", data={}) + second = EventMessage(event_type="a.b", source="test", data={}) + assert uuid.UUID(first.id).version == 4 + assert uuid.UUID(second.id).version == 4 + assert first.id != second.id + + def test_timestamp_is_iso8601_with_z_suffix(self): + event = EventMessage(event_type="a.b", source="test", data={}) + assert event.timestamp.endswith("Z") + # Must not raise; the exact value depends on datetime.utcnow(). + datetime.fromisoformat(event.timestamp[:-1]) + + def test_to_dict_returns_exactly_the_expected_keys(self): + event = EventMessage( + event_type="a.b", source="test", data={"x": 1}, node_name="node1" + ) + result = event.to_dict() + assert set(result) == { + "id", + "timestamp", + "event_type", + "source", + "node_name", + "data", + } + assert result["id"] == event.id + assert result["timestamp"] == event.timestamp + assert result["event_type"] == "a.b" + assert result["source"] == "test" + assert result["node_name"] == "node1" + assert result["data"] == {"x": 1} + + def test_to_json_round_trips_to_dict(self): + event = EventMessage( + event_type="a.b", source="test", data={"x": 1}, node_name="node1" + ) + assert json.loads(event.to_json()) == event.to_dict() + + +class TestMatchesFilters: + def make_event(self, event_type="a.b", node_name=None): + return EventMessage( + event_type=event_type, source="test", data={}, node_name=node_name + ) + + def test_no_filters_pass_all_events(self): + connection = WebSocketConnection(MagicMock()) + assert connection.matches_filters(self.make_event()) is True + + def test_event_filter_matching_event_type(self): + connection = WebSocketConnection(MagicMock()) + connection.event_filters = ["baremetal.node.power_set"] + assert ( + connection.matches_filters( + self.make_event(event_type="baremetal.node.power_set") + ) + is True + ) + + def test_event_filter_non_matching_event_type(self): + connection = WebSocketConnection(MagicMock()) + connection.event_filters = ["baremetal.node.power_set"] + assert ( + connection.matches_filters(self.make_event(event_type="other.event")) + is False + ) + + def test_node_filter_matching_node(self): + connection = WebSocketConnection(MagicMock()) + connection.node_filters = ["node1"] + assert connection.matches_filters(self.make_event(node_name="node1")) is True + + def test_node_filter_non_matching_node(self): + connection = WebSocketConnection(MagicMock()) + connection.node_filters = ["node1"] + assert connection.matches_filters(self.make_event(node_name="other")) is False + + def test_node_filter_rejects_event_without_node_name(self): + connection = WebSocketConnection(MagicMock()) + connection.node_filters = ["node1"] + assert connection.matches_filters(self.make_event(node_name=None)) is False + + def test_service_filter_matches_first_dot_segment(self): + connection = WebSocketConnection(MagicMock()) + connection.service_filters = ["baremetal"] + assert ( + connection.matches_filters( + self.make_event(event_type="baremetal.node.power_set") + ) + is True + ) + + def test_service_filter_non_matching_service(self): + connection = WebSocketConnection(MagicMock()) + connection.service_filters = ["baremetal"] + assert ( + connection.matches_filters( + self.make_event(event_type="compute.instance.update") + ) + is False + ) + + def test_empty_event_type_maps_to_unknown_service(self): + connection = WebSocketConnection(MagicMock()) + connection.service_filters = ["unknown"] + assert connection.matches_filters(self.make_event(event_type="")) is True + + def test_combined_filters_are_anded(self): + connection = WebSocketConnection(MagicMock()) + connection.event_filters = ["a.b"] + connection.node_filters = ["node1"] + event = self.make_event(event_type="a.b", node_name="other") + assert connection.matches_filters(event) is False + + +class TestConnectDisconnect: + @pytest.mark.asyncio + async def test_connect_accepts_and_registers_connection(self): + manager = WebSocketManager() + manager._broadcast_events = MagicMock(return_value="broadcast-coro") + websocket = make_websocket() + with patch( + "osism.services.websocket_manager.asyncio.create_task" + ) as create_task: + create_task.return_value = MagicMock(done=MagicMock(return_value=False)) + await manager.connect(websocket) + websocket.accept.assert_awaited_once() + assert isinstance(manager.connections[websocket], WebSocketConnection) + + @pytest.mark.asyncio + async def test_connect_starts_broadcaster_only_once(self): + manager = WebSocketManager() + manager._broadcast_events = MagicMock(return_value="broadcast-coro") + with patch( + "osism.services.websocket_manager.asyncio.create_task" + ) as create_task: + task = MagicMock() + task.done.return_value = False + create_task.return_value = task + await manager.connect(make_websocket()) + await manager.connect(make_websocket()) + create_task.assert_called_once_with("broadcast-coro") + + @pytest.mark.asyncio + async def test_connect_restarts_broadcaster_when_task_is_done(self): + manager = WebSocketManager() + manager._broadcast_events = MagicMock(return_value="broadcast-coro") + with patch( + "osism.services.websocket_manager.asyncio.create_task" + ) as create_task: + task = MagicMock() + task.done.return_value = True + create_task.return_value = task + await manager.connect(make_websocket()) + await manager.connect(make_websocket()) + assert create_task.call_count == 2 + + @pytest.mark.asyncio + async def test_disconnect_removes_connection(self): + manager = WebSocketManager() + websocket = make_websocket() + manager.connections[websocket] = WebSocketConnection(websocket) + await manager.disconnect(websocket) + assert websocket not in manager.connections + + @pytest.mark.asyncio + async def test_disconnect_unknown_websocket_is_a_noop(self): + manager = WebSocketManager() + await manager.disconnect(make_websocket()) + assert manager.connections == {} + + +class TestUpdateFilters: + @pytest.mark.asyncio + async def test_update_filters_sets_all_filter_lists(self): + manager = WebSocketManager() + websocket = make_websocket() + manager.connections[websocket] = WebSocketConnection(websocket) + await manager.update_filters( + websocket, + event_filters=["a.b"], + node_filters=["node1"], + service_filters=["baremetal"], + ) + connection = manager.connections[websocket] + assert connection.event_filters == ["a.b"] + assert connection.node_filters == ["node1"] + assert connection.service_filters == ["baremetal"] + + @pytest.mark.asyncio + async def test_update_filters_partial_update_keeps_other_lists(self): + manager = WebSocketManager() + websocket = make_websocket() + connection = WebSocketConnection(websocket) + connection.event_filters = ["a.b"] + connection.node_filters = ["node1"] + connection.service_filters = ["baremetal"] + manager.connections[websocket] = connection + await manager.update_filters(websocket, node_filters=["node2"]) + assert connection.event_filters == ["a.b"] + assert connection.node_filters == ["node2"] + assert connection.service_filters == ["baremetal"] + + @pytest.mark.asyncio + async def test_update_filters_empty_list_clears_a_filter(self): + manager = WebSocketManager() + websocket = make_websocket() + connection = WebSocketConnection(websocket) + connection.event_filters = ["a.b"] + manager.connections[websocket] = connection + await manager.update_filters(websocket, event_filters=[]) + assert connection.event_filters == [] + + @pytest.mark.asyncio + async def test_update_filters_unknown_websocket_is_a_noop(self): + manager = WebSocketManager() + await manager.update_filters(make_websocket(), event_filters=["a.b"]) + assert manager.connections == {} + + +class TestAddEventAndHeartbeat: + @pytest.mark.asyncio + async def test_add_event_puts_event_on_queue(self): + manager = WebSocketManager() + event = EventMessage(event_type="a.b", source="test", data={}) + await manager.add_event(event) + assert manager.event_queue.qsize() == 1 + assert manager.event_queue.get_nowait() is event + + @pytest.mark.asyncio + async def test_send_heartbeat_without_connections_queues_nothing(self): + manager = WebSocketManager() + await manager.send_heartbeat() + assert manager.event_queue.qsize() == 0 + + @pytest.mark.asyncio + async def test_send_heartbeat_queues_heartbeat_event(self): + manager = WebSocketManager() + websocket = make_websocket() + manager.connections[websocket] = WebSocketConnection(websocket) + await manager.send_heartbeat() + event = manager.event_queue.get_nowait() + assert event.event_type == "heartbeat" + assert event.source == "osism" + assert event.data == {"message": "ping"} + + +class TestBroadcastEventFromNotification: + @pytest.mark.asyncio + @pytest.mark.parametrize( + "event_type,payload,expected_node_name,expected_resource_id", + [ + pytest.param( + "baremetal.node.power_set.end", + {"ironic_object.data": {"name": "node1", "uuid": "abc-123"}}, + "node1", + "abc-123", + id="baremetal", + ), + pytest.param( + "compute.instance.update", + { + "nova_object.data": { + "host": "compute-1", + "name": "instance-a", + "uuid": "u-1", + } + }, + "compute-1", + "u-1", + id="compute-prefers-host", + ), + pytest.param( + "nova.instance.update", + {"nova_object.data": {"name": "instance-a", "uuid": "u-1"}}, + "instance-a", + "u-1", + id="nova-falls-back-to-name", + ), + pytest.param( + "network.floatingip.update.end", + {"neutron_object.data": {"id": "net-1", "name": "public"}}, + "public", + "net-1", + id="network", + ), + pytest.param( + "neutron.port.create.end", + {"neutron_object.data": {"uuid": "net-2", "device_id": "dev-1"}}, + "dev-1", + "net-2", + id="neutron-fallbacks", + ), + pytest.param( + "volume.volume.create.end", + {"cinder_object.data": {"id": "vol-1", "name": "volume-a"}}, + "volume-a", + "vol-1", + id="volume", + ), + pytest.param( + "volume.volume.update.end", + {"cinder_object.data": {"uuid": "vol-2", "display_name": "disp-a"}}, + "disp-a", + "vol-2", + id="volume-fallbacks", + ), + pytest.param( + "image.image.upload.end", + {"glance_object.data": {"id": "img-1", "name": "cirros"}}, + "cirros", + "img-1", + id="image", + ), + pytest.param( + "image.image.update.end", + {"glance_object.data": {"uuid": "img-2"}}, + None, + "img-2", + id="image-uuid-fallback", + ), + pytest.param( + "identity.project.created", + {"keystone_object.data": {"id": "proj-1", "name": "admin"}}, + "admin", + "proj-1", + id="identity", + ), + pytest.param( + "identity.user.updated", + {"keystone_object.data": {"uuid": "user-2"}}, + None, + "user-2", + id="identity-uuid-fallback", + ), + ], + ) + async def test_extracts_identifiers_per_service_type( + self, event_type, payload, expected_node_name, expected_resource_id + ): + manager = WebSocketManager() + await manager.broadcast_event_from_notification(event_type, payload) + event = manager.event_queue.get_nowait() + assert event.event_type == event_type + assert event.source == "openstack" + assert event.node_name == expected_node_name + assert event.data["resource_id"] == expected_resource_id + assert event.data["service_type"] == event_type.split(".")[0] + + @pytest.mark.asyncio + async def test_known_service_without_expected_payload_key(self): + manager = WebSocketManager() + await manager.broadcast_event_from_notification( + "baremetal.node.power_set.end", {"foo": "bar"} + ) + event = manager.event_queue.get_nowait() + assert event.node_name is None + assert event.data["resource_id"] is None + assert event.data["service_type"] == "baremetal" + + @pytest.mark.asyncio + async def test_unknown_service_type_queues_event_without_extraction(self): + manager = WebSocketManager() + await manager.broadcast_event_from_notification("foo.bar", {"x": 1}) + event = manager.event_queue.get_nowait() + assert event.node_name is None + assert event.data["service_type"] == "foo" + assert event.data["resource_id"] is None + assert event.data["x"] == 1 + + @pytest.mark.asyncio + async def test_empty_event_type_maps_to_unknown_service(self): + manager = WebSocketManager() + await manager.broadcast_event_from_notification("", {}) + event = manager.event_queue.get_nowait() + assert event.data["service_type"] == "unknown" + + @pytest.mark.asyncio + async def test_payload_is_copied_and_not_mutated(self): + manager = WebSocketManager() + payload = {"ironic_object.data": {"name": "node1", "uuid": "abc-123"}} + await manager.broadcast_event_from_notification( + "baremetal.node.power_set.end", payload + ) + assert payload == {"ironic_object.data": {"name": "node1", "uuid": "abc-123"}} + event = manager.event_queue.get_nowait() + assert event.data["ironic_object.data"] == payload["ironic_object.data"] + assert event.data["service_type"] == "baremetal" + assert event.data["resource_id"] == "abc-123" + + @pytest.mark.asyncio + async def test_error_is_caught_and_nothing_is_queued(self, caplog): + caplog.set_level(logging.ERROR, logger="osism.websocket") + manager = WebSocketManager() + # A list payload survives the "key in payload" checks and .copy(), + # but fails on the string-keyed item assignment afterwards. + await manager.broadcast_event_from_notification("baremetal.node.power_set", []) + assert manager.event_queue.qsize() == 0 + assert "Error creating event from notification" in caplog.text + + +class TestBroadcastEvents: + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_matching_event_is_sent_to_connection(self): + manager = WebSocketManager() + websocket = make_websocket() + manager.connections[websocket] = WebSocketConnection(websocket) + event = EventMessage(event_type="a.b", source="test", data={"x": 1}) + manager.event_queue.put_nowait(event) + + task = asyncio.create_task(manager._broadcast_events()) + try: + await wait_until(lambda: websocket.send_text.await_count == 1) + websocket.send_text.assert_awaited_once_with(event.to_json()) + finally: + task.cancel() + await task + + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_non_matching_connection_is_skipped(self): + manager = WebSocketManager() + filtered_websocket = make_websocket() + filtered_connection = WebSocketConnection(filtered_websocket) + filtered_connection.event_filters = ["other.event"] + manager.connections[filtered_websocket] = filtered_connection + open_websocket = make_websocket() + manager.connections[open_websocket] = WebSocketConnection(open_websocket) + event = EventMessage(event_type="a.b", source="test", data={}) + manager.event_queue.put_nowait(event) + + task = asyncio.create_task(manager._broadcast_events()) + try: + # The filtered connection is registered first, so it is always + # evaluated before the open one sends and satisfies this wait. + await wait_until(lambda: open_websocket.send_text.await_count == 1) + filtered_websocket.send_text.assert_not_awaited() + open_websocket.send_text.assert_awaited_once_with(event.to_json()) + finally: + task.cancel() + await task + + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_websocket_disconnect_removes_connection(self): + manager = WebSocketManager() + websocket = make_websocket() + websocket.send_text.side_effect = WebSocketDisconnect() + manager.connections[websocket] = WebSocketConnection(websocket) + manager.event_queue.put_nowait( + EventMessage(event_type="a.b", source="test", data={}) + ) + + task = asyncio.create_task(manager._broadcast_events()) + try: + await wait_until(lambda: websocket not in manager.connections) + finally: + task.cancel() + await task + + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_generic_send_error_logs_and_removes_connection(self, caplog): + caplog.set_level(logging.ERROR, logger="osism.websocket") + manager = WebSocketManager() + websocket = make_websocket() + websocket.send_text.side_effect = RuntimeError("boom") + manager.connections[websocket] = WebSocketConnection(websocket) + manager.event_queue.put_nowait( + EventMessage(event_type="a.b", source="test", data={}) + ) + + task = asyncio.create_task(manager._broadcast_events()) + try: + await wait_until(lambda: websocket not in manager.connections) + assert "Error sending message to WebSocket" in caplog.text + finally: + task.cancel() + await task + + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_event_without_connections_is_consumed(self): + manager = WebSocketManager() + manager.event_queue.put_nowait( + EventMessage(event_type="a.b", source="test", data={}) + ) + + task = asyncio.create_task(manager._broadcast_events()) + try: + await wait_until(lambda: manager.event_queue.qsize() == 0) + assert not task.done() + finally: + task.cancel() + await task + + @pytest.mark.asyncio + @pytest.mark.timeout(10) + async def test_cancellation_stops_the_loop_cleanly(self, caplog): + caplog.set_level(logging.INFO, logger="osism.websocket") + manager = WebSocketManager() + + task = asyncio.create_task(manager._broadcast_events()) + # Let the broadcaster start and park on the queue.get() await before + # cancelling. Cancelling a task that has not run yet delivers the + # CancelledError at the coroutine's entry, bypassing the loop's + # try/except; only once it is suspended inside the loop does the + # except catch it and exit via break, so awaiting the task must not + # raise. + await wait_until(lambda: "Starting WebSocket event broadcaster" in caplog.text) + task.cancel() + await task + assert task.done() + assert "WebSocket broadcaster task cancelled" in caplog.text