Adriano
466e63dc19
Wrapper async tipizzati sui sei servizi MCP HTTP che Cerbero Bite consuma in autonomia. 277 test pass, copertura clients 93%, mypy strict pulito, ruff clean. Base layer: - clients/_base.py: HttpToolClient con httpx + tenacity (retry esponenziale 3x, timeout 8s, mapping HTTP→eccezioni tipizzate). - clients/_exceptions.py: McpAuthError, McpServerError, McpToolError, McpDataAnomalyError, McpNotFoundError, McpTimeoutError. - config/mcp_endpoints.py: risoluzione URL via Docker DNS (mcp-deribit:9011, ...) con override per servizio via env var; caricamento bearer token da secrets/core.token o CERBERO_BITE_CORE_TOKEN_FILE. Wrapper: - clients/macro.py: next_high_severity_within() per filtro entry §2.5. - clients/sentiment.py: funding_cross_median_annualized() con annualizzazione per period nativo per exchange (Binance/Bybit/OKX 1095, Hyperliquid 8760). - clients/hyperliquid.py: funding_rate_annualized() per filtro §2.6. - clients/portfolio.py: total_equity_eur(), asset_pct_of_portfolio() per sizing engine + filtro §2.7. - clients/telegram.py: notify-only (no callback queue, no conferme — Bite auto-execute). - clients/deribit.py: environment_info, index_price_eth, latest_dvol, options_chain, get_tickers, orderbook_depth_top3, get_account_summary, get_positions, place_combo_order (combo atomico), cancel_order. CLI: - cerbero-bite ping: health-check parallelo di tutti gli MCP con tabella rich (OK/FAIL/SKIPPED). Docker: - Dockerfile multi-stage Python 3.13 + uv, user non-root. - docker-compose.yml con rete external "cerbero-suite", secret core_token montato a /run/secrets/core_token, env per ogni MCP. - secrets/README.md documenta il setup del token. Documentazione di intervento: - docs/12-mcp-deribit-changes.md: spec delle modifiche apportate al server mcp-deribit (place_combo_order + override testnet via DERIBIT_TESTNET). Dipendenze: - aggiunto pytest-httpx per i test HTTP. - rimosso mcp>=1.0 (non usiamo l'SDK MCP, parliamo via HTTP REST). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""Tests for the MCP endpoint and token resolver."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cerbero_bite.config.mcp_endpoints import (
|
|
DEFAULT_ENDPOINTS,
|
|
MCP_SERVICES,
|
|
load_endpoints,
|
|
load_token,
|
|
)
|
|
|
|
|
|
def test_defaults_match_known_docker_dns() -> None:
|
|
assert DEFAULT_ENDPOINTS["deribit"] == "http://mcp-deribit:9011"
|
|
assert DEFAULT_ENDPOINTS["telegram"] == "http://mcp-telegram:9017"
|
|
|
|
|
|
def test_load_endpoints_uses_defaults_when_env_empty() -> None:
|
|
endpoints = load_endpoints(env={})
|
|
for name, url in DEFAULT_ENDPOINTS.items():
|
|
assert endpoints.for_service(name) == url
|
|
|
|
|
|
def test_per_service_env_var_override() -> None:
|
|
endpoints = load_endpoints(
|
|
env={"CERBERO_BITE_MCP_DERIBIT_URL": "http://localhost:9911"}
|
|
)
|
|
assert endpoints.deribit == "http://localhost:9911"
|
|
assert endpoints.macro == DEFAULT_ENDPOINTS["macro"]
|
|
|
|
|
|
def test_per_service_env_var_strips_trailing_slash() -> None:
|
|
endpoints = load_endpoints(
|
|
env={"CERBERO_BITE_MCP_DERIBIT_URL": "http://localhost:9911///"}
|
|
)
|
|
assert endpoints.deribit == "http://localhost:9911"
|
|
|
|
|
|
def test_for_service_unknown_raises_key_error() -> None:
|
|
endpoints = load_endpoints(env={})
|
|
with pytest.raises(KeyError, match="unknown MCP service"):
|
|
endpoints.for_service("nope")
|
|
|
|
|
|
def test_load_token_uses_explicit_path(tmp_path: Path) -> None:
|
|
target = tmp_path / "core.token"
|
|
target.write_text("abcdef\n", encoding="utf-8")
|
|
assert load_token(path=target) == "abcdef"
|
|
|
|
|
|
def test_load_token_uses_env_var(tmp_path: Path) -> None:
|
|
target = tmp_path / "core.token"
|
|
target.write_text("xyz", encoding="utf-8")
|
|
token = load_token(env={"CERBERO_BITE_CORE_TOKEN_FILE": str(target)})
|
|
assert token == "xyz"
|
|
|
|
|
|
def test_load_token_raises_when_file_missing(tmp_path: Path) -> None:
|
|
with pytest.raises(FileNotFoundError):
|
|
load_token(path=tmp_path / "missing")
|
|
|
|
|
|
def test_load_token_raises_when_file_empty(tmp_path: Path) -> None:
|
|
target = tmp_path / "empty"
|
|
target.write_text("", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="empty"):
|
|
load_token(path=target)
|
|
|
|
|
|
def test_mcp_services_table_is_complete() -> None:
|
|
expected = {"deribit", "hyperliquid", "macro", "sentiment", "telegram", "portfolio"}
|
|
assert set(MCP_SERVICES) == expected
|