From 3fab53083dd760831add198cef30818680e1bfd8 Mon Sep 17 00:00:00 2001 From: Abhisek Das Date: Sat, 16 May 2026 01:16:49 +0530 Subject: [PATCH] test(streaming): cover defensive branches of _original_chunk_has_reasoning_content Adds a unit test that exercises each early-return guard in the helper (missing original_chunk, no choices, delta is None) so codecov/patch counts the helper as fully covered. --- .../test_streaming_handler.py | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/test_litellm/litellm_core_utils/test_streaming_handler.py b/tests/test_litellm/litellm_core_utils/test_streaming_handler.py index c88d30bdf0e..cb2f6b3686c 100644 --- a/tests/test_litellm/litellm_core_utils/test_streaming_handler.py +++ b/tests/test_litellm/litellm_core_utils/test_streaming_handler.py @@ -2,6 +2,7 @@ import json import os import sys import time +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest @@ -230,6 +231,36 @@ def test_is_chunk_non_empty_ignores_original_chunk_without_reasoning( ) +def test_original_chunk_has_reasoning_content_defensive_branches(): + """ + Cover the defensive guards inside _original_chunk_has_reasoning_content so + every early-return path is exercised. Each branch must return False without + raising on the surrounding malformed input. + """ + from litellm.litellm_core_utils.streaming_handler import ( + _original_chunk_has_reasoning_content, + ) + + # 1. Missing original_chunk entirely. + assert _original_chunk_has_reasoning_content({}) is False + + # 2. original_chunk present but no choices. + chunk_no_choices = ModelResponseStream(choices=[]) + assert ( + _original_chunk_has_reasoning_content({"original_chunk": chunk_no_choices}) + is False + ) + + # 3. original_chunk with a choice whose delta is None. ModelResponseStream + # always materialises a Delta, so emulate the missing-delta case with a + # SimpleNamespace standing in for the choice. + chunk_no_delta = SimpleNamespace(choices=[SimpleNamespace(delta=None)]) + assert ( + _original_chunk_has_reasoning_content({"original_chunk": chunk_no_delta}) + is False + ) + + def test_is_chunk_non_empty_with_annotations( initialized_custom_stream_wrapper: CustomStreamWrapper, ):