From 5ac545edd385eb9c9cd7e4281da0a6e5b10ca61a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Furga=C5=82a?= <83299832+00200200@users.noreply.github.com> Date: Tue, 29 Sep 2026 12:21:45 +0200 Subject: [PATCH] test(proxy): pin Anthropic SSE CRLF and multi-block guardrail rewrites Regression coverage for the Greptile P1s so CRLF-framed streams still hit post-call rewriting and each content-block index keeps its own text --- .../guardrail_translation/__init__.py | 0 .../guardrail_translation/test_handler.py | 145 ++++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 tests/unit/llms/anthropic/passthrough/guardrail_translation/__init__.py create mode 100644 tests/unit/llms/anthropic/passthrough/guardrail_translation/test_handler.py diff --git a/tests/unit/llms/anthropic/passthrough/guardrail_translation/__init__.py b/tests/unit/llms/anthropic/passthrough/guardrail_translation/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/llms/anthropic/passthrough/guardrail_translation/test_handler.py b/tests/unit/llms/anthropic/passthrough/guardrail_translation/test_handler.py new file mode 100644 index 00000000000..7fc896b9c4f --- /dev/null +++ b/tests/unit/llms/anthropic/passthrough/guardrail_translation/test_handler.py @@ -0,0 +1,145 @@ +""" +Regression tests for AnthropicPassthroughGuardrailHandler SSE rewriting. + +Pins the two Greptile P1s from PR #42585: +- CRLF-framed streams must still reach post-call rewriting +- Multi-index text blocks must keep their own rewrites (not merge into index 0) +""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock + +import pytest + +from litellm.llms.anthropic.passthrough.guardrail_translation.handler import ( + AnthropicPassthroughGuardrailHandler, + _parse_sse_blocks, +) + + +def _text_delta_frame(index: int, text: str, sep: bytes = b"\n\n", line_end: bytes | None = None) -> bytes: + if line_end is None: + line_end = sep[: len(sep) // 2] or b"\n" + payload = { + "type": "content_block_delta", + "index": index, + "delta": {"type": "text_delta", "text": text}, + } + return ( + b"event: content_block_delta" + + line_end + + b"data: " + + json.dumps(payload, separators=(",", ":")).encode() + + sep + ) + + +def _message_stop_frame(sep: bytes = b"\n\n", line_end: bytes | None = None) -> bytes: + if line_end is None: + line_end = sep[: len(sep) // 2] or b"\n" + return b"event: message_stop" + line_end + b'data: {"type":"message_stop"}' + sep + + +def _frame_payloads(body: bytes) -> list[dict]: + payloads: list[dict] = [] + for block in _parse_sse_blocks(body): + for line in block.decode().splitlines(): + if line.startswith("data:"): + payloads.append(json.loads(line[5:].strip())) + break + return payloads + + +class TestParseSseBlocks: + def test_splits_lf_crlf_and_cr_blank_lines(self): + body = ( + b"event: a\ndata: 1\n\n" + b"event: b\r\ndata: 2\r\n\r\n" + b"event: c\rdata: 3\r\r" + ) + blocks = _parse_sse_blocks(body) + assert len(blocks) == 3 + assert blocks[0].endswith(b"\n\n") + assert blocks[1].endswith(b"\r\n\r\n") + assert blocks[2].endswith(b"\r\r") + + +class TestDeAnonymizeEventStream: + def _proxy(self, mock_hook) -> MagicMock: + proxy_logging_obj = MagicMock() + proxy_logging_obj.post_call_success_hook = mock_hook + return proxy_logging_obj + + @pytest.mark.asyncio + async def test_crlf_framed_stream_still_invokes_guardrail(self): + """P1: CRLF frames must not merge so message_stop wins and deltas skip rewriting.""" + sse = _text_delta_frame(0, "", sep=b"\r\n\r\n") + _message_stop_frame(sep=b"\r\n\r\n") + hook_calls: list[dict] = [] + + async def mock_hook(data, user_api_key_dict, response): + hook_calls.append(response) + response = dict(response) + response["content"] = [{"type": "text", "text": "Alice"}] + return response + + # Precondition: a naive LF-only split would leave one merged block. + assert len(sse.split(b"\n\n")) == 1 + + result = await AnthropicPassthroughGuardrailHandler.de_anonymize_event_stream( + body_bytes=sse, + proxy_logging_obj=self._proxy(mock_hook), + user_api_key_dict=MagicMock(), + data={}, + ) + + assert len(hook_calls) == 1 + assert hook_calls[0]["content"][0]["text"] == "" + assert b"" not in result + assert b"Alice" in result + assert result.endswith(b'data: {"type":"message_stop"}\r\n\r\n') + + @pytest.mark.asyncio + async def test_multi_index_text_blocks_keep_their_own_rewrites(self): + """P1: rewrites must stay on their content-block index around tool/thinking events.""" + tool_delta = ( + b"event: content_block_delta\n" + b'data: {"type":"content_block_delta","index":1,' + b'"delta":{"type":"input_json_delta","partial_json":"{}"}}\n\n' + ) + sse = ( + _text_delta_frame(0, " said") + + tool_delta + + _text_delta_frame(2, "bye ") + ) + seen: dict[str, list[str]] = {} + + async def mock_hook(data, user_api_key_dict, response): + seen["texts"] = [block["text"] for block in response["content"]] + response = dict(response) + response["content"] = [ + {"type": "text", "text": "Alice said"}, + {"type": "text", "text": "bye Bob"}, + ] + return response + + result = await AnthropicPassthroughGuardrailHandler.de_anonymize_event_stream( + body_bytes=sse, + proxy_logging_obj=self._proxy(mock_hook), + user_api_key_dict=MagicMock(), + data={}, + ) + + assert seen["texts"] == [" said", "bye "] + payloads = _frame_payloads(result) + assert [(p["index"], p["delta"].get("text")) for p in payloads] == [ + (0, "Alice said"), + (0, ""), + (1, None), + (2, "bye Bob"), + ] + # Tool frame between text blocks must stay untouched and in order. + assert payloads[2]["delta"]["type"] == "input_json_delta" + assert payloads[2]["delta"]["partial_json"] == "{}"