From 1bfd7ae5458dd863951fc358eaf4e86f2bfbec17 Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Mon, 17 Aug 2026 10:24:31 -0300 Subject: [PATCH] feat(aicore): add transparent proxy routing and BTP Destination Service mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements Option 3 from the AFSDK-4306 security alignment meeting: SDK absorbs all routing complexity so agent code is identical in all environments. The deployer controls routing by choosing which env vars to inject. Two new modes in set_aicore_config(): Proxy mode (AICORE_PROXY_URL set): - Routes all LiteLLM calls through an external LiteLLM proxy - Sets litellm.api_base / litellm.api_key globally - Rewrites sap/ → litellm_proxy/ transparently in completion() and acompletion() wrappers (including on auth-error retry) - No AI Core credentials written to the process environment - JWT never reaches the agent process (proxy handles OAuth) Destination mode (AICORE_DESTINATION_NAME set): - Loads AI Core credentials at startup from a named BTP Destination Service destination via the existing sap_cloud_sdk.destination client - Deployer only injects Destination Service binding — AI Core client_secret is never in the K8s Secret, only in BTP Destination Service - Combined with _clear_client_secret() (PR #257), the secret is removed from env after the first successful LiteLLM call Direct mode (neither set): existing behaviour unchanged, including transparent TLS (AICORE_TRANSPARENT_TLS). Adds 30 unit tests covering both new modes and all edge cases. AFSDK-4306 --- src/sap_cloud_sdk/aicore/__init__.py | 141 ++++++++++++--- src/sap_cloud_sdk/aicore/completion.py | 42 ++++- tests/aicore/unit/test_aicore.py | 231 +++++++++++++++++++++++++ tests/aicore/unit/test_completion.py | 134 ++++++++++++++ 4 files changed, 524 insertions(+), 24 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py index 1b11b658..63209957 100644 --- a/src/sap_cloud_sdk/aicore/__init__.py +++ b/src/sap_cloud_sdk/aicore/__init__.py @@ -13,7 +13,7 @@ from sap_cloud_sdk.core.telemetry.metrics_decorator import record_metrics from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation -from .completion import acompletion, completion, reload_aicore_credentials +from .completion import acompletion, completion, reload_aicore_credentials, _set_proxy_active from .filtering import ( AzureContentFilter, ContentFilter, @@ -35,6 +35,12 @@ # No client_secret or certificate material is required in the service binding. TRANSPARENT_TLS_ENV_VAR = "AICORE_TRANSPARENT_TLS" +# Option 3 — transparent proxy routing. +# Deployer injects these; agent code is identical in all environments. +_PROXY_URL_ENV = "AICORE_PROXY_URL" +_PROXY_VIRTUAL_KEY_ENV = "AICORE_PROXY_VIRTUAL_KEY" +_DESTINATION_NAME_ENV = "AICORE_DESTINATION_NAME" + def _is_transparent_tls() -> bool: """Return True when transparent TLS proxy mode is active.""" @@ -128,19 +134,27 @@ def _get_aicore_base_url(instance_name: str = "aicore-instance") -> str: def set_aicore_config(instance_name: str = "aicore-instance") -> None: """Load AI Core credentials and activate content filtering. - Loads secrets from files or environment variables and sets them as - process env vars so ``litellm`` picks them up. + Detects which routing mode is active based on environment variables: + + - ``AICORE_PROXY_URL`` set → **proxy mode**: routes all LiteLLM calls + through a LiteLLM proxy; ``sap/`` is aliased to + ``litellm_proxy/`` transparently. No AI Core credentials + are written to the process environment. + + - ``AICORE_DESTINATION_NAME`` set → **destination mode**: loads AI Core + credentials from a BTP Destination Service destination at startup. + The deployer only needs to inject Destination Service binding credentials; + the AI Core ``client_secret`` never needs to be in the K8s Secret. + Combined with the ``_clear_client_secret()`` mechanism (PR #257), + the secret is removed from env after the first LiteLLM call. - File mappings based on the Kubernetes secret structure: - clientid → AICORE_CLIENT_ID - clientsecret → AICORE_CLIENT_SECRET (skipped in transparent TLS mode) - url → AICORE_AUTH_URL - serviceurls (JSON with AI_API_URL) → AICORE_BASE_URL + - Neither set → **direct mode** (existing behaviour): credentials are + loaded from a mounted K8s secret volume or environment variables. + ``AICORE_TRANSPARENT_TLS=true`` suppresses ``client_secret`` and + relies on an mTLS sidecar. - When ``AICORE_TRANSPARENT_TLS=true`` is set, the infrastructure sidecar - adds the mTLS certificate on the SDK's behalf. In this mode the SDK omits - ``AICORE_CLIENT_SECRET`` from the environment — LiteLLM will use plain - HTTPS to the token endpoint and the sidecar will attach the certificate. + Agent code is identical in all three modes — the deployer controls + routing by choosing which env vars to inject. After credentials are loaded, content filtering is activated on every ``sap/*`` LiteLLM call at the configured thresholds (default: severity @@ -150,9 +164,100 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: to turn filtering off at runtime, or set ``AICORE_FILTER_ENABLED=false`` to keep it off entirely. """ + proxy_url = os.environ.get(_PROXY_URL_ENV, "") + destination_name = os.environ.get(_DESTINATION_NAME_ENV, "") + + if proxy_url: + _configure_proxy_mode(proxy_url) + elif destination_name: + _configure_destination_mode(destination_name) + else: + _configure_direct_mode(instance_name) + + set_filtering() + + +def _configure_proxy_mode(proxy_url: str) -> None: + """Configure LiteLLM to route calls through an external proxy. + + Sets ``litellm.api_base`` / ``litellm.api_key`` globally and activates + the ``sap/`` → ``litellm_proxy/`` model alias rewrite in the + completion wrappers. No AI Core credentials are written to env. + """ + import litellm as _litellm + + virtual_key = os.environ.get(_PROXY_VIRTUAL_KEY_ENV, "") + _litellm.api_base = proxy_url + if virtual_key: + _litellm.api_key = virtual_key + _set_proxy_active(True) + logger.info("AI Core proxy mode active — routing via %s", proxy_url) + + +def _configure_destination_mode(name: str) -> None: + """Load AI Core credentials from a BTP Destination Service destination. + + Calls the Destination Service at startup to resolve the named destination + and extracts ``clientId``, ``clientSecret``, ``tokenServiceURL``, and the + AI Core ``URL`` from the destination configuration properties. These are + written to the standard ``AICORE_*`` env vars so that LiteLLM can fetch + an OAuth token from XSUAA as usual. + + Security: The deployer does NOT need to inject ``AICORE_CLIENT_SECRET`` + directly — only Destination Service binding credentials are required in + the agent environment. The AI Core ``client_secret`` is fetched here + and removed from env after the first successful LiteLLM call + (PR #257 ``_clear_client_secret()`` mechanism). + + Raises ``RuntimeError`` if the destination is not found or does not + return ``clientId`` / ``clientSecret``. + """ + from sap_cloud_sdk.destination import create_client # lazy import + + client = create_client() + dest = client.get_destination(name) + + if dest is None: + raise RuntimeError( + f"AI Core destination '{name}' not found in Destination Service. " + "Check that the destination exists and the binding has access." + ) + + base_url = dest.url or "" + if base_url and not base_url.endswith("/v2"): + base_url = base_url.rstrip("/") + "/v2" + if base_url: + os.environ["AICORE_BASE_URL"] = base_url + + resource_group = dest.properties.get("resource_group", "default") + os.environ["AICORE_RESOURCE_GROUP"] = resource_group + + client_id = dest.properties.get("clientId", "") + client_secret = dest.properties.get("clientSecret", "") + token_service_url = dest.properties.get("tokenServiceURL", "") + + if not client_id or not client_secret: + raise RuntimeError( + f"Destination '{name}' did not return clientId/clientSecret. " + "Ensure the destination uses OAuth2ClientCredentials authentication " + "and the calling app has the Destination Service technical-user scope." + ) + + os.environ["AICORE_CLIENT_ID"] = client_id + os.environ["AICORE_CLIENT_SECRET"] = client_secret # cleared after first LiteLLM call + + if token_service_url: + if not token_service_url.endswith("/oauth/token"): + token_service_url = token_service_url.rstrip("/") + "/oauth/token" + os.environ["AICORE_AUTH_URL"] = token_service_url + + logger.info("AI Core destination mode active — credentials loaded from '%s'", name) + + +def _configure_direct_mode(instance_name: str) -> None: + """Load AI Core credentials directly from mounted secrets or env vars.""" transparent_tls = _is_transparent_tls() - # Load secrets client_id = _get_secret("AICORE_CLIENT_ID", "clientid", instance_name=instance_name) auth_url = _get_secret("AICORE_AUTH_URL", "url", instance_name=instance_name) base_url = _get_aicore_base_url(instance_name) @@ -160,14 +265,12 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: "AICORE_RESOURCE_GROUP", default="default", instance_name=instance_name ) - # Ensure AICORE_AUTH_URL has /oauth/token suffix if auth_url and not auth_url.endswith("/oauth/token"): auth_url = auth_url.rstrip("/") + "/oauth/token" if base_url and not base_url.endswith("/v2"): base_url = base_url.rstrip("/") + "/v2" - # Set environment variables for LiteLLM if client_id: os.environ["AICORE_CLIENT_ID"] = client_id if auth_url: @@ -178,7 +281,6 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: os.environ["AICORE_RESOURCE_GROUP"] = resource_group if transparent_tls: - # Remove any stale client_secret — the sidecar provides the mTLS cert. os.environ.pop("AICORE_CLIENT_SECRET", None) logger.info("AI Core transparent TLS mode active — client_secret not required") else: @@ -188,15 +290,8 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: if client_secret: os.environ["AICORE_CLIENT_SECRET"] = client_secret - # Log configuration completion (excluding sensitive information) logger.info("AI Core configuration has been set successfully") - # Activate content filtering for all sap/* LiteLLM model calls. - # AICORE_FILTER_ENABLED=false disables; AICORE_FILTER_* tune thresholds. - # Errors propagate — filtering misconfiguration should surface at startup - # rather than be swallowed silently. - set_filtering() - __all__ = [ "set_aicore_config", diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index a389ebf6..8f3e00d2 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -71,6 +71,26 @@ _secret_lock = threading.Lock() _secret_cleared = False +# Proxy mode state — set by _configure_proxy_mode() in __init__.py. +# When active, completion() rewrites sap/ → litellm_proxy/. +_proxy_lock = threading.Lock() +_proxy_active: bool = False + + +def _set_proxy_active(value: bool) -> None: + """Activate or deactivate proxy model aliasing (called by set_aicore_config).""" + global _proxy_active + with _proxy_lock: + _proxy_active = value + + +def _rewrite_model_for_proxy(kwargs: dict) -> dict: + """Rewrite sap/ to litellm_proxy/ when proxy mode is active.""" + model = kwargs.get("model", "") + if isinstance(model, str) and model.startswith("sap/"): + return {**kwargs, "model": "litellm_proxy/" + model[4:]} + return kwargs + def _clear_client_secret() -> None: """Remove AICORE_CLIENT_SECRET from env after LiteLLM has cached the token. @@ -136,13 +156,24 @@ def completion(*args: Any, **kwargs: Any) -> Any: On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert), reloads credentials from the mounted secret volume and retries once. + + When proxy mode is active (``AICORE_PROXY_URL`` set), rewrites + ``sap/`` to ``litellm_proxy/`` transparently. """ + with _proxy_lock: + proxy = _proxy_active + if proxy: + kwargs = _rewrite_model_for_proxy(kwargs) try: result = litellm.completion(*args, **kwargs) _clear_client_secret() return result except litellm.AuthenticationError: reload_aicore_credentials() + with _proxy_lock: + proxy = _proxy_active + if proxy: + kwargs = _rewrite_model_for_proxy(kwargs) result = litellm.completion(*args, **kwargs) _clear_client_secret() return result @@ -156,14 +187,23 @@ def completion(*args: Any, **kwargs: Any) -> Any: async def acompletion(*args: Any, **kwargs: Any) -> Any: """Async wrapper around :func:`litellm.acompletion`. - Same credential-minimisation and rotation semantics as :func:`completion`. + Same credential-minimisation, rotation, and proxy aliasing semantics as + :func:`completion`. """ + with _proxy_lock: + proxy = _proxy_active + if proxy: + kwargs = _rewrite_model_for_proxy(kwargs) try: result = await litellm.acompletion(*args, **kwargs) _clear_client_secret() return result except litellm.AuthenticationError: reload_aicore_credentials() + with _proxy_lock: + proxy = _proxy_active + if proxy: + kwargs = _rewrite_model_for_proxy(kwargs) result = await litellm.acompletion(*args, **kwargs) _clear_client_secret() return result diff --git a/tests/aicore/unit/test_aicore.py b/tests/aicore/unit/test_aicore.py index 50acb264..112b14f8 100644 --- a/tests/aicore/unit/test_aicore.py +++ b/tests/aicore/unit/test_aicore.py @@ -4,6 +4,7 @@ import os from unittest.mock import mock_open, patch +import pytest from sap_cloud_sdk.aicore import ( _get_aicore_base_url, @@ -835,3 +836,233 @@ def test_transparent_tls_does_not_call_get_secret_for_client_secret(self): called_names = [c.args[0] for c in mock_get_secret.call_args_list] assert "AICORE_CLIENT_SECRET" not in called_names + + +# --------------------------------------------------------------------------- +# Proxy mode — set_aicore_config() with AICORE_PROXY_URL +# --------------------------------------------------------------------------- + + +class TestSetAICoreConfigProxyMode: + """set_aicore_config() routes via proxy when AICORE_PROXY_URL is present.""" + + def _base_proxy_env(self, **extra): + return {"AICORE_PROXY_URL": "https://proxy.example.com", **extra} + + def test_proxy_mode_sets_litellm_api_base(self): + import litellm + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._set_proxy_active"), + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + assert litellm.api_base == "https://proxy.example.com" + litellm.api_base = None # cleanup + + def test_proxy_mode_sets_litellm_api_key_when_virtual_key_present(self): + import litellm + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._set_proxy_active"), + patch.dict( + "os.environ", + self._base_proxy_env(AICORE_PROXY_VIRTUAL_KEY="sk-virt-123"), + clear=True, + ), + ): + set_aicore_config() + assert litellm.api_key == "sk-virt-123" + litellm.api_key = None # cleanup + + def test_proxy_mode_activates_proxy_flag(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._set_proxy_active") as mock_set_proxy, + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + mock_set_proxy.assert_called_once_with(True) + + def test_proxy_mode_does_not_write_aicore_credentials(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._set_proxy_active"), + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + for var in ("AICORE_CLIENT_ID", "AICORE_CLIENT_SECRET", "AICORE_AUTH_URL"): + assert var not in os.environ, f"{var} must not be written in proxy mode" + + def test_proxy_mode_takes_precedence_over_destination(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._set_proxy_active") as mock_set_proxy, + patch("sap_cloud_sdk.aicore._configure_destination_mode") as mock_dest, + patch.dict( + "os.environ", + self._base_proxy_env(AICORE_DESTINATION_NAME="aicore"), + clear=True, + ), + ): + set_aicore_config() + mock_set_proxy.assert_called_once_with(True) + mock_dest.assert_not_called() + + def test_proxy_mode_still_calls_set_filtering(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering") as mock_filter, + patch("sap_cloud_sdk.aicore._set_proxy_active"), + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + mock_filter.assert_called_once() + + def test_direct_mode_used_when_neither_proxy_nor_destination_set(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._configure_direct_mode") as mock_direct, + patch.dict("os.environ", {}, clear=True), + ): + set_aicore_config() + mock_direct.assert_called_once() + + +# --------------------------------------------------------------------------- +# Destination mode — set_aicore_config() with AICORE_DESTINATION_NAME +# --------------------------------------------------------------------------- + + +class TestSetAICoreConfigDestinationMode: + """set_aicore_config() loads credentials from BTP Destination Service.""" + + def _mock_destination( + self, + url="https://api.ai.prod.example.com", + properties=None, + auth_tokens=None, + ): + from unittest.mock import MagicMock + dest = MagicMock() + dest.url = url + dest.properties = properties or { + "clientId": "sb-client-id", + "clientSecret": "client-secret-value", + "tokenServiceURL": "https://auth.example.com/oauth/token", + } + dest.auth_tokens = auth_tokens or [] + return dest + + def test_destination_mode_sets_base_url_with_v2_suffix(self): + dest = self._mock_destination(url="https://api.ai.prod.example.com") + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch( + "sap_cloud_sdk.destination.create_client" + ) as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_BASE_URL"] == "https://api.ai.prod.example.com/v2" + + def test_destination_mode_does_not_double_v2(self): + dest = self._mock_destination(url="https://api.ai.prod.example.com/v2") + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_BASE_URL"] == "https://api.ai.prod.example.com/v2" + + def test_destination_mode_sets_resource_group_from_properties(self): + dest = self._mock_destination( + properties={ + "clientId": "id", + "clientSecret": "sec", + "tokenServiceURL": "https://auth.example.com/oauth/token", + "resource_group": "production", + } + ) + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_RESOURCE_GROUP"] == "production" + + def test_destination_mode_defaults_resource_group_to_default(self): + dest = self._mock_destination() # no resource_group in properties + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_RESOURCE_GROUP"] == "default" + + def test_destination_mode_sets_client_credentials(self): + dest = self._mock_destination() + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_CLIENT_ID"] == "sb-client-id" + assert os.environ["AICORE_CLIENT_SECRET"] == "client-secret-value" + + def test_destination_mode_appends_oauth_token_suffix(self): + dest = self._mock_destination( + properties={ + "clientId": "id", + "clientSecret": "sec", + "tokenServiceURL": "https://auth.example.com", # no /oauth/token + } + ) + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_AUTH_URL"] == "https://auth.example.com/oauth/token" + + def test_destination_mode_raises_when_destination_not_found(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "missing"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = None + with pytest.raises(RuntimeError, match="not found"): + set_aicore_config() + + def test_destination_mode_raises_when_no_client_credentials(self): + dest = self._mock_destination(properties={"resource_group": "default"}) + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + with pytest.raises(RuntimeError, match="clientId/clientSecret"): + set_aicore_config() + + def test_destination_mode_still_calls_set_filtering(self): + dest = self._mock_destination() + with ( + patch("sap_cloud_sdk.aicore.set_filtering") as mock_filter, + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + mock_filter.assert_called_once() diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 30c89005..427bfe52 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -443,3 +443,137 @@ async def fake_acompletion(*args, **kwargs): ): with pytest.raises(litellm.AuthenticationError): asyncio.run(acompletion(model="sap/x", messages=[])) + + +# --------------------------------------------------------------------------- +# Proxy mode — model aliasing in completion() / acompletion() +# --------------------------------------------------------------------------- + + +class TestCompletionProxyModeAliasing: + """completion() rewrites sap/ → litellm_proxy/ when proxy is active.""" + + def setup_method(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(False) + + def teardown_method(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(False) + + def test_sap_model_rewritten_when_proxy_active(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(True) + sentinel = object() + captured = {} + + with patch( + "sap_cloud_sdk.aicore.completion.litellm.completion", + side_effect=lambda *a, **kw: captured.update(kw) or sentinel, + ): + result = completion(model="sap/gpt-4o", messages=[]) + + assert result is sentinel + assert captured["model"] == "litellm_proxy/gpt-4o" + + def test_model_unchanged_when_proxy_not_active(self): + sentinel = object() + captured = {} + + with patch( + "sap_cloud_sdk.aicore.completion.litellm.completion", + side_effect=lambda *a, **kw: captured.update(kw) or sentinel, + ): + result = completion(model="sap/gpt-4o", messages=[]) + + assert result is sentinel + assert captured["model"] == "sap/gpt-4o" + + def test_non_sap_model_unchanged_even_when_proxy_active(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(True) + sentinel = object() + captured = {} + + with patch( + "sap_cloud_sdk.aicore.completion.litellm.completion", + side_effect=lambda *a, **kw: captured.update(kw) or sentinel, + ): + result = completion(model="openai/gpt-4o", messages=[]) + + assert result is sentinel + assert captured["model"] == "openai/gpt-4o" + + def test_proxy_rewrite_on_auth_error_retry(self): + """Model aliasing is applied on both the initial call and the retry.""" + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(True) + + sentinel = object() + auth_err = litellm.AuthenticationError( + message="401", llm_provider="sap", model="sap/x" + ) + call_models = [] + + def fake_completion(*args, **kwargs): + call_models.append(kwargs.get("model")) + if len(call_models) == 1: + raise auth_err + return sentinel + + with ( + patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=fake_completion), + patch("sap_cloud_sdk.aicore.set_aicore_config"), + ): + result = completion(model="sap/gpt-4o", messages=[]) + + assert result is sentinel + assert call_models == ["litellm_proxy/gpt-4o", "litellm_proxy/gpt-4o"] + + +class TestACompletionProxyModeAliasing: + """acompletion() proxy aliasing — async path.""" + + def setup_method(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(False) + + def teardown_method(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(False) + + def test_sap_model_rewritten_when_proxy_active(self): + from sap_cloud_sdk.aicore.completion import _set_proxy_active + _set_proxy_active(True) + sentinel = object() + captured = {} + + async def fake_acompletion(*args, **kwargs): + captured.update(kwargs) + return sentinel + + with patch( + "sap_cloud_sdk.aicore.completion.litellm.acompletion", + side_effect=fake_acompletion, + ): + result = asyncio.run(acompletion(model="sap/gpt-4o", messages=[])) + + assert result is sentinel + assert captured["model"] == "litellm_proxy/gpt-4o" + + def test_model_unchanged_when_proxy_not_active(self): + sentinel = object() + captured = {} + + async def fake_acompletion(*args, **kwargs): + captured.update(kwargs) + return sentinel + + with patch( + "sap_cloud_sdk.aicore.completion.litellm.acompletion", + side_effect=fake_acompletion, + ): + result = asyncio.run(acompletion(model="sap/gpt-4o", messages=[])) + + assert result is sentinel + assert captured["model"] == "sap/gpt-4o"