73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""mcp_server 测试：参数解析 + 非回环强制 token + ASGI 鉴权中间件。"""
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from ppclock.mcp_server import (TOKEN_ENV, TokenAuthMiddleware,
|
|
parse_args)
|
|
|
|
|
|
class TestParseArgs:
|
|
def test_defaults(self):
|
|
a = parse_args([])
|
|
assert a.transport == "stdio"
|
|
assert a.host == "127.0.0.1"
|
|
assert a.port == 8972
|
|
assert a.mac is None
|
|
assert a.connect_timeout == 90.0
|
|
assert a.idle_timeout == 300.0
|
|
|
|
def test_http_flags(self):
|
|
a = parse_args(["--transport", "http", "--host", "0.0.0.0",
|
|
"--port", "9000", "--mac", "AA:BB:CC:DD:EE:FF",
|
|
"--connect-timeout", "120", "--idle-timeout", "60"])
|
|
assert a.transport == "http" and a.host == "0.0.0.0"
|
|
assert a.port == 9000 and a.mac == "AA:BB:CC:DD:EE:FF"
|
|
assert a.connect_timeout == 120.0 and a.idle_timeout == 60.0
|
|
|
|
|
|
def run_middleware(token_set, auth_header):
|
|
"""驱动 TokenAuthMiddleware，返回 (status, app_called)。"""
|
|
captured = {}
|
|
|
|
async def app(scope, receive, send):
|
|
captured["called"] = True
|
|
await send({"type": "http.response.start", "status": 200, "headers": []})
|
|
await send({"type": "http.response.body", "body": b"ok"})
|
|
|
|
sent = []
|
|
|
|
async def send(msg):
|
|
sent.append(msg)
|
|
|
|
headers = []
|
|
if auth_header is not None:
|
|
headers.append((b"authorization", auth_header.encode()))
|
|
scope = {"type": "http", "headers": headers}
|
|
mw = TokenAuthMiddleware(app, token_set)
|
|
asyncio.run(mw(scope, None, send))
|
|
status = next(m["status"] for m in sent if m["type"] == "http.response.start")
|
|
return status, captured.get("called", False)
|
|
|
|
|
|
class TestTokenAuth:
|
|
def test_no_header_401(self):
|
|
status, called = run_middleware("secret", None)
|
|
assert status == 401 and called is False
|
|
|
|
def test_wrong_token_401(self):
|
|
status, called = run_middleware("secret", "Bearer nope")
|
|
assert status == 401 and called is False
|
|
|
|
def test_right_token_passes(self):
|
|
status, called = run_middleware("secret", "Bearer secret")
|
|
assert status == 200 and called is True
|
|
|
|
|
|
class TestBindGuard:
|
|
def test_non_loopback_requires_token(self, monkeypatch):
|
|
from ppclock.mcp_server import main
|
|
monkeypatch.delenv(TOKEN_ENV, raising=False)
|
|
with pytest.raises(SystemExit):
|
|
main(["--transport", "http", "--host", "0.0.0.0"])
|