AdrianoDev
993326136b
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>
113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
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
|