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

"""Tests for SocketIOService.__init_singleton and singleton behavior."""

from unittest.mock import patch
import pytest
from flask import Flask

from local_deep_research.web.services.socket_service import SocketIOService


@pytest.fixture(autouse=True)
def reset_singleton():
    """Reset singleton before each test."""
    SocketIOService._instance = None
    yield
    SocketIOService._instance = None


@pytest.fixture
def minimal_app():
    """Create a minimal Flask app (no create_app() which triggers SocketIOService)."""
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "test"
    app.config["TESTING"] = True
    return app


class TestSocketIOServiceSingleton:
    """Tests for singleton creation and behavior."""

    def test_requires_app_on_first_creation(self):
        """Should raise ValueError if no app provided on first init."""
        with pytest.raises(ValueError, match="Flask app must be specified"):
            SocketIOService()

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_creates_singleton_with_app(
        self, mock_env, mock_socketio, minimal_app
    ):
        mock_env.return_value = None

        service = SocketIOService(app=minimal_app)
        assert service is not None
        assert SocketIOService._instance is service

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_second_call_returns_same_instance(
        self, mock_env, mock_socketio, minimal_app
    ):
        mock_env.return_value = None

        first = SocketIOService(app=minimal_app)
        second = SocketIOService(app=minimal_app)
        assert first is second

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_second_call_without_app_returns_existing(
        self, mock_env, mock_socketio, minimal_app
    ):
        mock_env.return_value = None

        first = SocketIOService(app=minimal_app)
        second = SocketIOService()  # No app needed on second call
        assert first is second


class TestSocketIOServiceCorsConfig:
    """Tests for WebSocket CORS configuration in __init_singleton."""

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_default_cors_is_same_origin(
        self, mock_env, mock_socketio, minimal_app
    ):
        """No env var set -> secure same-origin-only default (None), per #3091."""
        mock_env.return_value = None

        SocketIOService(app=minimal_app)
        mock_socketio.assert_called_once()
        call_kwargs = mock_socketio.call_args[1]
        assert call_kwargs["cors_allowed_origins"] is None

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_wildcard_env_setting(self, mock_env, mock_socketio, minimal_app):
        mock_env.return_value = "*"

        SocketIOService(app=minimal_app)
        call_kwargs = mock_socketio.call_args[1]
        assert call_kwargs["cors_allowed_origins"] == "*"

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_specific_origins(self, mock_env, mock_socketio, minimal_app):
        mock_env.return_value = "https://a.com, https://b.com"

        SocketIOService(app=minimal_app)
        call_kwargs = mock_socketio.call_args[1]
        assert call_kwargs["cors_allowed_origins"] == [
            "https://a.com",
            "https://b.com",
        ]

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_empty_env_disables_cors(
        self, mock_env, mock_socketio, minimal_app
    ):
        """Empty string env var -> same-origin only (None)."""
        mock_env.return_value = ""

        SocketIOService(app=minimal_app)
        call_kwargs = mock_socketio.call_args[1]
        assert call_kwargs["cors_allowed_origins"] is None

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_socketio_async_mode_threading(
        self, mock_env, mock_socketio, minimal_app
    ):
        mock_env.return_value = None

        SocketIOService(app=minimal_app)
        call_kwargs = mock_socketio.call_args[1]
        assert call_kwargs["async_mode"] == "threading"

    @patch("local_deep_research.web.services.socket_service.SocketIO")
    @patch("local_deep_research.settings.env_registry.get_env_setting")
    def test_socketio_path(self, mock_env, mock_socketio, minimal_app):
        mock_env.return_value = None

        SocketIOService(app=minimal_app)
        call_kwargs = mock_socketio.call_args[1]
        assert call_kwargs["path"] == "/socket.io"


class TestOriginRejectionLogging:
    """The diagnostic hook that surfaces engine.io's silenced 'bad-origin'
    WebSocket rejections through loguru (so a misconfigured origin isn't a
    silent frozen UI)."""

    def test_logs_each_rejected_origin_once_and_ignores_other_errors(self):
        from loguru import logger
        from flask import Flask
        from flask_socketio import SocketIO
        from local_deep_research.web.services.socket_service import (
            _install_origin_rejection_logging,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test"
        socketio = SocketIO(
            app, async_mode="threading", cors_allowed_origins=None
        )

        assert _install_origin_rejection_logging(socketio) is True
        eio = socketio.server.eio

        # The package disables loguru by default (__init__.py); app startup
        # re-enables it. Enable it here so the warning emitted from inside the
        # socket_service module reaches our sink.
        logger.enable("local_deep_research")
        warnings: list[str] = []
        sink_id = logger.add(
            lambda m: warnings.append(m.record["message"]), level="WARNING"
        )
        try:
            eio._log_error_once(
                "https://evil.com is not an accepted origin.", "bad-origin"
            )
            # same origin again -> deduped (no second warning)
            eio._log_error_once(
                "https://evil.com is not an accepted origin.", "bad-origin"
            )
            # a different origin -> its own warning
            eio._log_error_once(
                "https://other.com is not an accepted origin.", "bad-origin"
            )
            # an unrelated engine.io error -> must NOT be turned into a warning
            eio._log_error_once("Invalid transport", "bad-transport")
        finally:
            logger.remove(sink_id)
            logger.disable("local_deep_research")  # restore import-time default

        origin_warnings = [
            w for w in warnings if "rejected a WebSocket handshake" in w
        ]
        assert len(origin_warnings) == 2
        assert any("evil.com" in w for w in origin_warnings)
        assert any("other.com" in w for w in origin_warnings)
        # the fix hint is present; the unrelated error never warned
        assert all(
            "LDR_SECURITY_WEBSOCKET_ALLOWED_ORIGINS" in w
            for w in origin_warnings
        )
        assert not any("Invalid transport" in w for w in warnings)

    def test_returns_false_when_engineio_internals_absent(self):
        from unittest.mock import Mock
        from local_deep_research.web.services.socket_service import (
            _install_origin_rejection_logging,
        )

        broken = Mock()
        # No .server.eio._log_error_once -> AttributeError path -> no-op.
        del broken.server
        assert _install_origin_rejection_logging(broken) is False

    def test_dedup_set_is_capped(self):
        """The per-origin dedup set is bounded: Origin is attacker-controlled at
        the pre-auth handshake, so feeding many distinct origins must not warn
        (or grow) without limit."""
        from loguru import logger
        from flask import Flask
        from flask_socketio import SocketIO
        from local_deep_research.web.services.socket_service import (
            _install_origin_rejection_logging,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test"
        socketio = SocketIO(
            app, async_mode="threading", cors_allowed_origins=None
        )
        assert _install_origin_rejection_logging(socketio) is True
        eio = socketio.server.eio

        logger.enable("local_deep_research")
        warnings: list[str] = []
        sink_id = logger.add(
            lambda m: warnings.append(m.record["message"]), level="WARNING"
        )
        try:
            for i in range(250):
                eio._log_error_once(
                    f"https://h{i}.example is not an accepted origin.",
                    "bad-origin",
                )
        finally:
            logger.remove(sink_id)
            logger.disable("local_deep_research")

        origin_warnings = [
            w for w in warnings if "rejected a WebSocket handshake" in w
        ]
        # Capped at 100 despite 250 distinct origins fed in.
        assert len(origin_warnings) == 100