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>
51 lines
2.1 KiB
Python
51 lines
2.1 KiB
Python
from dataclasses import dataclass
|
|
from typing import Final
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from tests.integration._support.client import Gateway
|
|
from tests.integration._support.workers import readiness_failure
|
|
|
|
HEALTHY: Final = {
|
|
"status": "healthy",
|
|
"cache_type": "redis",
|
|
"ping_response": True,
|
|
"set_cache_response": "success",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Subscribers:
|
|
count: int
|
|
|
|
def pubsub_numsub(self, *channels: str) -> tuple[tuple[str, int], ...]:
|
|
return tuple((channel, self.count) for channel in channels)
|
|
|
|
|
|
@pytest.mark.parametrize("subscribers", (0, 1, 2))
|
|
def test_readiness_requires_a_subscriber_for_each_proxy(subscribers: int) -> None:
|
|
def respond(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/cache/ping":
|
|
assert request.headers["authorization"] == "Bearer owned-key"
|
|
return httpx.Response(200, json=HEALTHY)
|
|
return httpx.Response(200)
|
|
|
|
with httpx.Client(base_url="http://127.0.0.1:4000", transport=httpx.MockTransport(respond)) as client:
|
|
primary: Final = Gateway(client, "owned-key", "http://127.0.0.1:8190")
|
|
failure: Final = readiness_failure(client, primary.upstream_url, (primary, primary), Subscribers(subscribers))
|
|
assert failure == (None if subscribers == 2 else f"auth-cache subscribers {subscribers}/2")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
(("status", "unhealthy"), ("cache_type", "local"), ("ping_response", False), ("set_cache_response", "failure")),
|
|
)
|
|
def test_readiness_rejects_a_proxy_without_working_shared_redis(field: str, value: str | bool) -> None:
|
|
def respond(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json={**HEALTHY, field: value})
|
|
|
|
with httpx.Client(base_url="http://127.0.0.1:4000", transport=httpx.MockTransport(respond)) as client:
|
|
primary: Final = Gateway(client, "owned-key", "http://127.0.0.1:8190")
|
|
failure: Final = readiness_failure(client, primary.upstream_url, (primary,), Subscribers(1))
|
|
assert failure == "proxy 0 Redis read/write health"
|