diff --git a/tests/integration/test_venice_web_search_wire.py b/tests/integration/test_venice_web_search_wire.py new file mode 100644 index 00000000..2636af1f --- /dev/null +++ b/tests/integration/test_venice_web_search_wire.py @@ -0,0 +1,131 @@ +"""What Routstr actually puts on the wire for a Venice web-search request. + +The unit tests stop at the kwargs handed to litellm. Everything that produced +the reported ``400 Unrecognized key(s) in object: 'web_search_options'`` +happened *after* that point, inside litellm's Anthropic adapter, so this test +runs the whole dispatch against a loopback OpenAI-compatible server and reads +the bytes Venice would have received. +""" + +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any, Iterator + +import pytest + +from routstr.payment.models import Architecture, Model, Pricing +from routstr.upstream.litellm_routing import configure_litellm +from routstr.upstream.venice import VeniceUpstreamProvider + +_CHUNKS = [ + { + "id": "chatcmpl-1", + "object": "chat.completion.chunk", + "created": 0, + "model": "deepseek-v4-flash-0731", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}], + }, + { + "id": "chatcmpl-1", + "object": "chat.completion.chunk", + "created": 0, + "model": "deepseek-v4-flash-0731", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7}, + }, +] + + +@pytest.fixture +def upstream() -> Iterator[tuple[str, dict[str, Any]]]: + """A loopback stand-in for ``api.venice.ai`` that records one request.""" + captured: dict[str, Any] = {} + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 - http.server's spelling + length = int(self.headers.get("Content-Length", 0)) + captured["path"] = self.path + captured["body"] = json.loads(self.rfile.read(length)) + + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + for chunk in _CHUNKS: + self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode()) + self.wfile.write(b"data: [DONE]\n\n") + + def log_message(self, *args: Any) -> None: + return None + + server = HTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}/v1", captured + finally: + server.shutdown() + thread.join(timeout=5) + + +def _model() -> Model: + return Model( + id="deepseek-v4-flash-0731", + name="deepseek-v4-flash-0731", + created=0, + description="", + context_length=8192, + architecture=Architecture( + modality="text->text", + input_modalities=["text"], + output_modalities=["text"], + tokenizer="Unknown", + instruct_type=None, + ), + pricing=Pricing(prompt=0.0, completion=0.0), + ) + + +@pytest.mark.asyncio +async def test_web_search_request_reaches_venice_in_its_own_shape( + upstream: tuple[str, dict[str, Any]], +) -> None: + base_url, captured = upstream + # The app applies this at startup; without it litellm posts the Anthropic + # body to /responses, which Venice serves only in alpha. + configure_litellm() + + provider = VeniceUpstreamProvider(api_key="sk-test") + provider.base_url = base_url + + await provider._dispatch_anthropic_messages( + request_body=json.dumps( + { + "model": "venice/deepseek-v4-flash-0731", + "messages": [{"role": "user", "content": "what shipped today?"}], + "max_tokens": 64, + "stream": True, + "tools": [ + {"type": "web_search_20250305", "name": "web_search"}, + { + "name": "lookup", + "description": "Look something up", + "input_schema": {"type": "object", "properties": {}}, + }, + ], + } + ).encode(), + model_obj=_model(), + ) + + body = captured["body"] + assert captured["path"] == "/v1/chat/completions" + # The reported 400, at the only place it could be observed. + assert "web_search_options" not in body + assert body["model"] == ( + "deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true" + ) + # The function tool still travels, in OpenAI's shape. + assert [tool["function"]["name"] for tool in body["tools"]] == ["lookup"]