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

"""
Tests for error handler *behavior* in app_factory.register_error_handlers().

Existing tests cover registration and 404 content; these test:
- 401 handler JSON vs redirect branching
- 413 handler JSON vs text branching
- 500 handler JSON vs text branching
- WebAPIException handler JSON conversion
- CSRF handler IP-based messaging (private/public/proxied/HTTPS)
- NewsAPIException handler JSON conversion
- is_private_ip edge cases (IPv4-mapped IPv6, link-local, multicast)
"""

import pytest
from unittest.mock import patch

from flask import abort


@pytest.fixture
def app():
    """Create a test app with error handlers 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
        # Must disable exception propagation so error handlers are invoked
        app.config["PROPAGATE_EXCEPTIONS"] = False
        app.config["TRAP_HTTP_EXCEPTIONS"] = False

        # Routes that trigger 401 errors
        @app.route("/api/trigger-401")
        def api_trigger_401():
            abort(401)

        @app.route("/settings/api/trigger-401")
        def settings_api_trigger_401():
            abort(401)

        @app.route("/trigger-401")
        def web_trigger_401():
            abort(401)

        # Routes that trigger 413 errors
        @app.route("/api/trigger-413")
        def api_trigger_413():
            abort(413)

        @app.route("/trigger-413")
        def web_trigger_413():
            abort(413)

        # Routes that trigger 500 errors
        @app.route("/api/trigger-500")
        def api_trigger_500():
            raise Exception("boom")

        @app.route("/trigger-500")
        def web_trigger_500():
            raise Exception("boom")

        # Route that raises WebAPIException
        @app.route("/api/trigger-web-api-error")
        def trigger_web_api_error():
            from local_deep_research.web.exceptions import (
                WebAPIException,
            )

            raise WebAPIException(
                message="Something went wrong",
                status_code=400,
                error_code="BAD_REQUEST",
                details={"field": "name"},
            )

        # Route that raises AuthenticationRequiredError
        @app.route("/api/trigger-auth-error")
        def trigger_auth_error():
            from local_deep_research.web.exceptions import (
                AuthenticationRequiredError,
            )

            raise AuthenticationRequiredError()

        # Route that raises NewsAPIException
        @app.route("/api/trigger-news-error")
        def trigger_news_error():
            from local_deep_research.news.exceptions import (
                NewsAPIException,
            )

            raise NewsAPIException(
                message="Feed unavailable",
                status_code=503,
                error_code="FEED_ERROR",
            )

        return app


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


class TestUnauthorizedHandler:
    """Tests for the 401 error handler behavior."""

    def test_401_returns_json_for_api_path(self, client):
        """API paths get JSON error response for 401."""
        response = client.get("/api/trigger-401")
        assert response.status_code == 401
        data = response.get_json()
        assert data is not None
        assert data["error"] == "Authentication required"

    def test_401_returns_json_for_settings_api_path(self, client):
        """/settings/api/ paths get JSON error response for 401."""
        response = client.get("/settings/api/trigger-401")
        assert response.status_code == 401
        data = response.get_json()
        assert data is not None
        assert data["error"] == "Authentication required"

    def test_401_redirects_for_web_path(self, client):
        """Non-API paths redirect to login page for 401."""
        response = client.get("/trigger-401")
        assert response.status_code == 302
        assert "/auth/login" in response.location


class TestRequestTooLargeHandler:
    """Tests for the 413 error handler behavior."""

    def test_413_returns_json_for_api_path(self, client):
        """API paths get JSON error response for 413."""
        response = client.get("/api/trigger-413")
        assert response.status_code == 413
        data = response.get_json()
        assert data is not None
        assert data["error"] == "Request too large"

    def test_413_returns_text_for_web_path(self, client):
        """Non-API paths get plain text error response for 413."""
        response = client.get("/trigger-413")
        assert response.status_code == 413
        assert response.data == b"Request too large"


class TestServerErrorHandler:
    """Tests for the 500 error handler behavior."""

    def test_500_returns_json_for_api_path(self, client):
        """API paths get JSON error response for 500."""
        response = client.get("/api/trigger-500")
        assert response.status_code == 500
        data = response.get_json()
        assert data is not None
        assert data["error"] == "Server error"

    def test_500_returns_text_for_web_path(self, client):
        """Non-API paths get plain text error response for 500."""
        response = client.get("/trigger-500")
        assert response.status_code == 500
        assert response.data == b"Server error"


class TestCsrfErrorHandler:
    """Tests for CSRF error handler IP-based branching."""

    def _trigger_csrf(
        self, app, remote_addr, is_secure=False, forwarded_for=None
    ):
        """Trigger a CSRF error with controlled request environment."""
        from flask_wtf.csrf import CSRFError

        # Must set PREFERRED_URL_SCHEME to match is_secure, because Flask
        # uses it to determine request.scheme / request.is_secure
        app.config["PREFERRED_URL_SCHEME"] = "https" if is_secure else "http"

        ctx_kwargs = {
            "environ_base": {
                "REMOTE_ADDR": remote_addr,
                "wsgi.url_scheme": "https" if is_secure else "http",
            },
        }
        if forwarded_for:
            ctx_kwargs["headers"] = {"X-Forwarded-For": forwarded_for}

        with app.test_request_context("/api/test", **ctx_kwargs):
            # CSRF handler is registered under status code 400
            handlers_400 = app.error_handler_spec.get(None, {}).get(400, {})
            handler = handlers_400.get(CSRFError)
            if handler is None:
                pytest.skip("CSRFError handler not registered")
            error = CSRFError("CSRF token missing")
            response = handler(error)
            return response

    def test_csrf_error_returns_description_as_json(self, app):
        """CSRF errors return the error description as JSON, regardless of
        source IP or scheme. The old IP-based 'use HTTPS' branch is gone."""
        response = self._trigger_csrf(app, "8.8.8.8")
        assert response.status_code == 400
        data = response.get_json()
        assert "error" in data
        # No IP-specific guidance is added anymore
        assert "Solutions" not in data["error"]
        assert "8.8.8.8" not in data["error"]

    def test_csrf_returns_400_status(self, app):
        """CSRF errors return 400 status code."""
        from flask_wtf.csrf import CSRFError

        with app.test_request_context("/api/test"):
            handlers_400 = app.error_handler_spec.get(None, {}).get(400, {})
            handler = handlers_400.get(CSRFError)
            if handler is None:
                pytest.skip("CSRFError handler not registered")
            response = handler(CSRFError("bad token"))
            assert response.status_code == 400


class TestNewsAPIExceptionHandler:
    """Tests for NewsAPIException error handler."""

    def test_news_exception_returns_json(self, client):
        """NewsAPIException is converted to JSON with error_code and status."""
        response = client.get("/api/trigger-news-error")
        assert response.status_code == 503
        data = response.get_json()
        assert data["error"] == "Feed unavailable"
        assert data["error_code"] == "FEED_ERROR"
        assert data["status_code"] == 503


class TestWebAPIExceptionHandler:
    """Tests for WebAPIException error handler."""

    def test_web_api_exception_returns_json(self, client):
        """WebAPIException is converted to JSON with to_dict() fields."""
        response = client.get("/api/trigger-web-api-error")
        assert response.status_code == 400
        data = response.get_json()
        assert data is not None
        assert data["status"] == "error"
        assert data["message"] == "Something went wrong"
        assert data["error_code"] == "BAD_REQUEST"
        assert data["details"] == {"field": "name"}

    def test_web_api_exception_preserves_status_code(self, client):
        """Response uses the exception's status_code, not a default."""
        response = client.get("/api/trigger-web-api-error")
        assert response.status_code == 400

    def test_authentication_required_error_returns_401(self, client):
        """AuthenticationRequiredError returns 401 with structured JSON."""
        response = client.get("/api/trigger-auth-error")
        assert response.status_code == 401
        data = response.get_json()
        assert data is not None
        assert data["status"] == "error"
        assert "Authentication required" in data["message"]
        assert data["error_code"] == "AUTHENTICATION_REQUIRED"


class TestIsPrivateIpEdgeCases:
    """Edge cases for is_private_ip not covered by existing tests."""

    def test_ipv4_mapped_ipv6_private(self):
        """IPv4-mapped IPv6 address with private IPv4 is treated as private."""
        from local_deep_research.security.network_utils import is_private_ip

        assert is_private_ip("::ffff:192.168.1.1") is True

    def test_link_local_address(self):
        """Link-local addresses (169.254.x.x) are private."""
        from local_deep_research.security.network_utils import is_private_ip

        assert is_private_ip("169.254.1.1") is True

    def test_multicast_address(self):
        """Multicast addresses (224.x.x.x) are not private."""
        from local_deep_research.security.network_utils import is_private_ip

        assert is_private_ip("224.0.0.1") is False