From cfc7d488cdd6b69df77f0788e74b82fb0d1e8f36 Mon Sep 17 00:00:00 2001 From: Daniel Phang Date: Sun, 27 Sep 2026 09:53:14 -0700 Subject: [PATCH] test(responses): complete recovery regression annotations --- ...itellm_responses_transformation_handler.py | 21 +++++++------- ...responses_transformation_transformation.py | 28 ++++++++++--------- 2 files changed, 26 insertions(+), 23 deletions(-) diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py index 95bf6e5e2e5..871f2a3ff70 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py @@ -1,4 +1,5 @@ import json +from collections.abc import Iterator from datetime import datetime from itertools import chain from typing import Final @@ -350,10 +351,10 @@ async def test_collect_response_from_async_stream_recovers_output_items(recovera ], ids=["validated-terminal", "partial-terminal"], ) -def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload): - event = ResponseOutputItemDoneEvent(**_output_item_done_event(), sequence_number=1) +def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload: dict[str, JsonValue]) -> None: + event: Final = ResponseOutputItemDoneEvent(**_output_item_done_event(), sequence_number=1) - response = ResponsesToCompletionBridgeHandler._coerce_response_object( + response: Final = ResponsesToCompletionBridgeHandler._coerce_response_object( terminal_payload, {"headers": {"x-request-id": "req_test"}}, (object(), event), @@ -364,23 +365,23 @@ def test_recovery_accepts_sdk_events_and_dictionary_terminals(terminal_payload): assert terminal_payload["output"] == [] -def test_recovery_preserves_complete_response_without_consuming_events(): - terminal = ResponsesAPIResponse.model_construct(output=[_output_item_done_event()["item"]]) +def test_recovery_preserves_complete_response_without_consuming_events() -> None: + terminal: Final = ResponsesAPIResponse.model_construct(output=[_output_item_done_event()["item"]]) - def unavailable_events(): + def unavailable_events() -> Iterator[object]: raise AssertionError("Complete terminal output must bypass event recovery") yield - response = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, unavailable_events()) + response: Final = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, unavailable_events()) assert response is terminal assert response.output == [_output_item_done_event()["item"]] -def test_recovery_keeps_empty_terminal_when_no_output_can_be_recovered(): - terminal = _empty_responses_response() +def test_recovery_keeps_empty_terminal_when_no_output_can_be_recovered() -> None: + terminal: Final = _empty_responses_response() - response = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, (object(),)) + response: Final = ResponsesToCompletionBridgeHandler._coerce_response_object(terminal, None, (object(),)) assert response is terminal assert response.output == [] diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index 85364d1fc69..6a87345e4a7 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -612,7 +612,9 @@ def test_transform_response_recovers_empty_output_from_raw_sse(): @pytest.mark.parametrize("nested_citation", [False, True]) @pytest.mark.parametrize("recovery_path", ["raw_sse", "bridge"]) -def test_transform_response_recovers_text_and_citation_offsets(nested_citation, recovery_path): +def test_transform_response_recovers_text_and_citation_offsets( + nested_citation: bool, recovery_path: Literal["raw_sse", "bridge"] +) -> None: from copy import deepcopy from litellm.completion_extras.litellm_responses_transformation.handler import ( @@ -622,12 +624,12 @@ def test_transform_response_recovers_text_and_citation_offsets(nested_citation, LiteLLMResponsesTransformationHandler, ) - handler = LiteLLMResponsesTransformationHandler() - citation = {"start_index": 0, "end_index": 5, "title": "Source", "url": "https://example.com"} - annotation = ( + handler: Final = LiteLLMResponsesTransformationHandler() + citation: Final = {"start_index": 0, "end_index": 5, "title": "Source", "url": "https://example.com"} + annotation: Final = ( {"type": "url_citation", "url_citation": citation} if nested_citation else {"type": "url_citation", **citation} ) - events = ( + events: Final = ( { "type": "response.output_text.done", "output_index": 0, @@ -645,18 +647,18 @@ def test_transform_response_recovers_text_and_citation_offsets(nested_citation, "annotations": [annotation], }, ) - original_events = deepcopy(events) - raw_sse = "\n".join(f"data: {json.dumps(event)}" for event in events) - raw_response = ( + original_events: Final = deepcopy(events) + raw_sse: Final = "\n".join(f"data: {json.dumps(event)}" for event in events) + raw_response: Final = ( ResponsesToCompletionBridgeHandler._coerce_response_object(_make_empty_responses_api_response(), None, events) if recovery_path == "bridge" else _make_empty_responses_api_response() ) - model_response = _make_empty_model_response() - logging_obj = Mock() + model_response: Final = _make_empty_model_response() + logging_obj: Final = Mock() logging_obj.model_call_details = {"original_response": raw_sse if recovery_path == "raw_sse" else None} - result = handler.transform_response( + result: Final = handler.transform_response( model="gpt-5.4", raw_response=raw_response, model_response=model_response, @@ -670,8 +672,8 @@ def test_transform_response_recovers_text_and_citation_offsets(nested_citation, assert len(result.choices) == 1 assert result.choices[0].message.content == "Hello, world!" - shifted_citation = {**citation, "start_index": 7, "end_index": 12} - shifted_annotation = ( + shifted_citation: Final = {**citation, "start_index": 7, "end_index": 12} + shifted_annotation: Final = ( {"type": "url_citation", "url_citation": shifted_citation} if nested_citation else {"type": "url_citation", **shifted_citation}