AdrianoDev
b71c66917c
- ruff: contextlib.suppress al posto di try/except/pass (client_registry, test_env_routing) - rimozione services/ legacy (residuo da git rm) - fix integration test fixture: rimosso sys.modules.pop che inquinava module references nei test successivi (test_audit, test_client_init_default_http) 254 test passano. Ruff: clean. Mypy: 68 warning preesistenti dal codice V1 migrato (strict=false). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
from fastapi import FastAPI, Request
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def test_health_no_auth_required():
|
|
from cerbero_mcp.auth import install_auth_middleware
|
|
fa = FastAPI()
|
|
install_auth_middleware(fa, testnet_token="t", mainnet_token="m")
|
|
|
|
@fa.get("/health")
|
|
def h():
|
|
return {"ok": True}
|
|
|
|
c = TestClient(fa)
|
|
r = c.get("/health")
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_apidocs_no_auth_required():
|
|
from cerbero_mcp.auth import install_auth_middleware
|
|
fa = FastAPI(docs_url="/apidocs")
|
|
install_auth_middleware(fa, testnet_token="t", mainnet_token="m")
|
|
|
|
c = TestClient(fa)
|
|
r = c.get("/apidocs")
|
|
assert r.status_code == 200
|
|
r = c.get("/openapi.json")
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_missing_authorization_header_401():
|
|
from cerbero_mcp.auth import install_auth_middleware
|
|
fa = FastAPI()
|
|
install_auth_middleware(fa, testnet_token="t", mainnet_token="m")
|
|
|
|
@fa.get("/mcp-deribit/health")
|
|
def h():
|
|
return {"ok": True}
|
|
|
|
c = TestClient(fa)
|
|
r = c.get("/mcp-deribit/health")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_invalid_bearer_401():
|
|
from cerbero_mcp.auth import install_auth_middleware
|
|
fa = FastAPI()
|
|
install_auth_middleware(fa, testnet_token="t", mainnet_token="m")
|
|
|
|
@fa.get("/mcp-deribit/health")
|
|
def h():
|
|
return {"ok": True}
|
|
|
|
c = TestClient(fa)
|
|
r = c.get("/mcp-deribit/health", headers={"Authorization": "Bearer wrong"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_testnet_token_sets_env_testnet():
|
|
from cerbero_mcp.auth import install_auth_middleware
|
|
|
|
fa = FastAPI()
|
|
install_auth_middleware(fa, testnet_token="tk_test", mainnet_token="tk_live")
|
|
|
|
@fa.get("/mcp-deribit/peek")
|
|
def peek(request: Request):
|
|
return {"env": request.state.environment}
|
|
|
|
c = TestClient(fa)
|
|
r = c.get("/mcp-deribit/peek", headers={"Authorization": "Bearer tk_test"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"env": "testnet"}
|
|
|
|
|
|
def test_mainnet_token_sets_env_mainnet():
|
|
from cerbero_mcp.auth import install_auth_middleware
|
|
|
|
fa = FastAPI()
|
|
install_auth_middleware(fa, testnet_token="tk_test", mainnet_token="tk_live")
|
|
|
|
@fa.get("/mcp-deribit/peek")
|
|
def peek(request: Request):
|
|
return {"env": request.state.environment}
|
|
|
|
c = TestClient(fa)
|
|
r = c.get("/mcp-deribit/peek", headers={"Authorization": "Bearer tk_live"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"env": "mainnet"}
|
|
|
|
|
|
def test_uses_compare_digest():
|
|
"""Verifica che _check_token usi secrets.compare_digest (timing-safe)."""
|
|
import inspect
|
|
|
|
from cerbero_mcp import auth
|
|
|
|
src = inspect.getsource(auth)
|
|
assert "compare_digest" in src, "auth.py deve usare secrets.compare_digest"
|