ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: const

Browse Source
This commit is contained in:
Yuta Saito 2026-01-19 12:03:26 +09:00
parent 20b6468222
commit 30c4a38179
2 changed files with 7 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/mcp_tests/mcp_server.py
View File

@ -1,6 +1,4 @@
# math_server.py
from __future__ import annotations
import argparse
import os

19
tests/mcp_tests/test_proxy_mcp_e2e.py
View File

@ -27,10 +27,7 @@ PROJECT_ROOT = Path(__file__).resolve().parents[2]
PROXY_START_TIMEOUT = 30
@pytest.fixture(scope="session")
def proxy_authorization_header() -> str:
"""Shared Authorization header value for proxy calls."""
return "Bearer sk-1234"
PROXY_AUTHORIZATION_HEADER = "Bearer sk-1234"
@pytest.fixture(scope="session", autouse=True)
@ -153,14 +150,12 @@ def proxy_server_url(
class TestProxyMcpSimpleConnections:
@pytest.mark.asyncio
async def test_proxy_mcp_stdio_roundtrip(
self, proxy_server_url: str, proxy_authorization_header: str
) -> None:
async def test_proxy_mcp_stdio_roundtrip(self, proxy_server_url: str) -> None:
async with asyncio.timeout(20):
async with streamablehttp_client(
url=f"{proxy_server_url}/mcp",
headers={
"Authorization": proxy_authorization_header,
"Authorization": PROXY_AUTHORIZATION_HEADER,
"x-mcp-servers": "math_stdio",
},
) as (read, write, _get_session_id):
@ -179,13 +174,13 @@ class TestProxyMcpSimpleConnections:
@pytest.mark.asyncio
async def test_proxy_mcp_streamable_http_roundtrip(
self, proxy_server_url: str, proxy_authorization_header: str
self, proxy_server_url: str
) -> None:
async with asyncio.timeout(20):
async with streamablehttp_client(
url=f"{proxy_server_url}/mcp",
headers={
"Authorization": proxy_authorization_header,
"Authorization": PROXY_AUTHORIZATION_HEADER,
"x-mcp-servers": "math_streamable_http",
},
) as (read, write, _get_session_id):
@ -204,12 +199,12 @@ class TestProxyMcpSimpleConnections:
@pytest.mark.asyncio
async def test_proxy_mcp_lists_all_servers_without_header(
self, proxy_server_url: str, proxy_authorization_header: str
self, proxy_server_url: str
) -> None:
async with asyncio.timeout(20):
async with streamablehttp_client(
url=f"{proxy_server_url}/mcp",
headers={"Authorization": proxy_authorization_header},
headers={"Authorization": PROXY_AUTHORIZATION_HEADER},
) as (read, write, _get_session_id):
async with ClientSession(read, write) as session:
await session.initialize()