From 27dcc4376a9a2297f595ee92ca80b6776a2dd84d Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Mon, 13 Jul 2026 22:12:50 +0200 Subject: [PATCH 1/4] Add unit tests for osism/services websocket_manager and event_bridge Create the tests/unit/services package with test modules for osism/services/websocket_manager.py and osism/services/event_bridge.py, which previously had no unit test coverage. test_websocket_manager.py covers the EventMessage value object, the per-connection filter logic in WebSocketConnection.matches_filters, the WebSocketManager connection registry (connect/disconnect/update_filters), the event queue helpers (add_event/send_heartbeat), the per-service identifier extraction in broadcast_event_from_notification and the _broadcast_events broadcaster loop including disconnect cleanup and cancellation handling. test_event_bridge.py covers Redis initialization including environment variable overrides and connection failures, the publish path in add_event with reconnect and local-queue fallback, thread startup guards in set_websocket_manager, the _redis_subscriber_loop including resubscribe and retry exhaustion, _process_single_event, the _process_events loop and shutdown. The thread loops are called synchronously and terminated via _shutdown_event-driven side effects, so no real threads, sockets or sleeps are involved. Most WebSocketManager methods are coroutines, so pytest-asyncio is added to the Pipfile dev packages (with asyncio_default_fixture_loop_scope set in setup.cfg); async tests are marked with pytest.mark.asyncio explicitly to keep --strict-markers happy. Loop-driving tests carry pytest-timeout markers so a regression cannot hang CI. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- Pipfile | 1 + Pipfile.lock | 19 +- setup.cfg | 1 + tests/unit/services/test_event_bridge.py | 511 +++++++++++++++ tests/unit/services/test_websocket_manager.py | 593 ++++++++++++++++++ 5 files changed, 1124 insertions(+), 1 deletion(-) create mode 100644 tests/unit/services/test_event_bridge.py create mode 100644 tests/unit/services/test_websocket_manager.py 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/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..fbf19c176 --- /dev/null +++ b/tests/unit/services/test_event_bridge.py @@ -0,0 +1,511 @@ +# 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. Note that a failing ``get_message`` only breaks the inner loop and +triggers a resubscribe; the retry counter and ``_shutdown_event.wait`` +back-off only engage when ``subscribe()`` itself raises. + +The module logs via the stdlib ``logging`` module (``osism.event_bridge``), +so the plain ``caplog`` fixture is used for log assertions. +""" + +import json +import logging +import queue +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from osism.services.event_bridge import EventBridge + + +@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_triggers_resubscribe(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: + raise ConnectionError("lost") + bridge._shutdown_event.set() + return None + + subscriber.get_message.side_effect = get_message + bridge._redis_subscriber_loop() + assert subscriber.subscribe.call_count == 2 + assert subscriber.close.call_count == 2 + assert "Error getting Redis message: lost" in caplog.text + + @pytest.mark.timeout(10) + def test_subscribe_error_waits_and_reinitializes_redis( + self, bridge, mocker, caplog + ): + caplog.set_level(logging.INFO, logger="osism.event_bridge") + subscriber = MagicMock() + subscriber.subscribe.side_effect = ConnectionError("down") + bridge._redis_subscriber = subscriber + init_redis = mocker.patch.object(bridge, "_init_redis") + + def wait(timeout=None): + bridge._shutdown_event.set() + return True + + wait_mock = mocker.patch.object( + bridge._shutdown_event, "wait", side_effect=wait + ) + bridge._redis_subscriber_loop() + wait_mock.assert_called_once_with(5) + init_redis.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): + caplog.set_level(logging.ERROR, logger="osism.event_bridge") + subscriber = MagicMock() + subscriber.subscribe.side_effect = ConnectionError("down") + bridge._redis_subscriber = subscriber + init_redis = mocker.patch.object(bridge, "_init_redis") + wait_mock = mocker.patch.object( + bridge._shutdown_event, "wait", return_value=False + ) + bridge._redis_subscriber_loop() + assert subscriber.subscribe.call_count == 5 + # 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() + 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") + ) + bridge._websocket_manager = manager + bridge._process_single_event({"event_type": "a.b", "payload": {}}) + assert "Error processing event via bridge: boom" in caplog.text + + +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 From c48a233480ab1707823e5cdc72efb67f7d99b2b5 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Wed, 22 Jul 2026 13:14:49 +0200 Subject: [PATCH 2/4] Close failed Redis subscriber before reconnect creates a new one When subscribe() raised in _redis_subscriber_loop, the except handler called _init_redis(), which reassigned self._redis_subscriber to a fresh pubsub object without closing the failed one. The per-iteration finally block then closed self._redis_subscriber - by that time the fresh replacement - so the failed subscriber leaked and the next iteration subscribed on an already closed object. Introduce _close_subscriber(), close the failed subscriber in the except handler before _init_redis() replaces it, and replace the per-iteration finally cleanup with a single close when the loop exits, so the freshly created subscriber is never closed prematurely. The reconnect tests now install successive distinct subscriber mocks via the _init_redis side effect and assert that each retry subscribes on the newly created instance and that every failed subscriber is closed exactly once. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- osism/services/event_bridge.py | 20 ++++++--- tests/unit/services/test_event_bridge.py | 54 ++++++++++++++++++------ 2 files changed, 55 insertions(+), 19 deletions(-) diff --git a/osism/services/event_bridge.py b/osism/services/event_bridge.py index a0c051856..78a5268ea 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 @@ -210,6 +219,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 +235,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") diff --git a/tests/unit/services/test_event_bridge.py b/tests/unit/services/test_event_bridge.py index fbf19c176..ac5fc8c45 100644 --- a/tests/unit/services/test_event_bridge.py +++ b/tests/unit/services/test_event_bridge.py @@ -329,44 +329,72 @@ def get_message(timeout=None): subscriber.get_message.side_effect = get_message bridge._redis_subscriber_loop() assert subscriber.subscribe.call_count == 2 - assert subscriber.close.call_count == 2 + # The subscriber stays open across the resubscribe and is only + # closed once when the loop exits. + assert subscriber.close.call_count == 1 assert "Error getting Redis message: 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") - subscriber = MagicMock() - subscriber.subscribe.side_effect = ConnectionError("down") - bridge._redis_subscriber = subscriber - init_redis = mocker.patch.object(bridge, "_init_redis") + failed = MagicMock() + failed.subscribe.side_effect = ConnectionError("down") + bridge._redis_subscriber = failed + fresh = MagicMock() - def wait(timeout=None): + def get_message(timeout=None): bridge._shutdown_event.set() - return True + 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", side_effect=wait + 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") - subscriber = MagicMock() - subscriber.subscribe.side_effect = ConnectionError("down") - bridge._redis_subscriber = subscriber - init_redis = mocker.patch.object(bridge, "_init_redis") + 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() - assert subscriber.subscribe.call_count == 5 + 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 From 569aa4e4b714a85dd8c0d4592dd9104dca62a733 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Wed, 22 Jul 2026 13:15:18 +0200 Subject: [PATCH 3/4] Route get_message failures through subscriber back-off A failing get_message() broke the inner loop of _redis_subscriber_loop and immediately resubscribed: the per-iteration cleanup closed the subscriber without recreating it, so the next iteration subscribed on a closed object, and the path bypassed the retry counter and _shutdown_event.wait() back-off entirely, so a persistently failing get_message() busy-looped forever. Re-raise the error instead so it runs through the same bounded back-off as a failing subscribe(): the failed subscriber is closed, retry_count is incremented and capped by max_retries, the loop waits retry_delay seconds, and _init_redis() provides a fresh subscriber before the resubscribe. The resubscribe test now installs a distinct fresh subscriber via the _init_redis side effect and asserts the back-off wait, the bounded retry accounting and that the resubscribe happens on the new instance. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- osism/services/event_bridge.py | 4 ++- tests/unit/services/test_event_bridge.py | 45 ++++++++++++++++-------- 2 files changed, 33 insertions(+), 16 deletions(-) diff --git a/osism/services/event_bridge.py b/osism/services/event_bridge.py index 78a5268ea..cf933c6fb 100644 --- a/osism/services/event_bridge.py +++ b/osism/services/event_bridge.py @@ -211,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 diff --git a/tests/unit/services/test_event_bridge.py b/tests/unit/services/test_event_bridge.py index ac5fc8c45..0b64e3359 100644 --- a/tests/unit/services/test_event_bridge.py +++ b/tests/unit/services/test_event_bridge.py @@ -14,9 +14,9 @@ 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. Note that a failing ``get_message`` only breaks the inner loop and -triggers a resubscribe; the retry counter and ``_shutdown_event.wait`` -back-off only engage when ``subscribe()`` itself raises. +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. @@ -313,26 +313,41 @@ def get_message(timeout=None): assert "Error processing Redis event: boom" in caplog.text @pytest.mark.timeout(10) - def test_get_message_error_triggers_resubscribe(self, bridge, caplog): + 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") - subscriber = MagicMock() - bridge._redis_subscriber = subscriber - calls = {"count": 0} + failed = MagicMock() + failed.get_message.side_effect = ConnectionError("lost") + bridge._redis_subscriber = failed + fresh = MagicMock() def get_message(timeout=None): - calls["count"] += 1 - if calls["count"] == 1: - raise ConnectionError("lost") bridge._shutdown_event.set() return None - subscriber.get_message.side_effect = get_message + 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() - assert subscriber.subscribe.call_count == 2 - # The subscriber stays open across the resubscribe and is only - # closed once when the loop exits. - assert subscriber.close.call_count == 1 + 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( From 46f6c45b4c8d1fe95e6aa39c0b87bd4ba40e9c27 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Wed, 22 Jul 2026 13:23:53 +0200 Subject: [PATCH 4/4] Marshal bridged events onto the broadcaster event loop _process_single_event() runs in the event bridge's worker threads and drove broadcast_event_from_notification() on a freshly created event loop per call. The coroutine awaits event_queue.put() on the manager's asyncio.Queue, whose broadcaster waiters live on the API loop; waking them from a worker thread is not thread-safe - it raises under asyncio debug mode and in production mutates the API loop's ready queue without waking it, so delivery was unreliable and the error was swallowed. Capture the running loop in WebSocketManager.connect() when the broadcaster task is started and expose it as WebSocketManager.loop. The event bridge now submits the coroutine to that loop with asyncio.run_coroutine_threadsafe() and waits for the result. The private-loop path remains only as a fallback for when no broadcaster loop exists yet; the queue has no waiters then, so it is safe. Add an integration-style test that runs a real WebSocketManager with an active broadcaster and calls _process_single_event() from a worker thread, asserting the event is delivered to a connected client. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- osism/services/event_bridge.py | 29 ++++++++++++------- osism/services/websocket_manager.py | 4 +++ tests/unit/services/test_event_bridge.py | 36 ++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 10 deletions(-) diff --git a/osism/services/event_bridge.py b/osism/services/event_bridge.py index cf933c6fb..bd7bdf478 100644 --- a/osism/services/event_bridge.py +++ b/osism/services/event_bridge.py @@ -253,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/tests/unit/services/test_event_bridge.py b/tests/unit/services/test_event_bridge.py index 0b64e3359..a05bef12e 100644 --- a/tests/unit/services/test_event_bridge.py +++ b/tests/unit/services/test_event_bridge.py @@ -22,6 +22,7 @@ so the plain ``caplog`` fixture is used for log assertions. """ +import asyncio import json import logging import queue @@ -30,6 +31,7 @@ import pytest from osism.services.event_bridge import EventBridge +from osism.services.websocket_manager import WebSocketManager @pytest.fixture @@ -439,6 +441,8 @@ def test_without_manager_warns_and_returns(self, bridge, caplog): 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( @@ -451,10 +455,42 @@ def test_coroutine_error_is_swallowed(self, bridge, caplog): 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)