"""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: `` 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("") == "" assert _strip_credentials(None) == "" # --------------------------------------------------------------------------- # 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 ""`` and ``model or ""`` 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 "" 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 "" 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 )