Cerbero-mcp/services/mcp-deribit/tests/test_server_acl.py

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi.testclient import TestClient
from mcp_deribit.server import create_app
from mcp_common.auth import Principal, TokenStore


@pytest.fixture
def mock_client():
    c = MagicMock()
    c.get_ticker = AsyncMock(return_value={"mark_price": 50000})
    c.get_instruments = AsyncMock(return_value=[])
    c.get_orderbook = AsyncMock(return_value={"bids": [], "asks": []})
    c.get_positions = AsyncMock(return_value=[])
    c.get_account_summary = AsyncMock(return_value={"equity": 1000})
    c.get_trade_history = AsyncMock(return_value=[])
    c.get_historical = AsyncMock(return_value={"candles": []})
    c.get_technical_indicators = AsyncMock(return_value={"rsi": 55.0})
    c.place_order = AsyncMock(return_value={"order_id": "x"})
    c.cancel_order = AsyncMock(return_value={"order_id": "x", "state": "cancelled"})
    c.set_stop_loss = AsyncMock(return_value={"order_id": "x", "stop_price": 45000})
    c.set_take_profit = AsyncMock(return_value={"order_id": "x", "tp_price": 55000})
    c.close_position = AsyncMock(return_value={"closed": True})
    c.set_leverage = AsyncMock(return_value={"state": "ok"})
    return c


@pytest.fixture
def http(mock_client):
    store = TokenStore(tokens={
        "ct": Principal("core", {"core"}),
        "ot": Principal("observer", {"observer"}),
    })
    app = create_app(client=mock_client, token_store=store)
    return TestClient(app)


def test_health(http):
    assert http.get("/health").status_code == 200


def test_get_ticker_core_ok(http):
    r = http.post(
        "/tools/get_ticker",
        headers={"Authorization": "Bearer ct"},
        json={"instrument_name": "BTC-PERPETUAL"},
    )
    assert r.status_code == 200
    assert r.json()["mark_price"] == 50000


def test_get_ticker_observer_ok(http):
    r = http.post(
        "/tools/get_ticker",
        headers={"Authorization": "Bearer ot"},
        json={"instrument_name": "BTC-PERPETUAL"},
    )
    assert r.status_code == 200


def test_get_ticker_no_auth_401(http):
    r = http.post("/tools/get_ticker", json={"instrument_name": "BTC-PERPETUAL"})
    assert r.status_code == 401


def test_get_ticker_alias_instrument_ok(http, mock_client):
    r = http.post(
        "/tools/get_ticker",
        headers={"Authorization": "Bearer ct"},
        json={"instrument": "ETH"},
    )
    assert r.status_code == 200
    mock_client.get_ticker.assert_awaited_with("ETH")


def test_place_order_core_ok(http):
    r = http.post(
        "/tools/place_order",
        headers={"Authorization": "Bearer ct"},
        json={"instrument_name": "BTC-PERPETUAL", "side": "buy", "amount": 10},
    )
    assert r.status_code == 200


def test_place_order_observer_forbidden(http):
    r = http.post(
        "/tools/place_order",
        headers={"Authorization": "Bearer ot"},
        json={"instrument_name": "BTC-PERPETUAL", "side": "buy", "amount": 10},
    )
    assert r.status_code == 403


def test_place_order_notional_cap_enforced(http):
    """CER-016: reject se notional > CERBERO_MAX_NOTIONAL (default 200)."""
    r = http.post(
        "/tools/place_order",
        headers={"Authorization": "Bearer ct"},
        json={
            "instrument_name": "ETH-PERPETUAL",
            "side": "buy",
            "amount": 335,  # USD — cap 200
        },
    )
    assert r.status_code == 403
    body = r.json()
    assert body["error"]["code"] == "HARD_PROHIBITION"


def test_place_order_leverage_cap_enforced(http):
    """CER-016: reject leverage > 3x."""
    r = http.post(
        "/tools/place_order",
        headers={"Authorization": "Bearer ct"},
        json={
            "instrument_name": "BTC-PERPETUAL",
            "side": "buy",
            "amount": 50,
            "leverage": 50,
        },
    )
    assert r.status_code == 403
    body = r.json()
    assert body["error"]["code"] == "HARD_PROHIBITION"


def test_place_order_reduce_only_skips_cap(http):
    """CER-016: reduce_only orders bypassano cap notional (è close)."""
    r = http.post(
        "/tools/place_order",
        headers={"Authorization": "Bearer ct"},
        json={
            "instrument_name": "ETH-PERPETUAL",
            "side": "sell",
            "amount": 10000,
            "reduce_only": True,
        },
    )
    assert r.status_code == 200


def test_close_position_core_ok(http):
    r = http.post(
        "/tools/close_position",
        headers={"Authorization": "Bearer ct"},
        json={"instrument_name": "BTC-PERPETUAL"},
    )
    assert r.status_code == 200


def test_close_position_observer_forbidden(http):
    r = http.post(
        "/tools/close_position",
        headers={"Authorization": "Bearer ot"},
        json={"instrument_name": "BTC-PERPETUAL"},
    )
    assert r.status_code == 403


def test_cancel_order_observer_forbidden(http):
    r = http.post(
        "/tools/cancel_order",
        headers={"Authorization": "Bearer ot"},
        json={"order_id": "abc123"},
    )
    assert r.status_code == 403


def test_set_stop_loss_observer_forbidden(http):
    r = http.post(
        "/tools/set_stop_loss",
        headers={"Authorization": "Bearer ot"},
        json={"order_id": "abc123", "stop_price": 45000.0},
    )
    assert r.status_code == 403


def test_get_account_summary_observer_ok(http):
    r = http.post(
        "/tools/get_account_summary",
        headers={"Authorization": "Bearer ot"},
        json={"currency": "USDC"},
    )
    assert r.status_code == 200
    assert r.json()["equity"] == 1000