mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
4b3355bdc6
* test(e2e): prove the virtual key lifecycle on every replica Walks one virtual key through create, read, partial update, clear, enforce and delete against a live proxy and database, reading every write back on every gateway replica. The management suite already had single write-then-read tests for keys, but none of them proved that a partial /key/update leaves the untouched fields alone, that an explicit null clears a field, or that a write is visible on more than the one gateway that took it. Adds read_back_everywhere to the shared ProxyClient: it polls a GET path on every URL in PROXY_REPLICA_URLS until each replica's parsed body satisfies the caller's predicate, and fails naming the replica that never converged. The CLEAR sentinel in the e2e models makes an explicit JSON null expressible in a body the transport otherwise strips of None fields. Documents /key/update's merge patch semantics on the endpoint docstring. * test(e2e): prove key revocation and field preservation on every replica Applies the findings from an adversarial review of the first commit. The delete step only checked that chat was refused on the gateway that took the write, so it would have passed while a sibling gateway kept serving the deleted key. It now serves one call from every replica first, so each has the key cached and the delete has something to revoke everywhere, then polls every replica for the refusal. The file also carried its own poll loop that tested the deadline before attempting, so it gave up one attempt early and skipped the attempt landing exactly on the deadline. It now shares the harness helper, which is generic over the polled value rather than over a parsed body, so the same loop covers both the info read-back and the chat refusal. The model the enforcement step registers now carries a unique marker in its alias, matching every other deployment this suite creates, so concurrent runs never share one model group. The docstring sentence claimed an explicit null clears any field. It does not: the metadata-backed fields merge into stored metadata, where a null is a silent no-op, and only the key's own columns clear. Regenerating the dashboard types picks up the corrected text. * fix(e2e): delete a deployment that never becomes servable Registering a model posts /model/new and then waits for every replica to list it. When that wait timed out the deployment already existed in the database but its id had never been returned, so no caller could delete it and the row outlived the run. It is now deleted before the failure propagates. Found by review on the key lifecycle suite, whose module fixture registers a deployment this way, but every caller of the shared helper had the same exposure. * docs(e2e): drop the duplicated notes from the lifecycle docstrings The delete method restated what the warm-up helper already explains, and the module restated the merge patch rule that the endpoint and the request model both document.
189 lines
6.7 KiB
Python
189 lines
6.7 KiB
Python
"""Harness coverage for the barriers that gate on every replica.
|
|
|
|
No proxy needed and no ``e2e`` marker: this pins that a model registered through
|
|
the control plane only counts as servable once every configured replica lists it
|
|
on /v1/models, and that a management write only counts as read back once every
|
|
replica's read satisfies the caller's predicate, which is what keeps a two-gateway
|
|
stack from handing a test a model or a key that one gateway has not caught up on
|
|
yet. The fakes are plain pollers standing in for each replica's transport plus an
|
|
injected clock, so nothing here monkeypatches anything.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass
|
|
from itertools import chain, repeat
|
|
from types import MappingProxyType
|
|
from typing import Final
|
|
|
|
import pytest
|
|
|
|
from e2e_config import parse_replica_urls
|
|
from e2e_http import Result, Success
|
|
from models import KeyInfo, KeyInfoResponse, ModelListEntry, ModelsListResponse
|
|
from proxy_client import (
|
|
Poller,
|
|
ConvergeOutcome,
|
|
Converged,
|
|
ModelsPoller,
|
|
NotConverged,
|
|
NotServableOn,
|
|
Servable,
|
|
await_converged_everywhere,
|
|
await_servable_everywhere,
|
|
first_lagging_replica,
|
|
converge_timeout_message,
|
|
)
|
|
|
|
MODEL: Final = "gpt-under-test"
|
|
TIMEOUT: Final = 10.0
|
|
INTERVAL: Final = 2.0
|
|
RPM_BEFORE_UPDATE: Final = 100
|
|
RPM_AFTER_UPDATE: Final = 200
|
|
|
|
|
|
@dataclass
|
|
class FakeClock:
|
|
elapsed: float = 0.0
|
|
|
|
def now(self) -> float:
|
|
return self.elapsed
|
|
|
|
def sleep(self, seconds: float) -> None:
|
|
self.elapsed += seconds
|
|
|
|
|
|
def _listing(*model_ids: str) -> Success[ModelsListResponse]:
|
|
entries: Final = tuple(ModelListEntry(id=model_id) for model_id in model_ids)
|
|
return Success(status_code=200, data=ModelsListResponse(data=entries))
|
|
|
|
|
|
def _poller(results: Iterable[Success[ModelsListResponse]]) -> ModelsPoller:
|
|
it: Final = iter(results)
|
|
return lambda _timeout: next(it)
|
|
|
|
|
|
def _await(pollers: Mapping[str, ModelsPoller]) -> Servable | NotServableOn:
|
|
clock: Final = FakeClock()
|
|
return await_servable_everywhere(
|
|
pollers,
|
|
model_name=MODEL,
|
|
timeout=TIMEOUT,
|
|
interval=INTERVAL,
|
|
request_timeout=5.0,
|
|
db_sync_seconds=0.0,
|
|
now=clock.now,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
|
|
class TestAwaitServableEverywhere:
|
|
@pytest.mark.parametrize("missing", ["gateway-1", "gateway-2"])
|
|
def test_fails_on_the_replica_that_never_lists_the_model(self, missing: str) -> None:
|
|
pollers: Final = {
|
|
"gateway-1": _poller(repeat(_listing(MODEL))),
|
|
"gateway-2": _poller(repeat(_listing(MODEL))),
|
|
} | {missing: _poller(repeat(_listing()))}
|
|
assert _await(pollers) == NotServableOn(replica=missing, last_result=_listing())
|
|
|
|
def test_passes_once_every_replica_lists_the_model(self) -> None:
|
|
pollers: Final = {
|
|
"gateway-1": _poller(repeat(_listing(MODEL))),
|
|
"gateway-2": _poller(chain(repeat(_listing(), 2), repeat(_listing(MODEL)))),
|
|
}
|
|
assert _await(pollers) == Servable()
|
|
|
|
|
|
def _key_info(rpm_limit: int) -> Success[KeyInfoResponse]:
|
|
return Success(status_code=200, data=KeyInfoResponse(info=KeyInfo(rpm_limit=rpm_limit)))
|
|
|
|
|
|
def _reads(results: Iterable[Result[KeyInfoResponse]]) -> Poller[Result[KeyInfoResponse]]:
|
|
it: Final = iter(results)
|
|
return lambda: next(it)
|
|
|
|
|
|
def _updated(result: Result[KeyInfoResponse]) -> bool:
|
|
return isinstance(result, Success) and result.data.info.rpm_limit == RPM_AFTER_UPDATE
|
|
|
|
|
|
def _converge(
|
|
pollers: Mapping[str, Poller[Result[KeyInfoResponse]]], clock: FakeClock
|
|
) -> Mapping[str, ConvergeOutcome[Result[KeyInfoResponse]]]:
|
|
return await_converged_everywhere(
|
|
pollers,
|
|
converged=_updated,
|
|
timeout=TIMEOUT,
|
|
interval=INTERVAL,
|
|
now=clock.now,
|
|
sleep=clock.sleep,
|
|
)
|
|
|
|
|
|
class TestAwaitConvergedEverywhere:
|
|
def test_waits_for_the_replica_that_lags_behind_the_write(self) -> None:
|
|
clock: Final = FakeClock()
|
|
pollers: Final = MappingProxyType(
|
|
{
|
|
"gateway-1": _reads(repeat(_key_info(RPM_AFTER_UPDATE))),
|
|
"gateway-2": _reads(
|
|
chain(repeat(_key_info(RPM_BEFORE_UPDATE), 2), repeat(_key_info(RPM_AFTER_UPDATE)))
|
|
),
|
|
}
|
|
)
|
|
outcomes: Final = _converge(pollers, clock)
|
|
assert outcomes == {
|
|
"gateway-1": Converged(result=_key_info(RPM_AFTER_UPDATE)),
|
|
"gateway-2": Converged(result=_key_info(RPM_AFTER_UPDATE)),
|
|
}
|
|
assert first_lagging_replica(outcomes) is None
|
|
assert clock.elapsed == 2 * INTERVAL
|
|
|
|
def test_names_the_replica_that_never_converges_with_its_last_read(self) -> None:
|
|
clock: Final = FakeClock()
|
|
pollers: Final = MappingProxyType(
|
|
{
|
|
"gateway-1": _reads(repeat(_key_info(RPM_AFTER_UPDATE))),
|
|
"gateway-2": _reads(repeat(_key_info(RPM_BEFORE_UPDATE))),
|
|
}
|
|
)
|
|
outcomes: Final = _converge(pollers, clock)
|
|
assert first_lagging_replica(outcomes) == (
|
|
"gateway-2",
|
|
NotConverged(last_result=_key_info(RPM_BEFORE_UPDATE)),
|
|
)
|
|
assert clock.elapsed == TIMEOUT
|
|
message: Final = converge_timeout_message(
|
|
what="GET /key/info",
|
|
replica="gateway-2",
|
|
timeout=TIMEOUT,
|
|
last_result=_key_info(RPM_BEFORE_UPDATE),
|
|
)
|
|
assert "gateway-2" in message and "/key/info" in message and str(RPM_BEFORE_UPDATE) in message
|
|
|
|
def test_each_replica_gets_its_own_full_budget(self) -> None:
|
|
"""A replica that converges late must not eat into the next replica's budget: both
|
|
need most of the timeout here, so one shared deadline would starve the second."""
|
|
clock: Final = FakeClock()
|
|
slow: Final = chain(repeat(_key_info(RPM_BEFORE_UPDATE), 3), repeat(_key_info(RPM_AFTER_UPDATE)))
|
|
pollers: Final = MappingProxyType(
|
|
{
|
|
"gateway-1": _reads(slow),
|
|
"gateway-2": _reads(
|
|
chain(repeat(_key_info(RPM_BEFORE_UPDATE), 3), repeat(_key_info(RPM_AFTER_UPDATE)))
|
|
),
|
|
}
|
|
)
|
|
outcomes: Final = _converge(pollers, clock)
|
|
assert first_lagging_replica(outcomes) is None
|
|
assert clock.elapsed == 2 * 3 * INTERVAL
|
|
|
|
|
|
class TestParseReplicaUrls:
|
|
def test_splits_and_trims_the_gateway_addresses(self) -> None:
|
|
raw: Final = " http://127.0.0.1:4010/, http://127.0.0.1:4011 "
|
|
assert parse_replica_urls(raw, "http://lb") == ("http://127.0.0.1:4010", "http://127.0.0.1:4011")
|
|
|
|
def test_falls_back_to_the_data_plane_address_when_unset(self) -> None:
|
|
assert parse_replica_urls("", "http://lb") == ("http://lb",)
|