mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-03 07:23:33 +00:00
d76db470f9
* test: update integration tests left stale by recent product changes * test: expect proxy Messages calls to stay on Python with Rust enabled
588 lines
26 KiB
Python
588 lines
26 KiB
Python
import json
|
|
import uuid
|
|
from collections.abc import Callable, Mapping
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
import anthropic
|
|
import httpx
|
|
import pytest
|
|
from integration._support.client import Gateway, eventually, object_value
|
|
from integration._support.database import read_rows
|
|
from integration._support.wire import Reply, Request, Wire, wire_server
|
|
from openai import OpenAI
|
|
from pydantic import JsonValue, TypeAdapter
|
|
|
|
_BACKEND: Final = "deepseek-ai/DeepSeek-V4-Flash"
|
|
_MODEL: Final = f"inferenceapis/{_BACKEND}"
|
|
_API_KEY: Final = "synthetic-inferenceapis-key"
|
|
_OPENAI_KEY: Final = "synthetic-openai-key"
|
|
_OVERFLOW_TEXT: Final = "Input too long: 150055 input tokens, limit is 131072 for this model"
|
|
_PARAM_ERROR_TEXT: Final = (
|
|
"Input should be 'error' or 'truncate', field: 'context_length_exceeded_behavior', value: 'foo'"
|
|
)
|
|
_WEATHER_TOOL: Final[JsonValue] = {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_weather",
|
|
"description": "Get the current weather for a city",
|
|
"parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
|
|
},
|
|
}
|
|
_COST_MAP_PATH: Final = Path(__file__).resolve().parents[3] / "model_prices_and_context_window.json"
|
|
_ENDPOINT_SUPPORT_PATH: Final = Path(__file__).resolve().parents[3] / "litellm/provider_endpoints_support_backup.json"
|
|
_COST_MAP: Final = TypeAdapter(dict[str, dict[str, JsonValue]])
|
|
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
|
|
_JSON_LIST: Final = TypeAdapter(list[JsonValue])
|
|
|
|
|
|
def _identity(label: str) -> str:
|
|
return f"inferenceapis-{label}-{uuid.uuid4().hex}"
|
|
|
|
|
|
def _catalog_cost(field: str) -> float:
|
|
value: Final = _COST_MAP.validate_json(_COST_MAP_PATH.read_bytes())[_MODEL][field]
|
|
assert isinstance(value, (int, float))
|
|
return float(value)
|
|
|
|
|
|
def _approx(value: float) -> object:
|
|
return pytest.approx(value, rel=1e-6) # pyright: ignore[reportUnknownMemberType] # pytest lacks typed approx stubs
|
|
|
|
|
|
def _sse(events: tuple[Mapping[str, JsonValue], ...], named: bool) -> tuple[bytes, ...]:
|
|
return tuple(
|
|
(f"event: {event['type']}\n" if named else "").encode() + f"data: {json.dumps(event)}\n\n".encode()
|
|
for event in events
|
|
)
|
|
|
|
|
|
def _chat_completion(identity: str, message: Mapping[str, JsonValue], finish: str = "stop") -> Reply:
|
|
return Reply(
|
|
body=json.dumps(
|
|
{
|
|
"id": identity,
|
|
"object": "chat.completion",
|
|
"created": 1,
|
|
"model": _BACKEND,
|
|
"choices": [{"index": 0, "message": message, "finish_reason": finish}],
|
|
"usage": {"prompt_tokens": 23, "completion_tokens": 41, "total_tokens": 64},
|
|
}
|
|
).encode()
|
|
)
|
|
|
|
|
|
def _chat_frames(identity: str, pieces: tuple[str, ...]) -> Reply:
|
|
chunk: Final[dict[str, JsonValue]] = {
|
|
"id": identity,
|
|
"object": "chat.completion.chunk",
|
|
"created": 1,
|
|
"model": _BACKEND,
|
|
}
|
|
deltas: Final[tuple[dict[str, JsonValue], ...]] = tuple(
|
|
{**chunk, "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}]} for piece in pieces
|
|
)
|
|
finish: Final[dict[str, JsonValue]] = {**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
|
|
return Reply(
|
|
content_type="text/event-stream",
|
|
chunks=(*_sse((*deltas, finish), named=False), b"data: [DONE]\n\n"),
|
|
)
|
|
|
|
|
|
def _response_object(identity: str, text: str, status: str) -> dict[str, JsonValue]:
|
|
return {
|
|
"id": f"resp_{identity}",
|
|
"object": "response",
|
|
"created_at": 1,
|
|
"status": status,
|
|
"model": _BACKEND,
|
|
"output": [
|
|
{
|
|
"type": "message",
|
|
"id": f"msg_{identity}",
|
|
"status": "completed",
|
|
"role": "assistant",
|
|
"content": [{"type": "output_text", "text": text, "annotations": []}],
|
|
}
|
|
]
|
|
if status == "completed"
|
|
else [],
|
|
"usage": {"input_tokens": 9, "output_tokens": 2, "total_tokens": 11},
|
|
}
|
|
|
|
|
|
def _responses_reply(identity: str, text: str, stream: bool) -> Reply:
|
|
completed: Final = _response_object(identity, text, "completed")
|
|
if not stream:
|
|
return Reply(body=json.dumps(completed).encode())
|
|
events: Final[tuple[Mapping[str, JsonValue], ...]] = (
|
|
{"type": "response.created", "sequence_number": 0, "response": _response_object(identity, text, "in_progress")},
|
|
{
|
|
"type": "response.output_text.delta",
|
|
"sequence_number": 1,
|
|
"item_id": f"msg_{identity}",
|
|
"output_index": 0,
|
|
"content_index": 0,
|
|
"delta": text,
|
|
},
|
|
{"type": "response.completed", "sequence_number": 2, "response": completed},
|
|
)
|
|
return Reply(content_type="text/event-stream", chunks=_sse(events, named=True))
|
|
|
|
|
|
def _message(identity: str, text: str) -> dict[str, JsonValue]:
|
|
return {
|
|
"id": identity,
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"model": _BACKEND,
|
|
"content": [{"type": "text", "text": text}],
|
|
"stop_reason": "end_turn",
|
|
"stop_sequence": None,
|
|
"usage": {"input_tokens": 11, "output_tokens": 3},
|
|
}
|
|
|
|
|
|
def _messages_reply(identity: str, text: str, stream: bool) -> Reply:
|
|
if not stream:
|
|
return Reply(body=json.dumps(_message(identity, text)).encode())
|
|
events: Final[tuple[Mapping[str, JsonValue], ...]] = (
|
|
{"type": "message_start", "message": {**_message(identity, ""), "content": [], "stop_reason": None}},
|
|
{"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
|
|
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}},
|
|
{"type": "content_block_stop", "index": 0},
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
|
|
"usage": {"output_tokens": 3},
|
|
},
|
|
{"type": "message_stop"},
|
|
)
|
|
return Reply(content_type="text/event-stream", chunks=_sse(events, named=True))
|
|
|
|
|
|
def _error(status: int, message: str, code: str | None, kind: str = "invalid_request_error") -> Reply:
|
|
return Reply(status=status, body=json.dumps({"error": {"message": message, "type": kind, "code": code}}).encode())
|
|
|
|
|
|
def _body(request: Request) -> dict[str, JsonValue]:
|
|
return _JSON_OBJECT.validate_json(request.body)
|
|
|
|
|
|
def _assert_vendor_request(request: Request, target: str) -> dict[str, JsonValue]:
|
|
assert (request.method, request.target) == ("POST", target), request.target
|
|
assert request.headers["authorization"] == f"Bearer {_API_KEY}"
|
|
body: Final = _body(request)
|
|
assert body["model"] == _BACKEND
|
|
return body
|
|
|
|
|
|
def _chat(model: str, prompt: str, **extra: JsonValue) -> dict[str, JsonValue]:
|
|
return {"model": model, "messages": [{"role": "user", "content": prompt}], **extra}
|
|
|
|
|
|
def _messages(model: str, prompt: str, **extra: JsonValue) -> dict[str, JsonValue]:
|
|
return {"model": model, "max_tokens": 32, "messages": [{"role": "user", "content": prompt}], **extra}
|
|
|
|
|
|
def _responses(model: str, prompt: str, **extra: JsonValue) -> dict[str, JsonValue]:
|
|
return {"model": model, "input": prompt, **extra}
|
|
|
|
|
|
def _consumed(gateway: Gateway, path: str, body: Mapping[str, JsonValue]) -> tuple[httpx.Response, str]:
|
|
with gateway.client.stream("POST", path, json=body, headers={"Authorization": f"Bearer {gateway.key}"}) as response:
|
|
text: Final = b"".join(response.iter_bytes()).decode()
|
|
return response, text
|
|
|
|
|
|
def _data_events(text: str) -> tuple[dict[str, JsonValue], ...]:
|
|
return tuple(
|
|
_JSON_OBJECT.validate_json(line.removeprefix("data: "))
|
|
for line in text.splitlines()
|
|
if line.startswith("data: ") and line != "data: [DONE]"
|
|
)
|
|
|
|
|
|
def _targets(wire: Wire) -> list[tuple[str, str]]:
|
|
return [(request.method, request.target) for request in wire.drain()]
|
|
|
|
|
|
def test_chat_completion_reaches_the_vendor_and_bills_the_catalog_price(gateway: Gateway) -> None:
|
|
identity: Final = _identity("chat")
|
|
prompt: Final = f"chat {identity}"
|
|
|
|
def respond(request: Request) -> Reply:
|
|
body: Final = _assert_vendor_request(request, "/v1/chat/completions")
|
|
assert body["messages"] == [{"role": "user", "content": prompt}]
|
|
return _chat_completion(identity, {"role": "assistant", "content": "43"})
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response: Final = gateway.request("POST", "/v1/chat/completions", _chat(model, prompt))
|
|
assert response.status_code == 200, response.text
|
|
payload: Final = _JSON_OBJECT.validate_json(response.content)
|
|
assert payload["id"] == identity
|
|
choice: Final = object_value(_JSON_LIST.validate_python(payload["choices"])[0])
|
|
assert object_value(choice["message"])["content"] == "43"
|
|
expected_cost: Final = 23 * _catalog_cost("input_cost_per_token") + 41 * _catalog_cost("output_cost_per_token")
|
|
assert float(response.headers["x-litellm-response-cost"]) == _approx(expected_cost)
|
|
assert _targets(wire) == [("POST", "/v1/chat/completions")]
|
|
rows: Final = eventually(
|
|
lambda: read_rows(
|
|
'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s',
|
|
(identity,),
|
|
),
|
|
lambda values: len(values) == 1,
|
|
seconds=70,
|
|
)
|
|
assert (rows[0]["prompt_tokens"], rows[0]["completion_tokens"]) == (23, 41)
|
|
spend: Final = rows[0]["spend"]
|
|
assert isinstance(spend, (int, float, str))
|
|
assert float(spend) == _approx(expected_cost)
|
|
|
|
|
|
def test_streamed_chat_completion_delivers_every_delta_and_one_done(gateway: Gateway) -> None:
|
|
identity: Final = _identity("chat-stream")
|
|
pieces: Final = ("1\n2\n", "3\n4\n", "5")
|
|
|
|
def respond(request: Request) -> Reply:
|
|
assert _assert_vendor_request(request, "/v1/chat/completions")["stream"] is True
|
|
return _chat_frames(identity, pieces)
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response, text = _consumed(gateway, "/v1/chat/completions", _chat(model, f"count {identity}", stream=True))
|
|
assert response.status_code == 200, text
|
|
assert text.count("data: [DONE]") == 1
|
|
events: Final = _data_events(text)
|
|
assert {str(event["id"]) for event in events} == {identity}
|
|
deltas: Final = tuple(
|
|
object_value(object_value(choice)["delta"]).get("content") or ""
|
|
for event in events
|
|
for choice in _JSON_LIST.validate_python(event["choices"])
|
|
)
|
|
assert "".join(str(delta) for delta in deltas) == "".join(pieces)
|
|
assert _targets(wire) == [("POST", "/v1/chat/completions")]
|
|
|
|
|
|
@pytest.mark.parametrize("stream", (False, True), ids=("json", "stream"))
|
|
def test_responses_call_goes_to_the_vendor_responses_endpoint(gateway: Gateway, stream: bool) -> None:
|
|
identity: Final = _identity("responses")
|
|
answer: Final = f"answer {identity}"
|
|
prompt: Final = f"responses {identity}"
|
|
|
|
def respond(request: Request) -> Reply:
|
|
body: Final = _assert_vendor_request(request, "/v1/responses")
|
|
assert body["input"] == prompt
|
|
assert bool(body.get("stream")) is stream
|
|
return _responses_reply(identity, answer, stream)
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response, text = _consumed(gateway, "/v1/responses", _responses(model, prompt, stream=stream))
|
|
assert response.status_code == 200, text
|
|
completed: Final = (
|
|
_JSON_OBJECT.validate_json(text)
|
|
if not stream
|
|
else object_value(
|
|
next(event for event in _data_events(text) if event["type"] == "response.completed")["response"]
|
|
)
|
|
)
|
|
assert completed["status"] == "completed"
|
|
assert answer in json.dumps(completed["output"])
|
|
assert _targets(wire) == [("POST", "/v1/responses")]
|
|
|
|
|
|
@pytest.mark.parametrize("stream", (False, True), ids=("json", "stream"))
|
|
def test_messages_call_goes_to_the_vendor_messages_endpoint_untranslated(gateway: Gateway, stream: bool) -> None:
|
|
identity: Final = _identity("messages")
|
|
answer: Final = f"answer {identity}"
|
|
prompt: Final = f"messages {identity}"
|
|
|
|
def respond(request: Request) -> Reply:
|
|
body: Final = _assert_vendor_request(request, "/v1/messages")
|
|
assert body["messages"] == [{"role": "user", "content": prompt}]
|
|
assert body["max_tokens"] == 32
|
|
assert "anthropic-version" in request.headers
|
|
assert bool(body.get("stream")) is stream
|
|
return _messages_reply(identity, answer, stream)
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response, text = _consumed(gateway, "/v1/messages", _messages(model, prompt, stream=stream))
|
|
assert response.status_code == 200, text
|
|
if stream:
|
|
events: Final = _data_events(text)
|
|
assert [event["type"] for event in events][-1] == "message_stop"
|
|
assert object_value(events[0]["message"])["id"] == identity
|
|
assert answer in text
|
|
else:
|
|
payload: Final = _JSON_OBJECT.validate_json(text)
|
|
assert payload["id"] == identity
|
|
assert payload["content"] == [{"type": "text", "text": answer}]
|
|
assert _targets(wire) == [("POST", "/v1/messages")]
|
|
|
|
|
|
def test_chat_tool_call_round_trips(gateway: Gateway) -> None:
|
|
identity: Final = _identity("tool")
|
|
tool_call: Final[JsonValue] = {
|
|
"id": "call_paris",
|
|
"type": "function",
|
|
"function": {"name": "get_weather", "arguments": json.dumps({"city": "Paris"})},
|
|
}
|
|
|
|
def respond(request: Request) -> Reply:
|
|
body: Final = _assert_vendor_request(request, "/v1/chat/completions")
|
|
assert body["tools"] == [_WEATHER_TOOL]
|
|
assert body["tool_choice"] == "auto"
|
|
message: Final[dict[str, JsonValue]] = {"role": "assistant", "content": None, "tool_calls": [tool_call]}
|
|
return _chat_completion(identity, message, "tool_calls")
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response: Final = gateway.request(
|
|
"POST",
|
|
"/v1/chat/completions",
|
|
_chat(model, "weather in Paris", tools=[_WEATHER_TOOL], tool_choice="auto"),
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
payload: Final = _JSON_OBJECT.validate_json(response.content)
|
|
assert payload["id"] == identity
|
|
choice: Final = object_value(_JSON_LIST.validate_python(payload["choices"])[0])
|
|
assert choice["finish_reason"] == "tool_calls"
|
|
assert object_value(choice["message"])["tool_calls"] == [tool_call]
|
|
assert _targets(wire) == [("POST", "/v1/chat/completions")]
|
|
|
|
|
|
def test_non_openai_params_reach_the_vendor_body_top_level(gateway: Gateway) -> None:
|
|
identity: Final = _identity("extra")
|
|
|
|
def respond(request: Request) -> Reply:
|
|
body: Final = _assert_vendor_request(request, "/v1/chat/completions")
|
|
assert body["top_k"] == 5
|
|
assert body["chat_template_kwargs"] == {"enable_thinking": False}
|
|
assert "extra_body" not in body
|
|
return _chat_completion(identity, {"role": "assistant", "content": "ok"})
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response: Final = gateway.request(
|
|
"POST",
|
|
"/v1/chat/completions",
|
|
_chat(model, "extra params", top_k=5, chat_template_kwargs={"enable_thinking": False}),
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert _JSON_OBJECT.validate_json(response.content)["id"] == identity
|
|
assert _targets(wire) == [("POST", "/v1/chat/completions")]
|
|
|
|
|
|
_OVERFLOW_CASES: Final = (
|
|
("/v1/chat/completions", "/v1/chat/completions", "context_length_exceeded"),
|
|
("/v1/chat/completions", "/v1/chat/completions", "upstream_rejected"),
|
|
("/v1/messages", "/v1/messages", "upstream_rejected"),
|
|
("/v1/responses", "/v1/responses", "upstream_rejected"),
|
|
)
|
|
_REQUEST_BUILDERS: Final[Mapping[str, Callable[[str, str], dict[str, JsonValue]]]] = {
|
|
"/v1/chat/completions": _chat,
|
|
"/v1/messages": _messages,
|
|
"/v1/responses": _responses,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(("route", "target", "code"), _OVERFLOW_CASES)
|
|
def test_vendor_overflow_reaches_the_caller_as_a_context_window_error(
|
|
gateway: Gateway, route: str, target: str, code: str
|
|
) -> None:
|
|
prompt: Final = f"overflow {uuid.uuid4().hex}"
|
|
|
|
def respond(request: Request) -> Reply:
|
|
_assert_vendor_request(request, target)
|
|
return _error(400, _OVERFLOW_TEXT, code)
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response: Final = gateway.request("POST", route, _REQUEST_BUILDERS[route](model, prompt))
|
|
assert response.status_code == 400, response.text
|
|
assert "ContextWindowExceededError" in response.text, response.text
|
|
assert _OVERFLOW_TEXT in response.text, response.text
|
|
assert _targets(wire) == [("POST", target)]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "message", "code", "kind", "caller_status", "caller_class"),
|
|
(
|
|
(400, _PARAM_ERROR_TEXT, None, "invalid_request_error", 400, "BadRequestError"),
|
|
(401, "Invalid API key", "invalid_api_key", "authentication_error", 401, "AuthenticationError"),
|
|
),
|
|
ids=("param-error-naming-the-overflow-field", "bad-key"),
|
|
)
|
|
def test_vendor_errors_that_are_not_overflow_keep_their_class(
|
|
gateway: Gateway, status: int, message: str, code: str | None, kind: str, caller_status: int, caller_class: str
|
|
) -> None:
|
|
def respond(request: Request) -> Reply:
|
|
_assert_vendor_request(request, "/v1/chat/completions")
|
|
return _error(status, message, code, kind)
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response: Final = gateway.request("POST", "/v1/chat/completions", _chat(model, f"error {uuid.uuid4().hex}"))
|
|
assert response.status_code == caller_status, response.text
|
|
assert caller_class in response.text, response.text
|
|
assert "ContextWindowExceededError" not in response.text, response.text
|
|
assert _targets(wire) == [("POST", "/v1/chat/completions")]
|
|
|
|
|
|
def _echoing(_: Request) -> Reply:
|
|
return _chat_completion(_identity("echo"), {"role": "assistant", "content": "ok"})
|
|
|
|
|
|
def _forwarded_top_k(wire: Wire) -> list[JsonValue]:
|
|
return [_body(request).get("top_k") for request in wire.drain()]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
(5, [1, 2], "", "x" * 5120),
|
|
ids=("int", "list", "empty-string", "five-kb"),
|
|
)
|
|
def test_a_non_openai_param_of_any_shape_forwards_like_it_does_for_an_openai_deployment(
|
|
gateway: Gateway, value: JsonValue
|
|
) -> None:
|
|
with (
|
|
wire_server(_echoing) as vendor,
|
|
wire_server(_echoing) as openai_peer,
|
|
gateway.scenario() as scenario,
|
|
):
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{vendor.url}/v1", api_key=_API_KEY)
|
|
control: Final = scenario.model(
|
|
model="openai/gpt-4o-mini", api_base=f"{openai_peer.url}/v1", api_key=_OPENAI_KEY
|
|
)
|
|
responses: Final = tuple(
|
|
gateway.request("POST", "/v1/chat/completions", _chat(name, "shape", top_k=value))
|
|
for name in (model, control)
|
|
)
|
|
assert [response.status_code for response in responses] == [200, 200], [r.text for r in responses]
|
|
assert _forwarded_top_k(vendor) == [value]
|
|
assert _forwarded_top_k(openai_peer) == [value]
|
|
|
|
|
|
def test_a_param_sent_twice_forwards_the_last_value_like_an_openai_deployment(gateway: Gateway) -> None:
|
|
with (
|
|
wire_server(_echoing) as vendor,
|
|
wire_server(_echoing) as openai_peer,
|
|
gateway.scenario() as scenario,
|
|
):
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{vendor.url}/v1", api_key=_API_KEY)
|
|
control: Final = scenario.model(
|
|
model="openai/gpt-4o-mini", api_base=f"{openai_peer.url}/v1", api_key=_OPENAI_KEY
|
|
)
|
|
responses: Final = tuple(
|
|
gateway.client.post(
|
|
"/v1/chat/completions",
|
|
content=(
|
|
f'{{"model": "{name}", "messages": [{{"role": "user", "content": "twice"}}], '
|
|
'"top_k": 1, "top_k": 2}'
|
|
).encode(),
|
|
headers={"Authorization": f"Bearer {gateway.key}", "Content-Type": "application/json"},
|
|
)
|
|
for name in (model, control)
|
|
)
|
|
assert [response.status_code for response in responses] == [200, 200], [r.text for r in responses]
|
|
assert _forwarded_top_k(vendor) == [2]
|
|
assert _forwarded_top_k(openai_peer) == [2]
|
|
|
|
|
|
def test_messages_on_a_json_provider_without_native_messages_forwards_top_k(gateway: Gateway) -> None:
|
|
identity: Final = _identity("bridge")
|
|
|
|
def respond(request: Request) -> Reply:
|
|
assert (request.method, request.target) == ("POST", "/v1/chat/completions"), request.target
|
|
assert _body(request)["top_k"] == 5
|
|
return _chat_completion(identity, {"role": "assistant", "content": "bridged"})
|
|
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model="veniceai/venice-uncensored", api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
response: Final = gateway.request("POST", "/v1/messages", _messages(model, "bridge", top_k=5))
|
|
assert response.status_code == 200, response.text
|
|
assert _JSON_OBJECT.validate_json(response.content)["content"] == [{"type": "text", "text": "bridged"}]
|
|
assert _targets(wire) == [("POST", "/v1/chat/completions")]
|
|
|
|
|
|
def _lists_inferenceapis(endpoint: Mapping[str, JsonValue]) -> bool:
|
|
providers: Final = _JSON_LIST.validate_python(endpoint["providers"])
|
|
return any(object_value(provider)["slug"] == "inferenceapis" for provider in providers)
|
|
|
|
|
|
def _bundled_inferenceapis_endpoints() -> frozenset[str]:
|
|
providers: Final = object_value(_JSON_OBJECT.validate_json(_ENDPOINT_SUPPORT_PATH.read_bytes())["providers"])
|
|
flags: Final = object_value(object_value(providers["inferenceapis"])["endpoints"])
|
|
return frozenset(key for key, supported in flags.items() if supported)
|
|
|
|
|
|
def test_public_routes_list_the_provider_with_its_endpoints_and_fields(gateway: Gateway) -> None:
|
|
providers: Final = gateway.request("GET", "/public/providers")
|
|
assert providers.status_code == 200, providers.text
|
|
assert "inferenceapis" in _JSON_LIST.validate_json(providers.content)
|
|
|
|
fields_response: Final = gateway.request("GET", "/public/providers/fields")
|
|
assert fields_response.status_code == 200, fields_response.text
|
|
entries: Final = tuple(
|
|
object_value(entry)
|
|
for entry in _JSON_LIST.validate_json(fields_response.content)
|
|
if object_value(entry)["litellm_provider"] == "inferenceapis"
|
|
)
|
|
assert len(entries) == 1
|
|
required: Final = {
|
|
object_value(field)["key"]: object_value(field)["required"]
|
|
for field in _JSON_LIST.validate_python(entries[0]["credential_fields"])
|
|
}
|
|
assert required == {"api_base": False, "api_key": True}
|
|
|
|
endpoints_response: Final = gateway.request("GET", "/public/endpoints")
|
|
assert endpoints_response.status_code == 200, endpoints_response.text
|
|
endpoints: Final = tuple(
|
|
object_value(endpoint)
|
|
for endpoint in _JSON_LIST.validate_python(_JSON_OBJECT.validate_json(endpoints_response.content)["endpoints"])
|
|
)
|
|
listed: Final = frozenset(str(endpoint["key"]) for endpoint in endpoints if _lists_inferenceapis(endpoint))
|
|
assert {"chat_completions", "messages", "responses"} <= listed, listed
|
|
assert listed == _bundled_inferenceapis_endpoints()
|
|
|
|
|
|
def test_openai_and_anthropic_sdk_clients_reach_the_vendor(gateway: Gateway) -> None:
|
|
chat_identity: Final = _identity("sdk-chat")
|
|
message_identity: Final = _identity("sdk-messages")
|
|
|
|
def respond(request: Request) -> Reply:
|
|
if request.target == "/v1/messages":
|
|
_assert_vendor_request(request, "/v1/messages")
|
|
return _messages_reply(message_identity, "from messages", stream=False)
|
|
_assert_vendor_request(request, "/v1/chat/completions")
|
|
return _chat_completion(chat_identity, {"role": "assistant", "content": "from chat"})
|
|
|
|
base_url: Final = str(gateway.client.base_url).rstrip("/")
|
|
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
|
model: Final = scenario.model(model=_MODEL, api_base=f"{wire.url}/v1", api_key=_API_KEY)
|
|
openai_client: Final = OpenAI(
|
|
api_key=gateway.key,
|
|
base_url=f"{base_url}/v1",
|
|
max_retries=0,
|
|
http_client=httpx.Client(timeout=30, trust_env=False),
|
|
)
|
|
completion: Final = openai_client.chat.completions.create(
|
|
model=model, messages=[{"role": "user", "content": "sdk chat"}]
|
|
)
|
|
assert completion.id == chat_identity
|
|
assert completion.choices[0].message.content == "from chat"
|
|
anthropic_client: Final = anthropic.Anthropic(
|
|
base_url=base_url, api_key=gateway.key, auth_token=gateway.key, max_retries=0
|
|
)
|
|
message: Final = anthropic_client.messages.create(
|
|
model=model, max_tokens=32, messages=[{"role": "user", "content": "sdk messages"}]
|
|
)
|
|
assert message.id == message_identity
|
|
assert len(message.content) == 1
|
|
block: Final = message.content[0]
|
|
assert isinstance(block, anthropic.types.TextBlock)
|
|
assert block.text == "from messages"
|
|
assert sorted(_targets(wire)) == [("POST", "/v1/chat/completions"), ("POST", "/v1/messages")]
|