Adriano
abf5a140e2
Each bot now manages its own notification + portfolio aggregation: * TelegramClient calls the public Bot API directly via httpx, reading CERBERO_BITE_TELEGRAM_BOT_TOKEN / CERBERO_BITE_TELEGRAM_CHAT_ID from env. No credentials → silent disabled mode. * PortfolioClient composes DeribitClient + HyperliquidClient + the new MacroClient.get_asset_price/eur_usd_rate to expose equity (EUR) and per-asset exposure as the bot's own slice (no cross-bot view). * mcp-telegram and mcp-portfolio removed from MCP_SERVICES / McpEndpoints and the cerbero-bite ping CLI; health_check no longer probes portfolio. Docs (02/04/06/07) and docker-compose updated to reflect the new architecture. 353/353 tests pass; ruff clean; mypy src clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""Tests for the runtime dependency container."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from cerbero_bite.clients.portfolio import PortfolioClient
|
|
from cerbero_bite.config import golden_config
|
|
from cerbero_bite.config.mcp_endpoints import load_endpoints
|
|
from cerbero_bite.runtime import build_runtime
|
|
from cerbero_bite.state import connect
|
|
|
|
|
|
def test_build_runtime_creates_state_and_audit_files(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.sqlite"
|
|
audit_path = tmp_path / "audit.log"
|
|
|
|
ctx = build_runtime(
|
|
cfg=golden_config(),
|
|
endpoints=load_endpoints(env={}),
|
|
token="t",
|
|
db_path=db_path,
|
|
audit_path=audit_path,
|
|
clock=lambda: datetime(2026, 4, 27, 14, 0, tzinfo=UTC),
|
|
)
|
|
|
|
assert db_path.exists()
|
|
assert ctx.audit_log.path == audit_path
|
|
# system_state singleton initialised
|
|
conn = connect(db_path)
|
|
try:
|
|
state = ctx.repository.get_system_state(conn)
|
|
finally:
|
|
conn.close()
|
|
assert state is not None
|
|
assert state.config_version == ctx.cfg.config_version
|
|
|
|
|
|
def test_build_runtime_clients_pinned_to_endpoints(tmp_path: Path) -> None:
|
|
ctx = build_runtime(
|
|
cfg=golden_config(),
|
|
endpoints=load_endpoints(
|
|
env={"CERBERO_BITE_MCP_DERIBIT_URL": "http://localhost:9911"}
|
|
),
|
|
token="t",
|
|
db_path=tmp_path / "state.sqlite",
|
|
audit_path=tmp_path / "audit.log",
|
|
)
|
|
# type checks: every client is the right concrete type
|
|
assert ctx.deribit.SERVICE == "deribit"
|
|
assert ctx.macro.SERVICE == "macro"
|
|
assert ctx.sentiment.SERVICE == "sentiment"
|
|
assert ctx.hyperliquid.SERVICE == "hyperliquid"
|
|
# Portfolio is now an in-process aggregator over deribit/hyperliquid/macro;
|
|
# it has no SERVICE attribute. Telegram is also in-process and disabled
|
|
# when env vars are unset.
|
|
assert isinstance(ctx.portfolio, PortfolioClient)
|
|
assert ctx.telegram.enabled is False
|