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

test: add chunk_parser unit tests for all 6 MiniMax streaming patterns

Browse source 
- 18 unit tests covering reasoning field, reasoning_details,
  <think>...</think>, <think> only, </think> only, plain content,
  edge cases, and full M2.7 sequence
- Mocked HTTP streaming tests for M3 and M2.7
- All tests run without requiring real API key
This commit is contained in:
renearaos 2026-06-18 21:23:24 -04:00
parent f04f81fcc0
commit 1868502e8a
1 changed files with 484 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

484
tests/test_litellm/llms/minimax/chat/test_minimax_streaming_handler.py Normal file
View file

@ -0,0 +1,484 @@
"""
Test MinimaxChatResponseIterator chunk_parser patterns.
Tests all 6 streaming patterns without requiring a real API key.
Uses direct chunk_parser calls and mocked HTTP handler for full stream testing.
"""
import json
import os
import sys
from typing import Any, Generator, List
from unittest.mock import AsyncMock, MagicMock, patch
sys.path.insert(0, os.path.abspath(os.path.join(__file__, *([os.pardir] * 6))))
try:
import pytest
except ImportError:
pytest = None # allow standalone execution via if __name__ == "__main__"
import litellm
from litellm import completion
from litellm.llms.custom_httpx.http_handler import HTTPHandler, AsyncHTTPHandler
from litellm.llms.minimax.chat.transformation import (
MinimaxChatConfig,
MinimaxChatResponseIterator,
)
from litellm.types.utils import ModelResponseStream
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_chunk(
delta: dict,
chunk_id: str = "test-123",
created: int = 1700000000,
model: str = "MiniMax-M3",
finish_reason: str = None,
) -> dict:
"""Build a chunk dict that mimics a MiniMax streaming SSE event."""
chunk = {
"id": chunk_id,
"object": "chat.completion.chunk",
"created": created,
"model": model,
"choices": [{"index": 0, "delta": delta}],
}
if finish_reason:
chunk["choices"][0]["finish_reason"] = finish_reason
if "usage" in delta:
chunk["usage"] = delta.pop("usage")
return chunk
def _make_iterator() -> MinimaxChatResponseIterator:
"""Create a fresh MinimaxChatResponseIterator for unit testing."""
return MinimaxChatResponseIterator(
streaming_response=iter([]), sync_stream=True
)
def _sse_line(chunk: dict) -> str:
"""Serialize a chunk dict to an SSE data line."""
return f"data: {json.dumps(chunk)}"
# ===========================================================================
# Unit tests: direct chunk_parser calls
# ===========================================================================
class TestChunkParserPatterns:
"""Verify each of the 6 input patterns produces the correct delta."""
# ------------------------------------------------------------------
# Pattern 1: M3 – reasoning field
# ------------------------------------------------------------------
def test_reasoning_field(self):
"""delta.reasoning → content=None, reasoning_content=value"""
it = _make_iterator()
chunk = _make_chunk({
"reasoning": "thinking text",
"content": "<think>\nthinking text",
"role": "assistant",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "thinking text"
assert delta.content is None
def test_reasoning_content_field(self):
"""delta.reasoning_content → content=None"""
it = _make_iterator()
chunk = _make_chunk({
"reasoning_content": "deep thought",
"content": "<think>\ndeep thought",
"role": "assistant",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "deep thought"
assert delta.content is None
# ------------------------------------------------------------------
# Pattern 2: M3 – reasoning_details array
# ------------------------------------------------------------------
def test_reasoning_details_array(self):
"""delta.reasoning_details → concatenated to reasoning_content"""
it = _make_iterator()
chunk = _make_chunk({
"reasoning_details": [
{"text": "step 1 "},
{"text": "step 2 "},
{"text": "step 3"},
],
"content": "",
})
result = it.chunk_parser(chunk)
assert result.choices[0].delta.reasoning_content == "step 1 step 2 step 3"
def test_reasoning_details_empty_array(self):
"""Empty reasoning_details array → nothing set."""
it = _make_iterator()
chunk = _make_chunk({
"reasoning_details": [],
})
result = it.chunk_parser(chunk)
assert getattr(result.choices[0].delta, "reasoning_content", None) is None
def test_reasoning_details_empty_text(self):
"""Array entries with empty text → nothing set."""
it = _make_iterator()
chunk = _make_chunk({
"reasoning_details": [{"text": ""}, {"text": ""}],
})
result = it.chunk_parser(chunk)
assert getattr(result.choices[0].delta, "reasoning_content", None) is None
# ------------------------------------------------------------------
# Pattern 3: both <think> and </think> in one chunk
# ------------------------------------------------------------------
def test_think_both_tags(self):
"""<think>reasoning</think> → strip tags, emit reasoning_content."""
it = _make_iterator()
chunk = _make_chunk({
"content": "<think>inner reasoning</think>\n\nanswer here",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "inner reasoning"
assert delta.content == "\n\nanswer here"
def test_think_both_tags_only_reasoning(self):
"""<think>...</think> without trailing answer → content=None."""
it = _make_iterator()
chunk = _make_chunk({
"content": "<think>just thinking</think>",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "just thinking"
assert delta.content is None
# ------------------------------------------------------------------
# Pattern 4: <think> opening tag only (no </think>)
# ------------------------------------------------------------------
def test_think_opening_only(self):
"""<think> without </think> → strip prefix, emit as reasoning."""
it = _make_iterator()
chunk = _make_chunk({
"content": "<think>\nthought process begins",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "thought process begins"
assert delta.content is None
assert it.started_reasoning_content is True
# ------------------------------------------------------------------
# Pattern 5: </think> closing tag only (no <think>)
# ------------------------------------------------------------------
def test_think_closing_only_with_content(self):
"""</think> + trailing answer → split correctly."""
it = _make_iterator()
it.started_reasoning_content = True # simulate previous reasoning
chunk = _make_chunk({
"content": "</think>\n\nHello there!",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
# No <think> before, so before="" → no extra reasoning_content
assert delta.content == "\n\nHello there!"
assert it.finished_reasoning_content is True
def test_think_closing_only_with_before(self):
"""Text before </think> → reasoning_content, after → content."""
it = _make_iterator()
it.started_reasoning_content = True
chunk = _make_chunk({
"content": "last thought</think>\n\nanswer",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "last thought"
assert delta.content == "\n\nanswer"
# ------------------------------------------------------------------
# Pattern 6: plain content – no tags, no reasoning field
# ------------------------------------------------------------------
def test_plain_content_becomes_reasoning(self):
"""Non-empty content without indicators → accumulate as reasoning."""
it = _make_iterator()
chunk = _make_chunk({
"content": "The user wants",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.reasoning_content == "The user wants"
assert delta.content is None
assert it.started_reasoning_content is True
def test_plain_content_skipped_empty_string(self):
"""Empty string content → skip (no reasoning flag)."""
it = _make_iterator()
chunk = _make_chunk({
"content": "",
"role": "assistant",
})
result = it.chunk_parser(chunk)
delta = result.choices[0].delta
assert delta.content == ""
assert getattr(delta, "reasoning_content", None) is None
def test_plain_content_accumulate_across_chunks(self):
"""Multiple plain-content chunks → accumulate in pending."""
it = _make_iterator()
c1 = _make_chunk({"content": "first "})
r1 = it.chunk_parser(c1)
assert r1.choices[0].delta.reasoning_content == "first"
c2 = _make_chunk({"content": "second "})
r2 = it.chunk_parser(c2)
assert r2.choices[0].delta.reasoning_content == "second"
# ------------------------------------------------------------------
# Edge: finish / empty delta
# ------------------------------------------------------------------
def test_finish_chunk_passthrough(self):
"""Empty delta with finish_reason → pass through unchanged."""
it = _make_iterator()
chunk = _make_chunk({}, finish_reason="stop")
result = it.chunk_parser(chunk)
assert result.choices[0].finish_reason == "stop"
def test_no_choices(self):
"""Chunk without choices → super().chunk_parser handles it."""
it = _make_iterator()
chunk = {"id": "test", "object": "chat.completion.chunk"}
result = it.chunk_parser(chunk)
assert result is not None
# ------------------------------------------------------------------
# Full M2.7-like sequence
# ------------------------------------------------------------------
def test_m27_full_sequence(self):
"""Simulate a MiniMax-M2.7 streaming session."""
it = _make_iterator()
# 1. role-only introductory chunk
r0 = it.chunk_parser(_make_chunk({
"content": "", "role": "assistant",
}))
assert getattr(r0.choices[0].delta, "reasoning_content", None) is None
# 2-4. plain reasoning chunks
r1 = it.chunk_parser(_make_chunk({"content": "The "}))
assert r1.choices[0].delta.reasoning_content == "The"
r2 = it.chunk_parser(_make_chunk({"content": "user wants "}))
assert r2.choices[0].delta.reasoning_content == "user wants"
r3 = it.chunk_parser(_make_chunk({"content": "a greeting."}))
assert r3.choices[0].delta.reasoning_content == "a greeting."
# 5. </think> delimiter + answer start
r4 = it.chunk_parser(_make_chunk({
"content": "</think>\n\nHello",
}))
delta = r4.choices[0].delta
assert delta.content == "\n\nHello"
# 6. answer continues
r5 = it.chunk_parser(_make_chunk({"content": " there!"}))
assert r5.choices[0].delta.content == " there!"
# 7. finish
r6 = it.chunk_parser(_make_chunk({}, finish_reason="stop"))
assert r6.choices[0].finish_reason == "stop"
# ===========================================================================
# Integration tests: mocked HTTP handler
# ===========================================================================
def _build_m3_chunks() -> List[str]:
"""Build SSE lines mimicking MiniMax-M3 streaming."""
base = {
"id": "stream-m3-001",
"object": "chat.completion.chunk",
"created": 1700000000,
"model": "MiniMax-M3",
}
deltas = [
{"reasoning": "thinking", "content": "<think>\nthinking", "role": "assistant"},
{"reasoning": " deeper", "content": " deeper"},
{"content": "\n</think>\n\nHello!"},
]
lines = []
for i, d in enumerate(deltas):
finish = "stop" if i == len(deltas) - 1 else None
chunk = {**base, "choices": [{"index": 0, "delta": d}]}
if finish:
chunk["choices"][0]["finish_reason"] = finish
lines.append(_sse_line(chunk))
lines.append("data: [DONE]")
return lines
def _build_m27_chunks() -> List[str]:
"""Build SSE lines mimicking MiniMax-M2.7 streaming."""
base = {
"id": "stream-m27-002",
"object": "chat.completion.chunk",
"created": 1700000000,
"model": "MiniMax-M2.7",
}
deltas = [
{"content": "", "role": "assistant"},
{"content": "The "},
{"content": "user "},
{"content": "wants "},
{"content": "a greeting."},
{"content": "</think>\n\nHello"},
{"content": " there!"},
]
lines = []
for i, d in enumerate(deltas):
finish = None
if d.get("content", "") != "" and not any(
k in d for k in ("reasoning", "finish_reason")
):
pass
chunk = {**base, "choices": [{"index": 0, "delta": d}]}
if i == len(deltas) - 1:
chunk["choices"][0]["finish_reason"] = "stop"
lines.append(_sse_line(chunk))
lines.append("data: [DONE]")
return lines
def test_mocked_m3_streaming():
"""Mock HTTPHandler.post for MiniMax-M3 and verify stream output."""
chunks = _build_m3_chunks()
def _iter_lines() -> Generator:
yield from chunks
mock_resp = MagicMock()
mock_resp.iter_lines.return_value = _iter_lines()
mock_resp.status_code = 200
mock_resp.headers = {"content-type": "text/event-stream"}
with patch.object(HTTPHandler, "post", return_value=mock_resp):
response = completion(
model="minimax/MiniMax-M3",
messages=[{"role": "user", "content": "Say hi"}],
api_key="sk-mock-key",
api_base="https://api.minimax.io/v1",
stream=True,
)
collected = list(response)
assert len(collected) == 4
reasoning_chunks = [
c for c in collected
if c.choices[0].delta.reasoning_content is not None
]
assert len(reasoning_chunks) >= 1
def test_mocked_m27_streaming():
"""Mock HTTPHandler.post for MiniMax-M2.7 and verify stream output."""
chunks = _build_m27_chunks()
def _iter_lines() -> Generator:
yield from chunks
mock_resp = MagicMock()
mock_resp.iter_lines.return_value = _iter_lines()
mock_resp.status_code = 200
mock_resp.headers = {"content-type": "text/event-stream"}
with patch.object(HTTPHandler, "post", return_value=mock_resp):
response = completion(
model="minimax/MiniMax-M2.7",
messages=[{"role": "user", "content": "Say hi"}],
api_key="sk-mock-key",
api_base="https://api.minimax.io/v1",
stream=True,
)
collected = list(response)
assert len(collected) > 0
# Find the answer chunk after </think>
answer_chunks = [
c for c in collected
if c.choices[0].delta.content is not None
and c.choices[0].finish_reason is None
]
assert len(answer_chunks) >= 1
if __name__ == "__main__":
t = TestChunkParserPatterns()
print("=== Pattern 1: reasoning field ===")
t.test_reasoning_field()
print(" ✓ reasoning field")
t.test_reasoning_content_field()
print(" ✓ reasoning_content field")
print("\n=== Pattern 2: reasoning_details ===")
t.test_reasoning_details_array()
print(" ✓ reasoning_details array")
t.test_reasoning_details_empty_array()
print(" ✓ empty array")
t.test_reasoning_details_empty_text()
print(" ✓ empty text entries")
print("\n=== Pattern 3: both think tags ===")
t.test_think_both_tags()
print(" ✓ both tags")
t.test_think_both_tags_only_reasoning()
print(" ✓ both tags (reasoning only)")
print("\n=== Pattern 4: opening tag only ===")
t.test_think_opening_only()
print(" ✓ <think> without </think>")
print("\n=== Pattern 5: closing tag only ===")
t.test_think_closing_only_with_content()
print(" ✓ </think> with trailing content")
t.test_think_closing_only_with_before()
print(" ✓ text before </think>")
print("\n=== Pattern 6: plain content ===")
t.test_plain_content_becomes_reasoning()
print(" ✓ becomes reasoning")
t.test_plain_content_skipped_empty_string()
print(" ✓ empty string skipped")
t.test_plain_content_accumulate_across_chunks()
print(" ✓ accumulate across chunks")
print("\n=== Edge cases ===")
t.test_finish_chunk_passthrough()
print(" ✓ finish chunk")
t.test_no_choices()
print(" ✓ no choices")
print("\n=== M2.7 full sequence ===")
t.test_m27_full_sequence()
print(" ✓ full sequence")
print("\n✅ All chunk_parser unit tests passed!")