Spaces:
Running on Zero
Running on Zero
| """Regression coverage for the ZeroGPU OpenAI-compatible tool-call contract.""" | |
| import json | |
| import os | |
| from unittest.mock import patch | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| import app | |
| from app import normalize_native_tool_calls, parse_model_tool_calls | |
| def test_parse_qwen_xml_tool_call_with_newlines(): | |
| text = """<think>Need project data.</think> | |
| <tool_call> | |
| <function=get_project_status> | |
| <parameter=project>flow</parameter> | |
| </function> | |
| </tool_call>""" | |
| calls, reasoning, content = parse_model_tool_calls(text) | |
| assert reasoning == "Need project data." | |
| assert content is None | |
| assert calls == [{ | |
| "index": 0, | |
| "id": calls[0]["id"], | |
| "type": "function", | |
| "function": {"name": "get_project_status", "arguments": '{"project": "flow"}'}, | |
| }] | |
| def test_parse_qwen_json_tool_call_without_space_padding(): | |
| calls, reasoning, content = parse_model_tool_calls( | |
| '<tool_call>{"name":"get_status","arguments":{"project":"flow"}}</tool_call>' | |
| ) | |
| assert reasoning is None | |
| assert content is None | |
| assert calls[0]["function"] == { | |
| "name": "get_status", | |
| "arguments": '{"project": "flow"}', | |
| } | |
| def test_normalize_native_llama_cpp_tool_calls(): | |
| calls = normalize_native_tool_calls([{ | |
| "id": "call_native", | |
| "type": "function", | |
| "function": {"name": "get_project_status", "arguments": {"project": "flow"}}, | |
| }]) | |
| assert calls == [{ | |
| "index": 0, | |
| "id": "call_native", | |
| "type": "function", | |
| "function": { | |
| "name": "get_project_status", | |
| "arguments": '{"project": "flow"}', | |
| }, | |
| }] | |
| def test_parse_two_calls_in_one_turn_keeps_indexes(): | |
| calls, _, _ = parse_model_tool_calls( | |
| '<tool_call><function=a><parameter=x>1</parameter></function></tool_call>' | |
| '<tool_call><function=b><parameter=y>{"nested": {"deep": 2}}</parameter></function></tool_call>' | |
| ) | |
| assert [c["index"] for c in calls] == [0, 1] | |
| assert [c["function"]["name"] for c in calls] == ["a", "b"] | |
| assert json.loads(calls[1]["function"]["arguments"]) == {"y": {"nested": {"deep": 2}}} | |
| def test_parse_nested_json_arguments_and_multiline_json(): | |
| calls, _, content = parse_model_tool_calls( | |
| '<tool_call>\n{"name":"search","arguments":{"filters":{"tag":"x","range":{"min":1,"max":9}}},' | |
| '"id":"c1"}\n</tool_call>' | |
| ) | |
| assert content is None | |
| assert calls[0]["function"]["name"] == "search" | |
| assert json.loads(calls[0]["function"]["arguments"]) == { | |
| "filters": {"tag": "x", "range": {"min": 1, "max": 9}} | |
| } | |
| def test_parse_handles_empty_and_none_input(): | |
| assert parse_model_tool_calls(None) == (None, None, "") | |
| assert parse_model_tool_calls("") == (None, None, "") | |
| assert parse_model_tool_calls("plain answer") == (None, None, "plain answer") | |
| def space_client(): | |
| headers = {} | |
| if os.environ.get("FLOW_API_KEY"): | |
| headers["Authorization"] = f"Bearer {os.environ['FLOW_API_KEY']}" | |
| return TestClient(app.fastapi_app, headers=headers) | |
| def _native_response(content, tool_calls=None, finish_reason="stop"): | |
| message = {"content": content} | |
| if tool_calls is not None: | |
| message["tool_calls"] = tool_calls | |
| return { | |
| "choices": [{"message": message, "finish_reason": finish_reason}], | |
| "usage": {"prompt_tokens": 5, "completion_tokens": 2}, | |
| } | |
| def test_endpoint_surfaces_native_tool_calls_when_content_is_none(space_client): | |
| raw = _native_response( | |
| None, | |
| tool_calls=[{ | |
| "id": "call_native", | |
| "type": "function", | |
| "function": {"name": "get_project_status", "arguments": {"project": "flow"}}, | |
| }], | |
| finish_reason="tool_calls", | |
| ) | |
| payload = { | |
| "model": "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", | |
| "messages": [{"role": "user", "content": "status"}], | |
| "tools": [{"type": "function", "function": {"name": "get_project_status", "parameters": {"type": "object"}}}], | |
| } | |
| with patch.object(app, "list_gguf_files", return_value=["Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf"]), \ | |
| patch.object(app, "gpu_tracked_call", return_value=raw): | |
| response = space_client.post("/v1/chat/completions", json=payload) | |
| choice = response.json()["choices"][0] | |
| assert response.status_code == 200 | |
| assert choice["finish_reason"] == "tool_calls" | |
| call = choice["message"]["tool_calls"][0] | |
| assert call["function"]["name"] == "get_project_status" | |
| assert json.loads(call["function"]["arguments"]) == {"project": "flow"} | |
| def test_endpoint_does_not_crash_on_none_content_without_tools(space_client): | |
| with patch.object(app, "list_gguf_files", return_value=["Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf"]), \ | |
| patch.object(app, "gpu_tracked_call", return_value=_native_response(None)): | |
| response = space_client.post( | |
| "/v1/chat/completions", | |
| json={"model": "Qwen3.8-27B-Uncensored.i1-Q4_K_M.gguf", "messages": [{"role": "user", "content": "hi"}]}, | |
| ) | |
| choice = response.json()["choices"][0] | |
| assert response.status_code == 200 | |
| assert choice["finish_reason"] == "stop" | |
| assert choice["message"].get("content") == "" | |