From 0124daa8b741c69c01c481849f6cc9a30ccfb065 Mon Sep 17 00:00:00 2001 From: Ali Khan Date: Sun, 23 Aug 2026 10:31:08 -0400 Subject: [PATCH] refactor(oauth): satisfy the new lint and test-quality gates The base branch grew LIT001/LIT002 mutability rules, ruff UP045, and a test-quality gate since this PR was opened. The resolver now takes a read-only Mapping, the anthropic dispatch stamps use_bearer_for_custom_base on the request's litellm_params instead of constructing a new dict, and the flagged tests are rewritten to fake at the HTTP boundary with respx instead of patching litellm internals; the injection tests now assert the outbound Authorization header and that no oauth_* field reaches the provider request body --- .../messages/transformation.py | 2 +- .../llms/openai_like/oauth_authenticator.py | 26 +- litellm/main.py | 10 +- litellm/proxy/litellm_pre_call_utils.py | 4 +- .../test_anthropic_messages_transformation.py | 100 +++--- .../openai_like/test_oauth_authenticator.py | 335 +++++++++--------- 6 files changed, 233 insertions(+), 244 deletions(-) diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/transformation.py b/litellm/llms/anthropic/experimental_pass_through/messages/transformation.py index 54b21c3df69..8f96c8220b0 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/messages/transformation.py @@ -13,6 +13,7 @@ from litellm.litellm_core_utils.litellm_logging import verbose_logger from litellm.llms.base_llm.anthropic_messages.transformation import ( BaseAnthropicMessagesConfig, ) +from litellm.llms.openai_like.oauth_authenticator import resolve_client_credentials_token from litellm.types.llms.anthropic import ( ANTHROPIC_ADVISOR_TOOL_TYPE, ANTHROPIC_BETA_HEADER_VALUES, @@ -21,7 +22,6 @@ from litellm.types.llms.anthropic import ( from litellm.types.llms.anthropic_messages.anthropic_response import ( AnthropicMessagesResponse, ) -from litellm.llms.openai_like.oauth_authenticator import resolve_client_credentials_token from litellm.types.llms.anthropic_tool_search import get_tool_search_beta_header from litellm.types.router import GenericLiteLLMParams diff --git a/litellm/llms/openai_like/oauth_authenticator.py b/litellm/llms/openai_like/oauth_authenticator.py index 408861c21c7..b0fff5e0afa 100644 --- a/litellm/llms/openai_like/oauth_authenticator.py +++ b/litellm/llms/openai_like/oauth_authenticator.py @@ -8,7 +8,7 @@ from config. """ import hashlib -from typing import Optional +from collections.abc import Mapping import httpx from pydantic import BaseModel, ValidationError @@ -34,12 +34,12 @@ class _TokenResponse(BaseModel): def get_client_credentials_token( - token_url: Optional[str], - client_id: Optional[str], - client_secret: Optional[str], - scope: Optional[str] = None, + token_url: str | None, + client_id: str | None, + client_secret: str | None, + scope: str | None = None, timeout: float = 30.0, - http_client: Optional[HTTPHandler] = None, + http_client: HTTPHandler | None = None, ) -> str: """ Fetch (and cache) an OAuth2 access token using the client_credentials grant. @@ -69,7 +69,7 @@ def get_client_credentials_token( if isinstance(cached, str): return cached - data = { + data = { # mutable-ok: HTTPHandler.post types the form payload as a plain dict "grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, @@ -115,15 +115,15 @@ def get_client_credentials_token( class _ClientCredentialsConfig(BaseModel): oauth_client_credentials: bool = False - oauth_token_url: Optional[str] = None - oauth_client_id: Optional[str] = None - oauth_client_secret: Optional[str] = None - oauth_scope: Optional[str] = None + oauth_token_url: str | None = None + oauth_client_id: str | None = None + oauth_client_secret: str | None = None + oauth_scope: str | None = None def resolve_client_credentials_token( - litellm_params: dict[str, object], -) -> Optional[str]: + litellm_params: Mapping[str, object], +) -> str | None: """Return an OAuth2 client_credentials bearer when the deployment opted in. Runs on every OpenAI-compatible completion, so the common no-OAuth case is a diff --git a/litellm/main.py b/litellm/main.py index 24e1e529e6d..0ddc0e06f8b 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -2483,7 +2483,7 @@ def _complete_custom_openai( resolve_client_credentials_token, ) - oauth_api_key = resolve_client_credentials_token(cast("dict[str, object]", litellm_params)) + oauth_api_key = resolve_client_credentials_token(litellm_params) if oauth_api_key is not None: api_key = oauth_api_key @@ -2760,12 +2760,10 @@ def _complete_anthropic(ctx: _CompletionDispatchContext) -> _CompletionDispatchR resolve_client_credentials_token, ) - oauth_api_key = resolve_client_credentials_token(cast("dict[str, object]", litellm_params)) + oauth_api_key = resolve_client_credentials_token(litellm_params) if oauth_api_key is not None: api_key = oauth_api_key - request_litellm_params: Final = ( - {**litellm_params, "use_bearer_for_custom_base": True} if oauth_api_key is not None else litellm_params - ) + litellm_params["use_bearer_for_custom_base"] = True custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict # call /messages @@ -2795,7 +2793,7 @@ def _complete_anthropic(ctx: _CompletionDispatchContext) -> _CompletionDispatchR model_response=model_response, print_verbose=print_verbose, optional_params=optional_params, - litellm_params=request_litellm_params, + litellm_params=litellm_params, logger_fn=logger_fn, encoding=_get_encoding(), # for calculating input/output tokens api_key=api_key, diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index d2d0905266b..fa6509c96ad 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -550,7 +550,9 @@ def _strip_client_pricing_overrides(data: dict[str, Any]) -> None: ) -def _strip_client_oauth_overrides(data: dict[str, Any]) -> None: +def _strip_client_oauth_overrides( + data: dict[str, Any], # mutable-ok: strips in place on the request body the pre-call pipeline threads through +) -> None: """Drop OAuth client-credentials fields from a client request body. These are deployment configuration; a client-supplied ``oauth_token_url`` diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_messages_transformation.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_messages_transformation.py index d45aa993742..fc251b1ea36 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_messages_transformation.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_messages_transformation.py @@ -5,9 +5,9 @@ Anthropic Messages API with an OAuth2 gateway; the minted bearer must replace x-api-key auth. """ -from unittest.mock import MagicMock, patch - +import httpx import pytest +import respx from litellm.llms.anthropic.experimental_pass_through.messages.transformation import ( AnthropicMessagesConfig, @@ -29,71 +29,59 @@ def _flush_token_cache(): _token_cache.flush_cache() -def _token_client(access_token="tok-messages", expires_in=3600): - client = MagicMock() - response = MagicMock() - response.raise_for_status.return_value = None - response.json.return_value = {"access_token": access_token, "expires_in": expires_in} - client.post.return_value = response - return client +def _mock_token_endpoint(respx_mock, access_token="tok-messages", expires_in=3600): + return respx_mock.post("https://idp.test/token").mock( + return_value=httpx.Response(200, json={"access_token": access_token, "expires_in": expires_in}) + ) class TestValidateAnthropicMessagesEnvironmentOAuth: - def test_flag_swaps_x_api_key_for_minted_bearer(self): - token_client = _token_client(access_token="tok-messages") - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ): - headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment( - headers={"x-api-key": "sk-config"}, - model="claude-sonnet-5", - messages=[{"role": "user", "content": "hi"}], - optional_params={}, - litellm_params=dict(OAUTH_LITELLM_PARAMS), - api_key="sk-config", - api_base="https://gateway.test", - ) + def test_flag_swaps_x_api_key_for_minted_bearer(self, respx_mock: respx.MockRouter): + token_route = _mock_token_endpoint(respx_mock, access_token="tok-messages") - token_client.post.assert_called_once() + headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment( + headers={"x-api-key": "sk-config"}, + model="claude-sonnet-5", + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params=dict(OAUTH_LITELLM_PARAMS), + api_key="sk-config", + api_base="https://gateway.test", + ) + + assert token_route.called assert "x-api-key" not in headers assert headers["authorization"] == "Bearer tok-messages" - def test_flag_sets_bearer_when_no_prior_auth_header(self): - token_client = _token_client(access_token="tok-fresh") - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ): - headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment( - headers={}, - model="claude-sonnet-5", - messages=[{"role": "user", "content": "hi"}], - optional_params={}, - litellm_params=dict(OAUTH_LITELLM_PARAMS), - api_key="sk-config", - api_base="https://gateway.test", - ) + def test_flag_sets_bearer_when_no_prior_auth_header(self, respx_mock: respx.MockRouter): + _mock_token_endpoint(respx_mock, access_token="tok-fresh") + + headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment( + headers={}, + model="claude-sonnet-5", + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params=dict(OAUTH_LITELLM_PARAMS), + api_key="sk-config", + api_base="https://gateway.test", + ) assert headers["authorization"] == "Bearer tok-fresh" assert "x-api-key" not in headers - def test_no_flag_keeps_x_api_key_and_fetches_nothing(self): - token_client = _token_client(access_token="should-not-be-used") - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ): - headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment( - headers={"x-api-key": "sk-config"}, - model="claude-sonnet-5", - messages=[{"role": "user", "content": "hi"}], - optional_params={}, - litellm_params={"oauth_token_url": "https://idp.test/token"}, - api_key="sk-config", - api_base="https://gateway.test", - ) + def test_no_flag_keeps_x_api_key_and_fetches_nothing(self, respx_mock: respx.MockRouter): + token_route = _mock_token_endpoint(respx_mock, access_token="should-not-be-used") - token_client.post.assert_not_called() + headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment( + headers={"x-api-key": "sk-config"}, + model="claude-sonnet-5", + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params={"oauth_token_url": "https://idp.test/token"}, + api_key="sk-config", + api_base="https://gateway.test", + ) + + assert not token_route.called assert headers["x-api-key"] == "sk-config" assert "authorization" not in headers diff --git a/tests/test_litellm/llms/openai_like/test_oauth_authenticator.py b/tests/test_litellm/llms/openai_like/test_oauth_authenticator.py index 6929fbdf83a..f5cff4afbad 100644 --- a/tests/test_litellm/llms/openai_like/test_oauth_authenticator.py +++ b/tests/test_litellm/llms/openai_like/test_oauth_authenticator.py @@ -5,10 +5,13 @@ OpenAI-compatible completion path. Regression coverage for https://github.com/BerriAI/litellm/issues/12367 """ -from unittest.mock import MagicMock, patch +import json +from unittest.mock import MagicMock +from urllib.parse import parse_qsl import httpx import pytest +import respx from litellm.llms.openai_like.oauth_authenticator import ( OAuthClientCredentialsError, @@ -18,6 +21,12 @@ from litellm.llms.openai_like.oauth_authenticator import ( ) +def _mock_token_endpoint(respx_mock, url="https://idp.test/token", access_token="tok-abc", expires_in=3600): + return respx_mock.post(url).mock( + return_value=httpx.Response(200, json={"access_token": access_token, "expires_in": expires_in}) + ) + + @pytest.fixture(autouse=True) def _flush_token_cache(): _token_cache.flush_cache() @@ -217,46 +226,41 @@ class TestClientCredentialsTokenFetch: class TestResolveClientCredentialsToken: - def test_flag_on_with_creds_returns_token(self): - client = _token_client(access_token="tok-r") - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=client, - ): - token = resolve_client_credentials_token( - { - "oauth_client_credentials": True, - "oauth_token_url": "https://idp.test/token", - "oauth_client_id": "cid-r", - "oauth_client_secret": "secret-r", - "oauth_scope": "scope-r", - } - ) + def test_flag_on_with_creds_returns_token(self, respx_mock: respx.MockRouter): + token_route = _mock_token_endpoint(respx_mock, access_token="tok-r") + + token = resolve_client_credentials_token( + { + "oauth_client_credentials": True, + "oauth_token_url": "https://idp.test/token", + "oauth_client_id": "cid-r", + "oauth_client_secret": "secret-r", + "oauth_scope": "scope-r", + } + ) assert token == "tok-r" - data = client.post.call_args.kwargs["data"] - assert data["client_id"] == "cid-r" - assert data["client_secret"] == "secret-r" - assert data["scope"] == "scope-r" + form = dict(parse_qsl(token_route.calls.last.request.read().decode())) + assert form["grant_type"] == "client_credentials" + assert form["client_id"] == "cid-r" + assert form["client_secret"] == "secret-r" + assert form["scope"] == "scope-r" - def test_flag_absent_returns_none_without_fetch(self): + def test_flag_absent_returns_none_without_fetch(self, respx_mock: respx.MockRouter): # Creds present but no flag -> OAuth must not engage; the deployment keeps # its configured api_key. Kills a mutant that triggers on creds presence. - client = _token_client() - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=client, - ): - result = resolve_client_credentials_token( - { - "oauth_token_url": "https://idp.test/token", - "oauth_client_id": "cid", - "oauth_client_secret": "secret", - } - ) + token_route = _mock_token_endpoint(respx_mock) + + result = resolve_client_credentials_token( + { + "oauth_token_url": "https://idp.test/token", + "oauth_client_id": "cid", + "oauth_client_secret": "secret", + } + ) assert result is None - client.post.assert_not_called() + assert not token_route.called def test_flag_false_returns_none(self): assert ( @@ -274,7 +278,7 @@ class TestResolveClientCredentialsToken: } ) - def test_env_vars_are_not_consulted(self, monkeypatch): + def test_env_vars_are_not_consulted(self, monkeypatch, respx_mock: respx.MockRouter): # Creds live ONLY in env; the resolver reads litellm_params exclusively, so # with the flag on but creds absent from litellm_params it must raise and # never mint a token from env. This is the exfiltration path the security @@ -282,18 +286,14 @@ class TestResolveClientCredentialsToken: monkeypatch.setenv("CUSTOM_OAUTH_TOKEN_URL", "https://idp.env/token") monkeypatch.setenv("CUSTOM_OAUTH_CLIENT_ID", "cid-env") monkeypatch.setenv("CUSTOM_OAUTH_CLIENT_SECRET", "secret-env") - client = _token_client(access_token="tok-env") + token_route = _mock_token_endpoint(respx_mock, url="https://idp.env/token", access_token="tok-env") - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=client, - ): - with pytest.raises(OAuthClientCredentialsError): - resolve_client_credentials_token({"oauth_client_credentials": True}) + with pytest.raises(OAuthClientCredentialsError): + resolve_client_credentials_token({"oauth_client_credentials": True}) - client.post.assert_not_called() + assert not token_route.called - def test_no_token_after_clientside_base_override_clear(self): + def test_no_token_after_clientside_base_override_clear(self, respx_mock: respx.MockRouter): # End-to-end with the router: a client api_base override clears the # deployment's oauth_client_credentials flag + creds, so the resolver # returns None (graceful fallback to the configured api_key) and no admin @@ -313,152 +313,153 @@ class TestResolveClientCredentialsToken: {"api_base": "https://client.example/v1"}, ) - client = _token_client(access_token="tok-leak") - with patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=client, - ): - assert resolve_client_credentials_token(cleared) is None + token_route = _mock_token_endpoint(respx_mock, url="https://idp.internal/token", access_token="tok-leak") - client.post.assert_not_called() + assert resolve_client_credentials_token(cleared) is None + assert not token_route.called + + +OPENAI_CHAT_RESPONSE = { + "id": "chatcmpl-oauth", + "object": "chat.completion", + "created": 1677652288, + "model": "gpt-4o", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "oauth works"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 9, "completion_tokens": 2, "total_tokens": 11}, +} + +ANTHROPIC_MESSAGES_RESPONSE = { + "id": "msg_oauth", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-5", + "content": [{"type": "text", "text": "oauth works"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 9, "output_tokens": 2}, +} class TestCompletionInjection: - def test_flag_injects_minted_token_as_api_key(self): - # End-to-end through litellm.completion on a plain openai/ model: the flag - # plus oauth_* in litellm_params mint a bearer that overrides the - # configured api_key, which the OpenAI SDK then sends as Authorization: - # Bearer. Fails if the injection in _complete_custom_openai is removed or - # the flag stops reaching litellm_params. + def test_flag_injects_minted_token_as_api_key(self, respx_mock: respx.MockRouter): + # End-to-end through litellm.completion on a plain openai/ model, faked at + # the HTTP boundary: the flag plus oauth_* in litellm_params mint a bearer + # that overrides the configured api_key, and the outbound chat request + # carries it as Authorization: Bearer while the oauth_* fields themselves + # never reach the provider request body. import litellm - import litellm.main as main_mod - from litellm.types.utils import ModelResponse - token_client = _token_client(access_token="tok-e2e") - fake = ModelResponse() - with ( - patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ), - patch.object( - main_mod.openai_chat_completions, "completion", return_value=fake - ) as chat, - ): - litellm.completion( - model="openai/gpt-4o", - messages=[{"role": "user", "content": "hi"}], - api_base="https://gateway.test/v1", - api_key="sk-should-be-overridden", - oauth_client_credentials=True, - oauth_token_url="https://idp.test/token", - oauth_client_id="cid-e2e", - oauth_client_secret="secret-e2e", - oauth_scope="scope-e2e", - ) + litellm.in_memory_llm_clients_cache.flush_cache() + _mock_token_endpoint(respx_mock, access_token="tok-e2e") + chat_route = respx_mock.post("https://gateway.test/v1/chat/completions").mock( + return_value=httpx.Response(200, json=OPENAI_CHAT_RESPONSE) + ) - token_client.post.assert_called_once() - assert chat.call_args.kwargs["api_key"] == "tok-e2e" + response = litellm.completion( + model="openai/gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_base="https://gateway.test/v1", + api_key="sk-should-be-overridden", + oauth_client_credentials=True, + oauth_token_url="https://idp.test/token", + oauth_client_id="cid-e2e", + oauth_client_secret="secret-e2e", + oauth_scope="scope-e2e", + ) - def test_no_flag_keeps_configured_api_key(self): + assert response.choices[0].message.content == "oauth works" + request = chat_route.calls.last.request + assert request.headers["authorization"] == "Bearer tok-e2e" + body = json.loads(request.read()) + assert not any(key.startswith("oauth_") for key in body) + + def test_no_flag_keeps_configured_api_key(self, respx_mock: respx.MockRouter): # Without the flag, even with oauth_* present, no token is fetched and the # configured api_key is sent unchanged. import litellm - import litellm.main as main_mod - from litellm.types.utils import ModelResponse - token_client = _token_client(access_token="should-not-be-used") - fake = ModelResponse() - with ( - patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ), - patch.object( - main_mod.openai_chat_completions, "completion", return_value=fake - ) as chat, - ): - litellm.completion( - model="openai/gpt-4o", - messages=[{"role": "user", "content": "hi"}], - api_base="https://gateway.test/v1", - api_key="sk-static", - oauth_token_url="https://idp.test/token", - oauth_client_id="cid", - oauth_client_secret="secret", - ) + litellm.in_memory_llm_clients_cache.flush_cache() + token_route = _mock_token_endpoint(respx_mock, access_token="should-not-be-used") + chat_route = respx_mock.post("https://gateway.test/v1/chat/completions").mock( + return_value=httpx.Response(200, json=OPENAI_CHAT_RESPONSE) + ) - token_client.post.assert_not_called() - assert chat.call_args.kwargs["api_key"] == "sk-static" + litellm.completion( + model="openai/gpt-4o", + messages=[{"role": "user", "content": "hi"}], + api_base="https://gateway.test/v1", + api_key="sk-static", + oauth_token_url="https://idp.test/token", + oauth_client_id="cid", + oauth_client_secret="secret", + ) + + assert not token_route.called + assert chat_route.calls.last.request.headers["authorization"] == "Bearer sk-static" class TestAnthropicCompletionInjection: - def test_flag_injects_minted_token_and_bearer_mode(self): - # End-to-end through litellm.completion on an anthropic/ model: the flag - # plus oauth_* in litellm_params mint a bearer that overrides the - # configured api_key, and use_bearer_for_custom_base is set so the - # anthropic header builder emits Authorization: Bearer for the custom - # gateway base instead of x-api-key. + def test_flag_injects_minted_token_and_bearer_mode(self, respx_mock: respx.MockRouter): + # End-to-end through litellm.completion on an anthropic/ model with a + # custom gateway base, faked at the HTTP boundary: the minted bearer + # replaces the configured api_key and use_bearer_for_custom_base makes the + # header builder send Authorization: Bearer instead of x-api-key. import litellm - import litellm.main as main_mod - from litellm.types.utils import ModelResponse - token_client = _token_client(access_token="tok-anthropic") - fake = ModelResponse() - with ( - patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ), - patch.object( - main_mod.anthropic_chat_completions, "completion", return_value=fake - ) as chat, - ): - litellm.completion( - model="anthropic/claude-sonnet-5", - messages=[{"role": "user", "content": "hi"}], - api_base="https://gateway.test", - api_key="sk-should-be-overridden", - oauth_client_credentials=True, - oauth_token_url="https://idp.test/token", - oauth_client_id="cid-anthropic", - oauth_client_secret="secret-anthropic", - oauth_scope="scope-anthropic", - ) + litellm.in_memory_llm_clients_cache.flush_cache() + _mock_token_endpoint(respx_mock, access_token="tok-anthropic") + messages_route = respx_mock.post("https://gateway.test/v1/messages").mock( + return_value=httpx.Response(200, json=ANTHROPIC_MESSAGES_RESPONSE) + ) - token_client.post.assert_called_once() - assert chat.call_args.kwargs["api_key"] == "tok-anthropic" - assert chat.call_args.kwargs["litellm_params"]["use_bearer_for_custom_base"] is True + response = litellm.completion( + model="anthropic/claude-sonnet-5", + messages=[{"role": "user", "content": "hi"}], + api_base="https://gateway.test", + api_key="sk-should-be-overridden", + oauth_client_credentials=True, + oauth_token_url="https://idp.test/token", + oauth_client_id="cid-anthropic", + oauth_client_secret="secret-anthropic", + oauth_scope="scope-anthropic", + ) - def test_no_flag_keeps_configured_api_key(self): + assert response.choices[0].message.content == "oauth works" + request = messages_route.calls.last.request + assert request.headers["authorization"] == "Bearer tok-anthropic" + assert "x-api-key" not in request.headers + body = json.loads(request.read()) + assert not any(key.startswith("oauth_") for key in body) + + def test_no_flag_keeps_configured_api_key(self, respx_mock: respx.MockRouter): import litellm - import litellm.main as main_mod - from litellm.types.utils import ModelResponse - token_client = _token_client(access_token="should-not-be-used") - fake = ModelResponse() - with ( - patch( - "litellm.llms.openai_like.oauth_authenticator._get_httpx_client", - return_value=token_client, - ), - patch.object( - main_mod.anthropic_chat_completions, "completion", return_value=fake - ) as chat, - ): - litellm.completion( - model="anthropic/claude-sonnet-5", - messages=[{"role": "user", "content": "hi"}], - api_base="https://gateway.test", - api_key="sk-static", - oauth_token_url="https://idp.test/token", - oauth_client_id="cid", - oauth_client_secret="secret", - ) + litellm.in_memory_llm_clients_cache.flush_cache() + token_route = _mock_token_endpoint(respx_mock, access_token="should-not-be-used") + messages_route = respx_mock.post("https://gateway.test/v1/messages").mock( + return_value=httpx.Response(200, json=ANTHROPIC_MESSAGES_RESPONSE) + ) - token_client.post.assert_not_called() - assert chat.call_args.kwargs["api_key"] == "sk-static" - assert "use_bearer_for_custom_base" not in chat.call_args.kwargs["litellm_params"] + litellm.completion( + model="anthropic/claude-sonnet-5", + messages=[{"role": "user", "content": "hi"}], + api_base="https://gateway.test", + api_key="sk-static", + oauth_token_url="https://idp.test/token", + oauth_client_id="cid", + oauth_client_secret="secret", + ) + + assert not token_route.called + request = messages_route.calls.last.request + assert request.headers["x-api-key"] == "sk-static" + assert "authorization" not in request.headers class TestOAuthParamsExcludedFromProviderBody: