learningcircuit--local-deep-research/tests/web/services/test_socket_service_concurrency.py

"""
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