test_tool_calls.py

"""Regression tests for Qwen/OpenClaude tool-call formats."""

from __future__ import annotations

import json
import unittest

from tool_calls import (
    extract_tool_call,
    extract_tool_calls,
    has_complete_tool_call,
    normalize_openai_tool_arguments,
    recover_forced_tool_call,
)


ALLOWED = {"Bash", "Read", "WebSearch"}


class ToolCallTests(unittest.TestCase):
    def parsed(self, text: str) -> dict:
        call, visible = extract_tool_call(text, ALLOWED)
        self.assertIsNotNone(call)
        self.assertEqual(visible, "")
        return call

    def test_json_wrapper(self) -> None:
        call = self.parsed(
            '<tool_call>{"name":"Bash","arguments":{"command":"pwd"}}</tool_call>'
        )
        self.assertEqual(call["function"]["name"], "Bash")
        self.assertEqual(json.loads(call["function"]["arguments"]), {"command": "pwd"})

    def test_python_literal_wrapper_from_dolphin(self) -> None:
        call = self.parsed(
            "<tool_call>{'name': 'WebSearch', "
            "'arguments': {'query': 'Gitlawb OpenClaude GitHub'}}</tool_call>"
        )
        self.assertEqual(call["function"]["name"], "WebSearch")
        self.assertEqual(
            json.loads(call["function"]["arguments"]),
            {"query": "Gitlawb OpenClaude GitHub"},
        )

    def test_python_literal_does_not_execute_expressions(self) -> None:
        text = (
            "<tool_call>{'name': 'Bash', "
            "'arguments': __import__('os').system('id')}</tool_call>"
        )
        call, visible = extract_tool_call(text, ALLOWED)
        self.assertIsNone(call)
        self.assertEqual(visible, text)

    def test_xml_wrapped_json_from_openclaude(self) -> None:
        call = self.parsed(
            '<xml>{"name":"Bash","arguments":{"command":"ls -la"}}</xml>'
        )
        self.assertEqual(call["function"]["name"], "Bash")
        self.assertEqual(
            json.loads(call["function"]["arguments"]), {"command": "ls -la"}
        )
        self.assertTrue(
            has_complete_tool_call(
                '<xml>{"name":"Bash","arguments":{"command":"pwd"}}</xml>'
            )
        )

    def test_fenced_self_closing_openclaude_tag(self) -> None:
        text = '''```xml
<Bash command="ls /tmp" description="List files"/>
```'''
        call = self.parsed(text)
        self.assertEqual(call["function"]["name"], "Bash")
        self.assertEqual(
            json.loads(call["function"]["arguments"]),
            {"command": "ls /tmp", "description": "List files"},
        )
        self.assertTrue(has_complete_tool_call('<Bash command="pwd"/>'))

    def test_bare_ampersand_in_tool_attribute_is_preserved(self) -> None:
        call = self.parsed(
            '<Bash command="curl https://api.example.test/forecast?latitude=0&current_weather=true"/>'
        )
        self.assertEqual(
            json.loads(call["function"]["arguments"]),
            {
                "command": (
                    "curl https://api.example.test/forecast?latitude=0"
                    "&current_weather=true"
                )
            },
        )

    def test_standard_xml_function(self) -> None:
        call = self.parsed(
            "<tool_call><function=Read><parameter=file_path>/tmp/a.txt"
            "</parameter></function></tool_call>"
        )
        self.assertEqual(call["function"]["name"], "Read")
        self.assertEqual(
            json.loads(call["function"]["arguments"]), {"file_path": "/tmp/a.txt"}
        )

    def test_literal_assistant_tool_text(self) -> None:
        call = self.parsed(
            '[Assistant called tool Bash with arguments {"command":"echo ok"}]'
        )
        self.assertEqual(
            json.loads(call["function"]["arguments"]), {"command": "echo ok"}
        )


    def test_function_call_alias_from_qwen_is_supported(self) -> None:
        text = (
            '<function_call>{"name":"Read","arguments":{"file_path":"app.py"}}'
            '</function_call>'
        )
        calls, visible = extract_tool_calls(text, {"Read"})
        self.assertEqual(visible, "")
        self.assertEqual(len(calls), 1)
        self.assertEqual(calls[0]["function"]["name"], "Read")
        self.assertEqual(
            json.loads(calls[0]["function"]["arguments"]),
            {"file_path": "app.py"},
        )
        self.assertTrue(has_complete_tool_call(text))

    def test_qwen_compact_textual_tool_call(self) -> None:
        call = self.parsed(
            '```python\nwebsearch with query="python 3.13 features and highlights"\n```'
        )
        self.assertEqual(call["function"]["name"], "WebSearch")
        self.assertEqual(
            json.loads(call["function"]["arguments"]),
            {"query": "python 3.13 features and highlights"},
        )
        self.assertTrue(has_complete_tool_call('websearch with query="python"'))

    def test_qwen_compact_textual_tool_call_supports_all_programming_tools(self) -> None:
        samples = {
            "Bash": 'bash with command="printf ok"',
            "Read": 'read with file_path="/tmp/fixture.py"',
            "Write": 'write with file_path="/tmp/new.py" content="pass"',
            "Edit": (
                'edit with file_path="/tmp/fixture.py" '
                'old_string="left" new_string="right"'
            ),
            "Glob": 'glob with pattern="**/*.py" path="/tmp"',
            "Grep": 'grep with pattern="TODO" path="/tmp"',
            "WebSearch": 'websearch with query="Qwen3 tool calling"',
            "WebFetch": (
                'webfetch with url="https://example.com" '
                'prompt="summarize"'
            ),
            "Task": 'agent with description="inspect the fixture"',
        }
        allowed = set(samples)
        for expected_name, text in samples.items():
            with self.subTest(tool=expected_name):
                call, visible = extract_tool_call(text, allowed)
                self.assertIsNotNone(call)
                self.assertEqual(visible, "")
                self.assertEqual(call["function"]["name"], expected_name)

    def test_forced_tool_recovers_argument_only_json(self) -> None:
        call = recover_forced_tool_call(
            '{"file_path":"/tmp/project/app.py"}',
            "Read",
        )
        self.assertIsNotNone(call)
        self.assertEqual(call["function"]["name"], "Read")
        self.assertEqual(
            json.loads(call["function"]["arguments"]),
            {"file_path": "/tmp/project/app.py"},
        )

    def test_forced_tool_recovery_rejects_prose_and_named_calls(self) -> None:
        self.assertIsNone(recover_forced_tool_call("I would read the file.", "Read"))
        self.assertIsNone(
            recover_forced_tool_call(
                '{"name":"Read","arguments":{"file_path":"/tmp/a"}}',
                "Read",
            )
        )

    def test_unknown_tool_is_not_exposed(self) -> None:
        call, visible = extract_tool_call('<Delete path="/"/>', ALLOWED)
        self.assertIsNone(call)
        self.assertIn("Delete", visible)

    def test_multiple_adjacent_calls_are_preserved_in_order(self) -> None:
        calls, visible = extract_tool_calls(
            '<tool_call>{"name":"Read","arguments":{"file_path":"/tmp/á.json"}}'
            '</tool_call>\n'
            '<tool_call>{"name":"Bash","arguments":{"command":"wc -c /tmp/á.json"}}'
            "</tool_call>",
            ALLOWED,
        )
        self.assertEqual(visible, "")
        self.assertEqual(
            [call["function"]["name"] for call in calls],
            ["Read", "Bash"],
        )
        self.assertEqual(
            json.loads(calls[0]["function"]["arguments"]),
            {"file_path": "/tmp/á.json"},
        )
        self.assertNotEqual(calls[0]["id"], calls[1]["id"])

    def test_mixed_parallel_formats_are_preserved(self) -> None:
        calls, visible = extract_tool_calls(
            '<Read file_path="/tmp/a &amp; b.txt"/>\n'
            "<tool-call><name>WebSearch</name><arguments>"
            '<argument name="query">Qwen3 unicode 日本語</argument>'
            "</arguments></tool-call>",
            ALLOWED,
        )
        self.assertEqual(visible, "")
        self.assertEqual(
            [call["function"]["name"] for call in calls],
            ["Read", "WebSearch"],
        )
        self.assertEqual(
            json.loads(calls[0]["function"]["arguments"]),
            {"file_path": "/tmp/a & b.txt"},
        )

    def test_extreme_parallel_batch_has_unique_ids(self) -> None:
        text = "".join(
            "<tool_call>"
            + json.dumps(
                {
                    "name": "Read",
                    "arguments": {"file_path": f"/tmp/file-{index}.txt"},
                }
            )
            + "</tool_call>"
            for index in range(64)
        )
        calls, visible = extract_tool_calls(text, ALLOWED)
        self.assertEqual(visible, "")
        self.assertEqual(len(calls), 64)
        self.assertEqual(len({call["id"] for call in calls}), 64)
        self.assertEqual(
            json.loads(calls[-1]["function"]["arguments"]),
            {"file_path": "/tmp/file-63.txt"},
        )

    def test_large_unicode_argument_is_not_truncated(self) -> None:
        value = ("á日本語&" * 8192) + "fim"
        text = (
            "<tool_call>"
            + json.dumps(
                {"name": "WebSearch", "arguments": {"query": value}},
                ensure_ascii=False,
            )
            + "</tool_call>"
        )
        call = self.parsed(text)
        self.assertEqual(json.loads(call["function"]["arguments"])["query"], value)

    def test_openai_history_arguments_are_mappings_for_qwen_template(self) -> None:
        self.assertEqual(
            normalize_openai_tool_arguments('{"command":"pwd"}'),
            {"command": "pwd"},
        )
        self.assertEqual(
            normalize_openai_tool_arguments({"file_path": "/tmp/a.txt"}),
            {"file_path": "/tmp/a.txt"},
        )
        self.assertEqual(normalize_openai_tool_arguments("not-json"), {})

    def test_official_qwen_json_format_round_trips_tricky_arguments_deterministically(self) -> None:
        cases = [
            {"file_path": "README.md"},
            {"path": "a/b c.py", "line": 17, "flag": True},
            {"query": "a & b ? x=1&y=2", "unicode": "ação — 東京 🚀"},
            {"content": "brace } inside string { and quote \" ok"},
            {"nested": {"items": [1, 2, {"x": "y"}], "empty": {}}, "none": None},
            {"command": "printf '%s\n' '{\"a\":1}' && echo done"},
        ]
        for arguments in cases:
            with self.subTest(arguments=arguments):
                payload = json.dumps(
                    {"name": "Read", "arguments": arguments},
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
                text = f"<tool_call>{payload}</tool_call>"
                calls, visible = extract_tool_calls(text, {"Read"})
                self.assertEqual(visible, "")
                self.assertEqual(len(calls), 1)
                self.assertEqual(calls[0]["function"]["name"], "Read")
                self.assertEqual(json.loads(calls[0]["function"]["arguments"]), arguments)
                self.assertTrue(has_complete_tool_call(text, {"Read"}))

    def test_stopping_signal_rejects_unadvertised_complete_tool(self) -> None:
        text = '<tool_call>{"name":"MadeUpTool","arguments":{}}</tool_call>'
        self.assertFalse(has_complete_tool_call(text, {"Read"}))

    def test_stopping_signal_accepts_advertised_complete_tool(self) -> None:
        text = '<tool_call>{"name":"Read","arguments":{"file_path":"README.md"}}</tool_call>'
        self.assertTrue(has_complete_tool_call(text, {"Read"}))


if __name__ == "__main__":
    unittest.main()