[misc] Remove unit test cases that fail the admission criteria (#30690)
This commit is contained in:
Liangsheng Yin
@@ -119,19 +119,6 @@ class TestConversationGetPrompt(CustomTestCase):
|
||||
self.assertIn("[USER]Hello\n", prompt)
|
||||
self.assertTrue(prompt.endswith("[ASST]"))
|
||||
|
||||
def test_none_message_in_prompt(self):
|
||||
"""Test that None message produces role-only output (no content)."""
|
||||
conv = Conversation(
|
||||
name="test",
|
||||
system_message="",
|
||||
roles=("User", "Assistant"),
|
||||
messages=[["User", "Q"], ["Assistant", None]],
|
||||
sep_style=SeparatorStyle.ADD_COLON_SINGLE,
|
||||
sep="\n",
|
||||
)
|
||||
prompt = conv.get_prompt()
|
||||
self.assertTrue(prompt.endswith("Assistant:"))
|
||||
|
||||
def test_empty_system_message(self):
|
||||
"""Test that empty system message produces empty prefix for LLAMA3."""
|
||||
conv = Conversation(
|
||||
@@ -189,22 +176,6 @@ class TestConversationGetPrompt(CustomTestCase):
|
||||
self.assertIn("[A]A<s2>", prompt)
|
||||
self.assertTrue(prompt.endswith("[U]"))
|
||||
|
||||
def test_llama2_with_system(self):
|
||||
"""Test LLAMA2 with system message."""
|
||||
conv = Conversation(
|
||||
name="test",
|
||||
system_message="<<SYS>>\nBe helpful\n<</SYS>>\n\n",
|
||||
system_template="[INST] {system_message}",
|
||||
roles=("[INST]", "[/INST]"),
|
||||
messages=[["[INST]", "Hi"], ["[/INST]", None]],
|
||||
sep_style=SeparatorStyle.LLAMA2,
|
||||
sep=" ",
|
||||
sep2=" </s><s>",
|
||||
)
|
||||
prompt = conv.get_prompt()
|
||||
self.assertIn("Be helpful", prompt)
|
||||
self.assertIn("Hi ", prompt)
|
||||
|
||||
def test_llama2_without_system(self):
|
||||
"""Test LLAMA2 without system message falls back to '[INST] ' prefix."""
|
||||
conv = Conversation(
|
||||
@@ -570,23 +541,6 @@ class TestConversationGetPrompt(CustomTestCase):
|
||||
self.assertIn("USER: <image>Describe this\n", prompt)
|
||||
self.assertIn("ASSISTANT: It shows a cat<eos>", prompt)
|
||||
|
||||
def test_mpt_with_tuple_message(self):
|
||||
"""Test MPT style extracts first element from tuple messages."""
|
||||
conv = Conversation(
|
||||
name="test",
|
||||
system_message="<|system|>",
|
||||
roles=("<|user|>", "<|assistant|>"),
|
||||
messages=[
|
||||
["<|user|>", ("Hello", "extra1", "extra2")],
|
||||
["<|assistant|>", None],
|
||||
],
|
||||
sep_style=SeparatorStyle.MPT,
|
||||
sep="\n",
|
||||
)
|
||||
prompt = conv.get_prompt()
|
||||
self.assertIn("<|user|>Hello\n", prompt)
|
||||
self.assertNotIn("extra1", prompt)
|
||||
|
||||
def test_invalid_sep_style_raises(self):
|
||||
"""Test that an invalid SeparatorStyle raises ValueError."""
|
||||
conv = Conversation(
|
||||
@@ -611,100 +565,6 @@ class TestConversationMethods(CustomTestCase):
|
||||
sep="\n",
|
||||
)
|
||||
|
||||
def test_append_message(self):
|
||||
"""Test appending messages to conversation."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Hello")
|
||||
conv.append_message("Assistant", "Hi")
|
||||
self.assertEqual(len(conv.messages), 2)
|
||||
self.assertEqual(conv.messages[0], ["User", "Hello"])
|
||||
|
||||
def test_set_system_message(self):
|
||||
"""Test setting the system message."""
|
||||
conv = self._make_conv()
|
||||
conv.set_system_message("Be helpful")
|
||||
self.assertEqual(conv.system_message, "Be helpful")
|
||||
|
||||
def test_update_last_message(self):
|
||||
"""Test updating the last message in-place."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Q")
|
||||
conv.append_message("Assistant", None)
|
||||
conv.update_last_message("Answer")
|
||||
self.assertEqual(conv.messages[-1][1], "Answer")
|
||||
|
||||
def test_to_openai_api_messages_with_system(self):
|
||||
"""Test conversion to OpenAI format with system message."""
|
||||
conv = self._make_conv()
|
||||
conv.system_message = "Be helpful"
|
||||
conv.append_message("User", "Hello")
|
||||
conv.append_message("Assistant", "Hi")
|
||||
result = conv.to_openai_api_messages()
|
||||
self.assertEqual(result[0], {"role": "system", "content": "Be helpful"})
|
||||
self.assertEqual(result[1], {"role": "user", "content": "Hello"})
|
||||
self.assertEqual(result[2], {"role": "assistant", "content": "Hi"})
|
||||
|
||||
def test_to_openai_api_messages_without_system(self):
|
||||
"""Test conversion to OpenAI format without system message."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Hello")
|
||||
result = conv.to_openai_api_messages()
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(result[0]["role"], "user")
|
||||
|
||||
def test_to_openai_api_messages_skips_none_assistant(self):
|
||||
"""Test that None assistant message is omitted from OpenAI format."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Hello")
|
||||
conv.append_message("Assistant", None)
|
||||
result = conv.to_openai_api_messages()
|
||||
self.assertEqual(len(result), 1) # only user message
|
||||
|
||||
def test_to_gradio_chatbot(self):
|
||||
"""Test conversion to Gradio chatbot format (user/assistant pairs)."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Q1")
|
||||
conv.append_message("Assistant", "A1")
|
||||
conv.append_message("User", "Q2")
|
||||
conv.append_message("Assistant", "A2")
|
||||
result = conv.to_gradio_chatbot()
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertEqual(result[0], ["Q1", "A1"])
|
||||
self.assertEqual(result[1], ["Q2", "A2"])
|
||||
|
||||
def test_to_gradio_chatbot_pending_response(self):
|
||||
"""Test Gradio format with pending assistant response (None)."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Q1")
|
||||
conv.append_message("Assistant", None)
|
||||
result = conv.to_gradio_chatbot()
|
||||
self.assertEqual(result, [["Q1", None]])
|
||||
|
||||
def test_append_image(self):
|
||||
"""Test appending image data to conversation."""
|
||||
conv = self._make_conv()
|
||||
conv.image_data = []
|
||||
conv.append_image("http://example.com/img.jpg", "auto")
|
||||
self.assertEqual(len(conv.image_data), 1)
|
||||
self.assertEqual(conv.image_data[0].url, "http://example.com/img.jpg")
|
||||
self.assertEqual(conv.image_data[0].detail, "auto")
|
||||
|
||||
def test_append_video(self):
|
||||
"""Test appending video data to conversation."""
|
||||
conv = self._make_conv()
|
||||
conv.video_data = []
|
||||
conv.append_video("http://example.com/vid.mp4")
|
||||
self.assertEqual(len(conv.video_data), 1)
|
||||
self.assertEqual(conv.video_data[0], "http://example.com/vid.mp4")
|
||||
|
||||
def test_append_audio(self):
|
||||
"""Test appending audio data to conversation."""
|
||||
conv = self._make_conv()
|
||||
conv.audio_data = []
|
||||
conv.append_audio("http://example.com/audio.wav")
|
||||
self.assertEqual(len(conv.audio_data), 1)
|
||||
self.assertEqual(conv.audio_data[0], "http://example.com/audio.wav")
|
||||
|
||||
def test_copy_is_independent(self):
|
||||
"""Test that copy() creates an independent conversation."""
|
||||
conv = self._make_conv()
|
||||
@@ -714,15 +574,6 @@ class TestConversationMethods(CustomTestCase):
|
||||
self.assertEqual(len(conv.messages), 1)
|
||||
self.assertEqual(len(copied.messages), 2)
|
||||
|
||||
def test_dict_serialization(self):
|
||||
"""Test dict() returns expected keys."""
|
||||
conv = self._make_conv()
|
||||
conv.append_message("User", "Hello")
|
||||
d = conv.dict()
|
||||
self.assertEqual(d["template_name"], "test")
|
||||
self.assertIn("messages", d)
|
||||
self.assertIn("roles", d)
|
||||
|
||||
|
||||
class TestTemplateRegistry(CustomTestCase):
|
||||
def test_builtin_templates_exist(self):
|
||||
@@ -730,24 +581,6 @@ class TestTemplateRegistry(CustomTestCase):
|
||||
self.assertTrue(chat_template_exists("chatml"))
|
||||
self.assertTrue(chat_template_exists("llama-2"))
|
||||
|
||||
def test_unregistered_template_not_found(self):
|
||||
"""Test that non-existent template returns False."""
|
||||
self.assertFalse(chat_template_exists("_nonexistent_template_xyz"))
|
||||
|
||||
def test_register_and_lookup(self):
|
||||
"""Test registering and looking up a custom template."""
|
||||
t = Conversation(
|
||||
name="_test_conv_template",
|
||||
roles=("A", "B"),
|
||||
messages=[],
|
||||
sep_style=SeparatorStyle.ADD_COLON_SINGLE,
|
||||
sep="\n",
|
||||
)
|
||||
register_conv_template(t)
|
||||
self.assertTrue(chat_template_exists("_test_conv_template"))
|
||||
# Cleanup
|
||||
del chat_templates["_test_conv_template"]
|
||||
|
||||
def test_register_duplicate_raises(self):
|
||||
"""Test that registering a duplicate name without override raises."""
|
||||
with self.assertRaises(AssertionError):
|
||||
@@ -841,32 +674,6 @@ class TestGenerateEmbeddingConvs(CustomTestCase):
|
||||
self.assertIn("Hello world", convs[0].messages[0][1])
|
||||
self.assertIsNone(convs[0].messages[1][1]) # assistant placeholder
|
||||
|
||||
def test_with_image(self):
|
||||
"""Test generating embedding conversations with image."""
|
||||
convs = generate_embedding_convs(
|
||||
texts=["Describe"],
|
||||
images=["http://example.com/img.jpg"],
|
||||
videos=[None],
|
||||
template_name="chatml",
|
||||
)
|
||||
self.assertEqual(len(convs), 1)
|
||||
msg = convs[0].messages[0][1]
|
||||
self.assertIn("<image>", msg)
|
||||
self.assertIn("Describe", msg)
|
||||
|
||||
def test_with_video(self):
|
||||
"""Test generating embedding conversations with video."""
|
||||
convs = generate_embedding_convs(
|
||||
texts=["Describe"],
|
||||
images=[None],
|
||||
videos=["http://example.com/vid.mp4"],
|
||||
template_name="chatml",
|
||||
)
|
||||
self.assertEqual(len(convs), 1)
|
||||
msg = convs[0].messages[0][1]
|
||||
self.assertIn("<video>", msg)
|
||||
self.assertIn("Describe", msg)
|
||||
|
||||
def test_with_image_and_video(self):
|
||||
"""Test embedding conv with both image and video."""
|
||||
convs = generate_embedding_convs(
|
||||
@@ -892,24 +699,8 @@ class TestGenerateEmbeddingConvs(CustomTestCase):
|
||||
# None text should not produce "None" string
|
||||
self.assertNotIn("None", msg)
|
||||
|
||||
def test_multiple_items(self):
|
||||
"""Test generating multiple embedding conversations."""
|
||||
convs = generate_embedding_convs(
|
||||
texts=["text1", "text2"],
|
||||
images=[None, None],
|
||||
videos=[None, None],
|
||||
template_name="chatml",
|
||||
)
|
||||
self.assertEqual(len(convs), 2)
|
||||
|
||||
|
||||
class TestGetFullMultimodalTextPrompt(CustomTestCase):
|
||||
def test_adds_missing_image_tokens(self):
|
||||
"""Test adding missing image tokens to prompt."""
|
||||
result = _get_full_multimodal_text_prompt("<image>", 3, "Describe this.")
|
||||
self.assertEqual(result.count("<image>"), 3)
|
||||
self.assertIn("Describe this.", result)
|
||||
|
||||
def test_preserves_existing_tokens(self):
|
||||
"""Test that existing tokens in prompt are preserved."""
|
||||
result = _get_full_multimodal_text_prompt(
|
||||
@@ -927,17 +718,6 @@ class TestGetFullMultimodalTextPrompt(CustomTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
_get_full_multimodal_text_prompt("<image>", 1, "<image> <image>")
|
||||
|
||||
def test_zero_count_with_no_tokens(self):
|
||||
"""Test zero modality count with no tokens in prompt."""
|
||||
result = _get_full_multimodal_text_prompt("<image>", 0, "Just text")
|
||||
self.assertEqual(result, "Just text")
|
||||
|
||||
def test_video_tokens(self):
|
||||
"""Test adding missing video tokens."""
|
||||
result = _get_full_multimodal_text_prompt("<video>", 2, "Describe:")
|
||||
self.assertEqual(result.count("<video>"), 2)
|
||||
self.assertIn("Describe:", result)
|
||||
|
||||
def test_tokens_joined_with_newline(self):
|
||||
"""Test that missing tokens are joined with newlines before prompt."""
|
||||
result = _get_full_multimodal_text_prompt("<image>", 3, "text")
|
||||
@@ -1071,16 +851,6 @@ class TestGenerateChatConv(CustomTestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
generate_chat_conv(request, "chatml")
|
||||
|
||||
def test_string_messages_raises(self):
|
||||
"""Test that passing messages as a raw string raises ValueError."""
|
||||
request = self._make_request(
|
||||
[ChatCompletionMessageUserParam(role="user", content="Hi")]
|
||||
)
|
||||
# Manually override messages to be a string to trigger validation
|
||||
request.__dict__["messages"] = "not a list"
|
||||
with self.assertRaises(ValueError):
|
||||
generate_chat_conv(request, "chatml")
|
||||
|
||||
def test_user_message_with_image(self):
|
||||
"""Test user message with image content part."""
|
||||
request = self._make_request(
|
||||
|
||||
@@ -4,10 +4,8 @@ import unittest
|
||||
|
||||
from sglang.srt.parser.harmony_parser import (
|
||||
CanonicalStrategy,
|
||||
Event,
|
||||
HarmonyParser,
|
||||
TextStrategy,
|
||||
Token,
|
||||
iter_tokens,
|
||||
prefix_hold,
|
||||
)
|
||||
@@ -18,36 +16,7 @@ register_cpu_ci(est_time=7, suite="base-a-test-cpu")
|
||||
register_cpu_ci(est_time=7, suite="base-c-test-cpu")
|
||||
|
||||
|
||||
class TestEvent(CustomTestCase):
|
||||
def test_init(self):
|
||||
"""Test Event dataclass initialization."""
|
||||
event = Event("reasoning", "content")
|
||||
self.assertEqual(event.event_type, "reasoning")
|
||||
self.assertEqual(event.content, "content")
|
||||
|
||||
|
||||
class TestToken(CustomTestCase):
|
||||
def test_init(self):
|
||||
"""Test Token dataclass initialization."""
|
||||
token = Token("START", 0, 7)
|
||||
self.assertEqual(token.type, "START")
|
||||
self.assertEqual(token.start, 0)
|
||||
self.assertEqual(token.end, 7)
|
||||
|
||||
|
||||
class TestPrefixHold(CustomTestCase):
|
||||
def test_empty_text(self):
|
||||
"""Test prefix_hold with empty text."""
|
||||
emit, hold = prefix_hold("", ["<|start|>"])
|
||||
self.assertEqual(emit, "")
|
||||
self.assertEqual(hold, "")
|
||||
|
||||
def test_no_matching_prefixes(self):
|
||||
"""Test prefix_hold with no matching prefixes."""
|
||||
emit, hold = prefix_hold("hello world", ["<|start|>", "<|end|>"])
|
||||
self.assertEqual(emit, "hello world")
|
||||
self.assertEqual(hold, "")
|
||||
|
||||
def test_partial_token_suffix(self):
|
||||
"""Test prefix_hold with partial token at end."""
|
||||
emit, hold = prefix_hold("hello <|ret", ["<|return|>"])
|
||||
@@ -68,11 +37,6 @@ class TestPrefixHold(CustomTestCase):
|
||||
|
||||
|
||||
class TestIterTokens(CustomTestCase):
|
||||
def test_empty_text(self):
|
||||
"""Test iter_tokens with empty text."""
|
||||
tokens = list(iter_tokens(""))
|
||||
self.assertEqual(tokens, [])
|
||||
|
||||
def test_plain_text(self):
|
||||
"""Test iter_tokens with plain text."""
|
||||
tokens = list(iter_tokens("hello world"))
|
||||
@@ -81,14 +45,6 @@ class TestIterTokens(CustomTestCase):
|
||||
self.assertEqual(tokens[0].start, 0)
|
||||
self.assertEqual(tokens[0].end, 11)
|
||||
|
||||
def test_single_token(self):
|
||||
"""Test iter_tokens with single structural token."""
|
||||
tokens = list(iter_tokens("<|start|>"))
|
||||
self.assertEqual(len(tokens), 1)
|
||||
self.assertEqual(tokens[0].type, "START")
|
||||
self.assertEqual(tokens[0].start, 0)
|
||||
self.assertEqual(tokens[0].end, 9)
|
||||
|
||||
def test_mixed_content(self):
|
||||
"""Test iter_tokens with mixed text and tokens."""
|
||||
tokens = list(iter_tokens("text<|start|>more text"))
|
||||
@@ -154,11 +110,6 @@ class TestCanonicalStrategy(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.strategy = CanonicalStrategy()
|
||||
|
||||
def test_init(self):
|
||||
"""Test CanonicalStrategy initialization."""
|
||||
self.assertIn("<|start|>", self.strategy.guard_tokens)
|
||||
self.assertIn("<|constrain|>", self.strategy.guard_tokens)
|
||||
|
||||
def test_extract_channel_type(self):
|
||||
"""Test _extract_channel_type method."""
|
||||
self.assertEqual(self.strategy._extract_channel_type("analysis"), "analysis")
|
||||
@@ -170,72 +121,6 @@ class TestCanonicalStrategy(CustomTestCase):
|
||||
self.assertEqual(self.strategy._extract_channel_type("ANALYSIS"), "analysis")
|
||||
self.assertIsNone(self.strategy._extract_channel_type("unknown"))
|
||||
|
||||
def test_parse_single_analysis_block(self):
|
||||
"""Test parsing single analysis block."""
|
||||
text = "<|channel|>analysis<|message|>Let me think about this<|end|>"
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "reasoning")
|
||||
self.assertEqual(events[0].content, "Let me think about this")
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_single_commentary_block(self):
|
||||
"""Test parsing single commentary block."""
|
||||
text = "<|channel|>commentary<|message|>User-visible message<|end|>"
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "normal")
|
||||
self.assertEqual(events[0].content, "User-visible message")
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_single_final_block(self):
|
||||
"""Test parsing single final block."""
|
||||
text = "<|start|>assistant<|channel|>final<|message|>The answer is 42<|return|>"
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "normal")
|
||||
self.assertEqual(events[0].content, "The answer is 42")
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_tool_call_commentary(self):
|
||||
"""Test parsing tool call on commentary channel."""
|
||||
text = '<|channel|>commentary to=functions.get_weather<|message|>{"location": "SF"}<|call|>'
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "tool_call")
|
||||
self.assertEqual(events[0].content, '{"location": "SF"}')
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_tool_call_analysis(self):
|
||||
"""Test parsing built-in tool call on analysis channel."""
|
||||
text = '<|channel|>analysis to=browser.search<|message|>{"query": "SGLang"}<|call|>'
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "tool_call")
|
||||
self.assertEqual(events[0].content, '{"query": "SGLang"}')
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_complex_sequence(self):
|
||||
"""Test parsing complex sequence with multiple blocks."""
|
||||
text = (
|
||||
"<|channel|>analysis<|message|>Need to use function get_weather.<|end|>"
|
||||
"<|start|>assistant<|channel|>commentary to=functions.get_weather<|message|>"
|
||||
'{"location":"San Francisco"}<|call|>'
|
||||
)
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 2)
|
||||
self.assertEqual(events[0].event_type, "reasoning")
|
||||
self.assertEqual(events[0].content, "Need to use function get_weather.")
|
||||
self.assertEqual(events[1].event_type, "tool_call")
|
||||
self.assertEqual(events[1].content, '{"location":"San Francisco"}')
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_with_interspersed_text(self):
|
||||
"""Test parsing with plain text between blocks."""
|
||||
text = (
|
||||
@@ -298,42 +183,11 @@ class TestCanonicalStrategy(CustomTestCase):
|
||||
self.assertEqual(events[0].content, "")
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
def test_parse_commentary_filler_between_blocks(self):
|
||||
"""Test that 'commentary' filler between <|call|> and <|channel|> is filtered out."""
|
||||
# This pattern occurs when the model generates malformed output
|
||||
text = (
|
||||
'<|channel|>commentary to=functions.get_weather<|message|>{"location":"SF"}<|call|>'
|
||||
"commentary" # This should be filtered out
|
||||
'<|channel|>commentary to=functions.get_temp<|message|>{"location":"NYC"}<|call|>'
|
||||
)
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
# Should have 2 tool calls, no "commentary" normal text
|
||||
self.assertEqual(len(events), 2)
|
||||
self.assertEqual(events[0].event_type, "tool_call")
|
||||
self.assertEqual(events[0].content, '{"location":"SF"}')
|
||||
self.assertEqual(events[1].event_type, "tool_call")
|
||||
self.assertEqual(events[1].content, '{"location":"NYC"}')
|
||||
self.assertEqual(remaining, "")
|
||||
|
||||
# Verify no "commentary" text was emitted as normal content
|
||||
normal_events = [e for e in events if e.event_type == "normal"]
|
||||
commentary_events = [
|
||||
e for e in normal_events if "commentary" in e.content.lower()
|
||||
]
|
||||
self.assertEqual(
|
||||
len(commentary_events), 0, "Commentary filler should be filtered out"
|
||||
)
|
||||
|
||||
|
||||
class TestTextStrategy(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.strategy = TextStrategy()
|
||||
|
||||
def test_init(self):
|
||||
"""Test TextStrategy initialization."""
|
||||
self.assertIn("analysis_then_final", self.strategy.patterns)
|
||||
|
||||
def test_parse_analysis_then_final(self):
|
||||
"""Test parsing analysis then final format."""
|
||||
text = "analysis I need to think about this. assistantfinal The answer is 42."
|
||||
@@ -387,16 +241,6 @@ class TestTextStrategy(CustomTestCase):
|
||||
self.assertEqual(len(events), 0)
|
||||
self.assertEqual(remaining, text) # Hold entire buffer
|
||||
|
||||
def test_parse_partial_analysis_streaming(self):
|
||||
"""Test streaming partial analysis content."""
|
||||
text = "analysis partial content"
|
||||
events, remaining = self.strategy.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "reasoning")
|
||||
self.assertEqual(events[0].content, " partial content") # Space preserved
|
||||
self.assertEqual(remaining, "analysis") # Hold header
|
||||
|
||||
def test_parse_case_insensitive(self):
|
||||
"""Test case insensitive parsing."""
|
||||
text = "ANALYSIS reasoning ASSISTANTFINAL answer"
|
||||
@@ -437,11 +281,6 @@ class TestHarmonyParser(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.parser = HarmonyParser()
|
||||
|
||||
def test_init(self):
|
||||
"""Test HarmonyParser initialization."""
|
||||
self.assertIsNone(self.parser.strategy)
|
||||
self.assertEqual(self.parser._buffer, "")
|
||||
|
||||
def test_strategy_selection_canonical(self):
|
||||
"""Test automatic strategy selection for canonical format."""
|
||||
events = self.parser.parse("<|channel|>analysis<|message|>test<|end|>")
|
||||
@@ -470,56 +309,6 @@ class TestHarmonyParser(CustomTestCase):
|
||||
self.assertIsInstance(self.parser.strategy, TextStrategy)
|
||||
self.assertEqual(len(events2), 1)
|
||||
|
||||
def test_streaming_canonical_format(self):
|
||||
"""Test streaming with canonical format."""
|
||||
chunks = [
|
||||
"<|channel|>analysis<|message|>",
|
||||
"reasoning content",
|
||||
"<|end|>",
|
||||
"<|start|>assistant<|channel|>final<|message|>",
|
||||
"final answer",
|
||||
"<|return|>",
|
||||
]
|
||||
|
||||
all_events = []
|
||||
for chunk in chunks:
|
||||
events = self.parser.parse(chunk)
|
||||
all_events.extend(events)
|
||||
|
||||
# Verify we get both reasoning and normal events
|
||||
reasoning_events = [e for e in all_events if e.event_type == "reasoning"]
|
||||
self.assertGreater(len(reasoning_events), 0)
|
||||
|
||||
normal_events = [e for e in all_events if e.event_type == "normal"]
|
||||
self.assertGreater(len(normal_events), 0)
|
||||
|
||||
# Verify content is eventually parsed correctly
|
||||
combined_reasoning = "".join(e.content for e in reasoning_events)
|
||||
combined_normal = "".join(
|
||||
e.content
|
||||
for e in normal_events
|
||||
if e.content and "<|return|>" not in e.content
|
||||
)
|
||||
|
||||
self.assertIn("reasoning content", combined_reasoning)
|
||||
self.assertIn("final answer", combined_normal)
|
||||
|
||||
def test_streaming_text_format(self):
|
||||
"""Test streaming with text format."""
|
||||
chunks = ["analysis reasoning", " content assistantfinal", " the answer"]
|
||||
|
||||
all_events = []
|
||||
for chunk in chunks:
|
||||
events = self.parser.parse(chunk)
|
||||
all_events.extend(events)
|
||||
|
||||
# Should have reasoning and normal events
|
||||
reasoning_events = [e for e in all_events if e.event_type == "reasoning"]
|
||||
normal_events = [e for e in all_events if e.event_type == "normal"]
|
||||
|
||||
self.assertGreater(len(reasoning_events), 0)
|
||||
self.assertGreater(len(normal_events), 0)
|
||||
|
||||
def test_streaming_commentary_filler(self):
|
||||
"""Test that 'commentary' filler is filtered in streaming case."""
|
||||
# Test when commentary arrives as a separate chunk after <|call|>
|
||||
@@ -679,35 +468,6 @@ class TestIntegrationScenarios(CustomTestCase):
|
||||
self.assertEqual(events[0].event_type, "tool_call")
|
||||
self.assertEqual(events[0].content, '{"query": "SGLang"}')
|
||||
|
||||
def test_tool_response_handling(self):
|
||||
"""Test tool response message handling."""
|
||||
parser = HarmonyParser()
|
||||
|
||||
text = '<|start|>functions.get_weather to=assistant<|channel|>commentary<|message|>{"sunny": true, "temperature": 20}<|end|>'
|
||||
|
||||
events = parser.parse(text)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(events[0].event_type, "normal")
|
||||
self.assertEqual(events[0].content, '{"sunny": true, "temperature": 20}')
|
||||
|
||||
def test_text_fallback_formats(self):
|
||||
"""Test various text fallback formats."""
|
||||
parser = HarmonyParser()
|
||||
|
||||
# Test analysis then final
|
||||
events1 = parser.parse("analysis thinking assistantfinal answer")
|
||||
self.assertEqual(len([e for e in events1 if e.event_type == "reasoning"]), 1)
|
||||
self.assertEqual(len([e for e in events1 if e.event_type == "normal"]), 1)
|
||||
|
||||
# Reset parser for next test
|
||||
parser = HarmonyParser()
|
||||
|
||||
# Test final only
|
||||
events2 = parser.parse("assistantfinal direct answer")
|
||||
self.assertEqual(len(events2), 1)
|
||||
self.assertEqual(events2[0].event_type, "normal")
|
||||
|
||||
def test_streaming_property_canonical(self):
|
||||
"""Test streaming property: chunked parsing produces same semantic content as one-shot parsing."""
|
||||
full_text = (
|
||||
@@ -819,12 +579,6 @@ class TestEdgeCases(CustomTestCase):
|
||||
self.assertEqual(len(reasoning_events), 1)
|
||||
self.assertGreater(len(normal_events), 0)
|
||||
|
||||
def test_empty_input(self):
|
||||
"""Test handling of empty input."""
|
||||
parser = HarmonyParser()
|
||||
events = parser.parse("")
|
||||
self.assertEqual(len(events), 0)
|
||||
|
||||
def test_whitespace_preservation(self):
|
||||
"""Test that whitespace is preserved correctly."""
|
||||
parser = HarmonyParser()
|
||||
@@ -878,14 +632,6 @@ class TestEdgeCases(CustomTestCase):
|
||||
class TestAdditionalEdgeCases(CustomTestCase):
|
||||
"""Additional tests to cover remaining edge cases."""
|
||||
|
||||
def test_prefix_hold_with_empty_token_in_list(self):
|
||||
"""Test that empty string token in the list is skipped."""
|
||||
from sglang.srt.parser.harmony_parser import prefix_hold
|
||||
|
||||
emit, hold = prefix_hold("hello", ["", "world"])
|
||||
self.assertEqual(emit, "hello")
|
||||
self.assertEqual(hold, "")
|
||||
|
||||
def test_iter_tokens_unknown_token_no_closing(self):
|
||||
"""Test iter_tokens with <| that has no closing |>."""
|
||||
from sglang.srt.parser.harmony_parser import iter_tokens
|
||||
@@ -894,72 +640,6 @@ class TestAdditionalEdgeCases(CustomTestCase):
|
||||
# Should emit TEXT tokens for the content after <|
|
||||
self.assertTrue(any(t.type == "TEXT" for t in tokens))
|
||||
|
||||
def test_canonical_commentary_filler_after_call(self):
|
||||
"""Test that MESSAGE token after CALL is filtered as commentary filler."""
|
||||
from sglang.srt.parser.harmony_parser import CanonicalStrategy
|
||||
|
||||
strategy = CanonicalStrategy()
|
||||
text = "<|start|><|channel|>analysis<|message|>thinking<|end|><|call|><|message|>noise<|return|><|channel|>final<|message|>answer<|end|>"
|
||||
events, remainder = strategy.parse(text)
|
||||
# The MESSAGE after CALL should be filtered, final answer should appear
|
||||
answers = [e.content for e in events if e.event_type == "normal"]
|
||||
self.assertTrue(any("answer" in a for a in answers))
|
||||
|
||||
def test_canonical_standalone_structural_token_filtered(self):
|
||||
"""Test that standalone structural tokens like <|end|> in TEXT position are filtered."""
|
||||
from sglang.srt.parser.harmony_parser import CanonicalStrategy
|
||||
|
||||
strategy = CanonicalStrategy()
|
||||
# A malformed sequence where an END token appears in an unexpected position
|
||||
text = "<|start|><|channel|>analysis<|message|>content<|end|>"
|
||||
events, remainder = strategy.parse(text)
|
||||
# Should parse without error
|
||||
self.assertTrue(len(events) >= 0)
|
||||
|
||||
def test_canonical_incomplete_block_returns_partial(self):
|
||||
"""Test parsing an incomplete channel block (no END token)."""
|
||||
from sglang.srt.parser.harmony_parser import CanonicalStrategy
|
||||
|
||||
strategy = CanonicalStrategy()
|
||||
text = "<|start|><|channel|>analysis<|message|>partial content"
|
||||
events, remainder = strategy.parse(text)
|
||||
# Incomplete block: should hold content as remainder or emit partial
|
||||
reasoning_events = [e for e in events if e.event_type == "reasoning"]
|
||||
# The partial content may be in events or remainder
|
||||
total = "".join(e.content for e in reasoning_events) + remainder
|
||||
self.assertIn("partial", total)
|
||||
|
||||
def test_text_strategy_commentary_channel(self):
|
||||
"""Test TextStrategy parsing commentary channel."""
|
||||
from sglang.srt.parser.harmony_parser import TextStrategy
|
||||
|
||||
strategy = TextStrategy()
|
||||
text = "commentary: some discussion\nassistantfinal: the answer"
|
||||
events, remainder = strategy.parse(text)
|
||||
normal = [e for e in events if e.event_type == "normal"]
|
||||
self.assertTrue(any("the answer" in e.content for e in normal))
|
||||
|
||||
def test_canonical_call_with_text_commentary_after(self):
|
||||
"""Test filtering of 'commentary' text after CALL token."""
|
||||
from sglang.srt.parser.harmony_parser import CanonicalStrategy
|
||||
|
||||
strategy = CanonicalStrategy()
|
||||
text = "<|start|><|channel|>analysis<|message|>think<|end|><|call|>commentary<|return|><|channel|>final<|message|>result<|end|>"
|
||||
events, remainder = strategy.parse(text)
|
||||
normal = [e for e in events if e.event_type == "normal"]
|
||||
self.assertTrue(any("result" in e.content for e in normal))
|
||||
|
||||
def test_canonical_return_without_final(self):
|
||||
"""Test that _parse_block returns None for block without proper end."""
|
||||
from sglang.srt.parser.harmony_parser import CanonicalStrategy
|
||||
|
||||
strategy = CanonicalStrategy()
|
||||
# Channel block that has no message content before end
|
||||
text = "<|start|><|channel|>final<|end|>"
|
||||
events, remainder = strategy.parse(text)
|
||||
# Should handle gracefully
|
||||
self.assertIsInstance(events, list)
|
||||
|
||||
def test_iter_tokens_unknown_at_end_no_next_marker(self):
|
||||
"""Test unknown token with |> close but no next <| marker after it."""
|
||||
from sglang.srt.parser.harmony_parser import iter_tokens
|
||||
@@ -984,18 +664,6 @@ class TestAdditionalEdgeCases(CustomTestCase):
|
||||
normal = [e.content for e in events if e.event_type == "normal"]
|
||||
self.assertTrue(any("answer" in c for c in normal))
|
||||
|
||||
def test_canonical_incomplete_parse_block_no_end(self):
|
||||
"""Test that a channel block without END/CALL/RETURN returns None (incomplete)."""
|
||||
from sglang.srt.parser.harmony_parser import CanonicalStrategy
|
||||
|
||||
strategy = CanonicalStrategy()
|
||||
# Channel with message but no end token
|
||||
text = "<|start|><|channel|>final<|message|>partial"
|
||||
events, remainder = strategy.parse(text)
|
||||
# Should be treated as incomplete
|
||||
total = "".join(e.content for e in events) + remainder
|
||||
self.assertIn("partial", total)
|
||||
|
||||
def test_text_strategy_commentary_only(self):
|
||||
"""Test TextStrategy with commentary-only pattern (no 'assistantfinal')."""
|
||||
from sglang.srt.parser.harmony_parser import TextStrategy
|
||||
|
||||
@@ -13,7 +13,6 @@ from sglang.srt.parser.reasoning_parser import (
|
||||
Nemotron3Detector,
|
||||
Qwen3Detector,
|
||||
ReasoningParser,
|
||||
StreamingParseResult,
|
||||
)
|
||||
from sglang.test.ci.ci_register import register_cpu_ci
|
||||
from sglang.test.test_utils import CustomTestCase
|
||||
@@ -21,20 +20,6 @@ from sglang.test.test_utils import CustomTestCase
|
||||
register_cpu_ci(est_time=7, suite="base-a-test-cpu")
|
||||
|
||||
|
||||
class TestStreamingParseResult(CustomTestCase):
|
||||
def test_init_default(self):
|
||||
"""Test default initialization of StreamingParseResult."""
|
||||
result = StreamingParseResult()
|
||||
self.assertEqual(result.normal_text, "")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_init_with_values(self):
|
||||
"""Test initialization with specific values."""
|
||||
result = StreamingParseResult("normal", "reasoning")
|
||||
self.assertEqual(result.normal_text, "normal")
|
||||
self.assertEqual(result.reasoning_text, "reasoning")
|
||||
|
||||
|
||||
class TestBaseReasoningFormatDetector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = BaseReasoningFormatDetector(
|
||||
@@ -44,15 +29,6 @@ class TestBaseReasoningFormatDetector(CustomTestCase):
|
||||
stream_reasoning=True,
|
||||
)
|
||||
|
||||
def test_init(self):
|
||||
"""Test initialization of BaseReasoningFormatDetector."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
self.assertEqual(self.detector._buffer, "")
|
||||
self.assertFalse(self.detector.stripped_think_start)
|
||||
|
||||
def test_detect_and_parse_normal_text(self):
|
||||
"""Test parsing normal text without reasoning."""
|
||||
text = "This is normal text"
|
||||
@@ -161,28 +137,6 @@ class TestDeepSeekR1Detector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = DeepSeekR1Detector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test DeepSeekR1Detector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertTrue(self.detector._in_reasoning) # force_reasoning=True
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_init_no_stream_reasoning(self):
|
||||
"""Test DeepSeekR1Detector with stream_reasoning=False."""
|
||||
detector = DeepSeekR1Detector(stream_reasoning=False)
|
||||
self.assertFalse(detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_r1_format(self):
|
||||
"""Test parsing DeepSeek-R1 format."""
|
||||
text = "I need to think about this. The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
# Should be treated as reasoning because force_reasoning=True
|
||||
self.assertEqual(
|
||||
result.reasoning_text, "I need to think about this. The answer is 42."
|
||||
)
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
def test_detect_and_parse_with_end_token(self):
|
||||
"""Test parsing with end token."""
|
||||
text = "I think this is the answer</think>The final answer is 42."
|
||||
@@ -203,20 +157,6 @@ class TestQwen3Detector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = Qwen3Detector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test Qwen3Detector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertFalse(self.detector._in_reasoning) # force_reasoning=False
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_qwen3_format(self):
|
||||
"""Test parsing Qwen3 format."""
|
||||
text = "<think>Let me think about this problem</think>The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "Let me think about this problem")
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_detect_and_parse_without_thinking(self):
|
||||
"""Test parsing without thinking (enable_thinking=False case)."""
|
||||
text = "Direct answer without thinking."
|
||||
@@ -225,70 +165,10 @@ class TestQwen3Detector(CustomTestCase):
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
|
||||
class TestQwen3ForcedReasoningDetector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = Qwen3Detector(force_reasoning=True)
|
||||
|
||||
def test_init(self):
|
||||
"""Test Qwen3ForcedReasoningDetector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertTrue(self.detector._in_reasoning) # force_reasoning=True
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_qwen3_forced_reasoning_format(self):
|
||||
"""Test parsing Qwen3-ForcedReasoning format (no <think> start tag)."""
|
||||
text = "I need to think about this step by step.</think>The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(
|
||||
result.reasoning_text, "I need to think about this step by step."
|
||||
)
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_detect_and_parse_with_start_token(self):
|
||||
"""Test parsing Qwen3-ForcedReasoning with optional <think> start tag."""
|
||||
text = "<think>I need to think about this.</think>The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
# Should work because base class logic handles both force_reasoning=True OR start token
|
||||
self.assertEqual(result.reasoning_text, "I need to think about this.")
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_streaming_qwen3_forced_reasoning_format(self):
|
||||
"""Test streaming parse of Qwen3-ForcedReasoning format."""
|
||||
# First chunk without <think> start
|
||||
result = self.detector.parse_streaming_increment("I need to")
|
||||
self.assertEqual(result.reasoning_text, "I need to")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
# More reasoning content
|
||||
result = self.detector.parse_streaming_increment(" think about this.")
|
||||
self.assertEqual(result.reasoning_text, " think about this.")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
# End token with normal text
|
||||
result = self.detector.parse_streaming_increment("</think>The answer is 42.")
|
||||
self.assertEqual(result.reasoning_text, "") # Buffer cleared
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
|
||||
class TestKimiDetector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = KimiDetector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test KimiDetector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "◁think▷")
|
||||
self.assertEqual(self.detector.think_end_token, "◁/think▷")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_kimi_format(self):
|
||||
"""Test parsing Kimi format."""
|
||||
text = "◁think▷Let me consider this carefully◁/think▷The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "Let me consider this carefully")
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_detect_and_parse_kimi_no_thinking(self):
|
||||
"""Test parsing Kimi format without thinking."""
|
||||
text = "Direct answer without thinking tokens."
|
||||
@@ -296,29 +176,6 @@ class TestKimiDetector(CustomTestCase):
|
||||
self.assertEqual(result.normal_text, text)
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_streaming_kimi_format(self):
|
||||
"""Test streaming parse of Kimi format."""
|
||||
# Test partial token
|
||||
result = self.detector.parse_streaming_increment("◁thi")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
# Complete start token
|
||||
result = self.detector.parse_streaming_increment("nk▷Start")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
self.assertEqual(result.reasoning_text, "Start")
|
||||
self.assertTrue(self.detector._in_reasoning)
|
||||
|
||||
# Add reasoning content
|
||||
result = self.detector.parse_streaming_increment("thinking...")
|
||||
self.assertEqual(result.reasoning_text, "thinking...")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
# End token - reasoning content is cleared when end token is processed
|
||||
result = self.detector.parse_streaming_increment("◁/think▷answer")
|
||||
self.assertEqual(result.reasoning_text, "") # Buffer cleared
|
||||
self.assertEqual(result.normal_text, "answer")
|
||||
|
||||
|
||||
class TestKimiK2Detector(CustomTestCase):
|
||||
"""Test cases for KimiK2 detector with tool interruption support."""
|
||||
@@ -334,36 +191,6 @@ class TestKimiK2Detector(CustomTestCase):
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_tool_interrupt(self):
|
||||
"""Test parsing with Kimi-K2 tool-section interruption."""
|
||||
text = "<think>thinking<|tool_calls_section_begin|><|tool_call_begin|>"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "thinking")
|
||||
self.assertEqual(
|
||||
result.normal_text, "<|tool_calls_section_begin|><|tool_call_begin|>"
|
||||
)
|
||||
|
||||
def test_streaming_tool_interrupt(self):
|
||||
"""Test streaming parse interrupted by tool section."""
|
||||
self.detector.parse_streaming_increment("<think>")
|
||||
result1 = self.detector.parse_streaming_increment("reasoning")
|
||||
self.assertEqual(result1.reasoning_text, "reasoning")
|
||||
self.assertEqual(result1.normal_text, "")
|
||||
|
||||
result2 = self.detector.parse_streaming_increment(
|
||||
"<|tool_calls_section_begin|>"
|
||||
)
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
self.assertEqual(result2.normal_text, "<|tool_calls_section_begin|>")
|
||||
|
||||
def test_streaming_after_interrupt_is_normal(self):
|
||||
"""After interruption, subsequent chunks should be normal text."""
|
||||
self.detector.parse_streaming_increment("<think>")
|
||||
self.detector.parse_streaming_increment("reasoning<|tool_calls_section_begin|>")
|
||||
result = self.detector.parse_streaming_increment("<|tool_call_begin|>")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
self.assertEqual(result.normal_text, "<|tool_call_begin|>")
|
||||
|
||||
|
||||
class TestGlm45Detector(CustomTestCase):
|
||||
"""Test cases for GLM45 detector with tool interruption support."""
|
||||
@@ -371,33 +198,6 @@ class TestGlm45Detector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = Glm45Detector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test Glm45Detector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertEqual(self.detector.tool_start_token, "<tool_call>")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_normal_reasoning(self):
|
||||
"""Test parsing normal reasoning block without tool interruption."""
|
||||
text = "<think>Let me think about this step by step</think>The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "Let me think about this step by step")
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_detect_and_parse_tool_interrupt(self):
|
||||
"""
|
||||
Test parsing with tool interruption.
|
||||
|
||||
GLM45 can interrupt reasoning with tool token (<tool_call>) without closing </think>.
|
||||
Should split at the first occurrence of tool_start_token using find().
|
||||
"""
|
||||
text = "<think>I need to think<tool_call>tool call data"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "I need to think")
|
||||
self.assertEqual(result.normal_text, "<tool_call>tool call data")
|
||||
|
||||
def test_detect_and_parse_multiple_tool_calls_find(self):
|
||||
"""
|
||||
Test that find() finds the FIRST occurrence of tool_start_token.
|
||||
@@ -413,17 +213,6 @@ class TestGlm45Detector(CustomTestCase):
|
||||
"<tool_call>first tool<tool_call>second tool<tool_call>final tool",
|
||||
)
|
||||
|
||||
def test_detect_and_parse_truncated_reasoning(self):
|
||||
"""
|
||||
Test truncated reasoning without tool or end tag.
|
||||
|
||||
Should return all content as reasoning_text.
|
||||
"""
|
||||
text = "<think>This is incomplete"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "This is incomplete")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
def test_detect_and_parse_normal_text_only(self):
|
||||
"""Test parsing text without reasoning block."""
|
||||
text = "Just the answer without any reasoning."
|
||||
@@ -431,50 +220,6 @@ class TestGlm45Detector(CustomTestCase):
|
||||
self.assertEqual(result.normal_text, text)
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_streaming_normal_flow(self):
|
||||
"""Test streaming with normal reasoning flow."""
|
||||
# Start reasoning
|
||||
result1 = self.detector.parse_streaming_increment("<think>")
|
||||
self.assertEqual(result1.normal_text, "")
|
||||
self.assertEqual(result1.reasoning_text, "")
|
||||
self.assertTrue(self.detector._in_reasoning)
|
||||
|
||||
# Reasoning content
|
||||
result2 = self.detector.parse_streaming_increment("thinking...")
|
||||
self.assertEqual(result2.normal_text, "")
|
||||
self.assertEqual(result2.reasoning_text, "thinking...")
|
||||
|
||||
# End reasoning
|
||||
result3 = self.detector.parse_streaming_increment("</think>answer")
|
||||
self.assertEqual(result3.normal_text, "answer")
|
||||
self.assertEqual(result3.reasoning_text, "")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
|
||||
def test_streaming_tool_interrupt_split_tokens(self):
|
||||
"""
|
||||
Test streaming with tool interruption where tool token is split across chunks.
|
||||
|
||||
This tests the buffer prefix logic that prevents partial emission of tool token.
|
||||
"""
|
||||
# Start reasoning
|
||||
self.detector.parse_streaming_increment("<think>")
|
||||
|
||||
# Add reasoning
|
||||
result1 = self.detector.parse_streaming_increment("thinking")
|
||||
self.assertEqual(result1.reasoning_text, "thinking")
|
||||
|
||||
# Send partial tool token (should be buffered, not emitted)
|
||||
result2 = self.detector.parse_streaming_increment("<tool_call>")
|
||||
# Tool token is in buffer, causing switch to normal mode
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
self.assertEqual(result2.normal_text, "<tool_call>")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
|
||||
# Send tool args
|
||||
result3 = self.detector.parse_streaming_increment("tool args")
|
||||
self.assertEqual(result3.reasoning_text, "")
|
||||
self.assertEqual(result3.normal_text, "tool args")
|
||||
|
||||
def test_streaming_no_stream_reasoning(self):
|
||||
"""Test streaming without stream_reasoning enabled."""
|
||||
detector = Glm45Detector(stream_reasoning=False)
|
||||
@@ -526,21 +271,6 @@ class TestHunyuanDetector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = HunyuanDetector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test HunyuanDetector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertEqual(self.detector.tool_start_token, "<tool_calls>")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_normal_reasoning(self):
|
||||
"""Test parsing normal reasoning block without tool interruption."""
|
||||
text = "<think>Let me think about this</think>The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "Let me think about this")
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_detect_and_parse_without_thinking(self):
|
||||
"""Test parsing without thinking tokens (no_think mode)."""
|
||||
text = "Direct answer without thinking."
|
||||
@@ -548,42 +278,6 @@ class TestHunyuanDetector(CustomTestCase):
|
||||
self.assertEqual(result.normal_text, text)
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_detect_and_parse_tool_interrupt(self):
|
||||
"""Test parsing with tool call interruption during reasoning."""
|
||||
text = "<think>I need to check<tool_calls><tool_call>get_weather<tool_sep></tool_call></tool_calls>"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "I need to check")
|
||||
self.assertIn("<tool_calls>", result.normal_text)
|
||||
|
||||
def test_streaming_normal_reasoning(self):
|
||||
"""Test streaming parse of normal reasoning block."""
|
||||
self.detector.parse_streaming_increment("<think>")
|
||||
result1 = self.detector.parse_streaming_increment("reasoning content")
|
||||
self.assertEqual(result1.reasoning_text, "reasoning content")
|
||||
|
||||
result2 = self.detector.parse_streaming_increment("</think>answer")
|
||||
self.assertEqual(result2.normal_text, "answer")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
|
||||
def test_streaming_tool_interrupt(self):
|
||||
"""Test streaming parse interrupted by tool call section."""
|
||||
self.detector.parse_streaming_increment("<think>")
|
||||
result1 = self.detector.parse_streaming_increment("thinking")
|
||||
self.assertEqual(result1.reasoning_text, "thinking")
|
||||
|
||||
result2 = self.detector.parse_streaming_increment("<tool_calls>")
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
self.assertEqual(result2.normal_text, "<tool_calls>")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
|
||||
def test_streaming_after_interrupt_is_normal(self):
|
||||
"""After tool interruption, subsequent chunks should be normal text."""
|
||||
self.detector.parse_streaming_increment("<think>")
|
||||
self.detector.parse_streaming_increment("reasoning<tool_calls>")
|
||||
result = self.detector.parse_streaming_increment("<tool_call>data")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
self.assertEqual(result.normal_text, "<tool_call>data")
|
||||
|
||||
def test_reasoning_parser_integration(self):
|
||||
"""Test Hunyuan through ReasoningParser API."""
|
||||
parser = ReasoningParser("hunyuan")
|
||||
@@ -617,21 +311,6 @@ class TestNemotron3Detector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = Nemotron3Detector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test Nemotron3Detector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<think>")
|
||||
self.assertEqual(self.detector.think_end_token, "</think>")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
self.assertFalse(self.detector._force_nonempty_content)
|
||||
|
||||
def test_detect_and_parse_complete_reasoning(self):
|
||||
"""Test parsing complete reasoning block."""
|
||||
text = "<think>Let me think about this</think>The answer is 42."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "Let me think about this")
|
||||
self.assertEqual(result.normal_text, "The answer is 42.")
|
||||
|
||||
def test_detect_and_parse_no_thinking(self):
|
||||
"""Test parsing without thinking tokens."""
|
||||
text = "Direct answer without thinking."
|
||||
@@ -671,28 +350,11 @@ class TestNemotron3Detector(CustomTestCase):
|
||||
self.assertEqual(result.normal_text, "Truncated reasoning without end token")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_force_nonempty_content_no_thinking_tokens(self):
|
||||
"""Test force_nonempty_content with plain text (no thinking tokens)."""
|
||||
detector = Nemotron3Detector(force_nonempty_content=True)
|
||||
text = "Plain text without any thinking."
|
||||
result = detector.detect_and_parse(text)
|
||||
# Normal text already exists, no swap needed
|
||||
self.assertEqual(result.normal_text, text)
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
|
||||
class TestGemma4Detector(CustomTestCase):
|
||||
def setUp(self):
|
||||
self.detector = Gemma4Detector()
|
||||
|
||||
def test_init(self):
|
||||
"""Test Gemma4Detector initialization."""
|
||||
self.assertEqual(self.detector.think_start_token, "<|channel>")
|
||||
self.assertEqual(self.detector.think_end_token, "<channel|>")
|
||||
self.assertEqual(self.detector.think_start_self_label, "thought\n")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertTrue(self.detector.stream_reasoning)
|
||||
|
||||
def test_detect_and_parse_complete_reasoning(self):
|
||||
"""Test parsing complete Gemma4 reasoning block (think_start_self_label is stripped)."""
|
||||
text = "<|channel>thought\nLet me think about this<channel|>The answer is 42."
|
||||
@@ -707,49 +369,6 @@ class TestGemma4Detector(CustomTestCase):
|
||||
self.assertEqual(result.normal_text, text)
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_detect_and_parse_reasoning_only(self):
|
||||
"""Test parsing when output is all reasoning (no end token yet)."""
|
||||
text = "<|channel>thought\nStill thinking..."
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "Still thinking...")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
def test_streaming_complete_flow(self):
|
||||
"""Test streaming parse of Gemma4 reasoning flow."""
|
||||
chunks = [
|
||||
"<|channel>",
|
||||
"thought\nreasoning content",
|
||||
"<channel|>",
|
||||
"final answer",
|
||||
]
|
||||
all_reasoning = ""
|
||||
all_normal = ""
|
||||
for chunk in chunks:
|
||||
result = self.detector.parse_streaming_increment(chunk)
|
||||
all_reasoning += result.reasoning_text
|
||||
all_normal += result.normal_text
|
||||
self.assertIn("reasoning content", all_reasoning)
|
||||
self.assertIn("final answer", all_normal)
|
||||
|
||||
def test_streaming_full_start_sequence(self):
|
||||
"""Test streaming with the full start sequence (token + self_label)."""
|
||||
# Gemma4 start sequence is "<|channel>thought\n", not just "<|channel>"
|
||||
result = self.detector.parse_streaming_increment("<|channel>thought\n")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
self.assertTrue(self.detector._in_reasoning)
|
||||
|
||||
result = self.detector.parse_streaming_increment("reasoning content")
|
||||
self.assertEqual(result.reasoning_text, "reasoning content")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
|
||||
def test_streaming_partial_start_buffered(self):
|
||||
"""Test that partial start sequence is buffered."""
|
||||
# "<|channel>" alone is a prefix of "<|channel>thought\n", so it's buffered
|
||||
result = self.detector.parse_streaming_increment("<|channel>")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
|
||||
def test_streaming_end_token_mid_chunk(self):
|
||||
"""Test end token arriving in the same chunk as reasoning content."""
|
||||
self.detector.parse_streaming_increment("<|channel>thought\n")
|
||||
@@ -760,18 +379,6 @@ class TestGemma4Detector(CustomTestCase):
|
||||
self.assertEqual(result.normal_text, "the answer")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
|
||||
def test_streaming_split_end_token(self):
|
||||
"""Test end token split across two chunks."""
|
||||
self.detector.parse_streaming_increment("<|channel>thought\n")
|
||||
self.detector.parse_streaming_increment("reasoning content")
|
||||
|
||||
result1 = self.detector.parse_streaming_increment("<chan")
|
||||
self.assertEqual(result1.normal_text, "")
|
||||
|
||||
result2 = self.detector.parse_streaming_increment("nel|>final answer")
|
||||
self.assertFalse(self.detector._in_reasoning)
|
||||
self.assertIn("final answer", result2.normal_text)
|
||||
|
||||
def test_streaming_self_label_split_across_chunks(self):
|
||||
"""Test self_label ('thought\\n') arriving separately from start token."""
|
||||
result1 = self.detector.parse_streaming_increment("<|channel>")
|
||||
@@ -784,37 +391,6 @@ class TestGemma4Detector(CustomTestCase):
|
||||
result3 = self.detector.parse_streaming_increment("reasoning here")
|
||||
self.assertEqual(result3.reasoning_text, "reasoning here")
|
||||
|
||||
def test_streaming_force_reasoning(self):
|
||||
"""Test streaming with force_reasoning=True (no start token needed)."""
|
||||
detector = Gemma4Detector(force_reasoning=True)
|
||||
|
||||
result1 = detector.parse_streaming_increment("reasoning content")
|
||||
self.assertEqual(result1.reasoning_text, "reasoning content")
|
||||
self.assertEqual(result1.normal_text, "")
|
||||
|
||||
result2 = detector.parse_streaming_increment("<channel|>the answer")
|
||||
self.assertFalse(detector._in_reasoning)
|
||||
self.assertIn("the answer", result2.normal_text)
|
||||
|
||||
def test_streaming_multiple_reasoning_chunks(self):
|
||||
"""Test reasoning content arriving in many small chunks."""
|
||||
self.detector.parse_streaming_increment("<|channel>thought\n")
|
||||
|
||||
all_reasoning = ""
|
||||
for chunk in ["Think", "ing ", "step ", "by ", "step."]:
|
||||
result = self.detector.parse_streaming_increment(chunk)
|
||||
all_reasoning += result.reasoning_text
|
||||
self.assertEqual(result.normal_text, "")
|
||||
self.assertEqual(all_reasoning, "Thinking step by step.")
|
||||
|
||||
def test_force_reasoning(self):
|
||||
"""Test Gemma4Detector with force_reasoning=True."""
|
||||
detector = Gemma4Detector(force_reasoning=True)
|
||||
text = "This should be reasoning<channel|>The answer."
|
||||
result = detector.detect_and_parse(text)
|
||||
self.assertEqual(result.reasoning_text, "This should be reasoning")
|
||||
self.assertEqual(result.normal_text, "The answer.")
|
||||
|
||||
|
||||
class TestReasoningParser(CustomTestCase):
|
||||
def test_init_valid_model(self):
|
||||
@@ -990,42 +566,6 @@ class TestReasoningParser(CustomTestCase):
|
||||
class TestIntegrationScenarios(CustomTestCase):
|
||||
"""Integration tests for realistic usage scenarios."""
|
||||
|
||||
def test_deepseek_r1_complete_response(self):
|
||||
"""Test complete DeepSeek-R1 response parsing."""
|
||||
parser = ReasoningParser("deepseek-r1")
|
||||
text = "I need to solve this step by step. First, I'll analyze the problem. The given equation is x + 2 = 5. To solve for x, I subtract 2 from both sides: x = 5 - 2 = 3.</think>The answer is x = 3."
|
||||
|
||||
reasoning, normal = parser.parse_non_stream(text)
|
||||
self.assertIn("step by step", reasoning)
|
||||
self.assertIn(
|
||||
"= 3", reasoning
|
||||
) # The reasoning contains "x = 5 - 2 = 3" which has "= 3"
|
||||
self.assertEqual(normal, "The answer is x = 3.")
|
||||
|
||||
def test_qwen3_streaming_scenario(self):
|
||||
"""Test Qwen3 streaming scenario."""
|
||||
parser = ReasoningParser("qwen3")
|
||||
|
||||
chunks = [
|
||||
"<think>",
|
||||
"Let me analyze this problem.",
|
||||
" I need to consider multiple factors.",
|
||||
"</think>",
|
||||
"Based on my analysis, the solution is to use a different approach.",
|
||||
]
|
||||
|
||||
all_reasoning = ""
|
||||
all_normal = ""
|
||||
|
||||
for chunk in chunks:
|
||||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||||
all_reasoning += reasoning
|
||||
all_normal += normal
|
||||
|
||||
self.assertIn("analyze", all_reasoning)
|
||||
self.assertIn("multiple factors", all_reasoning)
|
||||
self.assertIn("different approach", all_normal)
|
||||
|
||||
def test_kimi_streaming_scenario(self):
|
||||
"""Test Kimi streaming scenario."""
|
||||
parser = ReasoningParser("kimi")
|
||||
@@ -1049,35 +589,6 @@ class TestIntegrationScenarios(CustomTestCase):
|
||||
self.assertIn("multiple factors", all_reasoning)
|
||||
self.assertIn("42", all_normal)
|
||||
|
||||
def test_gemma4_complete_response(self):
|
||||
"""Test complete Gemma4 response parsing (think_start_self_label stripped)."""
|
||||
parser = ReasoningParser("gemma4")
|
||||
text = "<|channel>thought\nI need to solve x + 2 = 5. Subtracting 2: x = 3.<channel|>The answer is x = 3."
|
||||
reasoning, normal = parser.parse_non_stream(text)
|
||||
self.assertIn("x = 3", reasoning)
|
||||
self.assertNotIn("thought\n", reasoning)
|
||||
self.assertEqual(normal, "The answer is x = 3.")
|
||||
|
||||
def test_gemma4_streaming_scenario(self):
|
||||
"""Test Gemma4 streaming scenario."""
|
||||
parser = ReasoningParser("gemma4")
|
||||
chunks = [
|
||||
"<|channel>",
|
||||
"thought\nLet me analyze.",
|
||||
" Multiple factors.",
|
||||
"<channel|>",
|
||||
"The solution is 42.",
|
||||
]
|
||||
all_reasoning = ""
|
||||
all_normal = ""
|
||||
for chunk in chunks:
|
||||
reasoning, normal = parser.parse_stream_chunk(chunk)
|
||||
all_reasoning += reasoning
|
||||
all_normal += normal
|
||||
self.assertIn("analyze", all_reasoning)
|
||||
self.assertIn("Multiple factors", all_reasoning)
|
||||
self.assertIn("42", all_normal)
|
||||
|
||||
def test_empty_reasoning_blocks(self):
|
||||
"""Test handling of empty reasoning blocks."""
|
||||
parser = ReasoningParser("qwen3")
|
||||
@@ -1157,22 +668,6 @@ class TestBufferLossBugFix(CustomTestCase):
|
||||
self.assertEqual(result2.normal_text, "</answer")
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
|
||||
def test_partial_start_tag_buffer_preservation(self):
|
||||
"""
|
||||
Test that partial start tag fragments are properly preserved.
|
||||
"""
|
||||
detector = BaseReasoningFormatDetector("<think>", "</think>")
|
||||
|
||||
# Send partial start tag
|
||||
result1 = detector.parse_streaming_increment("<th")
|
||||
self.assertEqual(result1.normal_text, "")
|
||||
self.assertEqual(result1.reasoning_text, "")
|
||||
|
||||
# Complete with non-matching text
|
||||
result2 = detector.parse_streaming_increment("is is text")
|
||||
self.assertEqual(result2.normal_text, "<this is text")
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
|
||||
def test_partial_end_tag_in_reasoning_mode(self):
|
||||
"""
|
||||
Test partial end tag handling when already in reasoning mode.
|
||||
@@ -1194,25 +689,6 @@ class TestBufferLossBugFix(CustomTestCase):
|
||||
# The reasoning text should be empty since buffer was cleared when end tag was processed
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
|
||||
def test_multiple_partial_fragments(self):
|
||||
"""
|
||||
Test handling of multiple partial fragments that don't match any tokens.
|
||||
"""
|
||||
detector = BaseReasoningFormatDetector("<think>", "</think>")
|
||||
|
||||
# Send multiple partial fragments
|
||||
result1 = detector.parse_streaming_increment("<")
|
||||
self.assertEqual(result1.normal_text, "")
|
||||
self.assertEqual(result1.reasoning_text, "")
|
||||
|
||||
result2 = detector.parse_streaming_increment("/")
|
||||
self.assertEqual(result2.normal_text, "")
|
||||
self.assertEqual(result2.reasoning_text, "")
|
||||
|
||||
result3 = detector.parse_streaming_increment("random>")
|
||||
self.assertEqual(result3.normal_text, "</random>")
|
||||
self.assertEqual(result3.reasoning_text, "")
|
||||
|
||||
def test_edge_case_exact_token_match(self):
|
||||
"""
|
||||
Test edge case where buffer content exactly matches a token.
|
||||
@@ -1242,19 +718,6 @@ class TestGptOssDetector(CustomTestCase):
|
||||
|
||||
self.detector = GptOssDetector()
|
||||
|
||||
def test_detect_and_parse_with_analysis_and_final(self):
|
||||
"""Test one-shot parsing with analysis (reasoning) and final (normal) blocks."""
|
||||
text = "<|start|><|channel|>analysis<|message|>thinking hard<|end|><|channel|>final<|message|>the answer<|end|>"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertIn("thinking hard", result.reasoning_text)
|
||||
self.assertIn("the answer", result.normal_text)
|
||||
|
||||
def test_detect_and_parse_normal_only(self):
|
||||
"""Test one-shot parsing with only final block."""
|
||||
text = "<|start|><|channel|>final<|message|>just the answer<|end|>"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertIn("just the answer", result.normal_text)
|
||||
|
||||
def test_streaming_analysis_then_final(self):
|
||||
"""Test streaming parse across multiple chunks."""
|
||||
chunks = [
|
||||
@@ -1273,13 +736,6 @@ class TestGptOssDetector(CustomTestCase):
|
||||
self.assertIn("reasoning part", all_reasoning)
|
||||
self.assertIn("answer", all_normal)
|
||||
|
||||
def test_streaming_with_tool_call(self):
|
||||
"""Test streaming parse with tool call events."""
|
||||
text = "<|start|><|channel|>analysis<|message|>think<|end|><|call|>tool_data<|return|><|channel|>final<|message|>result<|end|>"
|
||||
result = self.detector.detect_and_parse(text)
|
||||
self.assertIn("think", result.reasoning_text)
|
||||
self.assertIn("result", result.normal_text)
|
||||
|
||||
|
||||
class TestMiniMaxAppendThinkDetector(CustomTestCase):
|
||||
"""Test cases for MiniMaxAppendThinkDetector."""
|
||||
@@ -1478,18 +934,6 @@ class TestContinueFinalMessage(CustomTestCase):
|
||||
self.assertEqual(result.reasoning_text, "new reasoning")
|
||||
self.assertEqual(result.normal_text, "new answer")
|
||||
|
||||
def test_streaming_returns_empty_when_in_reasoning_and_end_buffered(self):
|
||||
"""Test that streaming returns empty when buffer could be partial end token."""
|
||||
detector = BaseReasoningFormatDetector(
|
||||
"<think>", "</think>", force_reasoning=True, stream_reasoning=True
|
||||
)
|
||||
# In reasoning mode, send partial end token
|
||||
result = detector.parse_streaming_increment("</")
|
||||
self.assertEqual(result.reasoning_text, "")
|
||||
self.assertEqual(result.normal_text, "")
|
||||
# This goes through the path where _in_reasoning is True but buffer
|
||||
# is a prefix of think_end_token → returns empty
|
||||
|
||||
|
||||
class TestGptOssDetectorToolCall(CustomTestCase):
|
||||
"""Test GptOssDetector tool_call raw_text handling."""
|
||||
|
||||
Reference in New Issue
Block a user