Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: require api auth in server factory

Browse Source
This commit is contained in:
hamb1y
2026-07-08 12:16:12 +08:00
committed by Xubin Ren
parent 6c59332a8a
commit e86133c434
4 changed files with 90 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_openai_api.py
+47 -13
View File
@@ -27,6 +27,9 @@ except ImportError:
pytest_plugins = ("pytest_asyncio",)
API_KEY = "secret"
AUTH_HEADERS = {"Authorization": f"Bearer {API_KEY}"}
def _make_mock_agent(response_text: str = "mock response") -> MagicMock:
agent = MagicMock()
@@ -44,7 +47,7 @@ def mock_agent():
@pytest.fixture
def app(mock_agent):
return create_app(mock_agent, model_name="test-model", request_timeout=10.0)
return create_app(mock_agent, model_name="test-model", request_timeout=10.0, api_key=API_KEY)
@pytest_asyncio.fixture
@@ -104,20 +107,20 @@ def test_chat_completion_response_preserves_provider_total_usage() -> None:
@pytest.mark.asyncio
async def test_missing_messages_returns_400(aiohttp_client, app) -> None:
client = await aiohttp_client(app)
resp = await client.post("/v1/chat/completions", json={"model": "test"})
resp = await client.post("/v1/chat/completions", headers=AUTH_HEADERS, json={"model": "test"})
assert resp.status == 400
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_api_key_protects_api_routes_but_not_health(aiohttp_client, mock_agent) -> None:
app = create_app(mock_agent, model_name="test-model", api_key="secret")
app = create_app(mock_agent, model_name="test-model", api_key=API_KEY)
client = await aiohttp_client(app)
health = await client.get("/health")
missing = await client.get("/v1/models")
wrong = await client.get("/v1/models", headers={"Authorization": "Bearer wrong"})
ok = await client.get("/v1/models", headers={"Authorization": "Bearer secret"})
ok = await client.get("/v1/models", headers=AUTH_HEADERS)
assert health.status == 200
assert missing.status == 401
@@ -127,12 +130,33 @@ async def test_api_key_protects_api_routes_but_not_health(aiohttp_client, mock_a
assert (await wrong.json())["error"]["message"] == "Invalid API key"
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_api_routes_fail_closed_without_configured_api_key(aiohttp_client, mock_agent) -> None:
app = create_app(mock_agent, model_name="test-model")
client = await aiohttp_client(app)
health = await client.get("/health")
models = await client.get("/v1/models")
chat = await client.post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "hello"}]},
)
assert health.status == 200
assert models.status == 401
assert chat.status == 401
assert (await models.json())["error"]["message"] == "API key is not configured"
mock_agent.process_direct.assert_not_called()
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_no_user_message_returns_400(aiohttp_client, app) -> None:
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "system", "content": "you are a bot"}]},
)
assert resp.status == 400
@@ -144,6 +168,7 @@ async def test_stream_true_returns_sse(aiohttp_client, app) -> None:
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "hello"}], "stream": True},
)
assert resp.status == 200
@@ -220,10 +245,11 @@ async def test_single_user_message_must_have_user_role() -> None:
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_successful_request_uses_fixed_api_session(aiohttp_client, mock_agent) -> None:
app = create_app(mock_agent, model_name="test-model")
app = create_app(mock_agent, model_name="test-model", api_key=API_KEY)
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "hello"}]},
)
assert resp.status == 200
@@ -254,15 +280,17 @@ async def test_followup_requests_share_same_session_key(aiohttp_client) -> None:
agent.close_mcp = AsyncMock()
agent._last_usage = {}
app = create_app(agent, model_name="m")
app = create_app(agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
r1 = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "first"}]},
)
r2 = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "second"}]},
)
@@ -292,12 +320,13 @@ async def test_fixed_session_requests_are_serialized(aiohttp_client) -> None:
agent.close_mcp = AsyncMock()
agent._last_usage = {}
app = create_app(agent, model_name="m")
app = create_app(agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
async def send(msg: str):
return await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": msg}]},
)
@@ -318,7 +347,7 @@ async def test_fixed_session_requests_are_serialized(aiohttp_client) -> None:
@pytest.mark.asyncio
async def test_models_endpoint(aiohttp_client, app) -> None:
client = await aiohttp_client(app)
resp = await client.get("/v1/models")
resp = await client.get("/v1/models", headers=AUTH_HEADERS)
assert resp.status == 200
body = await resp.json()
assert body["object"] == "list"
@@ -338,10 +367,11 @@ async def test_health_endpoint(aiohttp_client, app) -> None:
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multimodal_content_extracts_text(aiohttp_client, mock_agent) -> None:
app = create_app(mock_agent, model_name="m")
app = create_app(mock_agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={
"messages": [
{
@@ -366,10 +396,11 @@ async def test_multimodal_content_extracts_text(aiohttp_client, mock_agent) -> N
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
@pytest.mark.asyncio
async def test_multimodal_remote_image_url_returns_400(aiohttp_client, mock_agent) -> None:
app = create_app(mock_agent, model_name="m")
app = create_app(mock_agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={
"messages": [
{
@@ -407,10 +438,11 @@ async def test_empty_response_retry_then_success(aiohttp_client) -> None:
agent.close_mcp = AsyncMock()
agent._last_usage = {}
app = create_app(agent, model_name="m")
app = create_app(agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "hello"}]},
)
assert resp.status == 200
@@ -434,10 +466,11 @@ async def test_empty_response_retry_does_not_duplicate_user_turn(aiohttp_client)
agent.close_mcp = AsyncMock()
agent._last_usage = {}
app = create_app(agent, model_name="m")
app = create_app(agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "hello"}]},
)
assert resp.status == 200
@@ -463,10 +496,11 @@ async def test_empty_response_falls_back(aiohttp_client) -> None:
agent.close_mcp = AsyncMock()
agent._last_usage = {}
app = create_app(agent, model_name="m")
app = create_app(agent, model_name="m", api_key=API_KEY)
client = await aiohttp_client(app)
resp = await client.post(
"/v1/chat/completions",
headers=AUTH_HEADERS,
json={"messages": [{"role": "user", "content": "hello"}]},
)
assert resp.status == 200