test(V2): migrazione test common/
Copiati e aggiornati i test da services/common/tests/ a tests/unit/common/. Import aggiornati da mcp_common a cerbero_mcp.common. Eliminati test di funzionalità V1-only (app_factory, environment, auth/Principal, server_base). Refactored test_audit.py (principal→actor str) e test_mcp_bridge.py (TokenStore→valid_tokens set). 71/71 test passano. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,112 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from cerbero_mcp.common.mcp_bridge import _derive_input_schemas, mount_mcp_endpoint
|
||||
from pydantic import BaseModel
|
||||
|
||||
VALID_TOKEN = "t"
|
||||
VALID_TOKENS: set[str] = {VALID_TOKEN}
|
||||
|
||||
|
||||
class EchoBody(BaseModel):
|
||||
msg: str
|
||||
n: int = 1
|
||||
|
||||
|
||||
def _make_app() -> FastAPI:
|
||||
app = FastAPI()
|
||||
|
||||
@app.post("/tools/echo")
|
||||
def echo(body: EchoBody):
|
||||
return {"echo": body.msg, "n": body.n}
|
||||
|
||||
@app.post("/tools/ping")
|
||||
def ping():
|
||||
return {"pong": True}
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def test_derive_input_schemas_resolves_lazy_annotations():
|
||||
app = _make_app()
|
||||
schemas = _derive_input_schemas(app, ["echo", "ping"])
|
||||
assert "echo" in schemas
|
||||
echo_schema = schemas["echo"]
|
||||
assert echo_schema["type"] == "object"
|
||||
assert "msg" in echo_schema["properties"]
|
||||
assert "n" in echo_schema["properties"]
|
||||
assert "msg" in echo_schema["required"]
|
||||
# ping has no Pydantic body → not in map (fallback applied by caller)
|
||||
assert "ping" not in schemas
|
||||
|
||||
|
||||
def test_mount_mcp_endpoint_exposes_derived_schemas():
|
||||
app = _make_app()
|
||||
mount_mcp_endpoint(
|
||||
app,
|
||||
name="test",
|
||||
version="1.0",
|
||||
valid_tokens=VALID_TOKENS,
|
||||
internal_base_url="http://localhost:0",
|
||||
tools=[
|
||||
{"name": "echo", "description": "Echo a message."},
|
||||
{"name": "ping", "description": "Ping."},
|
||||
],
|
||||
)
|
||||
c = TestClient(app)
|
||||
r = c.post(
|
||||
"/mcp",
|
||||
headers={"Authorization": f"Bearer {VALID_TOKEN}"},
|
||||
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
tools = r.json()["result"]["tools"]
|
||||
by_name = {t["name"]: t for t in tools}
|
||||
assert set(by_name["echo"]["inputSchema"]["required"]) == {"msg"}
|
||||
# ping fallback su schema generico
|
||||
assert by_name["ping"]["inputSchema"] == {
|
||||
"type": "object",
|
||||
"additionalProperties": True,
|
||||
}
|
||||
|
||||
|
||||
def test_mount_mcp_endpoint_requires_auth():
|
||||
app = _make_app()
|
||||
mount_mcp_endpoint(
|
||||
app,
|
||||
name="test",
|
||||
version="1.0",
|
||||
valid_tokens=VALID_TOKENS,
|
||||
internal_base_url="http://localhost:0",
|
||||
tools=[{"name": "echo"}],
|
||||
)
|
||||
c = TestClient(app)
|
||||
r = c.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
|
||||
assert r.status_code == 401
|
||||
r = c.post(
|
||||
"/mcp",
|
||||
headers={"Authorization": "Bearer WRONG"},
|
||||
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
|
||||
)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_explicit_input_schema_overrides_derived():
|
||||
app = _make_app()
|
||||
custom = {"type": "object", "properties": {"custom": {"type": "string"}}, "required": ["custom"]}
|
||||
mount_mcp_endpoint(
|
||||
app,
|
||||
name="test",
|
||||
version="1.0",
|
||||
valid_tokens=VALID_TOKENS,
|
||||
internal_base_url="http://localhost:0",
|
||||
tools=[{"name": "echo", "input_schema": custom}],
|
||||
)
|
||||
c = TestClient(app)
|
||||
r = c.post(
|
||||
"/mcp",
|
||||
headers={"Authorization": f"Bearer {VALID_TOKEN}"},
|
||||
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
|
||||
)
|
||||
assert r.json()["result"]["tools"][0]["inputSchema"] == custom
|
||||
Reference in New Issue
Block a user