learningcircuit--local-deep-research/tests/error_handling/test_openai_compat_errors.py

"""Tests for openai_compat_errors helper and its integration with
``ErrorReporter`` (#3878).

These tests construct ``openai`` / ``httpx`` exceptions directly (no network
calls) and check that the rewritten messages and ``Error type: <code>`` tokens
match the spec in the issue.
"""

from __future__ import annotations

import httpx
import pytest
from openai import (
    APIConnectionError,
    APITimeoutError,
    AuthenticationError,
    BadRequestError,
    NotFoundError,
    PermissionDeniedError,
    RateLimitError,
)

from local_deep_research.error_handling.error_reporter import (
    ErrorCategory,
    ErrorReporter,
)
from local_deep_research.error_handling.openai_compat_errors import (
    _strip_credentials,
    _walk_cause,
    friendly_openai_compatible_error,
    is_openai_compat_runtime_error,
)


def _req(
    url: str = "http://localhost:9999/v1/chat/completions",
) -> httpx.Request:
    return httpx.Request("POST", url)


def _resp(status: int, body: dict | None = None) -> httpx.Response:
    return httpx.Response(status, request=_req(), json=body or {})


# ---------------------------------------------------------------------------
# Acceptance criterion 4: redaction
# ---------------------------------------------------------------------------


class TestStripCredentials:
    def test_strips_userinfo(self):
        assert (
            _strip_credentials("https://user:secret@example.com/v1")
            == "https://example.com/v1"
        )

    def test_strips_userinfo_keeps_port(self):
        assert (
            _strip_credentials("https://u:p@example.com:8443/v1")
            == "https://example.com:8443/v1"
        )

    def test_no_userinfo_passes_through(self):
        assert (
            _strip_credentials("http://localhost:1234/v1")
            == "http://localhost:1234/v1"
        )

    def test_empty_returns_unknown_marker(self):
        assert _strip_credentials("") == "<unknown>"
        assert _strip_credentials(None) == "<unknown>"


# ---------------------------------------------------------------------------
# Cause-chain walker
# ---------------------------------------------------------------------------


class TestWalkCause:
    def test_returns_input_when_no_cause(self):
        exc = RuntimeError("flat")
        assert _walk_cause(exc) is exc

    def test_walks_to_deepest(self):
        root = APIConnectionError(message="conn", request=_req())
        try:
            try:
                raise root
            except Exception as e:
                raise RuntimeError("middle") from e
        except RuntimeError as e:
            try:
                raise ValueError("outer") from e
            except ValueError as outer:
                assert _walk_cause(outer) is root

    def test_cycle_safe(self):
        a = RuntimeError("a")
        b = RuntimeError("b")
        a.__cause__ = b
        b.__cause__ = a
        # Must terminate; the deepest reached before cycle is detected wins.
        assert _walk_cause(a) in (a, b)


# ---------------------------------------------------------------------------
# Acceptance criterion 1: connection-refused naming
# ---------------------------------------------------------------------------


class TestConnectionRefused:
    def test_openai_api_connection_error(self):
        exc = APIConnectionError(message="conn", request=_req())
        msg = friendly_openai_compatible_error(
            exc,
            provider="lmstudio",
            base_url="http://localhost:9999/v1",
            model="qwen2.5-7b",
        )
        assert "Cannot reach lmstudio at http://localhost:9999/v1" in msg
        assert "Error type: openai_connection_refused" in msg
        assert "Details:" in msg

    def test_httpx_connect_error_through_langchain_wrapper(self):
        root = httpx.ConnectError("All connection attempts failed")
        try:
            try:
                raise root
            except Exception as e:
                raise RuntimeError("LangChain wrapped") from e
        except RuntimeError as wrapped:
            msg = friendly_openai_compatible_error(
                wrapped,
                provider="openai_endpoint",
                base_url="http://localhost:1234/v1",
                model="any-model",
            )
        assert "Cannot reach openai_endpoint at http://localhost:1234/v1" in msg
        assert "Error type: openai_connection_refused" in msg


# ---------------------------------------------------------------------------
# Acceptance criterion 2: model-not-found naming
# ---------------------------------------------------------------------------


class TestModelNotFound:
    def test_notfound_names_provider_url_model(self):
        exc = NotFoundError(
            message="model 'typo-model' does not exist",
            response=_resp(404),
            body=None,
        )
        msg = friendly_openai_compatible_error(
            exc,
            provider="lmstudio",
            base_url="http://localhost:1234/v1",
            model="typo-model",
        )
        assert "lmstudio at http://localhost:1234/v1" in msg
        assert "'typo-model'" in msg
        assert "Error type: openai_model_not_found" in msg


# ---------------------------------------------------------------------------
# Acceptance criterion 3: auth naming
# ---------------------------------------------------------------------------


class TestAuth:
    def test_auth_names_provider_and_url(self):
        exc = AuthenticationError(
            message="invalid api key",
            response=_resp(401),
            body=None,
        )
        msg = friendly_openai_compatible_error(
            exc,
            provider="openai_endpoint",
            base_url="https://api.openai.com/v1",
            model="gpt-4o-mini",
        )
        assert "openai_endpoint rejected the API key" in msg
        assert "https://api.openai.com/v1" in msg
        assert "Error type: openai_auth" in msg


# ---------------------------------------------------------------------------
# Acceptance criterion 4: userinfo never leaks into the surfaced message
# ---------------------------------------------------------------------------


class TestNoCredentialLeak:
    def test_userinfo_stripped_from_friendly_text(self):
        exc = APIConnectionError(message="conn", request=_req())
        leaked_key = "supersecretkey1234567890"
        msg = friendly_openai_compatible_error(
            exc,
            provider="openai_endpoint",
            base_url=f"https://u:{leaked_key}@hosted.example.com/v1",
            model="m",
        )
        # The userinfo segment must NOT survive into the friendly portion of
        # the message. We split on the Details: suffix because the original
        # exception text is preserved there verbatim (and in practice does not
        # carry the URL, but if a future exception did, we'd still want this
        # test to guard the rewritten half).
        friendly_half = msg.split("| Details:")[0]
        assert leaked_key not in friendly_half
        assert "u:" not in friendly_half


# ---------------------------------------------------------------------------
# Acceptance criterion 5: ErrorReporter category mapping
# ---------------------------------------------------------------------------


class TestErrorReporterCategorisation:
    @pytest.fixture
    def reporter(self) -> ErrorReporter:
        return ErrorReporter()

    @pytest.mark.parametrize(
        ("token", "expected"),
        [
            ("openai_connection_refused", ErrorCategory.CONNECTION_ERROR),
            ("openai_timeout", ErrorCategory.CONNECTION_ERROR),
            ("openai_auth", ErrorCategory.MODEL_ERROR),
            ("openai_permission_denied", ErrorCategory.MODEL_ERROR),
            ("openai_model_not_found", ErrorCategory.MODEL_ERROR),
            ("openai_bad_request", ErrorCategory.MODEL_ERROR),
            ("openai_unknown", ErrorCategory.MODEL_ERROR),
            ("openai_rate_limit", ErrorCategory.RATE_LIMIT_ERROR),
        ],
    )
    def test_token_to_category(
        self, reporter: ErrorReporter, token: str, expected: ErrorCategory
    ):
        message = f"Some friendly text. (Error type: {token}) | Details: boom"
        assert reporter.categorize_error(message) == expected


# ---------------------------------------------------------------------------
# Helper-detector
# ---------------------------------------------------------------------------


class TestIsOpenAICompatRuntimeError:
    def test_yes_for_openai_class(self):
        exc = APITimeoutError(request=_req())
        assert is_openai_compat_runtime_error(exc) is True

    def test_yes_for_wrapped_openai_class(self):
        root = NotFoundError(message="missing", response=_resp(404), body=None)
        try:
            try:
                raise root
            except Exception as e:
                raise RuntimeError("wrap") from e
        except RuntimeError as wrapped:
            assert is_openai_compat_runtime_error(wrapped) is True

    def test_yes_for_httpx_connect_error(self):
        assert (
            is_openai_compat_runtime_error(httpx.ConnectError("nope")) is True
        )

    def test_no_for_unrelated_exception(self):
        assert is_openai_compat_runtime_error(ValueError("unrelated")) is False


# ---------------------------------------------------------------------------
# Additional class coverage (the four non-AC tokens still need to round-trip)
# ---------------------------------------------------------------------------


class TestAdditionalDispatch:
    def test_timeout(self):
        exc = APITimeoutError(request=_req())
        msg = friendly_openai_compatible_error(
            exc,
            provider="vllm",
            base_url="http://localhost:8000/v1",
            model="llama-3-8b",
        )
        assert "Error type: openai_timeout" in msg
        assert "did not respond in time" in msg

    def test_permission_denied(self):
        exc = PermissionDeniedError(
            message="forbidden", response=_resp(403), body=None
        )
        msg = friendly_openai_compatible_error(
            exc,
            provider="openai_endpoint",
            base_url="https://api.example.com/v1",
            model="gpt-4o",
        )
        assert "Error type: openai_permission_denied" in msg
        assert "'gpt-4o'" in msg

    def test_bad_request(self):
        exc = BadRequestError(message="bad", response=_resp(400), body=None)
        msg = friendly_openai_compatible_error(
            exc,
            provider="lmstudio",
            base_url="http://localhost:1234/v1",
            model="m",
        )
        assert "Error type: openai_bad_request" in msg

    def test_falls_back_to_unknown_for_unrelated(self):
        msg = friendly_openai_compatible_error(
            ValueError("just a value error"),
            provider="lmstudio",
            base_url="http://localhost:1234/v1",
            model="m",
        )
        assert "Error type: openai_unknown" in msg
        assert "Details: just a value error" in msg


# ---------------------------------------------------------------------------
# Edge cases pinned by this PR
# ---------------------------------------------------------------------------


class TestDispatchOrderingTimeoutBeforeConnection:
    """``openai.APITimeoutError`` subclasses ``APIConnectionError`` in
    openai>=1.x, so the dispatch table at openai_compat_errors.py:87 must
    check the timeout branch BEFORE the connection branch — otherwise
    every timeout would be mislabelled as ``openai_connection_refused``.

    The comment at lines 85-86 documents this constraint; this test pins it.

    Mutation: swap the two ``if`` blocks (lines 87-92 and 95-100) so the
    connection branch runs first, and ``APITimeoutError`` instances get the
    wrong token.
    """

    def test_timeout_subclass_dispatches_to_timeout_branch_not_connection(
        self,
    ):
        # Sanity: confirm the subclassing assumption that motivates the
        # ordering. If openai breaks this hierarchy in a future major
        # release, this assertion fails first and the test author can
        # decide whether the ordering rule still applies.
        assert issubclass(APITimeoutError, APIConnectionError)

        exc = APITimeoutError(request=_req())
        msg = friendly_openai_compatible_error(
            exc,
            provider="lmstudio",
            base_url="http://localhost:1234/v1",
            model="m",
        )
        assert "Error type: openai_timeout" in msg
        assert "Error type: openai_connection_refused" not in msg


class TestWalkCauseChainPreference:
    """``_walk_cause`` traverses ``cur.__cause__ or cur.__context__`` —
    i.e. explicit ``raise X from Y`` chains win over implicit
    ``__context__`` chains. Pins the preference.

    Mutation: swap to ``cur.__context__ or cur.__cause__`` at
    openai_compat_errors.py:60 — the test would point at the wrong root.
    """

    def test_cause_preferred_over_context_when_both_set(self):
        explicit_root = APIConnectionError(message="conn", request=_req())
        implicit_root = RuntimeError("implicit context")

        wrapper = RuntimeError("wrapper")
        wrapper.__cause__ = explicit_root
        wrapper.__context__ = implicit_root

        assert _walk_cause(wrapper) is explicit_root


class TestStripCredentialsEdgeCases:
    def test_ipv6_host_brackets_preserved(self):
        # ``urlparse`` returns ``hostname`` without brackets; the
        # implementation reassembles ``netloc`` and must re-add the
        # brackets, or the resulting URL is unparseable downstream —
        # ``http://::1:8080/`` is ambiguous about where the host ends
        # and the port begins.
        result = _strip_credentials("http://[::1]:8080/v1")
        assert result == "http://[::1]:8080/v1"

    def test_userinfo_stripped_with_ipv6_host(self):
        # Combine userinfo and IPv6 host. The key must be removed AND
        # the brackets must survive.
        result = _strip_credentials(
            "http://user:secret-key-12345@[::1]:8080/v1"
        )
        assert "secret-key-12345" not in result
        assert "user:" not in result
        assert result == "http://[::1]:8080/v1"

    def test_url_with_no_netloc_passed_through(self):
        # A bare path (no scheme/netloc) is returned as-is. The function
        # short-circuits at ``if not parsed.netloc:``. Worth pinning so
        # someone "fixing" the path-only case doesn't break it.
        assert _strip_credentials("/v1/chat/completions") == (
            "/v1/chat/completions"
        )


class TestFriendlyErrorNoneArgs:
    """``friendly_openai_compatible_error`` uses ``provider or
    "<unknown provider>"`` and ``model or "<unspecified>"`` to surface a
    legible message even when the caller doesn't know the values. Pin
    both placeholders so future refactors don't drop them.
    """

    def test_none_provider_replaced_with_placeholder(self):
        msg = friendly_openai_compatible_error(
            APIConnectionError(message="conn", request=_req()),
            provider=None,
            base_url="http://localhost:9999/v1",
            model="m",
        )
        assert "<unknown provider>" in msg

    def test_none_model_replaced_with_placeholder(self):
        msg = friendly_openai_compatible_error(
            NotFoundError(
                message="missing",
                response=_resp(404),
                body=None,
            ),
            provider="lmstudio",
            base_url="http://localhost:1234/v1",
            model=None,
        )
        assert "<unspecified>" in msg


# ---------------------------------------------------------------------------
# RateLimitError dispatch (follow-up to #3878)
# ---------------------------------------------------------------------------


class TestRateLimitErrorDispatch:
    """``openai.RateLimitError`` subclasses ``APIError`` via
    ``APIStatusError``.  It must dispatch to ``openai_rate_limit`` (mapped to
    RATE_LIMIT_ERROR) instead of falling through to the ``openai_unknown``
    catch-all (mapped to MODEL_ERROR).

    Without the explicit branch, a 429 is mis-categorised as MODEL_ERROR,
    producing the wrong suggestions and skipping the API_QUOTA_WARNING
    notification.
    """

    def test_rate_limit_dispatches_to_rate_limit_token(self):
        exc = RateLimitError(
            message="Rate limit exceeded",
            response=_resp(429),
            body=None,
        )
        msg = friendly_openai_compatible_error(
            exc,
            provider="openrouter",
            base_url="https://openrouter.ai/api/v1",
            model="gpt-4o",
        )
        assert "Error type: openai_rate_limit" in msg
        assert "Error type: openai_unknown" not in msg
        assert "rate-limited" in msg

    def test_rate_limit_categorised_as_rate_limit_error(self):
        reporter = ErrorReporter()
        message = (
            "openrouter at https://openrouter.ai/api/v1 rate-limited the "
            "request for model 'gpt-4o'. (Error type: openai_rate_limit) "
            "| Details: Error code: 429"
        )
        assert (
            reporter.categorize_error(message) == ErrorCategory.RATE_LIMIT_ERROR
        )