From ff3ef9d7e869c845f7f2d1e60fec223446d0710e Mon Sep 17 00:00:00 2001 From: Tin Chi Lo Date: Mon, 13 Jul 2026 12:35:50 -0700 Subject: [PATCH] refactor(e2e): keep the concurrency test self-contained with local tuning values --- tests/e2e/mcp/test_mcp_gateway_e2e.py | 90 +++++++++++---------------- 1 file changed, 37 insertions(+), 53 deletions(-) diff --git a/tests/e2e/mcp/test_mcp_gateway_e2e.py b/tests/e2e/mcp/test_mcp_gateway_e2e.py index 196d6462373..a45295cd2f8 100644 --- a/tests/e2e/mcp/test_mcp_gateway_e2e.py +++ b/tests/e2e/mcp/test_mcp_gateway_e2e.py @@ -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" )