learningcircuit--local-deep-research/tests/web/test_websocket_middleware.py

"""
Tests for handle_websocket_requests middleware and DiskSpoolingRequest.

Source: app_factory.py lines 599-608 (middleware), 45-57 (DiskSpoolingRequest).

Existing tests cover DiskSpoolingRequest attributes and _is_private_ip;
these test the websocket middleware *behavior* through the test client.
"""

import pytest
from unittest.mock import patch


@pytest.fixture
def app():
    """Create a test app with middleware registered."""
    from local_deep_research.web.app_factory import create_app

    with patch("local_deep_research.web.app_factory.SocketIOService"):
        app, _ = create_app()
        app.config["TESTING"] = True
        app.config["WTF_CSRF_ENABLED"] = False

        # Add a simple test route to verify passthrough
        @app.route("/api/research")
        def api_research():
            return "ok"

        return app


@pytest.fixture
def client(app):
    return app.test_client()


class TestHandleWebsocketRequests:
    """Tests for the handle_websocket_requests before_request hook."""

    def test_non_socket_path_passes_through(self, client):
        """Non-socket.io paths are unaffected by the middleware."""
        response = client.get("/api/research")
        assert response.status_code == 200
        assert response.data == b"ok"

    def test_socket_path_without_werkzeug_socket(self, client):
        """socket.io path without werkzeug.socket returns None (pass-through).

        The middleware checks environ.get('werkzeug.socket') — if absent,
        it returns None which lets Flask continue to the next handler.
        Since there's no actual socket.io route registered, this results
        in 404 from Flask's normal routing.
        """
        response = client.get("/socket.io/")
        # The middleware returns None (pass-through), so Flask routes normally
        # and returns 404 since there's no actual socket.io route
        assert response.status_code in (200, 404)

    def test_socket_path_with_werkzeug_socket(self, app):
        """socket.io path with werkzeug.socket set goes through normally."""
        # audit: PUNCHLIST reviewed 2026-05 — KEEP (ASSERT_TRUE).
        with app.test_request_context(
            "/socket.io/",
            environ_base={"werkzeug.socket": True},
        ):
            # Middleware should not interfere when werkzeug.socket is present
            from flask import request

            assert request.path == "/socket.io/"

    def test_socket_path_exception_returns_empty_200(self, app):
        """If werkzeug.socket check raises, middleware returns ("", 200)."""
        # We need to test the middleware directly by simulating the exception path.
        # The middleware does: if not request.environ.get("werkzeug.socket")
        # which can't normally raise. But the outer try/except catches any exception.

        # Create a custom environ that raises on .get()
        class ExplodingDict(dict):
            def get(self, key, default=None):
                if key == "werkzeug.socket":
                    raise RuntimeError("socket check failed")
                return super().get(key, default)

        with app.test_request_context("/socket.io/"):
            from flask import request

            # Replace environ with our exploding version
            original_environ = request.environ
            exploding = ExplodingDict(original_environ)
            request.environ = exploding

            # Find and call the before_request handler
            # The handle_websocket_requests function is registered as before_request
            for func in app.before_request_funcs.get(None, []):
                if func.__name__ == "handle_websocket_requests":
                    result = func()
                    assert result == ("", 200)
                    break
            else:
                pytest.fail(
                    "handle_websocket_requests not found in before_request_funcs"
                )