mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-10 03:28:53 +00:00
refactor(e2e): keep the concurrency test self-contained with local tuning values
This commit is contained in:
parent
23dffc7453
commit
ff3ef9d7e8
1 changed files with 37 additions and 53 deletions
|
|
@ -18,13 +18,12 @@ is accepted on LLM routes but 401s here).
|
|||
The concurrency test's observable is the stub's own per-marker in-flight
|
||||
counter, read back through the proxy via the stub's `stats` tool. The
|
||||
proxy-side semaphore queues excess tool calls instead of rejecting them, so
|
||||
with a cap of MAX_CONCURRENT a burst of BURST simultaneous slow_echo calls
|
||||
must all succeed while the stub never sees more than MAX_CONCURRENT of them
|
||||
overlap. The uncapped control server drives the identical burst through the
|
||||
identical machinery and must see all BURST calls overlap; that is the
|
||||
fail-before-fix evidence built into the test, since a broken (or skipped)
|
||||
semaphore makes the capped server record BURST in-flight exactly like the
|
||||
control and the equality assertion fail.
|
||||
under a cap every call in a simultaneous burst must still succeed while the
|
||||
stub never sees more than the cap overlap. The uncapped control server drives
|
||||
the identical burst through the identical machinery and must see the whole
|
||||
burst overlap; that is the fail-before-fix evidence built into the test,
|
||||
since a broken (or skipped) semaphore makes the capped server record the full
|
||||
burst exactly like the control and the equality assertion fail.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -35,15 +34,12 @@ import pytest
|
|||
|
||||
from e2e_config import MCP_STUB_URL, unique_marker
|
||||
from lifecycle import ResourceManager
|
||||
from mcp_client import McpAuth, McpClient, McpDenied
|
||||
from mcp_client import McpAuth, McpClient, McpDenied, McpToolText
|
||||
from models import McpServerCreateBody
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
STUB_TOOLS = ("echo", "slow_echo", "stats")
|
||||
MAX_CONCURRENT = 2
|
||||
BURST = 6
|
||||
SLOW_CALL_SECONDS = 2.0
|
||||
|
||||
|
||||
class TestMcpToolAccess:
|
||||
|
|
@ -111,12 +107,15 @@ class TestMcpServerMaxConcurrency:
|
|||
def test_max_concurrent_requests_caps_in_flight_upstream_calls(
|
||||
self, client: McpClient, resources: ResourceManager, scoped_key: str
|
||||
) -> None:
|
||||
"""Two servers against the same stub: one capped at MAX_CONCURRENT, one
|
||||
uncapped control. The tools/list polls are the settle step; they hold
|
||||
until the just-created records are servable on the gateway (and the
|
||||
fresh key has cleared the auth cache) so the bursts measure the
|
||||
semaphore, not propagation. Each burst is BURST simultaneous slow_echo
|
||||
calls, one per thread, each over its own MCP session."""
|
||||
"""Two servers against the same stub: one capped at 2 concurrent
|
||||
requests, one uncapped control. The tools/list polls are the settle
|
||||
step; they hold until the just-created records are servable on the
|
||||
gateway (and the fresh key has cleared the auth cache) so the bursts
|
||||
measure the semaphore, not propagation. Each burst is 6 simultaneous
|
||||
slow_echo calls, one per thread, each over its own MCP session."""
|
||||
max_concurrent = 2
|
||||
burst_size = 6
|
||||
slow_call_seconds = 2.0
|
||||
auth = McpAuth(header_name="x-litellm-api-key", key=scoped_key)
|
||||
|
||||
capped_alias = f"e2emcpcap{unique_marker()}"
|
||||
|
|
@ -125,7 +124,7 @@ class TestMcpServerMaxConcurrency:
|
|||
alias=capped_alias,
|
||||
url=MCP_STUB_URL,
|
||||
allow_all_keys=True,
|
||||
max_concurrent_requests=MAX_CONCURRENT,
|
||||
max_concurrent_requests=max_concurrent,
|
||||
)
|
||||
)
|
||||
resources.defer(lambda: client.delete_server(capped.server_id))
|
||||
|
|
@ -134,57 +133,42 @@ class TestMcpServerMaxConcurrency:
|
|||
control = client.create_server(McpServerCreateBody(alias=control_alias, url=MCP_STUB_URL, allow_all_keys=True))
|
||||
resources.defer(lambda: client.delete_server(control.server_id))
|
||||
|
||||
assert client.server_info(capped.server_id).max_concurrent_requests == MAX_CONCURRENT
|
||||
assert client.server_info(capped.server_id).max_concurrent_requests == max_concurrent
|
||||
assert client.server_info(control.server_id).max_concurrent_requests is None
|
||||
|
||||
_ = client.poll_tool_names(capped_alias, auth)
|
||||
_ = client.poll_tool_names(control_alias, auth)
|
||||
|
||||
capped_marker = unique_marker()
|
||||
with ThreadPoolExecutor(max_workers=BURST) as pool:
|
||||
futures = [
|
||||
pool.submit(
|
||||
client.call_tool,
|
||||
capped_alias,
|
||||
auth,
|
||||
f"{capped_alias}-slow_echo",
|
||||
{"text": "capped", "marker": capped_marker, "sleep_seconds": SLOW_CALL_SECONDS},
|
||||
)
|
||||
for _ in range(BURST)
|
||||
]
|
||||
capped_results = [future.result() for future in futures]
|
||||
def burst_of_slow_calls(alias: str, text: str, marker: str) -> list[McpToolText]:
|
||||
arguments = {"text": text, "marker": marker, "sleep_seconds": slow_call_seconds}
|
||||
with ThreadPoolExecutor(max_workers=burst_size) as pool:
|
||||
futures = [
|
||||
pool.submit(client.call_tool, alias, auth, f"{alias}-slow_echo", arguments)
|
||||
for _ in range(burst_size)
|
||||
]
|
||||
return [future.result() for future in futures]
|
||||
|
||||
assert len(capped_results) == BURST
|
||||
capped_marker = unique_marker()
|
||||
capped_results = burst_of_slow_calls(capped_alias, "capped", capped_marker)
|
||||
assert len(capped_results) == burst_size
|
||||
assert all(result.is_error is False and result.text == "capped" for result in capped_results), (
|
||||
f"queued calls must all still succeed under the cap: {capped_results}"
|
||||
)
|
||||
|
||||
capped_stats = client.stub_stats(capped_alias, auth, f"{capped_alias}-stats", capped_marker)
|
||||
assert capped_stats.completed == BURST
|
||||
assert capped_stats.max_in_flight == MAX_CONCURRENT, (
|
||||
f"stub saw {capped_stats.max_in_flight} overlapping calls; the cap of {MAX_CONCURRENT} "
|
||||
f"must be the exact ceiling AND be reached ({BURST} queued calls saturate it)"
|
||||
assert capped_stats.completed == burst_size
|
||||
assert capped_stats.max_in_flight == max_concurrent, (
|
||||
f"stub saw {capped_stats.max_in_flight} overlapping calls; the cap of {max_concurrent} "
|
||||
f"must be the exact ceiling AND be reached ({burst_size} queued calls saturate it)"
|
||||
)
|
||||
|
||||
control_marker = unique_marker()
|
||||
with ThreadPoolExecutor(max_workers=BURST) as pool:
|
||||
futures = [
|
||||
pool.submit(
|
||||
client.call_tool,
|
||||
control_alias,
|
||||
auth,
|
||||
f"{control_alias}-slow_echo",
|
||||
{"text": "control", "marker": control_marker, "sleep_seconds": SLOW_CALL_SECONDS},
|
||||
)
|
||||
for _ in range(BURST)
|
||||
]
|
||||
control_results = [future.result() for future in futures]
|
||||
|
||||
control_results = burst_of_slow_calls(control_alias, "control", control_marker)
|
||||
assert all(result.is_error is False and result.text == "control" for result in control_results)
|
||||
|
||||
control_stats = client.stub_stats(control_alias, auth, f"{control_alias}-stats", control_marker)
|
||||
assert control_stats.completed == BURST
|
||||
assert control_stats.max_in_flight == BURST, (
|
||||
f"uncapped control saw {control_stats.max_in_flight} overlapping calls, expected all {BURST}; "
|
||||
assert control_stats.completed == burst_size
|
||||
assert control_stats.max_in_flight == burst_size, (
|
||||
f"uncapped control saw {control_stats.max_in_flight} overlapping calls, expected all {burst_size}; "
|
||||
f"if this fails the instrument cannot detect over-cap concurrency and the capped assertion is vacuous"
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue