diff --git a/tests/test_litellm/llms/minimax/chat/test_minimax_streaming_handler.py b/tests/test_litellm/llms/minimax/chat/test_minimax_streaming_handler.py
new file mode 100644
index 00000000000..8ab8fd3c13f
--- /dev/null
+++ b/tests/test_litellm/llms/minimax/chat/test_minimax_streaming_handler.py
@@ -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": "\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": "\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 and in one chunk
+ # ------------------------------------------------------------------
+
+ def test_think_both_tags(self):
+ """reasoning → strip tags, emit reasoning_content."""
+ it = _make_iterator()
+ chunk = _make_chunk({
+ "content": "inner reasoning\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):
+ """... without trailing answer → content=None."""
+ it = _make_iterator()
+ chunk = _make_chunk({
+ "content": "just thinking",
+ })
+ result = it.chunk_parser(chunk)
+ delta = result.choices[0].delta
+ assert delta.reasoning_content == "just thinking"
+ assert delta.content is None
+
+ # ------------------------------------------------------------------
+ # Pattern 4: opening tag only (no )
+ # ------------------------------------------------------------------
+
+ def test_think_opening_only(self):
+ """ without → strip prefix, emit as reasoning."""
+ it = _make_iterator()
+ chunk = _make_chunk({
+ "content": "\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: closing tag only (no )
+ # ------------------------------------------------------------------
+
+ def test_think_closing_only_with_content(self):
+ """ + trailing answer → split correctly."""
+ it = _make_iterator()
+ it.started_reasoning_content = True # simulate previous reasoning
+ chunk = _make_chunk({
+ "content": "\n\nHello there!",
+ })
+ result = it.chunk_parser(chunk)
+ delta = result.choices[0].delta
+ # No 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 → reasoning_content, after → content."""
+ it = _make_iterator()
+ it.started_reasoning_content = True
+ chunk = _make_chunk({
+ "content": "last thought\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. delimiter + answer start
+ r4 = it.chunk_parser(_make_chunk({
+ "content": "\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": "\nthinking", "role": "assistant"},
+ {"reasoning": " deeper", "content": " deeper"},
+ {"content": "\n\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": "\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
+ 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(" ✓ without ")
+
+ print("\n=== Pattern 5: closing tag only ===")
+ t.test_think_closing_only_with_content()
+ print(" ✓ with trailing content")
+ t.test_think_closing_only_with_before()
+ print(" ✓ text before ")
+
+ 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!")