Spaces:
Running on Zero
Running on Zero
File size: 5,278 Bytes
09957a9 a86614a 09957a9 a86614a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | """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")
@pytest.fixture
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") == ""
|