scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-11 03:38:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 84

test(guardrails): cover an observe-only stream that no handler understands

Browse source
This commit is contained in:
Caduri Katzav 2026-10-03 17:04:11 +03:00
parent de065327e9
commit de597a5228
1 changed files with 14 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
tests/unit/proxy/guardrails/guardrail_hooks/unified_guardrail/test_unified_guardrail.py
View file

@ -53,7 +53,7 @@ async def _upstream(
def _guarded_stream(
observer: _Observer, upstream: AsyncIterator[ModelResponseStream], *, route: str | None = "/chat/completions"
observer: _Observer, upstream: AsyncIterator[object], *, route: str | None = "/chat/completions"
) -> AsyncIterator[object]:
return UnifiedLLMGuardrails().async_post_call_streaming_iterator_hook(
user_api_key_dict=UserAPIKeyAuth(api_key="test", request_route=route),
@ -96,6 +96,19 @@ async def test_a_failing_observer_never_breaks_the_stream(warning_messages: Call
]
async def test_a_stream_no_handler_understands_is_forwarded_unobserved(
warning_messages: Callable[[str], list[str]],
) -> None:
observer: Final = _Observer()
async def raw_bytes() -> AsyncIterator[bytes]:
yield b"data: raw\n\n"
sent: Final = [chunk async for chunk in _guarded_stream(observer, raw_bytes(), route=None)]
assert (sent, observer.observed, warning_messages("observe-only stream check")) == ([b"data: raw\n\n"], [], [])
async def test_an_empty_stream_is_not_observed(warning_messages: Callable[[str], list[str]]) -> None:
observer: Final = _Observer()