Files
Cerbero-mcp/tests/unit/test_settings.py
T
Adriano bc75d3980a feat(V2): unified /mcp interface; retire Bybit/Alpaca to old/
Add a common cross-exchange interface (/mcp) over the integrated venues
(deribit, hyperliquid):

- get_instruments: uniform schema where each row carries its own
  `exchange`, `fees` (maker/taker, live from Deribit, null where the
  venue has no per-instrument schedule) and `history_start` (listing
  date, live from Deribit creation_timestamp), plus type/tick_size and a
  lossless `native` blob. Optional `exchange` filter; fan-out otherwise.
- get_historical: generalized to {exchange, instrument, interval,
  start_date, end_date}, returning a single chosen venue's candles.
  Consensus merge stays available on /mcp-cross.

New: routers/unified.py, exchanges/cross/instruments.py (normalizers),
UnifiedClient in cross/client.py, schemas in cross/tools.py. Deribit
get_instruments now also surfaces maker/taker_commission and
creation_timestamp (additive).

Retire Bybit and Alpaca from the API surface: move clients, routers,
settings classes and their tests under old/ (history preserved via
git mv); drop them from the builder, /mcp-cross dispatch and symbol_map.
Bybit remains a public funding/OI data source in sentiment (not the
trading client). IBKR is intentionally excluded from /mcp for now.

Docs: rewrite API_REFERENCE.md (remove Bybit/Alpaca, document /mcp,
clarify that data_timestamp is injected globally by middleware).

Tests: add unified-interface coverage; update cross/settings/builder/boot
tests for the reduced venue set. Fix a pre-existing flaky assertion in
the Hyperliquid signing test (r/s use eth_utils.to_hex like the official
SDK, so a leading zero byte yields <66 chars ~1/256 of the time).

323 passed, ruff clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-29 09:09:22 +00:00

206 lines
7.0 KiB
Python

from __future__ import annotations
import os
import pytest
from pydantic import ValidationError
def _minimal_env(**overrides) -> dict:
base = {
"TESTNET_TOKEN": "t_test_123",
"MAINNET_TOKEN": "t_live_456",
"DERIBIT_CLIENT_ID": "id",
"DERIBIT_CLIENT_SECRET": "secret",
"DERIBIT_URL_LIVE": "https://www.deribit.com/api/v2",
"DERIBIT_URL_TESTNET": "https://test.deribit.com/api/v2",
"HYPERLIQUID_WALLET_ADDRESS": "0xabc",
"HYPERLIQUID_API_WALLET_ADDRESS": "0xdef",
"HYPERLIQUID_PRIVATE_KEY": "0x123",
"HYPERLIQUID_URL_LIVE": "https://api.hyperliquid.xyz",
"HYPERLIQUID_URL_TESTNET": "https://api.hyperliquid-testnet.xyz",
"FRED_API_KEY": "x",
"FINNHUB_API_KEY": "y",
"CRYPTOPANIC_KEY": "z",
"LUNARCRUSH_KEY": "w",
}
base.update(overrides)
return base
def test_settings_load_minimal(monkeypatch):
for k, v in _minimal_env().items():
monkeypatch.setenv(k, v)
monkeypatch.setenv("PORT", "9123")
from cerbero_mcp.settings import Settings
s = Settings()
assert s.port == 9123
assert s.host == "0.0.0.0"
assert s.testnet_token.get_secret_value() == "t_test_123"
assert s.mainnet_token.get_secret_value() == "t_live_456"
assert s.deribit.url_testnet.endswith("test.deribit.com/api/v2")
assert s.deribit.max_leverage == 3
def test_settings_missing_token_fails(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path) # isola dal .env reale del working dir
env = _minimal_env()
env.pop("TESTNET_TOKEN")
for k, v in env.items():
monkeypatch.setenv(k, v)
monkeypatch.delenv("TESTNET_TOKEN", raising=False)
from cerbero_mcp.settings import Settings
with pytest.raises(ValidationError):
Settings()
def test_settings_extras_ignored(monkeypatch):
for k, v in _minimal_env().items():
monkeypatch.setenv(k, v)
monkeypatch.setenv("UNRELATED_VAR", "ignored")
from cerbero_mcp.settings import Settings
s = Settings()
assert s.testnet_token.get_secret_value() == "t_test_123"
def test_settings_secret_str_no_leak(monkeypatch):
for k, v in _minimal_env().items():
monkeypatch.setenv(k, v)
from cerbero_mcp.settings import Settings
s = Settings()
assert "t_test_123" not in repr(s)
assert "t_live_456" not in repr(s)
def _isolated(monkeypatch, tmp_path, env: dict) -> None:
"""Isola Settings dal .env reale in working dir e setta solo env passato."""
monkeypatch.chdir(tmp_path)
for k in (
"DERIBIT_CLIENT_ID", "DERIBIT_CLIENT_SECRET",
"DERIBIT_CLIENT_ID_TESTNET", "DERIBIT_CLIENT_SECRET_TESTNET",
"DERIBIT_CLIENT_ID_LIVE", "DERIBIT_CLIENT_SECRET_LIVE",
):
monkeypatch.delenv(k, raising=False)
for k, v in env.items():
monkeypatch.setenv(k, v)
def test_deribit_credentials_legacy_single_pair(monkeypatch, tmp_path):
"""Solo DERIBIT_CLIENT_ID/SECRET → entrambi gli env usano la stessa coppia."""
_isolated(monkeypatch, tmp_path, _minimal_env())
from cerbero_mcp.settings import Settings
s = Settings()
assert s.deribit.credentials("testnet") == ("id", "secret")
assert s.deribit.credentials("mainnet") == ("id", "secret")
def test_deribit_credentials_per_env_pairs(monkeypatch, tmp_path):
"""Coppie _TESTNET e _LIVE → ognuna serve l'env corrispondente."""
env = _minimal_env()
env.pop("DERIBIT_CLIENT_ID")
env.pop("DERIBIT_CLIENT_SECRET")
env["DERIBIT_CLIENT_ID_TESTNET"] = "tid"
env["DERIBIT_CLIENT_SECRET_TESTNET"] = "tsec"
env["DERIBIT_CLIENT_ID_LIVE"] = "lid"
env["DERIBIT_CLIENT_SECRET_LIVE"] = "lsec"
_isolated(monkeypatch, tmp_path, env)
from cerbero_mcp.settings import Settings
s = Settings()
assert s.deribit.credentials("testnet") == ("tid", "tsec")
assert s.deribit.credentials("mainnet") == ("lid", "lsec")
def test_deribit_credentials_env_specific_overrides_fallback(monkeypatch, tmp_path):
"""_LIVE presente prevale sulla coppia base anche se entrambe configurate."""
env = _minimal_env()
env["DERIBIT_CLIENT_ID_LIVE"] = "lid"
env["DERIBIT_CLIENT_SECRET_LIVE"] = "lsec"
_isolated(monkeypatch, tmp_path, env)
from cerbero_mcp.settings import Settings
s = Settings()
assert s.deribit.credentials("mainnet") == ("lid", "lsec")
assert s.deribit.credentials("testnet") == ("id", "secret") # fallback
def test_deribit_credentials_missing_raises(monkeypatch, tmp_path):
"""Nessuna coppia configurata → ValueError esplicito."""
env = _minimal_env()
env.pop("DERIBIT_CLIENT_ID")
env.pop("DERIBIT_CLIENT_SECRET")
_isolated(monkeypatch, tmp_path, env)
from cerbero_mcp.settings import Settings
s = Settings()
with pytest.raises(ValueError, match="not configured for env=mainnet"):
s.deribit.credentials("mainnet")
def test_ibkr_settings_prefer_testnet_specific(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
for k in list(os.environ):
if k.startswith("IBKR_"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("IBKR_CONSUMER_KEY", "base_consumer")
monkeypatch.setenv("IBKR_CONSUMER_KEY_TESTNET", "paper_consumer")
monkeypatch.setenv("IBKR_ACCESS_TOKEN_TESTNET", "paper_token")
monkeypatch.setenv("IBKR_ACCESS_TOKEN_SECRET_TESTNET", "paper_secret")
monkeypatch.setenv("IBKR_SIGNATURE_KEY_PATH_TESTNET", "/secrets/sig_paper.pem")
monkeypatch.setenv("IBKR_ENCRYPTION_KEY_PATH_TESTNET", "/secrets/enc_paper.pem")
monkeypatch.setenv("IBKR_ACCOUNT_ID_TESTNET", "DU1234567")
monkeypatch.setenv("IBKR_DH_PRIME", "ffff")
from cerbero_mcp.settings import IBKRSettings
s = IBKRSettings()
creds = s.credentials("testnet")
assert creds["consumer_key"] == "paper_consumer"
assert creds["access_token"] == "paper_token"
assert creds["account_id"] == "DU1234567"
def test_ibkr_settings_fallback_to_base(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
for k in list(os.environ):
if k.startswith("IBKR_"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("IBKR_CONSUMER_KEY", "base_consumer")
monkeypatch.setenv("IBKR_ACCESS_TOKEN", "base_token")
monkeypatch.setenv("IBKR_ACCESS_TOKEN_SECRET", "base_secret")
monkeypatch.setenv("IBKR_SIGNATURE_KEY_PATH", "/secrets/sig.pem")
monkeypatch.setenv("IBKR_ENCRYPTION_KEY_PATH", "/secrets/enc.pem")
monkeypatch.setenv("IBKR_ACCOUNT_ID_TESTNET", "DU1234567")
monkeypatch.setenv("IBKR_DH_PRIME", "ffff")
from cerbero_mcp.settings import IBKRSettings
s = IBKRSettings()
creds = s.credentials("testnet")
assert creds["consumer_key"] == "base_consumer"
def test_ibkr_settings_missing_raises(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
for k in list(os.environ):
if k.startswith("IBKR_"):
monkeypatch.delenv(k, raising=False)
from cerbero_mcp.settings import IBKRSettings
s = IBKRSettings()
with pytest.raises(ValueError, match="IBKR credentials not configured"):
s.credentials("testnet")