scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 85

test(mcp): assert the aggregate connect passes unchallenged instead of counting token probes

Browse source 
The unselected aggregate connect test asserted only on a patched has_user_oauth_token
call count, which the test-quality gate flags as mock-echo (TQ002). It now calls the
preflight unpatched and asserts it returns without a sign-in challenge, so the mutant
that probes every registered server is still killed by the 401 it would raise.
Also applies ruff format to test_caller_sign_in.py, a file new in this PR

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-10-02 22:04:18 +00:00
parent ae7b0fb8ea
commit 5a7ec362a0
2 changed files with 10 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
tests/unit/proxy/_experimental/mcp_server/test_caller_sign_in.py
View file

@ -105,9 +105,7 @@ def test_provider_returning_none_contributes_nothing(registered):
def test_obo_server_contributes_jwt_issuers_and_own_scopes(monkeypatch):
monkeypatch.setenv("JWT_ISSUER", "https://jwt-idp.test")
sign_in: Final = caller_sign_in_for(
_server(auth_type=MCPAuth.oauth2_token_exchange, scopes=["read"]), None
)
sign_in: Final = caller_sign_in_for(_server(auth_type=MCPAuth.oauth2_token_exchange, scopes=["read"]), None)
assert sign_in == CallerSignIn(issuers=("https://jwt-idp.test",), scopes=("read",))

19
tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py
View file

@ -10458,17 +10458,16 @@ class TestPreemptive401ModeAware:
server = _make_oauth2_server(alias, oauth2_flow="authorization_code", delegate_auth_to_upstream=delegate)
manager.registry[server.server_id] = server
with patch.object(manager, "has_user_oauth_token", new_callable=AsyncMock, return_value=False) as tokens:
await server_module._raise_preemptive_401_for_unauthenticated_servers(
scope={"type": "http", "method": "POST", "path": "/mcp", "headers": [(b"host", b"testserver")]},
mcp_servers=None,
oauth2_headers=None,
mcp_server_auth_headers=None,
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-virtual-key"),
client_ip=None,
)
outcome = await server_module._raise_preemptive_401_for_unauthenticated_servers(
scope={"type": "http", "method": "POST", "path": "/mcp", "headers": [(b"host", b"testserver")]},
mcp_servers=None,
oauth2_headers=None,
mcp_server_auth_headers=None,
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-virtual-key"),
client_ip=None,
)
assert tokens.await_count == 0, "an unselected aggregate connect must not probe any server for a token"
assert outcome is None, "an unselected aggregate connect must pass without a sign-in challenge"
@pytest.mark.asyncio
async def test_deferred_discovery_runs_before_delegate_challenge(self):