From 91ab31e736a54dbbba5b07b2973d7df10af11439 Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Thu, 16 Apr 2026 18:15:23 +0000 Subject: [PATCH] Add review tests for PR #5061 --- .../test_pr5061_r2_anthropic_nonstream_502.py | 79 +++++++++++ ...st_pr5061_r2_anthropic_tool_choice_warn.py | 49 +++++++ tests/test_pr5061_r2_auth_headers.py | 40 ++++++ ..._pr5061_r2_enable_tools_tool_choice_400.py | 75 +++++++++++ tests/test_pr5061_r2_image_dedup_last_user.py | 74 +++++++++++ .../test_pr5061_r2_nonstream_success_shape.py | 92 +++++++++++++ .../test_pr5061_r2_nonstream_upstream_log.py | 74 +++++++++++ .../test_pr5061_r2_role_tool_nonpass_guard.py | 123 ++++++++++++++++++ tests/test_pr5061_r2_sse_done_on_error.py | 117 +++++++++++++++++ .../test_pr5061_r2_verbatim_stream_chunks.py | 122 +++++++++++++++++ 10 files changed, 845 insertions(+) create mode 100644 tests/test_pr5061_r2_anthropic_nonstream_502.py create mode 100644 tests/test_pr5061_r2_anthropic_tool_choice_warn.py create mode 100644 tests/test_pr5061_r2_auth_headers.py create mode 100644 tests/test_pr5061_r2_enable_tools_tool_choice_400.py create mode 100644 tests/test_pr5061_r2_image_dedup_last_user.py create mode 100644 tests/test_pr5061_r2_nonstream_success_shape.py create mode 100644 tests/test_pr5061_r2_nonstream_upstream_log.py create mode 100644 tests/test_pr5061_r2_role_tool_nonpass_guard.py create mode 100644 tests/test_pr5061_r2_sse_done_on_error.py create mode 100644 tests/test_pr5061_r2_verbatim_stream_chunks.py diff --git a/tests/test_pr5061_r2_anthropic_nonstream_502.py b/tests/test_pr5061_r2_anthropic_nonstream_502.py new file mode 100644 index 0000000000..f40887e6dc --- /dev/null +++ b/tests/test_pr5061_r2_anthropic_nonstream_502.py @@ -0,0 +1,79 @@ +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 diff --git a/tests/test_pr5061_r2_anthropic_tool_choice_warn.py b/tests/test_pr5061_r2_anthropic_tool_choice_warn.py new file mode 100644 index 0000000000..955063d675 --- /dev/null +++ b/tests/test_pr5061_r2_anthropic_tool_choice_warn.py @@ -0,0 +1,49 @@ +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() diff --git a/tests/test_pr5061_r2_auth_headers.py b/tests/test_pr5061_r2_auth_headers.py new file mode 100644 index 0000000000..8ba2939400 --- /dev/null +++ b/tests/test_pr5061_r2_auth_headers.py @@ -0,0 +1,40 @@ +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 diff --git a/tests/test_pr5061_r2_enable_tools_tool_choice_400.py b/tests/test_pr5061_r2_enable_tools_tool_choice_400.py new file mode 100644 index 0000000000..d1247beb89 --- /dev/null +++ b/tests/test_pr5061_r2_enable_tools_tool_choice_400.py @@ -0,0 +1,75 @@ +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 diff --git a/tests/test_pr5061_r2_image_dedup_last_user.py b/tests/test_pr5061_r2_image_dedup_last_user.py new file mode 100644 index 0000000000..60f9e0ff6e --- /dev/null +++ b/tests/test_pr5061_r2_image_dedup_last_user.py @@ -0,0 +1,74 @@ +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) diff --git a/tests/test_pr5061_r2_nonstream_success_shape.py b/tests/test_pr5061_r2_nonstream_success_shape.py new file mode 100644 index 0000000000..143c510188 --- /dev/null +++ b/tests/test_pr5061_r2_nonstream_success_shape.py @@ -0,0 +1,92 @@ +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" diff --git a/tests/test_pr5061_r2_nonstream_upstream_log.py b/tests/test_pr5061_r2_nonstream_upstream_log.py new file mode 100644 index 0000000000..ad4f077d06 --- /dev/null +++ b/tests/test_pr5061_r2_nonstream_upstream_log.py @@ -0,0 +1,74 @@ +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 diff --git a/tests/test_pr5061_r2_role_tool_nonpass_guard.py b/tests/test_pr5061_r2_role_tool_nonpass_guard.py new file mode 100644 index 0000000000..943896e13c --- /dev/null +++ b/tests/test_pr5061_r2_role_tool_nonpass_guard.py @@ -0,0 +1,123 @@ +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 "") diff --git a/tests/test_pr5061_r2_sse_done_on_error.py b/tests/test_pr5061_r2_sse_done_on_error.py new file mode 100644 index 0000000000..cc15d0a6b2 --- /dev/null +++ b/tests/test_pr5061_r2_sse_done_on_error.py @@ -0,0 +1,117 @@ +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 diff --git a/tests/test_pr5061_r2_verbatim_stream_chunks.py b/tests/test_pr5061_r2_verbatim_stream_chunks.py new file mode 100644 index 0000000000..0d694ba557 --- /dev/null +++ b/tests/test_pr5061_r2_verbatim_stream_chunks.py @@ -0,0 +1,122 @@ +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)