(url, test_case, stream)
| 268 | |
| 269 | |
| 270 | def run_test(url, test_case, stream): |
| 271 | name = test_case["name"] |
| 272 | mode = f"{'stream' if stream else 'non-stream'}" |
| 273 | apply_stage = test_case.get("apply_stage", "always") |
| 274 | print_header(f"{name} [{mode}] ({apply_stage})") |
| 275 | |
| 276 | response_format = test_case["response_format"] |
| 277 | print_schema_note(apply_stage, response_format) |
| 278 | |
| 279 | tools = test_case.get("tools") |
| 280 | mocks = test_case.get("mock_tool_responses") or {} |
| 281 | |
| 282 | all_tcs: list[dict] = [] |
| 283 | final_content = None |
| 284 | |
| 285 | if apply_stage == "always": |
| 286 | all_tcs, _msgs, final_content = run_tool_loop( |
| 287 | url, |
| 288 | messages=list(test_case["messages"]), |
| 289 | tools=tools, |
| 290 | mock_tool_responses=mocks, |
| 291 | stream=stream, |
| 292 | response_format=response_format, |
| 293 | ) |
| 294 | elif apply_stage == "after_tools": |
| 295 | # Phase 1: plain tool loop, no response_format applied yet. |
| 296 | all_tcs, msgs, interim_content = run_tool_loop( |
| 297 | url, |
| 298 | messages=list(test_case["messages"]), |
| 299 | tools=tools, |
| 300 | mock_tool_responses=mocks, |
| 301 | stream=stream, |
| 302 | response_format=None, |
| 303 | ) |
| 304 | if interim_content: |
| 305 | msgs.append({"role": "assistant", "content": interim_content}) |
| 306 | followup = test_case.get( |
| 307 | "followup", |
| 308 | "Now output the answer strictly as JSON matching the provided schema. " |
| 309 | "Do not include commentary.", |
| 310 | ) |
| 311 | msgs.append({"role": "user", "content": followup}) |
| 312 | |
| 313 | # Phase 2: request final structured output. Tools are not passed so the |
| 314 | # model focuses on producing the schema-constrained answer. |
| 315 | _print(f"\n{DIM}{MAGENTA} ⟐ follow-up turn with response_format applied{RESET}") |
| 316 | result = chat_completion( |
| 317 | url, msgs, tools=None, response_format=response_format, stream=stream |
| 318 | ) |
| 319 | final_content = result["content"] if result else None |
| 320 | else: |
| 321 | print_fail(f"Unknown apply_stage: {apply_stage}") |
| 322 | return False |
| 323 | |
| 324 | if final_content is None: |
| 325 | print_fail("No final content from server.") |
| 326 | return False |
| 327 |
no test coverage detected