| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import json |
| 5 | from dataclasses import dataclass, field |
| 6 | from types import NoneType |
| 7 | from typing import Any |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | from tests.tool_parsers.utils import run_tool_extraction |
| 12 | from vllm.tokenizers import TokenizerLike |
| 13 | from vllm.tool_parsers import ToolParserManager |
| 14 | |
| 15 | |
| 16 | @dataclass |
| 17 | class ToolParserTestConfig: |
| 18 | """Configuration for a tool parser's common tests. |
| 19 | |
| 20 | This dataclass contains all the test data and expected results needed |
| 21 | to run the common test suite for a parser. Each parser test file |
| 22 | creates one instance of this config with parser-specific values. |
| 23 | |
| 24 | Attributes: |
| 25 | parser_name: Name used with ToolParserManager (e.g., "mistral") |
| 26 | |
| 27 | Test data (model outputs): |
| 28 | no_tool_calls_output: Plain text without any tool syntax |
| 29 | single_tool_call_output: One tool call with simple arguments |
| 30 | parallel_tool_calls_output: Multiple tool calls in one response |
| 31 | various_data_types_output: Tool with various data types |
| 32 | empty_arguments_output: Tool call with no parameters |
| 33 | surrounding_text_output: Tool call mixed with regular text |
| 34 | escaped_strings_output: Tool call with escaped chars |
| 35 | malformed_input_outputs: List of invalid inputs |
| 36 | |
| 37 | Expected results: |
| 38 | single_tool_call_expected_name: Expected function name |
| 39 | single_tool_call_expected_args: Expected arguments dict |
| 40 | parallel_tool_calls_count: Number of tools in parallel test |
| 41 | parallel_tool_calls_names: Function names in order |
| 42 | single_tool_call_expected_content: Content field when tool called |
| 43 | parallel_tool_calls_expected_content: Content for parallel test |
| 44 | |
| 45 | xfail markers: |
| 46 | xfail_streaming: Mapping test name to xfail reason (streaming only) |
| 47 | xfail_nonstreaming: Mapping test name to xfail reason (non-streaming) |
| 48 | |
| 49 | Special flags: |
| 50 | allow_empty_or_json_empty_args: True if "" or "{}" both valid for empty args |
| 51 | supports_typed_arguments: True if the parser supports typed function arguments |
| 52 | |
| 53 | """ |
| 54 | |
| 55 | # Parser identification |
| 56 | parser_name: str |
| 57 | |
| 58 | # Test data - model outputs for each common test |
| 59 | no_tool_calls_output: str |
| 60 | single_tool_call_output: str |
| 61 | parallel_tool_calls_output: str |
| 62 | various_data_types_output: str |
| 63 | empty_arguments_output: str |
| 64 | surrounding_text_output: str |
| 65 | escaped_strings_output: str |
| 66 | malformed_input_outputs: list[str] |
| 67 | |
| 68 | # Expected results for specific tests (optional overrides) |
| 69 | single_tool_call_expected_name: str = "get_weather" |
| 70 | single_tool_call_expected_args: dict[str, Any] = field( |
| 71 | default_factory=lambda: {"city": "Tokyo"} |
| 72 | ) |
| 73 | parallel_tool_calls_count: int = 2 |
| 74 | parallel_tool_calls_names: list[str] = field( |
| 75 | default_factory=lambda: ["get_weather", "get_time"] |
| 76 | ) |
| 77 | |
| 78 | # xfail configuration - maps test name to xfail reason |
| 79 | xfail_streaming: dict[str, str] = field(default_factory=dict) |
| 80 | xfail_nonstreaming: dict[str, str] = field(default_factory=dict) |
| 81 | |
| 82 | # Content expectations (some parsers strip content, others don't) |
| 83 | single_tool_call_expected_content: str | None = None |
| 84 | parallel_tool_calls_expected_content: str | None = None |
| 85 | |
| 86 | # Special assertions for edge cases |
| 87 | allow_empty_or_json_empty_args: bool = True # "{}" or "" for empty args |
| 88 | supports_typed_arguments: bool = True |
| 89 | |
| 90 | |
| 91 | class ToolParserTests: |
| 92 | """Mixin class providing common test suite for tool parsers. |
| 93 | |
| 94 | To use this mixin in a parser test file: |
| 95 | |
| 96 | 1. Create a test_config fixture that returns a ToolParserTestConfig instance |
| 97 | 2. Inherit from this class |
| 98 | 3. Add parser-specific tests as additional methods |
| 99 | |
| 100 | Example: |
| 101 | class TestMistralToolParser(ToolParserTests): |
| 102 | @pytest.fixture |
| 103 | def test_config(self) -> ToolParserTestConfig: |
| 104 | return ToolParserTestConfig( |
| 105 | parser_name="mistral", |
| 106 | no_tool_calls_output="Plain text...", |
| 107 | # ... other config ... |
| 108 | ) |
| 109 | |
| 110 | # Parser-specific tests |
| 111 | def test_mistral_specific_feature(self, tool_parser): |
| 112 | # Custom test logic |
| 113 | pass |
| 114 | |
| 115 | """ |
| 116 | |
| 117 | @pytest.fixture |
| 118 | def test_config(self) -> ToolParserTestConfig: |
| 119 | """Override this to provide parser-specific configuration.""" |
| 120 | raise NotImplementedError( |
| 121 | "Subclass must provide test_config fixture returning ToolParserTestConfig" |
| 122 | ) |
| 123 | |
| 124 | @pytest.fixture |
| 125 | def tokenizer(self, default_tokenizer: TokenizerLike) -> TokenizerLike: |
| 126 | """Override this to provide parser-specific tokenizer.""" |
| 127 | return default_tokenizer |
| 128 | |
| 129 | @pytest.fixture |
| 130 | def tool_parser(self, test_config: ToolParserTestConfig, tokenizer: TokenizerLike): |
| 131 | return ToolParserManager.get_tool_parser(test_config.parser_name)(tokenizer) |
| 132 | |
| 133 | @pytest.fixture(params=[True, False]) |
| 134 | def streaming(self, request: pytest.FixtureRequest) -> bool: |
| 135 | return request.param |
| 136 | |
| 137 | def test_no_tool_calls( |
| 138 | self, |
| 139 | request: pytest.FixtureRequest, |
| 140 | tool_parser: Any, |
| 141 | test_config: ToolParserTestConfig, |
| 142 | streaming: bool, |
| 143 | ): |
| 144 | """Verify parser handles plain text without tool syntax.""" |
| 145 | # Apply xfail markers if configured |
| 146 | test_name = "test_no_tool_calls" |
| 147 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 148 | |
| 149 | content, tool_calls = run_tool_extraction( |
| 150 | tool_parser, test_config.no_tool_calls_output, streaming=streaming |
| 151 | ) |
| 152 | assert content == test_config.no_tool_calls_output, ( |
| 153 | f"Expected content to match input, got {content}" |
| 154 | ) |
| 155 | assert len(tool_calls) == 0, f"Expected no tool calls, got {len(tool_calls)}" |
| 156 | |
| 157 | def test_single_tool_call_simple_args( |
| 158 | self, |
| 159 | request: pytest.FixtureRequest, |
| 160 | tool_parser: Any, |
| 161 | test_config: ToolParserTestConfig, |
| 162 | streaming: bool, |
| 163 | ): |
| 164 | """Verify parser extracts one tool with simple arguments.""" |
| 165 | # Apply xfail markers if configured |
| 166 | test_name = "test_single_tool_call_simple_args" |
| 167 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 168 | |
| 169 | content, tool_calls = run_tool_extraction( |
| 170 | tool_parser, test_config.single_tool_call_output, streaming=streaming |
| 171 | ) |
| 172 | |
| 173 | # Content check (some parsers strip it) |
| 174 | if test_config.single_tool_call_expected_content is not None: |
| 175 | assert content == test_config.single_tool_call_expected_content |
| 176 | |
| 177 | assert len(tool_calls) == 1, f"Expected 1 tool call, got {len(tool_calls)}" |
| 178 | assert tool_calls[0].type == "function" |
| 179 | assert tool_calls[0].function.name == test_config.single_tool_call_expected_name |
| 180 | |
| 181 | args = json.loads(tool_calls[0].function.arguments) |
| 182 | for key, value in test_config.single_tool_call_expected_args.items(): |
| 183 | assert args.get(key) == value, ( |
| 184 | f"Expected {key}={value}, got {args.get(key)}" |
| 185 | ) |
| 186 | |
| 187 | def test_parallel_tool_calls( |
| 188 | self, |
| 189 | request: pytest.FixtureRequest, |
| 190 | tool_parser: Any, |
| 191 | test_config: ToolParserTestConfig, |
| 192 | streaming: bool, |
| 193 | ): |
| 194 | """Verify parser handles multiple tools in one response.""" |
| 195 | # Apply xfail markers if configured |
| 196 | test_name = "test_parallel_tool_calls" |
| 197 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 198 | |
| 199 | content, tool_calls = run_tool_extraction( |
| 200 | tool_parser, |
| 201 | test_config.parallel_tool_calls_output, |
| 202 | streaming=streaming, |
| 203 | ) |
| 204 | |
| 205 | assert len(tool_calls) == test_config.parallel_tool_calls_count, ( |
| 206 | f"Expected {test_config.parallel_tool_calls_count} " |
| 207 | f"tool calls, got {len(tool_calls)}" |
| 208 | ) |
| 209 | |
| 210 | # Verify tool names match expected |
| 211 | for i, expected_name in enumerate(test_config.parallel_tool_calls_names): |
| 212 | assert tool_calls[i].type == "function" |
| 213 | assert tool_calls[i].function.name == expected_name |
| 214 | |
| 215 | # Verify unique IDs |
| 216 | ids = [tc.id for tc in tool_calls] |
| 217 | assert len(ids) == len(set(ids)), "Tool call IDs should be unique" |
| 218 | |
| 219 | def test_various_data_types( |
| 220 | self, |
| 221 | request: pytest.FixtureRequest, |
| 222 | tool_parser: Any, |
| 223 | test_config: ToolParserTestConfig, |
| 224 | streaming: bool, |
| 225 | ): |
| 226 | """Verify parser handles all JSON types in arguments.""" |
| 227 | # Apply xfail markers if configured |
| 228 | test_name = "test_various_data_types" |
| 229 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 230 | |
| 231 | content, tool_calls = run_tool_extraction( |
| 232 | tool_parser, |
| 233 | test_config.various_data_types_output, |
| 234 | streaming=streaming, |
| 235 | ) |
| 236 | assert len(tool_calls) == 1, f"Expected 1 tool call, got {len(tool_calls)}" |
| 237 | |
| 238 | args = json.loads(tool_calls[0].function.arguments) |
| 239 | # Verify all expected fields present |
| 240 | required_fields_types = { |
| 241 | "string_field": str, |
| 242 | "int_field": int, |
| 243 | "float_field": float, |
| 244 | "bool_field": bool, |
| 245 | "null_field": NoneType, |
| 246 | "array_field": list, |
| 247 | "object_field": dict, |
| 248 | } |
| 249 | for required_field, expected_type in required_fields_types.items(): |
| 250 | assert required_field in args, ( |
| 251 | f"Expected field '{required_field}' in arguments" |
| 252 | ) |
| 253 | if test_config.supports_typed_arguments: |
| 254 | found_type = type(args[required_field]) |
| 255 | assert found_type is expected_type, ( |
| 256 | f"Expected field '{required_field}' to have type {expected_type}, " |
| 257 | f"got {found_type}" |
| 258 | ) |
| 259 | |
| 260 | def test_empty_arguments( |
| 261 | self, |
| 262 | request: pytest.FixtureRequest, |
| 263 | tool_parser: Any, |
| 264 | test_config: ToolParserTestConfig, |
| 265 | streaming: bool, |
| 266 | ): |
| 267 | """Verify parser handles parameterless tool calls.""" |
| 268 | # Apply xfail markers if configured |
| 269 | test_name = "test_empty_arguments" |
| 270 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 271 | |
| 272 | content, tool_calls = run_tool_extraction( |
| 273 | tool_parser, test_config.empty_arguments_output, streaming=streaming |
| 274 | ) |
| 275 | assert len(tool_calls) == 1, f"Expected 1 tool call, got {len(tool_calls)}" |
| 276 | |
| 277 | args = tool_calls[0].function.arguments |
| 278 | if test_config.allow_empty_or_json_empty_args: |
| 279 | assert args in ["{}", ""], f"Expected empty args, got {args}" |
| 280 | else: |
| 281 | assert args == "{}", f"Expected {{}}, got {args}" |
| 282 | |
| 283 | def test_surrounding_text( |
| 284 | self, |
| 285 | request: pytest.FixtureRequest, |
| 286 | tool_parser: Any, |
| 287 | test_config: ToolParserTestConfig, |
| 288 | streaming: bool, |
| 289 | ): |
| 290 | """Verify parser extracts tools from mixed content.""" |
| 291 | # Apply xfail markers if configured |
| 292 | test_name = "test_surrounding_text" |
| 293 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 294 | |
| 295 | content, tool_calls = run_tool_extraction( |
| 296 | tool_parser, test_config.surrounding_text_output, streaming=streaming |
| 297 | ) |
| 298 | assert len(tool_calls) >= 1, ( |
| 299 | f"Expected at least 1 tool call, got {len(tool_calls)}" |
| 300 | ) |
| 301 | |
| 302 | def test_escaped_strings( |
| 303 | self, |
| 304 | request: pytest.FixtureRequest, |
| 305 | tool_parser: Any, |
| 306 | test_config: ToolParserTestConfig, |
| 307 | streaming: bool, |
| 308 | ): |
| 309 | """Verify parser handles escaped characters in arguments.""" |
| 310 | # Apply xfail markers if configured |
| 311 | test_name = "test_escaped_strings" |
| 312 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 313 | |
| 314 | content, tool_calls = run_tool_extraction( |
| 315 | tool_parser, test_config.escaped_strings_output, streaming=streaming |
| 316 | ) |
| 317 | assert len(tool_calls) == 1, f"Expected 1 tool call, got {len(tool_calls)}" |
| 318 | |
| 319 | args = json.loads(tool_calls[0].function.arguments) |
| 320 | # At minimum, verify we can parse and have expected fields |
| 321 | # Exact escaping behavior varies by parser |
| 322 | assert len(args) > 0, "Expected some arguments with escaped strings" |
| 323 | |
| 324 | def test_malformed_input( |
| 325 | self, |
| 326 | request: pytest.FixtureRequest, |
| 327 | tool_parser: Any, |
| 328 | test_config: ToolParserTestConfig, |
| 329 | streaming: bool, |
| 330 | ): |
| 331 | """Verify parser gracefully handles invalid syntax.""" |
| 332 | # Apply xfail markers if configured |
| 333 | test_name = "test_malformed_input" |
| 334 | self.apply_xfail_mark(request, test_config, test_name, streaming) |
| 335 | |
| 336 | for malformed_input in test_config.malformed_input_outputs: |
| 337 | # Should not raise exception |
| 338 | content, tool_calls = run_tool_extraction( |
| 339 | tool_parser, malformed_input, streaming=streaming |
| 340 | ) |
| 341 | # Parser should handle gracefully (exact behavior varies) |
| 342 | |
| 343 | def test_streaming_reconstruction( |
| 344 | self, |
| 345 | request: pytest.FixtureRequest, |
| 346 | tool_parser: Any, |
| 347 | test_config: ToolParserTestConfig, |
| 348 | ): |
| 349 | """Verify streaming produces same result as non-streaming.""" |
| 350 | test_name = "test_streaming_reconstruction" |
| 351 | self.apply_xfail_mark(request, test_config, test_name, True) |
| 352 | |
| 353 | test_output = test_config.single_tool_call_output |
| 354 | |
| 355 | # Non-streaming result |
| 356 | content_non, tools_non = run_tool_extraction( |
| 357 | tool_parser, test_output, streaming=False |
| 358 | ) |
| 359 | |
| 360 | # Streaming result |
| 361 | content_stream, tools_stream = run_tool_extraction( |
| 362 | tool_parser, test_output, streaming=True |
| 363 | ) |
| 364 | |
| 365 | # Compare results |
| 366 | assert content_non == content_stream, "Content should match between modes" |
| 367 | assert len(tools_non) == len(tools_stream), "Tool count should match" |
| 368 | if len(tools_non) > 0: |
| 369 | assert tools_non[0].function.name == tools_stream[0].function.name |
| 370 | assert tools_non[0].function.arguments == tools_stream[0].function.arguments |
| 371 | |
| 372 | def apply_xfail_mark(self, request, test_config, test_name, streaming): |
| 373 | reason = None |
| 374 | if streaming and test_name in test_config.xfail_streaming: |
| 375 | reason = test_config.xfail_streaming[test_name] |
| 376 | elif not streaming and test_name in test_config.xfail_nonstreaming: |
| 377 | reason = test_config.xfail_nonstreaming[test_name] |
| 378 | if reason is not None: |
| 379 | mark = pytest.mark.xfail(reason=reason, strict=True) |
| 380 | request.node.add_marker(mark) |