ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test(studio): consolidate PR #5061 review tests into the passthrough suite

Browse source 
The 10 test_pr5061_r2_*.py files at repo root tests/ had three issues:
wrong directory (studio tests live at studio/backend/tests/), PR number
baked into filenames, and two files (anthropic_nonstream_502,
anthropic_tool_choice_warn) duplicated coverage already in Roland's
TestFriendlyErrorHttpx / TestAnthropicToolChoiceToOpenAI. A third file
(nonstream_upstream_log) pinned logger-call shape, which is brittle
diagnostic noise.

Dropped the three low-value files and merged the rest into
studio/backend/tests/test_openai_tool_passthrough.py as generic classes:

- TestLlamaAuthHeaders
- TestOpenAIMessagesForPassthrough
- TestOpenAIChatCompletionsToolGuards
- TestAnthropicEnableToolsToolChoiceConflict
- TestOpenAIPassthroughNonStreaming  (verbatim body + httpx 502 mapping)
- TestOpenAIPassthroughStreamVerbatim
- TestOpenAIPassthroughStreamErrorTermination

Test count: 65 tests across 8 Test* classes. Net -316 lines.
This commit is contained in:
Daniel Han 2026-04-16 18:39:10 +00:00
commit b735c3a006
11 changed files with 634 additions and 950 deletions
Download patch file Download diff file Expand all files Collapse all files

640
studio/backend/tests/test_openai_tool_passthrough.py
View file

@ -14,28 +14,47 @@ Covers:
defaults to "auto" when unset.
- _friendly_error() maps httpx transport errors to a "Lost connection"
message so passthrough failures are legible instead of bare 500s.
- _llama_auth_headers() returns a Bearer header only when an API key is set.
- _openai_messages_for_passthrough() splices legacy image_base64 into the
last user message and skips the splice when one is already inline.
- openai_chat_completions() rejects role="tool" / tool_calls-only messages
when the request does not take the passthrough path.
- anthropic_messages() rejects the enable_tools + tool_choice combination.
- _openai_passthrough_non_streaming() wraps httpx transport errors as 502
and returns the upstream JSON body verbatim on success.
- _openai_passthrough_stream() relays data: lines verbatim, breaks on
[DONE], and always emits [DONE] after an upstream error.
No running server or GPU required.
"""
import os
import sys
_backend = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, _backend)
import asyncio
import threading
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from fastapi import HTTPException
from pydantic import ValidationError
# conftest.py adds the backend root to sys.path so these flat imports resolve.
from models.inference import (
AnthropicMessagesRequest,
ChatCompletionRequest,
ChatMessage,
)
from core.inference.anthropic_compat import (
anthropic_tool_choice_to_openai,
)
from routes.inference import _build_passthrough_payload, _friendly_error
from routes import inference as inference_module
from routes.inference import (
_build_passthrough_payload,
_friendly_error,
_llama_auth_headers,
_openai_messages_for_passthrough,
_openai_passthrough_non_streaming,
_openai_passthrough_stream,
)
# =====================================================================
@ -410,3 +429,612 @@ class TestFriendlyErrorHttpx:
assert (
_friendly_error(RuntimeError("unrelated")) == "An internal error occurred"
)
# =====================================================================
# _llama_auth_headers — Bearer header only when API key is set
# =====================================================================
class TestLlamaAuthHeaders:
def test_returns_bearer_header_when_api_key_set(self):
class _Backend:
_api_key = "k_secret_123"
assert _llama_auth_headers(_Backend()) == {
"Authorization": "Bearer k_secret_123",
}
def test_returns_none_when_api_key_none(self):
class _Backend:
_api_key = None
assert _llama_auth_headers(_Backend()) is None
def test_returns_none_when_api_key_attribute_missing(self):
class _Backend:
pass
assert _llama_auth_headers(_Backend()) is None
def test_returns_none_when_api_key_empty_string(self):
class _Backend:
_api_key = ""
assert _llama_auth_headers(_Backend()) is None
# =====================================================================
# _openai_messages_for_passthrough — legacy image_base64 splice
# =====================================================================
_TINY_PNG_B64 = (
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYA"
"AjCB0C8AAAAASUVORK5CYII="
)
def _inline_image_part(b64 = _TINY_PNG_B64):
return {
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{b64}"},
}
class _PayloadWithImage:
"""Minimal stand-in for ChatCompletionRequest; only attributes read by
_openai_messages_for_passthrough matter here."""
def __init__(self, messages, image_base64 = None):
self.messages = messages
self.image_base64 = image_base64
class TestOpenAIMessagesForPassthrough:
def test_prior_user_image_does_not_block_new_base64(self):
payload = _PayloadWithImage(
messages = [
ChatMessage(
role = "user",
content = [{"type": "text", "text": "q1"}, _inline_image_part()],
),
ChatMessage(role = "assistant", content = "a1"),
ChatMessage(role = "user", content = "q2 no inline"),
],
image_base64 = _TINY_PNG_B64,
)
out = _openai_messages_for_passthrough(payload)
assert isinstance(out[-1]["content"], list)
assert any(part.get("type") == "image_url" for part in out[-1]["content"])
def test_last_user_with_inline_image_skips_splice(self):
payload = _PayloadWithImage(
messages = [
ChatMessage(role = "user", content = "earlier"),
ChatMessage(
role = "user",
content = [
{"type": "text", "text": "last"},
_inline_image_part(),
],
),
],
image_base64 = _TINY_PNG_B64,
)
out = _openai_messages_for_passthrough(payload)
last_parts = out[-1]["content"]
assert sum(1 for part in last_parts if part.get("type") == "image_url") == 1
def test_no_user_messages_appends_trailing_user(self):
payload = _PayloadWithImage(
messages = [ChatMessage(role = "system", content = "sys")],
image_base64 = _TINY_PNG_B64,
)
out = _openai_messages_for_passthrough(payload)
assert out[-1]["role"] == "user"
assert any(part.get("type") == "image_url" for part in out[-1]["content"])
def test_only_last_of_multiple_user_turns_receives_splice(self):
payload = _PayloadWithImage(
messages = [
ChatMessage(role = "user", content = "u1"),
ChatMessage(role = "assistant", content = "a1"),
ChatMessage(role = "user", content = "u2"),
ChatMessage(role = "assistant", content = "a2"),
ChatMessage(role = "user", content = "u3"),
],
image_base64 = _TINY_PNG_B64,
)
out = _openai_messages_for_passthrough(payload)
assert isinstance(out[0]["content"], str)
assert isinstance(out[2]["content"], str)
assert isinstance(out[4]["content"], list)
# =====================================================================
# openai_chat_completions — tool-shape guards on non-passthrough paths
# =====================================================================
class _FakeLlamaBackend:
def __init__(self, is_loaded = True, supports_tools = True, is_vision = False):
self.is_loaded = is_loaded
self.supports_tools = supports_tools
self.is_vision = is_vision
self._is_audio = False
self.model_identifier = "stub"
self.base_url = "http://127.0.0.1:0"
self._api_key = None
class _FakeInferenceBackend:
active_model_name = "hf"
models = {"hf": {}}
class _FakeFastAPIRequest:
async def is_disconnected(self):
return False
async def _marker_stream(*args, **kwargs):
return ("passthrough_stream", None)
async def _marker_nonstream(*args, **kwargs):
return ("passthrough_nonstream", None)
async def _call_openai_chat_completions(payload, llama):
with (
patch.object(inference_module, "get_llama_cpp_backend", return_value = llama),
patch.object(
inference_module, "get_inference_backend", return_value = _FakeInferenceBackend(),
),
patch.object(inference_module, "_openai_passthrough_stream", new = _marker_stream),
patch.object(
inference_module, "_openai_passthrough_non_streaming", new = _marker_nonstream,
),
):
return await inference_module.openai_chat_completions(
payload, _FakeFastAPIRequest(), current_subject = "u",
)
class TestOpenAIChatCompletionsToolGuards:
"""When the request does NOT take the tool-passthrough path, messages
with role="tool" or assistant tool_calls-only must be rejected at the
route boundary rather than producing an opaque upstream error."""
def _tool_result_payload(self, **extra):
return ChatCompletionRequest(
messages = [
{"role": "user", "content": "q"},
{"role": "tool", "tool_call_id": "c1", "content": "r"},
],
**extra,
)
def test_role_tool_on_non_gguf_rejected(self):
payload = self._tool_result_payload()
with pytest.raises(HTTPException) as exc_info:
asyncio.run(
_call_openai_chat_completions(
payload, _FakeLlamaBackend(is_loaded = False),
)
)
assert exc_info.value.status_code == 400
def test_role_tool_on_gguf_without_tool_support_rejected(self):
payload = self._tool_result_payload()
with pytest.raises(HTTPException) as exc_info:
asyncio.run(
_call_openai_chat_completions(
payload, _FakeLlamaBackend(supports_tools = False),
)
)
assert exc_info.value.status_code == 400
def test_role_tool_with_enable_tools_true_rejected(self):
payload = self._tool_result_payload(enable_tools = True)
with pytest.raises(HTTPException) as exc_info:
asyncio.run(_call_openai_chat_completions(payload, _FakeLlamaBackend()))
assert exc_info.value.status_code == 400
def test_assistant_tool_calls_only_rejected_when_no_passthrough(self):
payload = ChatCompletionRequest(
messages = [
{"role": "user", "content": "q"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "c1",
"type": "function",
"function": {"name": "f", "arguments": "{}"},
}
],
},
],
)
with pytest.raises(HTTPException) as exc_info:
asyncio.run(
_call_openai_chat_completions(
payload, _FakeLlamaBackend(supports_tools = False),
)
)
assert exc_info.value.status_code == 400
def test_plain_user_message_does_not_trip_tool_guard(self):
payload = ChatCompletionRequest(
messages = [{"role": "user", "content": "plain"}],
)
try:
asyncio.run(
_call_openai_chat_completions(
payload, _FakeLlamaBackend(supports_tools = False),
)
)
except HTTPException as exc:
# Downstream paths may fail for unrelated reasons; only assert
# the tool-shape guard did NOT fire.
assert "role='tool'" not in (exc.detail or "")
assert "tool_calls-only" not in (exc.detail or "")
# =====================================================================
# anthropic_messages — enable_tools + tool_choice conflict
# =====================================================================
class TestAnthropicEnableToolsToolChoiceConflict:
"""Server-side agentic loop (enable_tools=True) does not honor
tool_choice. Reject the combination at the route boundary with 400
so callers don't silently see their tool_choice dropped."""
def _payload(self, *, enable_tools, tool_choice):
return AnthropicMessagesRequest(
model = "default",
max_tokens = 64,
messages = [{"role": "user", "content": "q"}],
tool_choice = tool_choice,
enable_tools = enable_tools,
)
def test_enable_tools_true_with_tool_choice_raises_400(self):
payload = self._payload(enable_tools = True, tool_choice = {"type": "any"})
with patch.object(
inference_module,
"get_llama_cpp_backend",
return_value = _FakeLlamaBackend(),
):
with pytest.raises(HTTPException) as exc_info:
asyncio.run(
inference_module.anthropic_messages(
payload, _FakeFastAPIRequest(), current_subject = "u",
)
)
assert exc_info.value.status_code == 400
assert "tool_choice" in exc_info.value.detail
assert "enable_tools" in exc_info.value.detail
def test_enable_tools_false_with_tool_choice_skips_guard(self):
payload = self._payload(enable_tools = False, tool_choice = {"type": "any"})
with patch.object(
inference_module,
"get_llama_cpp_backend",
return_value = _FakeLlamaBackend(),
):
try:
asyncio.run(
inference_module.anthropic_messages(
payload, _FakeFastAPIRequest(), current_subject = "u",
)
)
except HTTPException as exc:
assert not (
exc.status_code == 400
and "tool_choice is not honored" in (exc.detail or "")
)
except Exception:
# Any other failure downstream is fine; we only pin that
# the enable_tools+tool_choice guard does NOT fire.
pass
# =====================================================================
# _openai_passthrough_non_streaming — verbatim body + httpx 502 mapping
# =====================================================================
class _FakeLlamaBase:
base_url = "http://127.0.0.1:0"
_api_key = None
def _openai_tools_payload(stream = False):
return ChatCompletionRequest(
messages = [{"role": "user", "content": "q"}],
tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
stream = stream,
)
def _mock_async_client_post(status_code, *, json_body = None, text_body = ""):
class _Client:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
async def post(self, *args, **kwargs):
resp = MagicMock()
resp.status_code = status_code
resp.text = text_body
resp.json = lambda: (json_body if json_body is not None else {})
return resp
return _Client
def _mock_async_client_raise(exc_factory):
class _Client:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
async def post(self, *args, **kwargs):
raise exc_factory()
return _Client
class TestOpenAIPassthroughNonStreaming:
def test_verbatim_json_body_returned_on_success(self):
native = {
"id": "chatcmpl-foo",
"object": "chat.completion",
"model": "qwen-native",
"choices": [
{
"index": 0,
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
},
}
],
"usage": {"prompt_tokens": 42, "completion_tokens": 7, "total_tokens": 49},
}
client_cls = _mock_async_client_post(200, json_body = native)
with patch.object(inference_module.httpx, "AsyncClient", client_cls):
resp = asyncio.run(
_openai_passthrough_non_streaming(_FakeLlamaBase(), _openai_tools_payload())
)
import json
body = json.loads(resp.body.decode("utf-8"))
assert body == native
assert body["choices"][0]["finish_reason"] == "tool_calls"
def test_preserves_native_id_and_model(self):
native = {
"id": "chatcmpl-native-xyz",
"model": "llama-native",
"choices": [
{
"finish_reason": "stop",
"message": {"role": "assistant", "content": "ok"},
}
],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
client_cls = _mock_async_client_post(200, json_body = native)
with patch.object(inference_module.httpx, "AsyncClient", client_cls):
resp = asyncio.run(
_openai_passthrough_non_streaming(_FakeLlamaBase(), _openai_tools_payload())
)
import json
body = json.loads(resp.body.decode("utf-8"))
assert body["id"] == "chatcmpl-native-xyz"
assert body["model"] == "llama-native"
def test_httpx_connect_error_mapped_to_502(self):
client_cls = _mock_async_client_raise(
lambda: httpx.ConnectError(
"refused", request = httpx.Request("POST", "http://x"),
)
)
with patch.object(inference_module.httpx, "AsyncClient", client_cls):
with pytest.raises(HTTPException) as exc_info:
asyncio.run(
_openai_passthrough_non_streaming(
_FakeLlamaBase(), _openai_tools_payload(),
)
)
assert exc_info.value.status_code == 502
assert "Lost connection" in exc_info.value.detail
def test_httpx_read_error_mapped_to_502(self):
client_cls = _mock_async_client_raise(
lambda: httpx.ReadError("eof", request = httpx.Request("POST", "http://x")),
)
with patch.object(inference_module.httpx, "AsyncClient", client_cls):
with pytest.raises(HTTPException) as exc_info:
asyncio.run(
_openai_passthrough_non_streaming(
_FakeLlamaBase(), _openai_tools_payload(),
)
)
assert exc_info.value.status_code == 502
# =====================================================================
# _openai_passthrough_stream — verbatim data: lines, [DONE] on error
# =====================================================================
class _FakeStreamResponse:
"""Stand-in for httpx.Response that yields a fixed list of SSE lines."""
def __init__(self, lines, status_code = 200):
self.status_code = status_code
self._lines = list(lines)
def aiter_lines(self):
parent = self
class _Iter:
def __init__(self):
self._idx = 0
def __aiter__(self):
return self
async def __anext__(self):
if self._idx >= len(parent._lines):
raise StopAsyncIteration
v = parent._lines[self._idx]
self._idx += 1
return v
async def aclose(self):
pass
return _Iter()
async def aread(self):
return b""
async def aclose(self):
pass
def _run_passthrough_stream(*, send_returns = None, send_raises = None):
"""Drive _openai_passthrough_stream with a fake httpx client and return
the list of emitted SSE chunks (decoded to str)."""
async def _send(req, stream):
if send_raises is not None:
raise send_raises()
return send_returns
with patch.object(inference_module.httpx, "AsyncClient") as mock_cls:
instance = MagicMock()
instance.build_request = MagicMock(return_value = MagicMock())
instance.send = _send
instance.aclose = AsyncMock()
mock_cls.return_value = instance
resp = asyncio.run(
_openai_passthrough_stream(
_FakeFastAPIRequest(),
threading.Event(),
_FakeLlamaBase(),
_openai_tools_payload(stream = True),
)
)
async def _collect():
return [
chunk if isinstance(chunk, str) else chunk.decode("utf-8")
async for chunk in resp.body_iterator
]
return asyncio.run(_collect())
class TestOpenAIPassthroughStreamVerbatim:
def test_relays_data_lines_verbatim(self):
chunks = _run_passthrough_stream(
send_returns = _FakeStreamResponse([
'data: {"id":"abc","choices":[{"delta":{"content":"hi"}}]}',
"data: [DONE]",
]),
)
assert any('"id":"abc"' in c for c in chunks)
assert any("data: [DONE]" in c for c in chunks)
def test_ignores_blank_and_non_data_lines(self):
chunks = _run_passthrough_stream(
send_returns = _FakeStreamResponse([
"",
": heartbeat",
'data: {"x":1}',
"data: [DONE]",
]),
)
for chunk in chunks:
assert chunk.startswith("data: ") or chunk == ""
assert any('"x":1' in c for c in chunks)
def test_breaks_on_done(self):
chunks = _run_passthrough_stream(
send_returns = _FakeStreamResponse([
'data: {"a":1}',
"data: [DONE]",
'data: {"should_not_appear":true}',
]),
)
assert not any("should_not_appear" in c for c in chunks)
class TestOpenAIPassthroughStreamErrorTermination:
"""On upstream failure (non-200 or transport exception), the stream
must emit an SSE error chunk followed by `data: [DONE]` so clients
that wait for [DONE] don't hang."""
def test_done_emitted_after_non_200_error(self):
class _Resp:
status_code = 500
async def aread(self):
return b"server oops"
async def aclose(self):
pass
chunks = _run_passthrough_stream(send_returns = _Resp())
assert any('"error"' in c for c in chunks)
assert any(c.strip() == "data: [DONE]" for c in chunks)
def test_done_emitted_after_transport_exception(self):
chunks = _run_passthrough_stream(
send_raises = lambda: httpx.ConnectError(
"boom", request = httpx.Request("POST", "http://x"),
),
)
assert any('"error"' in c for c in chunks)
assert any(c.strip() == "data: [DONE]" for c in chunks)
def test_done_comes_after_error_chunk(self):
chunks = _run_passthrough_stream(
send_raises = lambda: httpx.ReadError(
"reset", request = httpx.Request("POST", "http://x"),
),
)
err_idx = next(i for i, c in enumerate(chunks) if '"error"' in c)
done_idx = next(i for i, c in enumerate(chunks) if c.strip() == "data: [DONE]")
assert done_idx > err_idx

97
tests/test_pr5061_r2_anthropic_nonstream_502.py
View file

@ -1,97 +0,0 @@
import asyncio
import os, sys
from unittest.mock import patch
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
import httpx
import pytest
from fastapi import HTTPException
class _Llama:
base_url = "http://127.0.0.1:0"
_api_key = None
def test_httpx_connect_error_mapped_to_502():
from routes import inference as inf_mod
class _BadClient:
def __init__(self, *a, **kw):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def post(self, *a, **kw):
raise httpx.ConnectError(
"refused", request = httpx.Request("POST", "http://x")
)
with patch.object(inf_mod.httpx, "AsyncClient", _BadClient):
with pytest.raises(HTTPException) as ei:
asyncio.run(
inf_mod._anthropic_passthrough_non_streaming(
_Llama(),
openai_messages = [{"role": "user", "content": "q"}],
openai_tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
temperature = 0.6,
top_p = 0.95,
top_k = 20,
max_tokens = None,
message_id = "msg_1",
model_name = "stub",
tool_choice = "auto",
)
)
assert ei.value.status_code == 502
assert "Lost connection" in ei.value.detail
def test_httpx_read_error_mapped_to_502():
from routes import inference as inf_mod
class _BadClient:
def __init__(self, *a, **kw):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def post(self, *a, **kw):
raise httpx.ReadError("eof", request = httpx.Request("POST", "http://x"))
with patch.object(inf_mod.httpx, "AsyncClient", _BadClient):
with pytest.raises(HTTPException) as ei:
asyncio.run(
inf_mod._anthropic_passthrough_non_streaming(
_Llama(),
openai_messages = [{"role": "user", "content": "q"}],
openai_tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
temperature = 0.6,
top_p = 0.95,
top_k = 20,
max_tokens = None,
message_id = "msg_1",
model_name = "stub",
)
)
assert ei.value.status_code == 502

49
tests/test_pr5061_r2_anthropic_tool_choice_warn.py
View file

@ -1,49 +0,0 @@
import os, sys
from unittest.mock import patch, MagicMock
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
from routes import inference as inf_mod
def _simulate_coerce(tool_choice):
"""Mimic the exact snippet that lives at the top of anthropic_messages()."""
translated = inf_mod.anthropic_tool_choice_to_openai(tool_choice)
if translated is None:
if tool_choice is not None:
inf_mod.logger.warning(
"anthropic_messages.tool_choice_unrecognized",
tool_choice = tool_choice,
)
translated = "auto"
return translated
def test_unrecognized_dict_warns_and_falls_back_to_auto():
with patch.object(inf_mod.logger, "warning") as mock_warn:
result = _simulate_coerce({"type": "wibble"})
assert result == "auto"
mock_warn.assert_called_once()
assert mock_warn.call_args.args[0] == "anthropic_messages.tool_choice_unrecognized"
def test_non_dict_input_warns_and_falls_back():
with patch.object(inf_mod.logger, "warning") as mock_warn:
result = _simulate_coerce("auto_string")
assert result == "auto"
mock_warn.assert_called_once()
def test_none_input_no_warning():
with patch.object(inf_mod.logger, "warning") as mock_warn:
result = _simulate_coerce(None)
assert result == "auto"
mock_warn.assert_not_called()
def test_recognized_input_no_warning():
with patch.object(inf_mod.logger, "warning") as mock_warn:
result = _simulate_coerce({"type": "any"})
assert result == "required"
mock_warn.assert_not_called()

40
tests/test_pr5061_r2_auth_headers.py
View file

@ -1,40 +0,0 @@
import os, sys
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
from routes.inference import _llama_auth_headers
class _WithKey:
_api_key = "k_secret_123"
class _NoKey:
_api_key = None
class _Missing:
pass
class _EmptyKey:
_api_key = ""
def test_returns_bearer_header_when_api_key_set():
headers = _llama_auth_headers(_WithKey())
assert headers == {"Authorization": "Bearer k_secret_123"}
def test_returns_none_when_api_key_none():
assert _llama_auth_headers(_NoKey()) is None
def test_returns_none_when_api_key_attribute_missing():
assert _llama_auth_headers(_Missing()) is None
def test_returns_none_when_api_key_empty_string():
# Empty string is falsy; the helper should treat it as absent.
assert _llama_auth_headers(_EmptyKey()) is None

82
tests/test_pr5061_r2_enable_tools_tool_choice_400.py
View file

@ -1,82 +0,0 @@
import asyncio
import os, sys
from unittest.mock import patch
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
import pytest
from fastapi import HTTPException
class _Llama:
is_loaded = True
supports_tools = True
is_vision = False
_is_audio = False
model_identifier = "stub"
base_url = "http://127.0.0.1:0"
_api_key = None
class _Req:
async def is_disconnected(self):
return False
def _build_payload(enable_tools, tool_choice):
from models.inference import AnthropicMessagesRequest
return AnthropicMessagesRequest(
model = "default",
max_tokens = 64,
messages = [{"role": "user", "content": "q"}],
tool_choice = tool_choice,
enable_tools = enable_tools,
)
def test_enable_tools_true_with_tool_choice_raises_400():
from routes import inference as inf_mod
payload = _build_payload(enable_tools = True, tool_choice = {"type": "any"})
with patch.object(inf_mod, "get_llama_cpp_backend", return_value = _Llama()):
with pytest.raises(HTTPException) as ei:
asyncio.run(
inf_mod.anthropic_messages(payload, _Req(), current_subject = "u")
)
assert ei.value.status_code == 400
assert "tool_choice" in ei.value.detail
assert "enable_tools" in ei.value.detail
def test_guard_logic_unit():
# Re-implement the guard predicate exactly and pin it.
def would_raise(enable_tools, supports_tools, tool_choice):
server_tools = enable_tools and supports_tools
return bool(server_tools and tool_choice is not None)
assert would_raise(True, True, {"type": "any"}) is True
assert would_raise(True, True, None) is False
assert would_raise(False, True, {"type": "any"}) is False
assert would_raise(True, False, {"type": "any"}) is False
def test_enable_tools_false_with_tool_choice_does_not_raise_the_guard():
from routes import inference as inf_mod
payload = _build_payload(enable_tools = False, tool_choice = {"type": "any"})
with patch.object(inf_mod, "get_llama_cpp_backend", return_value = _Llama()):
try:
asyncio.run(
inf_mod.anthropic_messages(payload, _Req(), current_subject = "u")
)
except HTTPException as e:
assert not (
e.status_code == 400
and "tool_choice is not honored" in (e.detail or "")
)
except Exception:
# Any other failure downstream is fine; we only pin that the
# enable_tools+tool_choice guard does NOT fire when enable_tools is False.
pass

79
tests/test_pr5061_r2_image_dedup_last_user.py
View file

@ -1,79 +0,0 @@
import os, sys
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
from models.inference import ChatMessage
from routes.inference import _openai_messages_for_passthrough
_TINY = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
class _P:
def __init__(self, messages, image_base64 = None):
self.messages = messages
self.image_base64 = image_base64
def _img_url_part(b64 = _TINY):
return {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
def test_prior_user_image_does_not_block_new_base64():
p = _P(
messages = [
ChatMessage(
role = "user", content = [{"type": "text", "text": "q1"}, _img_url_part()]
),
ChatMessage(role = "assistant", content = "a1"),
ChatMessage(role = "user", content = "q2 no inline"),
],
image_base64 = _TINY,
)
out = _openai_messages_for_passthrough(p)
# last user must receive spliced image
assert isinstance(out[-1]["content"], list)
assert any(x.get("type") == "image_url" for x in out[-1]["content"])
def test_last_user_with_inline_image_skips_splice():
p = _P(
messages = [
ChatMessage(role = "user", content = "earlier"),
ChatMessage(
role = "user", content = [{"type": "text", "text": "last"}, _img_url_part()]
),
],
image_base64 = _TINY,
)
out = _openai_messages_for_passthrough(p)
last_parts = out[-1]["content"]
assert sum(1 for x in last_parts if x.get("type") == "image_url") == 1
def test_no_user_messages_appends_trailing_user():
p = _P(
messages = [ChatMessage(role = "system", content = "sys")],
image_base64 = _TINY,
)
out = _openai_messages_for_passthrough(p)
assert out[-1]["role"] == "user"
assert any(x.get("type") == "image_url" for x in out[-1]["content"])
def test_three_user_turns_only_last_receives_splice():
p = _P(
messages = [
ChatMessage(role = "user", content = "u1"),
ChatMessage(role = "assistant", content = "a1"),
ChatMessage(role = "user", content = "u2"),
ChatMessage(role = "assistant", content = "a2"),
ChatMessage(role = "user", content = "u3"),
],
image_base64 = _TINY,
)
out = _openai_messages_for_passthrough(p)
assert isinstance(out[0]["content"], str)
assert isinstance(out[2]["content"], str)
assert isinstance(out[4]["content"], list)

112
tests/test_pr5061_r2_nonstream_success_shape.py
View file

@ -1,112 +0,0 @@
import asyncio
import os, sys
from unittest.mock import patch, MagicMock
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
from models.inference import ChatCompletionRequest
class _Llama:
base_url = "http://127.0.0.1:0"
_api_key = None
def _payload():
return ChatCompletionRequest(
messages = [{"role": "user", "content": "q"}],
tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
)
def _build_mock_client(status, payload_json):
class _Client:
def __init__(self, *a, **kw):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def post(self, *a, **kw):
m = MagicMock()
m.status_code = status
m.text = ""
m.json = lambda: payload_json
return m
return _Client
def test_verbatim_json_body_returned():
from routes import inference as inf_mod
native = {
"id": "chatcmpl-foo",
"object": "chat.completion",
"model": "qwen-native",
"choices": [
{
"index": 0,
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
},
}
],
"usage": {"prompt_tokens": 42, "completion_tokens": 7, "total_tokens": 49},
}
with patch.object(inf_mod.httpx, "AsyncClient", _build_mock_client(200, native)):
resp = asyncio.run(
inf_mod._openai_passthrough_non_streaming(_Llama(), _payload())
)
import json
body = json.loads(resp.body.decode("utf-8"))
assert body == native # verbatim
assert body["choices"][0]["finish_reason"] == "tool_calls"
assert body["usage"]["prompt_tokens"] == 42
def test_preserves_native_id_and_model_fields():
from routes import inference as inf_mod
native = {
"id": "chatcmpl-native-xyz",
"model": "llama-native",
"choices": [
{"finish_reason": "stop", "message": {"role": "assistant", "content": "ok"}}
],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
with patch.object(inf_mod.httpx, "AsyncClient", _build_mock_client(200, native)):
resp = asyncio.run(
inf_mod._openai_passthrough_non_streaming(_Llama(), _payload())
)
import json
body = json.loads(resp.body.decode("utf-8"))
assert body["id"] == "chatcmpl-native-xyz"
assert body["model"] == "llama-native"

93
tests/test_pr5061_r2_nonstream_upstream_log.py
View file

@ -1,93 +0,0 @@
import asyncio
import os, sys
from unittest.mock import patch, MagicMock
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
import pytest
from fastapi import HTTPException
from models.inference import ChatCompletionRequest
class _Llama:
base_url = "http://127.0.0.1:0"
_api_key = None
def _payload():
return ChatCompletionRequest(
messages = [{"role": "user", "content": "q"}],
tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
)
def _mk_client(status, text):
class _Client:
def __init__(self, *a, **kw):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def post(self, *a, **kw):
m = MagicMock()
m.status_code = status
m.text = text
m.json = lambda: {}
return m
return _Client
def test_nonstream_non_200_calls_logger_error_with_status_and_body():
from routes import inference as inf_mod
with (
patch.object(
inf_mod.httpx, "AsyncClient", _mk_client(503, "backend overloaded detail")
),
patch.object(inf_mod.logger, "error") as mock_error,
):
with pytest.raises(HTTPException) as ei:
asyncio.run(inf_mod._openai_passthrough_non_streaming(_Llama(), _payload()))
assert ei.value.status_code == 503
mock_error.assert_called()
# Check any error call mentions upstream status and body
found = False
for call in mock_error.call_args_list:
msg = call.args[0] if call.args else ""
combined = f"{msg} {call.args} {call.kwargs}"
if (
"upstream error" in combined
and "503" in combined
and "backend overloaded" in combined
):
found = True
break
assert found, f"expected upstream error log with status and body, got {mock_error.call_args_list}"
def test_nonstream_200_does_not_call_logger_error():
from routes import inference as inf_mod
with (
patch.object(inf_mod.httpx, "AsyncClient", _mk_client(200, "{}")),
patch.object(inf_mod.logger, "error") as mock_error,
):
asyncio.run(inf_mod._openai_passthrough_non_streaming(_Llama(), _payload()))
# no upstream-error log on 200
for call in mock_error.call_args_list:
msg = call.args[0] if call.args else ""
assert "upstream error" not in msg

134
tests/test_pr5061_r2_role_tool_nonpass_guard.py
View file

@ -1,134 +0,0 @@
import asyncio
import os, sys
from unittest.mock import patch
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
import pytest
from fastapi import HTTPException
from models.inference import ChatCompletionRequest
class _Llama:
def __init__(self, is_loaded = True, supports_tools = True, is_vision = False):
self.is_loaded = is_loaded
self.supports_tools = supports_tools
self.is_vision = is_vision
self._is_audio = False
self.model_identifier = "stub"
self.base_url = "http://127.0.0.1:0"
self._api_key = None
class _Inf:
active_model_name = "hf"
models = {"hf": {}}
class _Req:
async def is_disconnected(self):
return False
async def _marker_stream(*a, **kw):
return ("passthrough_stream", None)
async def _marker_nonstream(*a, **kw):
return ("passthrough_nonstream", None)
async def _call(payload, llama):
from routes import inference as inf_mod
with (
patch.object(inf_mod, "get_llama_cpp_backend", return_value = llama),
patch.object(inf_mod, "get_inference_backend", return_value = _Inf()),
patch.object(inf_mod, "_openai_passthrough_stream", new = _marker_stream),
patch.object(
inf_mod, "_openai_passthrough_non_streaming", new = _marker_nonstream
),
):
return await inf_mod.openai_chat_completions(
payload, _Req(), current_subject = "u"
)
def test_role_tool_on_non_gguf_rejected():
payload = ChatCompletionRequest(
messages = [
{"role": "user", "content": "q"},
{"role": "tool", "tool_call_id": "c1", "content": "r"},
],
)
llama = _Llama(is_loaded = False)
with pytest.raises(HTTPException) as ei:
asyncio.run(_call(payload, llama))
assert ei.value.status_code == 400
assert "role='tool'" in ei.value.detail or "tool" in ei.value.detail.lower()
def test_role_tool_on_gguf_without_tool_support_rejected():
payload = ChatCompletionRequest(
messages = [
{"role": "user", "content": "q"},
{"role": "tool", "tool_call_id": "c1", "content": "r"},
],
)
llama = _Llama(supports_tools = False)
with pytest.raises(HTTPException) as ei:
asyncio.run(_call(payload, llama))
assert ei.value.status_code == 400
def test_role_tool_with_enable_tools_true_rejected():
payload = ChatCompletionRequest(
messages = [
{"role": "user", "content": "q"},
{"role": "tool", "tool_call_id": "c1", "content": "r"},
],
enable_tools = True,
)
llama = _Llama()
with pytest.raises(HTTPException) as ei:
asyncio.run(_call(payload, llama))
assert ei.value.status_code == 400
def test_assistant_tool_only_still_rejected_on_non_passthrough():
payload = ChatCompletionRequest(
messages = [
{"role": "user", "content": "q"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "c1",
"type": "function",
"function": {"name": "f", "arguments": "{}"},
}
],
},
],
)
llama = _Llama(supports_tools = False)
with pytest.raises(HTTPException) as ei:
asyncio.run(_call(payload, llama))
assert ei.value.status_code == 400
def test_no_tool_messages_passes_through():
payload = ChatCompletionRequest(
messages = [{"role": "user", "content": "plain"}],
)
llama = _Llama(supports_tools = False)
# Should not raise the tool-shape guard; may raise later for non-GGUF/inf path,
# but specifically this guard must not fire.
try:
asyncio.run(_call(payload, llama))
except HTTPException as e:
assert "role='tool'" not in (e.detail or "")
assert "tool_calls-only" not in (e.detail or "")

125
tests/test_pr5061_r2_sse_done_on_error.py
View file

@ -1,125 +0,0 @@
import asyncio
import os, sys, threading
from unittest.mock import patch, MagicMock, AsyncMock
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
import httpx
from models.inference import ChatCompletionRequest
class _Llama:
base_url = "http://127.0.0.1:0"
_api_key = None
class _Req:
async def is_disconnected(self):
return False
def _collect(stream_response):
async def _run():
out = []
async for chunk in stream_response.body_iterator:
out.append(chunk if isinstance(chunk, str) else chunk.decode("utf-8"))
return out
return asyncio.run(_run())
def _make_payload():
return ChatCompletionRequest(
messages = [{"role": "user", "content": "q"}],
tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
stream = True,
)
def test_done_emitted_after_non_200_error():
from routes import inference as inf_mod
class _Resp:
status_code = 500
async def aread(self):
return b"server oops"
async def aclose(self):
pass
async def _send(req, stream):
return _Resp()
with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
inst = MagicMock()
inst.build_request = MagicMock(return_value = MagicMock())
inst.send = _send
inst.aclose = AsyncMock()
mock_cls.return_value = inst
resp = asyncio.run(
inf_mod._openai_passthrough_stream(
_Req(), threading.Event(), _Llama(), _make_payload()
)
)
chunks = _collect(resp)
assert any('"error"' in c for c in chunks)
assert any(c.strip() == "data: [DONE]" for c in chunks)
def test_done_emitted_after_exception():
from routes import inference as inf_mod
async def _send(req, stream):
raise httpx.ConnectError("boom", request = httpx.Request("POST", "http://x"))
with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
inst = MagicMock()
inst.build_request = MagicMock(return_value = MagicMock())
inst.send = _send
inst.aclose = AsyncMock()
mock_cls.return_value = inst
resp = asyncio.run(
inf_mod._openai_passthrough_stream(
_Req(), threading.Event(), _Llama(), _make_payload()
)
)
chunks = _collect(resp)
assert any('"error"' in c for c in chunks)
assert any(c.strip() == "data: [DONE]" for c in chunks)
def test_done_comes_after_error_chunk():
from routes import inference as inf_mod
async def _send(req, stream):
raise httpx.ReadError("reset", request = httpx.Request("POST", "http://x"))
with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
inst = MagicMock()
inst.build_request = MagicMock(return_value = MagicMock())
inst.send = _send
inst.aclose = AsyncMock()
mock_cls.return_value = inst
resp = asyncio.run(
inf_mod._openai_passthrough_stream(
_Req(), threading.Event(), _Llama(), _make_payload()
)
)
chunks = _collect(resp)
err_idx = next(i for i, c in enumerate(chunks) if '"error"' in c)
done_idx = next(i for i, c in enumerate(chunks) if c.strip() == "data: [DONE]")
assert done_idx > err_idx

133
tests/test_pr5061_r2_verbatim_stream_chunks.py
View file

@ -1,133 +0,0 @@
import asyncio
import os, sys, threading
from unittest.mock import patch, MagicMock
_backend = os.path.join(os.path.dirname(__file__), "..", "studio", "backend")
sys.path.insert(0, _backend)
from models.inference import ChatCompletionRequest
class _Llama:
base_url = "http://127.0.0.1:0"
_api_key = None
class _Req:
async def is_disconnected(self):
return False
def _payload():
return ChatCompletionRequest(
messages = [{"role": "user", "content": "q"}],
tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
stream = True,
)
class _FakeResp:
def __init__(self, lines):
self.status_code = 200
self._lines = list(lines)
def aiter_lines(self):
parent = self
class _It:
def __init__(self):
self._idx = 0
def __aiter__(self):
return self
async def __anext__(self):
if self._idx >= len(parent._lines):
raise StopAsyncIteration
v = parent._lines[self._idx]
self._idx += 1
return v
async def aclose(self):
pass
return _It()
async def aread(self):
return b""
async def aclose(self):
pass
def _run_stream(fake_lines):
from routes import inference as inf_mod
async def _send(req, stream):
return _FakeResp(fake_lines)
async def _aclose_noop():
return None
with patch.object(inf_mod.httpx, "AsyncClient") as mock_cls:
inst = MagicMock()
inst.build_request = MagicMock(return_value = MagicMock())
inst.send = _send
inst.aclose = _aclose_noop
mock_cls.return_value = inst
resp = asyncio.run(
inf_mod._openai_passthrough_stream(
_Req(), threading.Event(), _Llama(), _payload()
)
)
async def _collect():
return [
c if isinstance(c, str) else c.decode("utf-8")
async for c in resp.body_iterator
]
return asyncio.run(_collect())
def test_passthrough_relays_data_lines_verbatim():
chunks = _run_stream(
[
'data: {"id":"abc","choices":[{"delta":{"content":"hi"}}]}',
"data: [DONE]",
]
)
assert any('"id":"abc"' in c for c in chunks)
assert any("data: [DONE]" in c for c in chunks)
def test_passthrough_ignores_blank_and_non_data_lines():
chunks = _run_stream(
[
"",
": heartbeat",
'data: {"x":1}',
"data: [DONE]",
]
)
# Only data: lines propagate.
for c in chunks:
assert c.startswith("data: ") or c == ""
assert any('"x":1' in c for c in chunks)
def test_passthrough_breaks_on_done():
chunks = _run_stream(
[
'data: {"a":1}',
"data: [DONE]",
'data: {"should_not_appear":true}',
]
)
assert not any("should_not_appear" in c for c in chunks)