# SPDX-License-Identifier: Apache-2.0
"""Comprehensive tests for tool call parsers."""

import json

import pytest

from vllm_mlx.tool_parsers import (
    AutoToolParser,
    DeepSeekToolParser,
    FunctionaryToolParser,
    Gemma4ToolParser,
    Glm47ToolParser,
    GraniteToolParser,
    HermesToolParser,
    KimiToolParser,
    LlamaToolParser,
    MistralToolParser,
    NemotronToolParser,
    QwenToolParser,
    ToolParserManager,
    xLAMToolParser,
)


class TestToolParserManager:
    """Test the ToolParserManager registry."""

    def test_list_registered(self):
        """Test that all expected parsers are registered."""
        parsers = ToolParserManager.list_registered()
        expected = [
            "auto",
            "mistral",
            "qwen",
            "llama",
            "hermes",
            "deepseek",
            "kimi",
            "granite",
            "nemotron",
            "xlam",
            "functionary",
            "gemma4",
        ]
        for p in expected:
            assert p in parsers, f"Parser '{p}' not found"

    def test_get_tool_parser_by_name(self):
        """Test getting parsers by name."""
        test_cases = [
            ("mistral", MistralToolParser),
            ("qwen", QwenToolParser),
            ("qwen3", QwenToolParser),
            ("llama", LlamaToolParser),
            ("llama3", LlamaToolParser),
            ("llama4", LlamaToolParser),
            ("auto", AutoToolParser),
            ("deepseek", DeepSeekToolParser),
            ("deepseek_v3", DeepSeekToolParser),
            ("deepseek_r1", DeepSeekToolParser),
            ("kimi", KimiToolParser),
            ("kimi_k2", KimiToolParser),
            ("moonshot", KimiToolParser),
            ("granite", GraniteToolParser),
            ("granite3", GraniteToolParser),
            ("nemotron", NemotronToolParser),
            ("nemotron3", NemotronToolParser),
            ("xlam", xLAMToolParser),
            ("functionary", FunctionaryToolParser),
            ("meetkai", FunctionaryToolParser),
            ("hermes", HermesToolParser),
            ("nous", HermesToolParser),
            ("gemma4", Gemma4ToolParser),
        ]
        for name, expected_cls in test_cases:
            parser_cls = ToolParserManager.get_tool_parser(name)
            assert parser_cls == expected_cls, f"Parser '{name}' returned wrong class"

    def test_get_unknown_parser_raises(self):
        """Test that unknown parser raises KeyError."""
        with pytest.raises(KeyError):
            ToolParserManager.get_tool_parser("unknown_parser")

    def test_parser_instantiation(self):
        """Test that all parsers can be instantiated without tokenizer."""
        for name in [
            "auto",
            "mistral",
            "qwen",
            "llama",
            "hermes",
            "deepseek",
            "kimi",
            "granite",
            "nemotron",
            "xlam",
            "functionary",
        ]:
            parser_cls = ToolParserManager.get_tool_parser(name)
            parser = parser_cls()  # Should not raise
            assert parser is not None


class TestMistralToolParser:
    """Test the Mistral tool parser."""

    @pytest.fixture
    def parser(self):
        return MistralToolParser()

    def test_old_format_single(self, parser):
        """Test parsing old Mistral format with single tool call."""
        text = '[TOOL_CALLS] [{"name": "get_weather", "arguments": {"city": "Paris"}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["city"] == "Paris"

    def test_old_format_multiple(self, parser):
        """Test parsing old Mistral format with multiple tool calls."""
        text = '[TOOL_CALLS] [{"name": "get_weather", "arguments": {"city": "Paris"}}, {"name": "get_time", "arguments": {"timezone": "UTC"}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2
        assert result.tool_calls[0]["name"] == "get_weather"
        assert result.tool_calls[1]["name"] == "get_time"

    def test_new_format(self, parser):
        """Test parsing new Mistral format."""
        text = '[TOOL_CALLS]get_weather{"city": "London"}'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"

    def test_no_tool_call(self, parser):
        """Test that regular text is not parsed as tool call."""
        text = "Hello, how can I help you today?"
        result = parser.extract_tool_calls(text)

        assert not result.tools_called
        assert result.content == text

    def test_content_with_tool_call(self, parser):
        """Test content before tool call is preserved."""
        text = 'Let me check the weather for you.[TOOL_CALLS] [{"name": "get_weather", "arguments": {}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.content == "Let me check the weather for you."


class TestQwenToolParser:
    """Test the Qwen tool parser."""

    @pytest.fixture
    def parser(self):
        return QwenToolParser()

    def test_xml_format(self, parser):
        """Test parsing Qwen XML format."""
        text = '<tool_call>{"name": "calculate", "arguments": {"x": 1, "y": 2}}</tool_call>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "calculate"

    def test_bracket_format(self, parser):
        """Test parsing Qwen bracket format (Qwen3 style)."""
        text = '[Calling tool: add({"a": 5, "b": 3})]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "add"

    def test_multiple_xml_calls(self, parser):
        """Test multiple XML tool calls."""
        text = '<tool_call>{"name": "func1", "arguments": {}}</tool_call><tool_call>{"name": "func2", "arguments": {}}</tool_call>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "I can help you with that question."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestLlamaToolParser:
    """Test the Llama tool parser."""

    @pytest.fixture
    def parser(self):
        return LlamaToolParser()

    def test_function_format(self, parser):
        """Test parsing Llama function format."""
        text = '<function=multiply>{"x": 3, "y": 4}</function>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "multiply"

    def test_multiple_functions(self, parser):
        """Test parsing multiple function calls."""
        text = '<function=add>{"a": 1}</function><function=multiply>{"x": 3}</function>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2
        assert result.tool_calls[0]["name"] == "add"
        assert result.tool_calls[1]["name"] == "multiply"

    def test_content_with_function(self, parser):
        """Test content before function call."""
        text = 'Computing result<function=calc>{"n": 5}</function>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.content == "Computing result"


class TestHermesToolParser:
    """Test the Hermes tool parser."""

    @pytest.fixture
    def parser(self):
        return HermesToolParser()

    def test_tool_call_format(self, parser):
        """Test parsing Hermes format."""
        text = (
            '<tool_call>{"name": "search", "arguments": {"query": "test"}}</tool_call>'
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "search"

    def test_with_reasoning(self, parser):
        """Test with reasoning block."""
        text = '<tool_call_reasoning>I need to search for this</tool_call_reasoning><tool_call>{"name": "search", "arguments": {}}</tool_call>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert "Reasoning" in (result.content or "")


class TestDeepSeekToolParser:
    """Test the DeepSeek tool parser."""

    @pytest.fixture
    def parser(self):
        return DeepSeekToolParser()

    def test_deepseek_format(self, parser):
        """Test parsing DeepSeek V3 format."""
        text = """<｜tool▁calls▁begin｜>
<｜tool▁call▁begin｜>function<｜tool▁sep｜>get_weather
```json
{"city": "Tokyo"}
```<｜tool▁call▁end｜>
<｜tool▁calls▁end｜>"""
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"

    def test_multiple_calls(self, parser):
        """Test multiple DeepSeek tool calls."""
        text = """<｜tool▁calls▁begin｜>
<｜tool▁call▁begin｜>function<｜tool▁sep｜>func1
```json
{"a": 1}
```<｜tool▁call▁end｜>
<｜tool▁call▁begin｜>function<｜tool▁sep｜>func2
```json
{"b": 2}
```<｜tool▁call▁end｜>
<｜tool▁calls▁end｜>"""
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2

    def test_content_before_tools(self, parser):
        """Test content before tool calls is preserved."""
        text = """Let me help you with that.<｜tool▁calls▁begin｜>
<｜tool▁call▁begin｜>function<｜tool▁sep｜>search
```json
{}
```<｜tool▁call▁end｜>
<｜tool▁calls▁end｜>"""
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.content == "Let me help you with that."

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "Here is my response without any tool calls."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestKimiToolParser:
    """Test the Kimi tool parser."""

    @pytest.fixture
    def parser(self):
        return KimiToolParser()

    def test_kimi_format(self, parser):
        """Test parsing Kimi K2 format."""
        text = """<|tool_calls_section_begin|>
<|tool_call_begin|>functions.get_weather:0<|tool_call_argument_begin|>{"city": "Beijing"}<|tool_call_end|>
<|tool_calls_section_end|>"""
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"

    def test_simple_function_name(self, parser):
        """Test with simple function name (no functions. prefix)."""
        text = (
            "<|tool_call_begin|>search:0<|tool_call_argument_begin|>{}<|tool_call_end|>"
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "I'll answer your question directly."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestGraniteToolParser:
    """Test the Granite tool parser."""

    @pytest.fixture
    def parser(self):
        return GraniteToolParser()

    def test_granite_30_format(self, parser):
        """Test parsing Granite 3.0 format."""
        text = (
            '<|tool_call|>[{"name": "calculate", "arguments": {"expression": "2+2"}}]'
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "calculate"

    def test_granite_31_format(self, parser):
        """Test parsing Granite 3.1 format."""
        text = '<tool_call>[{"name": "search", "arguments": {"query": "test"}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_multiple_calls(self, parser):
        """Test multiple tool calls."""
        text = '<|tool_call|>[{"name": "func1", "arguments": {}}, {"name": "func2", "arguments": {}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "The answer is 42."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestNemotronToolParser:
    """Test the Nemotron tool parser."""

    @pytest.fixture
    def parser(self):
        return NemotronToolParser()

    def test_parameter_format(self, parser):
        """Test parsing Nemotron parameter format."""
        text = "<tool_call><function=get_weather><parameter=city>Paris</parameter><parameter=units>celsius</parameter></function></tool_call>"
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["city"] == "Paris"
        assert args["units"] == "celsius"

    def test_json_format(self, parser):
        """Test parsing Nemotron with JSON arguments."""
        text = '<tool_call><function=calculate>{"expression": "2*3"}</function></tool_call>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "calculate"

    def test_multiple_calls(self, parser):
        """Test multiple Nemotron tool calls."""
        text = "<tool_call><function=func1><parameter=a>1</parameter></function></tool_call><tool_call><function=func2><parameter=b>2</parameter></function></tool_call>"
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "Here is the information you requested."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestXLAMToolParser:
    """Test the xLAM tool parser."""

    @pytest.fixture
    def parser(self):
        return xLAMToolParser()

    def test_json_array(self, parser):
        """Test parsing JSON array format."""
        text = '[{"name": "search", "arguments": {"query": "AI"}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_code_block(self, parser):
        """Test parsing markdown code block."""
        text = '```json\n[{"name": "calculate", "arguments": {"x": 5}}]\n```'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "calculate"

    def test_after_think(self, parser):
        """Test parsing after </think> tag."""
        text = (
            '<think>Let me search for this</think>[{"name": "search", "arguments": {}}]'
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_tool_calls_tag(self, parser):
        """Test [TOOL_CALLS] tag format."""
        text = '[TOOL_CALLS][{"name": "func", "arguments": {}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "I don't need to use any tools for this."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestFunctionaryToolParser:
    """Test the Functionary tool parser."""

    @pytest.fixture
    def parser(self):
        return FunctionaryToolParser()

    def test_recipient_format(self, parser):
        """Test parsing Functionary v3 recipient format."""
        text = '<|from|>assistant\n<|recipient|>get_weather\n<|content|>{"city": "NYC"}'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "get_weather"

    def test_function_format(self, parser):
        """Test parsing function format."""
        text = '<function=search>{"query": "test"}</function>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_json_array(self, parser):
        """Test parsing JSON array."""
        text = '[{"name": "func1", "arguments": {}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "Let me explain that to you."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called


class TestAutoToolParser:
    """Test the auto-detecting tool parser."""

    @pytest.fixture
    def parser(self):
        return AutoToolParser()

    def test_detects_mistral(self, parser):
        """Test auto detection of Mistral format."""
        text = '[TOOL_CALLS] [{"name": "search", "arguments": {}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_detects_qwen_xml(self, parser):
        """Test auto detection of Qwen XML format."""
        text = '<tool_call>{"name": "calculate", "arguments": {}}</tool_call>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "calculate"

    def test_detects_qwen_bracket(self, parser):
        """Test auto detection of Qwen bracket format."""
        text = '[Calling tool: add({"a": 1})]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "add"

    def test_detects_bare_bracket(self, parser):
        """Test auto detection of bare bracket format."""
        text = '[read({"file_path": "/tmp/test.py"})]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "read"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["file_path"] == "/tmp/test.py"

    def test_detects_llama(self, parser):
        """Test auto detection of Llama format."""
        text = '<function=multiply>{"x": 2}</function>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "multiply"

    def test_detects_nemotron(self, parser):
        """Test auto detection of Nemotron format."""
        text = "<tool_call><function=search><parameter=q>test</parameter></function></tool_call>"
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "search"

    def test_detects_raw_json(self, parser):
        """Test auto detection of raw JSON format."""
        text = '{"name": "test_func", "arguments": {"key": "value"}}'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "test_func"

    def test_no_tool_call(self, parser):
        """Test text without tool calls."""
        text = "This is just a regular response."
        result = parser.extract_tool_calls(text)

        assert not result.tools_called

    @pytest.mark.parametrize(
        "text",
        [
            "Read [the docs](https://example.test) before changing this.",
            "The output array was [1, 2, 3], not a tool call.",
            'Here is JSON: [{"label": "alpha", "score": 0.7}]',
            "Use [square brackets] for optional text.",
            "This prose mentions [read(file_path)] without JSON arguments.",
        ],
    )
    def test_no_false_positive_for_ordinary_bracket_text(self, parser, text):
        """Ordinary bracket text should remain content, not auto tool calls."""
        result = parser.extract_tool_calls(text)

        assert not result.tools_called
        assert result.content == text


class TestEdgeCases:
    """Test edge cases and error handling."""

    def test_empty_input(self):
        """Test with empty input."""
        parsers = [
            MistralToolParser(),
            QwenToolParser(),
            LlamaToolParser(),
            DeepSeekToolParser(),
            AutoToolParser(),
        ]
        for parser in parsers:
            result = parser.extract_tool_calls("")
            assert not result.tools_called

    def test_malformed_json(self):
        """Test with malformed JSON."""
        parser = MistralToolParser()
        text = '[TOOL_CALLS] [{"name": "func", "arguments": {invalid json}]'
        result = parser.extract_tool_calls(text)
        # Should not crash, may or may not parse

    def test_nested_arguments(self):
        """Test with deeply nested arguments."""
        parser = AutoToolParser()
        args = {"level1": {"level2": {"level3": [1, 2, 3]}}}
        text = f'{{"name": "complex", "arguments": {json.dumps(args)}}}'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        parsed_args = json.loads(result.tool_calls[0]["arguments"])
        assert parsed_args["level1"]["level2"]["level3"] == [1, 2, 3]

    def test_unicode_in_arguments(self):
        """Test with unicode characters in arguments."""
        parser = MistralToolParser()
        text = '[TOOL_CALLS] [{"name": "translate", "arguments": {"text": "日本語"}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["text"] == "日本語"

    def test_special_characters_in_name(self):
        """Test function names with special characters."""
        parser = LlamaToolParser()
        text = '<function=get_user_info>{"user_id": 123}</function>'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "get_user_info"

    def test_tool_call_id_uniqueness(self):
        """Test that each tool call gets a unique ID."""
        parser = MistralToolParser()
        text = '[TOOL_CALLS] [{"name": "func1", "arguments": {}}, {"name": "func2", "arguments": {}}]'
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        ids = [tc["id"] for tc in result.tool_calls]
        assert len(ids) == len(set(ids)), "Tool call IDs should be unique"


class TestBareBracketStreaming:
    """Test streaming for bare bracket tool calls."""

    def test_auto_streaming_bare_bracket(self):
        """Auto parser should emit structured tool calls for bare bracket streaming."""
        parser = AutoToolParser()

        chunks = [
            "[read(",
            '{"file_path": "/tmp/test.py"}',
            ")]",
        ]
        accumulated = ""
        tool_calls_found = False

        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                tool_calls_found = True
                assert r["tool_calls"][0]["function"]["name"] == "read"
                args = json.loads(r["tool_calls"][0]["function"]["arguments"])
                assert args["file_path"] == "/tmp/test.py"
                break

        assert tool_calls_found

    @pytest.mark.parametrize(
        "delta",
        [
            "See [the docs](https://example.test)",
            "Values: [1, 2, 3]",
            'JSON data: [{"label": "alpha"}]',
            "Literal [square brackets] in text",
        ],
    )
    def test_auto_streaming_ordinary_brackets_emit_content(self, delta):
        """Auto streaming should not suppress ordinary bracket text."""
        parser = AutoToolParser()

        result = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text=delta,
            delta_text=delta,
        )

        assert result == {"content": delta}


class TestStreamingParsing:
    """Test streaming tool call parsing."""

    def test_mistral_streaming(self):
        """Test Mistral streaming parsing."""
        parser = MistralToolParser()

        # Simulate streaming
        result1 = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text="Let me",
            delta_text="Let me",
        )
        assert result1 == {"content": "Let me"}

        result2 = parser.extract_tool_calls_streaming(
            previous_text="Let me",
            current_text="Let me[TOOL_CALLS]",
            delta_text="[TOOL_CALLS]",
        )
        # Should start tool call parsing

    def test_auto_streaming(self):
        """Test auto parser streaming."""
        parser = AutoToolParser()

        result = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text="Hello world",
            delta_text="Hello world",
        )
        assert result == {"content": "Hello world"}


class TestThinkTagStripping:
    """Test <think> tag stripping in tool parsers (Issue #26)."""

    def test_strip_think_tags_utility(self):
        """Test the strip_think_tags static method."""
        from vllm_mlx.tool_parsers.abstract_tool_parser import ToolParser

        # Basic stripping
        text = "<think>Let me analyze this</think>The answer is 42"
        assert ToolParser.strip_think_tags(text) == "The answer is 42"

        # Multi-line thinking
        text = "<think>Step 1\nStep 2\nStep 3</think>Result"
        assert ToolParser.strip_think_tags(text) == "Result"

        # No think tags
        text = "Just regular text"
        assert ToolParser.strip_think_tags(text) == "Just regular text"

        # Empty think tags
        text = "<think></think>Content"
        assert ToolParser.strip_think_tags(text) == "Content"

    def test_hermes_with_think_tags(self):
        """Test Hermes parser strips think tags before parsing tool calls."""
        parser = HermesToolParser()

        # Model output with think tags AND tool call (Ring-Mini-Linear-2.0 style)
        output = """<think>Let me search for that information.</think>
<tool_call>{"name": "search", "arguments": {"query": "weather"}}</tool_call>"""

        result = parser.extract_tool_calls(output)
        assert result.tools_called is True
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "search"

    def test_qwen_with_think_tags(self):
        """Test Qwen parser strips think tags before parsing tool calls."""
        parser = QwenToolParser()

        # Model output with think tags AND tool call
        output = """<think>I need to get the weather data.</think>
[Calling tool: get_weather({"city": "Tokyo"})]"""

        result = parser.extract_tool_calls(output)
        assert result.tools_called is True
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"

    def test_think_tags_with_no_tool_call(self):
        """Test that think tags are stripped even when no tool call is present."""
        parser = HermesToolParser()

        output = "<think>Let me think about this</think>The answer is 42."
        result = parser.extract_tool_calls(output)

        assert result.tools_called is False
        assert result.content == "The answer is 42."


class TestQwen3CoderParser:
    """Test Qwen3-Coder tool call parsing (Issue #47)."""

    def test_qwen3_coder_alias_registered(self):
        """Test that qwen3_coder is registered as an alias for Qwen3XMLToolParser."""
        from vllm_mlx.tool_parsers.qwen3_xml_tool_parser import Qwen3XMLToolParser

        parser_cls = ToolParserManager.get_tool_parser("qwen3_coder")
        assert parser_cls == Qwen3XMLToolParser

    def test_qwen3_coder_xml_format(self):
        """Test parsing Qwen3-Coder XML format (Nemotron-style)."""
        parser = HermesToolParser()
        text = (
            "<tool_call>\n"
            "<function=get_weather>\n"
            "<parameter=city>Paris</parameter>\n"
            "<parameter=units>celsius</parameter>\n"
            "</function>\n"
            "</tool_call>"
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["city"] == "Paris"
        assert args["units"] == "celsius"

    def test_qwen3_coder_with_think_tags(self):
        """Test Qwen3-Coder XML format with think tags."""
        parser = HermesToolParser()
        text = (
            "<think>I need to read this file.</think>\n"
            "<tool_call>\n"
            "<function=read_file>\n"
            "<parameter=path>/src/main.py</parameter>\n"
            "</function>\n"
            "</tool_call>"
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "read_file"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["path"] == "/src/main.py"

    def test_qwen3_coder_multiline_parameter(self):
        """Test Qwen3-Coder with multi-line parameter values (code)."""
        parser = HermesToolParser()
        text = (
            "<tool_call>\n"
            "<function=write_file>\n"
            "<parameter=path>/src/hello.py</parameter>\n"
            "<parameter=content>def hello():\n"
            "    print('hello')\n</parameter>\n"
            "</function>\n"
            "</tool_call>"
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert result.tool_calls[0]["name"] == "write_file"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["path"] == "/src/hello.py"
        assert "def hello():" in args["content"]

    def test_bare_function_without_tool_call_wrapper(self):
        """Test bare <function=...> blocks without <tool_call> wrapper."""
        parser = HermesToolParser()
        text = "<function=get_weather><parameter=city>Berlin</parameter></function>"
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_weather"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["city"] == "Berlin"

    def test_bare_multi_function_without_wrapper(self):
        """Test multiple bare <function=...> blocks without <tool_call> wrapper."""
        parser = HermesToolParser()
        text = (
            "<function=read_file>"
            "<parameter=path>/a.py</parameter>"
            "</function>\n"
            "<function=write_file>"
            "<parameter=path>/b.py</parameter>"
            "<parameter=content>hello</parameter>"
            "</function>"
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2
        assert result.tool_calls[0]["name"] == "read_file"
        assert result.tool_calls[1]["name"] == "write_file"

    def test_qwen3_coder_multiple_tool_calls(self):
        """Test multiple Nemotron XML tool calls (multi-tool scenario)."""
        parser = HermesToolParser()
        text = (
            "<tool_call>\n"
            "<function=read_file>\n"
            "<parameter=path>/src/main.py</parameter>\n"
            "</function>\n"
            "</tool_call>\n"
            "<tool_call>\n"
            "<function=list_files>\n"
            "<parameter=directory>/src</parameter>\n"
            "</function>\n"
            "</tool_call>"
        )
        result = parser.extract_tool_calls(text)

        assert result.tools_called
        assert len(result.tool_calls) == 2
        assert result.tool_calls[0]["name"] == "read_file"
        assert result.tool_calls[1]["name"] == "list_files"


class TestHermesStreamingFixes:
    """Test streaming fixes for Hermes parser (Issue #47)."""

    def test_streaming_complete_tool_call(self):
        """Test streaming with complete tool call in accumulated text."""
        parser = HermesToolParser()

        # Simulate token-by-token streaming
        # Token 1: regular content
        r = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text="Sure, let me check.",
            delta_text="Sure, let me check.",
        )
        assert r == {"content": "Sure, let me check."}

        # Token 2: opening tag
        r = parser.extract_tool_calls_streaming(
            previous_text="Sure, let me check.",
            current_text='Sure, let me check.<tool_call>{"name":',
            delta_text='<tool_call>{"name":',
        )
        assert r is None  # Inside tool call, suppress

        # Token 3: more JSON
        r = parser.extract_tool_calls_streaming(
            previous_text='Sure, let me check.<tool_call>{"name":',
            current_text='Sure, let me check.<tool_call>{"name": "search", "arguments": {"q": "test"}}',
            delta_text=' "search", "arguments": {"q": "test"}}',
        )
        assert r is None  # Still inside, no closing tag yet

        # Token 4: closing tag
        r = parser.extract_tool_calls_streaming(
            previous_text='Sure, let me check.<tool_call>{"name": "search", "arguments": {"q": "test"}}',
            current_text='Sure, let me check.<tool_call>{"name": "search", "arguments": {"q": "test"}}</tool_call>',
            delta_text="</tool_call>",
        )
        assert r is not None
        assert "tool_calls" in r
        assert r["tool_calls"][0]["function"]["name"] == "search"

    def test_streaming_split_closing_tag(self):
        """Test that </tool_call> split across deltas is detected via current_text."""
        parser = HermesToolParser()

        # Accumulated text has <tool_call> but </tool_call is split
        r = parser.extract_tool_calls_streaming(
            previous_text='<tool_call>{"name": "func", "arguments": {}}',
            current_text='<tool_call>{"name": "func", "arguments": {}}</tool_call',
            delta_text="</tool_call",
        )
        # Not yet complete (missing >)
        assert r is None

        # Now the > arrives, completing </tool_call>
        r = parser.extract_tool_calls_streaming(
            previous_text='<tool_call>{"name": "func", "arguments": {}}</tool_call',
            current_text='<tool_call>{"name": "func", "arguments": {}}</tool_call>',
            delta_text=">",
        )
        assert r is not None
        assert "tool_calls" in r
        assert r["tool_calls"][0]["function"]["name"] == "func"

    def test_streaming_content_after_tool_call(self):
        """Test that content after </tool_call> is not suppressed."""
        parser = HermesToolParser()

        # Complete tool call already in text
        full = '<tool_call>{"name": "func", "arguments": {}}</tool_call>'
        r = parser.extract_tool_calls_streaming(
            previous_text=full[:-1],  # everything except last >
            current_text=full,
            delta_text=">",
        )
        assert "tool_calls" in r

        # Now content comes after the tool call
        r = parser.extract_tool_calls_streaming(
            previous_text=full,
            current_text=full + "\nHere is the result.",
            delta_text="\nHere is the result.",
        )
        # Should NOT be suppressed
        assert r is not None
        assert r.get("content") == "\nHere is the result."

    def test_streaming_nemotron_xml_format(self):
        """Test streaming with Qwen3-Coder/Nemotron XML format."""
        parser = HermesToolParser()

        chunks = [
            "<tool_call>\n",
            "<function=get_weather>\n",
            "<parameter=city>London</parameter>\n",
            "</function>\n",
            "</tool_call>",
        ]

        accumulated = ""
        tool_calls_found = False
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                tool_calls_found = True
                assert r["tool_calls"][0]["function"]["name"] == "get_weather"
                break

        assert tool_calls_found, "Tool call should have been detected"

    def test_streaming_no_false_positives(self):
        """Test that regular text with < doesn't trigger false positives."""
        parser = HermesToolParser()

        r = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text="Use x < 10 and y > 5",
            delta_text="Use x < 10 and y > 5",
        )
        assert r == {"content": "Use x < 10 and y > 5"}

    def test_streaming_multi_tool_calls(self):
        """Test streaming with two sequential tool calls (Issue #47)."""
        parser = HermesToolParser()

        # First tool call arrives token by token
        chunks = [
            '<tool_call>{"name": "read_file", "arguments": {"path": "/a.py"}}',
            "</tool_call>",
            "\n",
            '<tool_call>{"name": "list_dir", "arguments": {"dir": "/src"}}',
            "</tool_call>",
        ]

        accumulated = ""
        emitted_calls = []
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                emitted_calls.extend(r["tool_calls"])

        assert len(emitted_calls) == 2
        assert emitted_calls[0]["function"]["name"] == "read_file"
        assert emitted_calls[1]["function"]["name"] == "list_dir"
        # Verify indexes are correct
        assert emitted_calls[0]["index"] == 0
        assert emitted_calls[1]["index"] == 1

    def test_streaming_multi_tool_no_duplicates(self):
        """Test that completed tool calls are not re-emitted on second completion."""
        parser = HermesToolParser()

        first_complete = '<tool_call>{"name": "func1", "arguments": {}}</tool_call>'
        # After first tool call is emitted, content between calls passes through
        between = "\n"
        second_start = '<tool_call>{"name": "func2", "arguments": {}}'
        second_end = "</tool_call>"

        # First tool call completed
        r1 = parser.extract_tool_calls_streaming(
            previous_text=first_complete[:-1],
            current_text=first_complete,
            delta_text=">",
        )
        assert r1 is not None and "tool_calls" in r1
        assert len(r1["tool_calls"]) == 1
        assert r1["tool_calls"][0]["function"]["name"] == "func1"

        # Whitespace between
        r2 = parser.extract_tool_calls_streaming(
            previous_text=first_complete,
            current_text=first_complete + between,
            delta_text=between,
        )
        assert r2 == {"content": between}

        # Second tool call building (inside block, suppress)
        r3 = parser.extract_tool_calls_streaming(
            previous_text=first_complete + between,
            current_text=first_complete + between + second_start,
            delta_text=second_start,
        )
        assert r3 is None  # Inside incomplete second tool call

        # Second tool call completed
        r4 = parser.extract_tool_calls_streaming(
            previous_text=first_complete + between + second_start,
            current_text=first_complete + between + second_start + second_end,
            delta_text=second_end,
        )
        assert r4 is not None and "tool_calls" in r4
        assert len(r4["tool_calls"]) == 1  # Only the NEW tool call
        assert r4["tool_calls"][0]["function"]["name"] == "func2"
        assert r4["tool_calls"][0]["index"] == 1  # Second index

    def test_streaming_multi_nemotron_xml(self):
        """Test streaming with multiple Nemotron XML tool calls."""
        parser = HermesToolParser()

        chunks = [
            "<tool_call>\n<function=read_file>\n",
            "<parameter=path>/a.py</parameter>\n",
            "</function>\n</tool_call>\n",
            "<tool_call>\n<function=write_file>\n",
            "<parameter=path>/b.py</parameter>\n",
            "<parameter=content>hello</parameter>\n",
            "</function>\n</tool_call>",
        ]

        accumulated = ""
        emitted_calls = []
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                emitted_calls.extend(r["tool_calls"])

        assert len(emitted_calls) == 2
        assert emitted_calls[0]["function"]["name"] == "read_file"
        assert emitted_calls[1]["function"]["name"] == "write_file"

    def test_streaming_bare_function_blocks(self):
        """Test streaming with bare <function= blocks without <tool_call> wrapper."""
        parser = HermesToolParser()

        chunks = [
            "<function=read_file>",
            "<parameter=path>/src/main.py</parameter>",
            "</function>",
        ]

        accumulated = ""
        tool_calls_found = False
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                tool_calls_found = True
                assert r["tool_calls"][0]["function"]["name"] == "read_file"
                args = json.loads(r["tool_calls"][0]["function"]["arguments"])
                assert args["path"] == "/src/main.py"
                break

        assert tool_calls_found, "Bare function block should have been detected"

    def test_streaming_bare_multi_function_blocks(self):
        """Test streaming with multiple bare <function= blocks."""
        parser = HermesToolParser()

        chunks = [
            "<function=func1><parameter=a>1</parameter></function>",
            "\n",
            "<function=func2>",
            "<parameter=b>2</parameter>",
            "</function>",
        ]

        accumulated = ""
        emitted_calls = []
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                emitted_calls.extend(r["tool_calls"])

        assert len(emitted_calls) == 2
        assert emitted_calls[0]["function"]["name"] == "func1"
        assert emitted_calls[1]["function"]["name"] == "func2"


class TestQwenFunctionFormat:
    """Test Qwen parser's <function=name> format support."""

    @pytest.fixture
    def parser(self):
        return QwenToolParser()

    def test_function_format_with_parameters(self, parser):
        """Test <function=name><parameter=key>value</parameter></function>."""
        text = "<function=get_weather><parameter=city>Prague</parameter></function>"
        result = parser.extract_tool_calls(text)
        assert result.tools_called
        assert result.tool_calls[0]["name"] == "get_weather"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["city"] == "Prague"

    def test_function_format_with_json(self, parser):
        """Test <function=name>{"key": "val"}</function>."""
        text = '<function=get_weather>{"city": "Prague"}</function>'
        result = parser.extract_tool_calls(text)
        assert result.tools_called
        assert result.tool_calls[0]["name"] == "get_weather"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["city"] == "Prague"

    def test_function_format_multiple(self, parser):
        """Test multiple <function=...> blocks."""
        text = (
            '<function=read_file>{"path": "/a.py"}</function>'
            '<function=write_file>{"path": "/b.py", "content": "hello"}</function>'
        )
        result = parser.extract_tool_calls(text)
        assert result.tools_called
        assert len(result.tool_calls) == 2
        assert result.tool_calls[0]["name"] == "read_file"
        assert result.tool_calls[1]["name"] == "write_file"

    def test_function_format_with_think_tags(self, parser):
        """Test <function=...> with think tags."""
        text = (
            "<think>I need to check the weather.</think>\n"
            '<function=get_weather>{"city": "Prague"}</function>'
        )
        result = parser.extract_tool_calls(text)
        assert result.tools_called
        assert result.tool_calls[0]["name"] == "get_weather"


class TestQwenStreamingBuffering:
    """Test Qwen parser streaming with partial-marker buffering."""

    @pytest.fixture
    def parser(self):
        return QwenToolParser()

    def test_streaming_function_format_complete(self, parser):
        """Test streaming with <function=name>...</function> format."""
        chunks = [
            "<function=get_weather>",
            "<parameter=city>Prague</parameter>",
            "</function>",
        ]
        accumulated = ""
        tool_calls_found = False
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                tool_calls_found = True
                assert r["tool_calls"][0]["function"]["name"] == "get_weather"
                break
        assert tool_calls_found

    def test_streaming_bracket_call_closing_marker_split(self, parser):
        """Qwen bracket calls should complete when ')' and ']' split chunks."""
        chunks = [
            '[Calling tool: add({"a": 1, "b": 2})',
            "]",
        ]

        accumulated = ""
        emitted = None
        for chunk in chunks:
            previous = accumulated
            accumulated += chunk
            emitted = parser.extract_tool_calls_streaming(
                previous_text=previous,
                current_text=accumulated,
                delta_text=chunk,
            )

        assert emitted is not None
        assert "tool_calls" in emitted
        assert emitted["tool_calls"][0]["function"]["name"] == "add"
        assert emitted["tool_calls"][0]["function"]["arguments"] == ('{"a": 1, "b": 2}')

    def test_streaming_partial_marker_buffered(self, parser):
        """Test that partial '<function' is buffered (not leaked as content)."""
        r = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text="Sure.",
            delta_text="Sure.",
        )
        assert r == {"content": "Sure."}

        # Partial marker "<function" — should be buffered
        r = parser.extract_tool_calls_streaming(
            previous_text="Sure.",
            current_text="Sure.<function",
            delta_text="<function",
        )
        assert r is None  # Buffered, not emitted

        # "=" confirms tool call marker
        r = parser.extract_tool_calls_streaming(
            previous_text="Sure.<function",
            current_text="Sure.<function=get_weather>",
            delta_text="=get_weather>",
        )
        assert r is None  # Inside incomplete function block

    def test_streaming_false_positive_functional(self, parser):
        """Regression: '<functional' across chunk boundary must NOT be suppressed.

        When a delta contains 'Look at <function', the content before the
        partial marker ('Look at ') must be emitted immediately. Only the
        marker suffix is buffered. On recovery, the marker prefix is
        re-emitted with the next delta so no text is lost.
        """
        # Token 1: "Look at <function" — emit "Look at ", buffer "<function"
        r = parser.extract_tool_calls_streaming(
            previous_text="",
            current_text="Look at <function",
            delta_text="Look at <function",
        )
        assert r == {"content": "Look at "}

        # Token 2: "al interface" — confirms it's NOT a tool call
        r = parser.extract_tool_calls_streaming(
            previous_text="Look at <function",
            current_text="Look at <functional interface",
            delta_text="al interface",
        )
        assert r == {"content": "<functional interface"}

    def test_streaming_false_positive_single_angle_bracket(self, parser):
        """Test that a single '<' followed by non-marker text is recovered."""
        # "<" alone — partial marker
        r = parser.extract_tool_calls_streaming(
            previous_text="Hello",
            current_text="Hello<",
            delta_text="<",
        )
        assert r is None  # Buffered

        # "div>" — not a tool marker
        r = parser.extract_tool_calls_streaming(
            previous_text="Hello<",
            current_text="Hello<div>",
            delta_text="div>",
        )
        assert r is not None
        assert "content" in r
        assert "<" in r["content"]
        assert "div>" in r["content"]

    def test_streaming_multiple_function_blocks(self, parser):
        """Test streaming with multiple <function= blocks."""
        chunks = [
            '<function=func1>{"a": 1}</function>',
            "\n",
            "<function=func2>",
            "<parameter=b>2</parameter>",
            "</function>",
        ]
        accumulated = ""
        emitted_calls = []
        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                emitted_calls.extend(r["tool_calls"])
        assert len(emitted_calls) == 2
        assert emitted_calls[0]["function"]["name"] == "func1"
        assert emitted_calls[1]["function"]["name"] == "func2"


class TestGLM47ToolParser:
    """Tests for GLM47 tool parser."""

    def test_zero_arguments_tool_call(self):
        """Test Fix 2: Handle zero-argument tool calls without crashing."""
        parser = Glm47ToolParser()

        output = "<tool_call>get_current_time</tool_call>"

        result = parser.extract_tool_calls(output)

        assert result.tools_called is True
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "get_current_time"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args == {}

    def test_with_arguments(self):
        """Test tool call with arguments."""
        parser = Glm47ToolParser()

        output = "<tool_call>search\n<arg_key>query</arg_key><arg_value>Python</arg_value></tool_call>"

        result = parser.extract_tool_calls(output)

        assert result.tools_called is True
        assert len(result.tool_calls) == 1
        assert result.tool_calls[0]["name"] == "search"
        args = json.loads(result.tool_calls[0]["arguments"])
        assert args["query"] == "Python"

    def test_streaming_zero_args(self):
        """Test Fix 2: Streaming with zero-argument tool call."""
        parser = Glm47ToolParser()

        chunks = ["<tool_call>", "get_status", "</tool_call>"]
        accumulated = ""
        tool_calls_found = False

        for chunk in chunks:
            prev = accumulated
            accumulated += chunk
            r = parser.extract_tool_calls_streaming(
                previous_text=prev,
                current_text=accumulated,
                delta_text=chunk,
            )
            if r is not None and "tool_calls" in r:
                tool_calls_found = True
                assert r["tool_calls"][0]["function"]["name"] == "get_status"
                args = json.loads(r["tool_calls"][0]["function"]["arguments"])
                assert args == {}

        assert tool_calls_found, "Zero-argument tool call should have been detected"
