diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index 70dc6266..d97dc633 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -5,15 +5,30 @@ calls ``raise_for_status()`` on 4xx responses **before** our patched ``transform_response`` runs, so the 400 surfaces in user code as a ``litellm.APIConnectionError`` whose ``str(exc)`` contains the JSON body of -the rejection. Output-filter rejections (HTTP 200 with -``finish_reason == "content_filter"``) go through ``transform_response`` -and surface as :class:`ContentFilteredError`. +the rejection. Output-filter rejections in buffered (non-streaming) mode +(HTTP 200 with ``finish_reason == "content_filter"``) go through +``transform_response`` and surface as :class:`ContentFilteredError`. That asymmetry would force callers to catch two exception types — these wrappers fix it by catching the wrapped exception inside the SDK and re-raising as :class:`ContentFilteredError` so callers can rely on a single exception type for "filter blocked you." +Streaming + output filtering +---------------------------- +When output filtering is active and ``stream=True``, the v2 API streams +chunks normally and signals a filter rejection via an in-stream error event +(no ``[DONE]``). LiteLLM's ``SAPStreamIterator`` surfaces this as a +``ValueError`` raised from inside the iterator. These wrappers detect when +output filtering is active and return a *filtering wrapper* around the +iterator that catches that ``ValueError`` and re-raises it as +:class:`ContentFilteredError`. + +Agent frameworks (``ChatLiteLLM``, LangGraph, etc.) that call +``litellm.completion`` directly — bypassing this wrapper — will see the raw +``ValueError`` from the iterator. They must use +:func:`sap_cloud_sdk.aicore.completion` to get the typed :class:`ContentFilteredError`. + Credential rotation handling ---------------------------- When a credential (client_secret) is rotated while the pod is running, @@ -40,9 +55,9 @@ ... These are intentionally thin — every other keyword argument is forwarded -verbatim to ``litellm.completion`` / ``litellm.acompletion``, including -``stream=True`` (the streaming iterator is returned unchanged). No -telemetry is recorded here; the wrappers fire on every LLM call and a +verbatim to ``litellm.completion`` / ``litellm.acompletion``. + +No telemetry is recorded here; the wrappers fire on every LLM call and a counter at this level would be both noisy and uninformative about adoption of the SDK. """ @@ -50,6 +65,7 @@ from __future__ import annotations import logging +from collections.abc import AsyncIterator, Iterator from typing import Any import litellm @@ -58,6 +74,8 @@ from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation from .filtering.filters import _parse_input_filter_error +from .filtering import _patch as _filtering_patch +from .filtering._patch import _parse_output_filter_stream_error logger = logging.getLogger(__name__) @@ -84,6 +102,59 @@ def _maybe_translate_filter_error(exc: BaseException) -> BaseException: return blocked if blocked is not None else exc +def _output_filtering_active() -> bool: + cfg = _filtering_patch._active_cfg + return cfg is not None and cfg.output_filtering is not None + + +class _FilteringStreamWrapper: + """Sync iterator wrapper that converts in-stream output-filter errors.""" + + def __init__(self, stream: Iterator) -> None: + self._stream = stream + + def __iter__(self) -> Iterator: + return self + + def __next__(self) -> Any: + try: + return next(self._stream) + except StopIteration: + raise + except ValueError as exc: + translated = _parse_output_filter_stream_error(exc) + if translated is not None: + raise translated from exc + raise + + def __getattr__(self, name: str) -> Any: + return getattr(self._stream, name) + + +class _AsyncFilteringStreamWrapper: + """Async iterator wrapper that converts in-stream output-filter errors.""" + + def __init__(self, stream: AsyncIterator) -> None: + self._stream = stream + + def __aiter__(self) -> AsyncIterator: + return self + + async def __anext__(self) -> Any: + try: + return await self._stream.__anext__() + except StopAsyncIteration: + raise + except (ValueError, Exception) as exc: + translated = _parse_output_filter_stream_error(exc) + if translated is not None: + raise translated from exc + raise + + def __getattr__(self, name: str) -> Any: + return getattr(self._stream, name) + + def completion(*args: Any, **kwargs: Any) -> Any: """Wrapper around :func:`litellm.completion` that normalises filter errors and handles credential rotation transparently. @@ -91,38 +162,53 @@ 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 output filtering is active and ``stream=True``, wraps the returned + iterator so that in-stream output-filter rejections surface as + :class:`ContentFilteredError` rather than a raw ``ValueError``. + Model strings (e.g. ``sap/``) are passed verbatim to LiteLLM in all routing modes — proxy routing is handled by ``litellm.api_base`` configured in :func:`set_aicore_config`, not by rewriting the model name. """ + is_streaming = kwargs.get("stream", False) try: - return litellm.completion(*args, **kwargs) + result = litellm.completion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - return litellm.completion(*args, **kwargs) + result = litellm.completion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc + if is_streaming and _output_filtering_active(): + return _FilteringStreamWrapper(result) + return result + async def acompletion(*args: Any, **kwargs: Any) -> Any: """Async wrapper around :func:`litellm.acompletion`. - Same credential-rotation semantics as :func:`completion`. + Same credential-rotation and output-filter stream-wrapping semantics as + :func:`completion`. Model strings are passed verbatim to LiteLLM in all routing modes. """ + is_streaming = kwargs.get("stream", False) try: - return await litellm.acompletion(*args, **kwargs) + result = await litellm.acompletion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - return await litellm.acompletion(*args, **kwargs) + result = await litellm.acompletion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc + if is_streaming and _output_filtering_active(): + return _AsyncFilteringStreamWrapper(result) + return result + __all__ = ["completion", "acompletion"] diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index c3c1ab1d..bd08e210 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -4,14 +4,14 @@ - Injects ``modules.filtering`` into every v2 completion request body via ``transform_request``. -- Detects filter rejections in responses and raises ``ContentFilteredError`` - via ``transform_response``. +- Detects filter rejections in buffered responses and raises + ``ContentFilteredError`` via ``transform_response``. The patch is applied by :func:`_install` (called by ``set_filtering`` / ``disable_filtering`` in :mod:`.filters`) and is idempotent — calling it multiple times with the same config is safe. -Two filter-rejection shapes (from the v2 API) are handled: +Two filter-rejection shapes (from the v2 API) are handled in buffered mode: - Input rejection: HTTP 4xx with ``error.location`` startswith ``"Filtering Module - Input Filter"``. @@ -26,10 +26,30 @@ back into :class:`ContentFilteredError` before it reaches caller code; the :func:`_parse_input_filter_error` helper in :mod:`.filters` does the JSON parsing. + +Streaming + output filtering +---------------------------- +The SAP AI Core Orchestration v2 API supports output filtering with streaming. +The server filters each chunk and signals a rejection via an in-stream error +event (no ``[DONE]``). LiteLLM's ``SAPStreamIterator`` surfaces this error +event as a ``ValueError(json.dumps(event_obj))`` from inside the iterator. + +:func:`sap_cloud_sdk.aicore.completion` / :func:`acompletion` wrap the +returned stream iterator when output filtering is active, catching that +``ValueError`` and translating it to :class:`ContentFilteredError`. This gives +callers a single, typed exception for filter rejections in both streaming and +non-streaming modes. + +Note: agent frameworks (``ChatLiteLLM``, LangGraph, etc.) that call +``litellm.completion`` directly — bypassing the SDK's completion wrapper — +will see the raw ``ValueError`` from the iterator. They must either use +:func:`sap_cloud_sdk.aicore.completion` to get the translated error, or +handle the ``ValueError`` themselves. """ from __future__ import annotations +import json import logging from typing import Any @@ -75,8 +95,8 @@ def transform_request( modules = body["config"]["modules"] if isinstance(modules, list): - # Fallback mode (list of configs) — inject into primary config only. - modules[0]["filtering"] = filtering_dict + for m in modules: + m["filtering"] = filtering_dict else: modules["filtering"] = filtering_dict @@ -172,3 +192,46 @@ def _install(cfg: Any) -> None: # cfg: ContentFiltering | None else: litellm.GenAIHubOrchestrationConfig = FilteringOrchestrationConfig logger.info("content filtering active (FilteringOrchestrationConfig)") + + +def _parse_output_filter_stream_error( + exc: BaseException, +) -> ContentFilteredError | None: + """Try to parse an in-stream output-filter rejection from a ValueError. + + LiteLLM's ``SAPStreamIterator`` raises ``ValueError(json.dumps(event_obj))`` + when the server sends an error event inside the stream. The v2 error shape + for an output-filter rejection is:: + + {"error": {"request_id": "...", "code": 500, + "message": "...", + "location": "Filtering Module - Output Filter", + "intermediate_results": {"output_filtering": {"data": {...}}}}} + + Returns a :class:`ContentFilteredError` when the pattern matches, otherwise + ``None`` so the caller can re-raise the original exception unchanged. + """ + if not isinstance(exc, (ValueError, Exception)): + return None + msg = str(exc) + brace = msg.find("{") + if brace == -1: + return None + try: + payload = json.loads(msg[brace:]) + except (ValueError, TypeError): + return None + err = payload.get("error", {}) + if not isinstance(err, dict): + return None + location = err.get("location") or "" + if not location.startswith("Filtering Module - Output Filter"): + return None + data = ( + err.get("intermediate_results", {}).get("output_filtering", {}).get("data", {}) + ) + return ContentFilteredError( + direction="output", + details=data, + request_id=err.get("request_id"), + ) diff --git a/src/sap_cloud_sdk/aicore/filtering/exceptions.py b/src/sap_cloud_sdk/aicore/filtering/exceptions.py index 22430561..51673d28 100644 --- a/src/sap_cloud_sdk/aicore/filtering/exceptions.py +++ b/src/sap_cloud_sdk/aicore/filtering/exceptions.py @@ -20,8 +20,11 @@ class ContentFilteredError(OrchestrationError): Attributes: direction: ``"input"`` if the user prompt was rejected (HTTP 4xx, ``error.location`` startswith ``"Filtering Module - Input Filter"``), - or ``"output"`` if the model response was rejected (HTTP 200, - ``finish_reason == "content_filter"``). + or ``"output"`` if the model response was rejected. In non-streaming + responses this is signalled by ``finish_reason == "content_filter"`` + (HTTP 200). In streaming responses, the server sends an in-stream + error event which :func:`sap_cloud_sdk.aicore.completion` translates + into this exception from inside the iterator. details: Severity scalars from ``intermediate_results.{input,output}_filtering.data``. Safe to log; does NOT include raw prompt or completion content. request_id: ``error.request_id`` (input) or top-level ``request_id`` (output). diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 095a9204..efc95d38 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -360,10 +360,12 @@ except ContentFilteredError as e: return "Your request was blocked by content safety policy." ``` -The wrapper forwards every argument verbatim to `litellm.completion` -(including `stream=True`), and only intercepts the wrapped -`APIConnectionError` shape that LiteLLM produces for input-filter -rejections. All other exceptions surface unchanged. +The wrapper forwards every argument verbatim to `litellm.completion`, +including `stream=True`. Both input and output filtering work with streaming — +filter rejections surface as `ContentFilteredError` regardless. See the +[Streaming Responses](#streaming-responses) section for details. + +All other exceptions surface unchanged. `ContentFilteredError` exposes three attributes — `direction`, `details`, `request_id`. The `details` field contains severity scalars from the server, @@ -444,21 +446,33 @@ print(response.choices[0].message.content) ### Streaming Responses +`stream=True` is fully supported, including when output filtering is active. +When the server rejects a streamed response via an in-stream filter event, the +SDK raises `ContentFilteredError(direction="output")` from inside the iterator +— the same exception type used for buffered (non-streaming) rejections. + +**Important:** this translation only applies when you use +`sap_cloud_sdk.aicore.completion` / `acompletion`. If agent frameworks such as +`ChatLiteLLM` or LangGraph call `litellm.completion` directly, mid-stream +output-filter rejections surface as a raw `ValueError` from the iterator. + ```python -from sap_cloud_sdk.aicore import set_aicore_config -from litellm import completion +from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config set_aicore_config() -response = completion( - model="sap/gpt-4", - messages=[{"role": "user", "content": "Tell me a story"}], - stream=True, -) +try: + response = completion( + model="sap/gpt-4", + messages=[{"role": "user", "content": "Tell me a story"}], + stream=True, + ) -for chunk in response: - if chunk.choices[0].delta.content: - print(chunk.choices[0].delta.content, end="") + for chunk in response: + if chunk.choices[0].delta.content: + print(chunk.choices[0].delta.content, end="") +except ContentFilteredError as e: + print(f"Blocked ({e.direction}): {e.details}") ``` ### Using Different Models diff --git a/tests/aicore/filtering/unit/test_filtering.py b/tests/aicore/filtering/unit/test_filtering.py index 0f0fe6c9..6cb16024 100644 --- a/tests/aicore/filtering/unit/test_filtering.py +++ b/tests/aicore/filtering/unit/test_filtering.py @@ -133,3 +133,60 @@ def test_disable_when_never_enabled(self, monkeypatch): disable_filtering() assert litellm.GenAIHubOrchestrationConfig is _ORIGINAL_CONFIG assert _filters_mod._active_cfg is None + + +class TestTransformRequestMultiModule: + """Regression tests: filtering must be injected into every fallback module, not just modules[0].""" + + def _make_cfg(self) -> ContentFiltering: + return ContentFiltering( + input_filtering=InputFiltering(filters=[AzureContentFilter()]) + ) + + def _call_transform_request(self, modules_value): + """Call the filtering injection logic directly, bypassing the LiteLLM base class.""" + from unittest.mock import patch + + body = {"config": {"modules": modules_value}} + instance = FilteringOrchestrationConfig() + + with patch( + "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", + return_value=body, + ): + return instance.transform_request( + model="sap/x", + messages=[], + optional_params={}, + litellm_params={}, + headers={}, + ) + + def test_single_module_dict_gets_filtering(self, monkeypatch): + _clear_aicore_env(monkeypatch) + set_filtering(self._make_cfg()) + modules = {"model_id": "primary-llm"} + result = self._call_transform_request(modules) + assert "filtering" in result["config"]["modules"] + + def test_multi_module_list_all_get_filtering(self, monkeypatch): + _clear_aicore_env(monkeypatch) + set_filtering(self._make_cfg()) + modules = [ + {"model_id": "primary-llm"}, + {"model_id": "fallback-llm-1"}, + {"model_id": "fallback-llm-2"}, + ] + result = self._call_transform_request(modules) + for i, m in enumerate(result["config"]["modules"]): + assert "filtering" in m, f"modules[{i}] missing filtering block" + + def test_multi_module_filtering_dict_is_identical_across_modules(self, monkeypatch): + _clear_aicore_env(monkeypatch) + set_filtering(self._make_cfg()) + modules = [{"model_id": "a"}, {"model_id": "b"}] + result = self._call_transform_request(modules) + assert ( + result["config"]["modules"][0]["filtering"] + == result["config"]["modules"][1]["filtering"] + ) diff --git a/tests/aicore/filtering/unit/test_patch.py b/tests/aicore/filtering/unit/test_patch.py index 1d6555d0..34140aba 100644 --- a/tests/aicore/filtering/unit/test_patch.py +++ b/tests/aicore/filtering/unit/test_patch.py @@ -1,4 +1,4 @@ -"""Unit tests for the LiteLLM patch (FilteringOrchestrationConfig, _install, _parse_input_filter_error).""" +"""Unit tests for the LiteLLM patch (FilteringOrchestrationConfig, _install, _parse_input_filter_error, _parse_output_filter_stream_error).""" import json import pytest @@ -17,6 +17,7 @@ FilteringOrchestrationConfig, _install, _ORIGINAL_CONFIG, + _parse_output_filter_stream_error, ) from sap_cloud_sdk.aicore.filtering.config import load_from_env from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError @@ -284,3 +285,59 @@ def test_install_idempotent(self): _install(cfg) # second call — no error assert litellm.GenAIHubOrchestrationConfig is FilteringOrchestrationConfig _install(None) + + +# --------------------------------------------------------------------------- +# _parse_output_filter_stream_error tests +# --------------------------------------------------------------------------- + + +OUTPUT_FILTER_STREAM_ERROR = { + "error": { + "request_id": "req-stream-output", + "code": 500, + "message": "Content filtered by output filter.", + "location": "Filtering Module - Output Filter", + "intermediate_results": { + "output_filtering": { + "data": { + "azure_content_safety": {"Hate": 0, "Violence": 6} + } + } + }, + } +} + + +class TestParseOutputFilterStreamError: + def _make_exc(self, payload: dict) -> ValueError: + return ValueError(json.dumps(payload)) + + def test_extracts_output_filter(self): + exc = self._make_exc(OUTPUT_FILTER_STREAM_ERROR) + blocked = _parse_output_filter_stream_error(exc) + assert blocked is not None + assert blocked.direction == "output" + assert blocked.request_id == "req-stream-output" + assert blocked.details["azure_content_safety"]["Violence"] == 6 + + def test_returns_none_for_non_output_location(self): + body = {"error": {"location": "Filtering Module - Input Filter", "message": "x"}} + assert _parse_output_filter_stream_error(self._make_exc(body)) is None + + def test_returns_none_for_non_filter_error(self): + body = {"error": {"location": "Model Module", "message": "model error"}} + assert _parse_output_filter_stream_error(self._make_exc(body)) is None + + def test_returns_none_for_no_brace(self): + assert _parse_output_filter_stream_error(ValueError("network error")) is None + + def test_returns_none_for_malformed_json(self): + assert _parse_output_filter_stream_error(ValueError("{not valid json}")) is None + + def test_returns_none_for_non_exception(self): + assert _parse_output_filter_stream_error(KeyboardInterrupt()) is None + + def test_stream_with_no_output_filter_location_passes_through(self): + body = {"code": 500, "message": "some server error"} + assert _parse_output_filter_stream_error(self._make_exc(body)) is None diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 8238e922..fc9cadf1 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -5,7 +5,8 @@ whether the rejection happened on input (litellm wraps it in ``APIConnectionError`` because the 4xx triggers ``raise_for_status()`` before our transport patch runs) or output (already raised by the -transport patch as :class:`ContentFilteredError`). Test focus: +transport patch as :class:`ContentFilteredError` in buffered mode, or +raised from the stream iterator in streaming mode). Test focus: - Successful calls pass through verbatim. - An input-filter-shaped ``APIConnectionError`` is re-raised as @@ -17,13 +18,16 @@ - ``acompletion`` exhibits the same behaviour on the async path. - On ``AuthenticationError``, credentials are reloaded and the call is retried once (credential rotation without pod restart). +- With output filtering active and ``stream=True``, the returned iterator + catches in-stream ``ValueError`` filter errors and raises + :class:`ContentFilteredError`. """ from __future__ import annotations import asyncio import json -from unittest.mock import MagicMock, call, patch +from unittest.mock import MagicMock, patch import litellm import pytest @@ -275,6 +279,265 @@ def test_non_auth_error_does_not_trigger_reload(self): mock_reload.assert_not_called() +# --------------------------------------------------------------------------- +# Streaming guard — stream=True + output filtering now supported +# --------------------------------------------------------------------------- + + +class TestStreamingGuard: + """stream=True is now allowed with output filtering active. + The returned iterator wraps filter errors into ContentFilteredError. + Input-only filtering: stream=True passes through unmodified. + """ + + @pytest.fixture(autouse=True) + def restore_filtering(self): + from sap_cloud_sdk.aicore.filtering._patch import _install + _install(None) + yield + _install(None) + + def _activate_input_only(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, InputFiltering, set_filtering, + ) + set_filtering(ContentFiltering( + input_filtering=InputFiltering(filters=[AzureContentFilter()]) + )) + + def _activate_output(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, OutputFiltering, set_filtering, + ) + set_filtering(ContentFiltering( + output_filtering=OutputFiltering(filters=[AzureContentFilter()]) + )) + + def _activate_input_and_output(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, InputFiltering, OutputFiltering, + set_filtering, + ) + set_filtering(ContentFiltering( + input_filtering=InputFiltering(filters=[AzureContentFilter()]), + output_filtering=OutputFiltering(filters=[AzureContentFilter()]), + )) + + # --- no filtering or input-only: stream=True passes through -------------------- + + def test_no_filtering_stream_allowed(self): + sentinel = object() + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel): + result = completion(model="sap/x", messages=[], stream=True) + assert result is sentinel + + def test_input_only_filtering_stream_allowed(self): + self._activate_input_only() + sentinel = object() + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel): + result = completion(model="sap/x", messages=[], stream=True) + assert result is sentinel + + def test_acompletion_input_only_stream_allowed(self): + sentinel = object() + + async def _fake_acompletion(*_, **__): + return sentinel + + with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): + result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + assert result is sentinel + + # --- output filtering + stream=True: iterator returned (not blocked upfront) -- + + def test_output_filter_stream_returns_wrapper(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + self._activate_output() + fake_stream = iter([]) + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=fake_stream): + result = completion(model="sap/x", messages=[], stream=True) + assert isinstance(result, _FilteringStreamWrapper) + + def test_input_and_output_filter_stream_returns_wrapper(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + self._activate_input_and_output() + fake_stream = iter([]) + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=fake_stream): + result = completion(model="sap/x", messages=[], stream=True) + assert isinstance(result, _FilteringStreamWrapper) + + def test_output_filter_no_stream_not_wrapped(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + self._activate_output() + sentinel = object() + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel): + result = completion(model="sap/x", messages=[]) + assert result is sentinel + assert not isinstance(result, _FilteringStreamWrapper) + + def test_acompletion_output_filter_stream_returns_async_wrapper(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + self._activate_output() + + async def _fake_stream_aiter(): + return + yield # make it an async generator + + async def _fake_acompletion(*_, **__): + return _fake_stream_aiter() + + with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): + result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + assert isinstance(result, _AsyncFilteringStreamWrapper) + + +# --------------------------------------------------------------------------- +# Stream output-filter iterator — ValueError from stream → ContentFilteredError +# --------------------------------------------------------------------------- + + +def _output_filter_stream_error_json() -> str: + """In-stream error event JSON for an output-filter rejection (v2 shape).""" + return json.dumps({ + "error": { + "request_id": "req-stream-out", + "code": 500, + "message": "Content filtered by output filter.", + "location": "Filtering Module - Output Filter", + "intermediate_results": { + "output_filtering": { + "data": { + "azure_content_safety": {"Violence": 6, "Hate": 0} + } + } + }, + } + }) + + +class TestStreamOutputFilterIterator: + """The filtering stream wrapper catches output-filter ValueError from the iterator.""" + + @pytest.fixture(autouse=True) + def restore_filtering(self): + from sap_cloud_sdk.aicore.filtering._patch import _install + _install(None) + yield + _install(None) + + def _activate_output(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, OutputFiltering, set_filtering, + ) + set_filtering(ContentFiltering( + output_filtering=OutputFiltering(filters=[AzureContentFilter()]) + )) + + # --- sync wrapper --- + + def test_sync_wrapper_raises_content_filtered_on_output_filter_error(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + def _bad_stream(): + raise ValueError(_output_filter_stream_error_json()) + yield # make it a generator + + wrapper = _FilteringStreamWrapper(_bad_stream()) + with pytest.raises(ContentFilteredError) as ei: + next(wrapper) + assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-out" + assert ei.value.details["azure_content_safety"]["Violence"] == 6 + + def test_sync_wrapper_yields_normal_chunks(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + chunks = ["chunk1", "chunk2"] + wrapper = _FilteringStreamWrapper(iter(chunks)) + assert list(wrapper) == chunks + + def test_sync_wrapper_passes_through_non_filter_value_error(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + def _bad_stream(): + raise ValueError("some unrelated error") + yield + + wrapper = _FilteringStreamWrapper(_bad_stream()) + with pytest.raises(ValueError, match="some unrelated error"): + next(wrapper) + + def test_completion_stream_raises_content_filtered_on_iteration(self): + """End-to-end: completion() + output filtering + stream=True → ContentFilteredError on next().""" + self._activate_output() + + def _bad_stream(): + raise ValueError(_output_filter_stream_error_json()) + yield + + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=_bad_stream()): + stream = completion(model="sap/x", messages=[], stream=True) + + with pytest.raises(ContentFilteredError) as ei: + next(stream) + assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-out" + + # --- async wrapper --- + + def test_async_wrapper_raises_content_filtered_on_output_filter_error(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + async def _bad_async_stream(): + raise ValueError(_output_filter_stream_error_json()) + yield # make it an async generator + + async def _run(): + wrapper = _AsyncFilteringStreamWrapper(_bad_async_stream()) + async for _ in wrapper: + pass + + with pytest.raises(ContentFilteredError) as ei: + asyncio.run(_run()) + assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-out" + + def test_async_wrapper_yields_normal_chunks(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + async def _good_stream(): + for item in ["a", "b", "c"]: + yield item + + async def _run(): + result = [] + wrapper = _AsyncFilteringStreamWrapper(_good_stream()) + async for chunk in wrapper: + result.append(chunk) + return result + + assert asyncio.run(_run()) == ["a", "b", "c"] + + def test_async_wrapper_passes_through_non_filter_value_error(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + async def _bad_async_stream(): + raise ValueError("unrelated async error") + yield + + async def _run(): + wrapper = _AsyncFilteringStreamWrapper(_bad_async_stream()) + async for _ in wrapper: + pass + + with pytest.raises(ValueError, match="unrelated async error"): + asyncio.run(_run()) + + # --------------------------------------------------------------------------- # Reactive reload on AuthenticationError — async # ---------------------------------------------------------------------------