""" Tests for SocketIOService concurrency and edge cases. Covers: emit_to_subscribers snapshot safety, logging restoration, empty subscriptions, mixed success/failure, remove_subscriptions, and emit_socket_event modes. Source: socket_service.py lines 129-216. """ from unittest.mock import patch, Mock class MockSocketIO: """Mock SocketIO that records emitted events.""" def __init__(self, app=None, **kwargs): self.emitted_events = [] self._handlers = {} def emit(self, event, data, room=None): self.emitted_events.append({"event": event, "data": data, "room": room}) def on(self, event): def decorator(f): self._handlers[event] = f return f return decorator @property def on_error(self): def decorator(f): self._handlers["error"] = f return f return decorator @property def on_error_default(self): def decorator(f): self._handlers["error_default"] = f return f return decorator def run(self, app, **kwargs): pass def _make_service(): """Create a fresh SocketIOService with mocked dependencies.""" from local_deep_research.web.services.socket_service import SocketIOService # Reset singleton SocketIOService._instance = None try: mock_app = Mock() mock_app.config = {} mock_app.debug = False with ( patch( "local_deep_research.web.services.socket_service.SocketIO", MockSocketIO, ), patch( "local_deep_research.settings.env_registry.get_env_setting", return_value=None, ), ): service = SocketIOService(app=mock_app) return service finally: # Restore singleton after each test via cleanup pass def _cleanup_singleton(original): from local_deep_research.web.services.socket_service import SocketIOService SocketIOService._instance = original class TestEmitToSubscribersConcurrency: """Tests for emit_to_subscribers concurrency edge cases.""" def setup_method(self): from local_deep_research.web.services.socket_service import ( SocketIOService, ) self._original = SocketIOService._instance def teardown_method(self): _cleanup_singleton(self._original) def test_snapshot_prevents_runtime_error_during_iteration(self): """Modifying subscriptions during emit doesn't cause RuntimeError.""" service = _make_service() research_id = "r1" # Add subscriptions with service._SocketIOService__lock: service._SocketIOService__socket_subscriptions[research_id] = { "sid1", "sid2", "sid3", } # Mock emit to modify subscriptions mid-iteration original_emit = service._SocketIOService__socketio.emit call_count = 0 def emit_and_modify(event, data, room=None): nonlocal call_count call_count += 1 if call_count == 1: # Modify the original set during iteration with service._SocketIOService__lock: service._SocketIOService__socket_subscriptions[ research_id ].add("sid4") service._SocketIOService__socket_subscriptions[ research_id ].discard("sid3") original_emit(event, data, room=room) service._SocketIOService__socketio.emit = emit_and_modify # Should not raise RuntimeError result = service.emit_to_subscribers( "event", research_id, {"key": "val"} ) assert result is True def test_logging_re_enabled_after_subscriber_exception(self): """__logging_enabled is restored to True even when subscriber emit raises.""" service = _make_service() research_id = "r2" with service._SocketIOService__lock: service._SocketIOService__socket_subscriptions[research_id] = { "sid1" } # Make emit raise — caught by inner per-subscriber try/except def raise_on_emit(event, data, room=None): raise ConnectionError("socket broken") service._SocketIOService__socketio.emit = raise_on_emit result = service.emit_to_subscribers( "event", research_id, {}, enable_logging=False ) # logging should be re-enabled via finally block assert service._SocketIOService__logging_enabled is True # Inner exception is caught, outer try returns True assert result is True def test_logging_re_enabled_after_outer_exception(self): """__logging_enabled is restored even if the outer try block fails.""" service = _make_service() # Force an exception in the outer try block by making the lock # context manager's __enter__ raise after the first call original_lock = service._SocketIOService__lock class ExplodingLock: def __enter__(self): raise RuntimeError("lock exploded") def __exit__(self, *args): pass service._SocketIOService__lock = ExplodingLock() result = service.emit_to_subscribers( "event", "r_outer", {}, enable_logging=False ) # Restore lock for cleanup service._SocketIOService__lock = original_lock # logging should still be re-enabled via finally assert service._SocketIOService__logging_enabled is True assert result is False def test_empty_subscription_set_drops_event(self): """With an empty subscriber set the event is dropped, not broadcast. Cross-user broadcast on empty subscriptions was a cross-user leak — see emit_to_subscribers in socket_service.py. The catch-up snapshot in __handle_subscribe replays the latest progress on subscribe so events that arrive before any client subscribes are recovered correctly. """ service = _make_service() research_id = "r3" with service._SocketIOService__lock: service._SocketIOService__socket_subscriptions[research_id] = set() result = service.emit_to_subscribers("event", research_id, {"data": 1}) assert result is True # success: dropping is intentional, not failure events = service._SocketIOService__socketio.emitted_events assert events == [] def test_one_subscriber_fails_others_succeed(self): """If one subscriber's emit raises, others still receive the event.""" service = _make_service() research_id = "r4" with service._SocketIOService__lock: service._SocketIOService__socket_subscriptions[research_id] = { "sid1", "sid2", "sid3", } call_log = [] def selective_emit(event, data, room=None): call_log.append(room) if room == "sid2": raise RuntimeError("connection lost") service._SocketIOService__socketio.emit = selective_emit # Should still return True (outer try succeeds) result = service.emit_to_subscribers("event", research_id, {}) assert result is True # All 3 subscribers were attempted assert len(call_log) == 3 def test_no_subscription_entry_drops_event(self): """Missing subscription entry → drop, do not fall back to broadcast. The broadcast fallback was removed to close a cross-user leak. """ service = _make_service() result = service.emit_to_subscribers("event", "nonexistent", {"x": 1}) assert result is True # success: dropping is intentional events = service._SocketIOService__socketio.emitted_events assert events == [] class TestRemoveSubscriptions: """Tests for remove_subscriptions_for_research.""" def setup_method(self): from local_deep_research.web.services.socket_service import ( SocketIOService, ) self._original = SocketIOService._instance def teardown_method(self): _cleanup_singleton(self._original) def test_removes_existing_subscriptions(self): """Removes subscriptions for a known research_id.""" service = _make_service() rid = "r10" with service._SocketIOService__lock: service._SocketIOService__socket_subscriptions[rid] = {"s1", "s2"} service.remove_subscriptions_for_research(rid) with service._SocketIOService__lock: assert rid not in service._SocketIOService__socket_subscriptions def test_remove_nonexistent_no_error(self): """Removing subscriptions for unknown research_id doesn't raise.""" service = _make_service() service.remove_subscriptions_for_research("unknown_id") # Should not raise class TestEmitSocketEvent: """Tests for emit_socket_event broadcast vs room modes.""" def setup_method(self): from local_deep_research.web.services.socket_service import ( SocketIOService, ) self._original = SocketIOService._instance def teardown_method(self): _cleanup_singleton(self._original) def test_broadcast_mode_no_room(self): """Without room, socketio.emit is called without room kwarg.""" service = _make_service() result = service.emit_socket_event("test_event", {"key": "val"}) assert result is True events = service._SocketIOService__socketio.emitted_events assert len(events) == 1 assert events[0]["room"] is None def test_room_mode(self): """With room specified, socketio.emit is called with room=room.""" service = _make_service() result = service.emit_socket_event( "test_event", {"key": "val"}, room="abc123" ) assert result is True events = service._SocketIOService__socketio.emitted_events assert len(events) == 1 assert events[0]["room"] == "abc123" def test_exception_returns_false(self): """If socketio.emit raises, emit_socket_event returns False.""" service = _make_service() service._SocketIOService__socketio.emit = Mock( side_effect=RuntimeError("boom") ) result = service.emit_socket_event("test_event", {}) assert result is False