mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-13 03:59:12 +00:00
3090f4b23e
* test(integration): wait for customer spend rows and use owned OpenRouter cost * ci: isolate integration workers and clean up resources after crashes * docs: describe isolated extensions workers and shard census * fix(ci): restore integration service and fixture guarantees * fix(integration): preserve replay IDs and native route fixtures * fix(tests): scope integration harness imports to each unit case * fix(ci): preserve fixtures and wait for owned workers * fix(ci): isolate nested runs and telemetry cases * fix(ci): make worker readiness and fault tests deterministic * chore(ci): drop MCP fixture repair already covered by main * chore(ci): keep the local catalog fix from main --------- Co-authored-by: yuneng <yuneng@berri.ai>
59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
import base64
|
|
import json
|
|
from collections.abc import Mapping
|
|
from typing import Final
|
|
|
|
from pydantic import JsonValue
|
|
|
|
from integration._support.client import Gateway
|
|
from integration._support.provider import SharedProvider
|
|
from integration._support.wire import Reply
|
|
from integration.translation.case import TranslationTestCase
|
|
from litellm.responses.utils import ResponsesAPIRequestUtils
|
|
|
|
TRANSPORT_HEADERS: Final = frozenset(
|
|
{"host", "accept", "accept-encoding", "connection", "content-length", "user-agent"}
|
|
)
|
|
|
|
|
|
def expected_response_for_deployment(
|
|
expected: Mapping[str, JsonValue], model_id: str | None
|
|
) -> Mapping[str, JsonValue]:
|
|
identifier: Final = expected.get("id")
|
|
if not isinstance(identifier, str) or not ResponsesAPIRequestUtils._is_litellm_encoded_response_id(identifier):
|
|
return expected
|
|
assert model_id, "An encoded response ID must identify the selected deployment"
|
|
provider, _, response = base64.b64decode(identifier.removeprefix("resp_")).decode().split(";", 2)
|
|
updated: Final = f"{provider};model_id:{model_id};{response}"
|
|
return {
|
|
**expected,
|
|
"id": "resp_" + base64.b64encode(updated.encode()).decode(),
|
|
}
|
|
|
|
|
|
def assert_translation(case: TranslationTestCase, gateway: Gateway, provider: SharedProvider) -> None:
|
|
provider.expect(Reply(status=case.mock_provider_status_code, body=json.dumps(case.mock_provider_response).encode()))
|
|
response: Final = gateway.request("POST", case.litellm_endpoint, case.litellm_request)
|
|
received: Final = provider.received()
|
|
assert [(request.method, request.target) for request in received] == [("POST", case.expected_provider_endpoint)]
|
|
sent: Final = received[0]
|
|
assert {
|
|
name: value
|
|
for name, value in sent.headers.items()
|
|
if name not in TRANSPORT_HEADERS and not name.startswith("x-stainless-")
|
|
} == dict(case.expected_provider_headers)
|
|
assert json.loads(sent.body) == case.expected_provider_request
|
|
assert response.status_code == case.expected_litellm_status_code, response.text
|
|
model_id: Final = response.headers.get("x-litellm-model-id")
|
|
expected: Final = expected_response_for_deployment(case.expected_litellm_response, model_id)
|
|
if expected is not case.expected_litellm_response:
|
|
deployments: Final = gateway.get("/model/info")["data"]
|
|
assert isinstance(deployments, list), deployments
|
|
assert any(
|
|
isinstance(deployment, dict)
|
|
and deployment.get("model_name") == case.litellm_request["model"]
|
|
and isinstance(deployment.get("model_info"), dict)
|
|
and deployment["model_info"].get("id") == model_id
|
|
for deployment in deployments
|
|
), "The encoded response ID must match the configured deployment"
|
|
assert response.json() == expected
|