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

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock

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


@pytest.fixture
def token_store():
    return TokenStore(
        tokens={
            "core-tok": Principal("core", {"core"}),
            "obs-tok": Principal("observer", {"observer"}),
        }
    )


@pytest.fixture
def mock_client():
    c = MagicMock()
    c.get_ticker = AsyncMock(return_value={"symbol": "BTCUSDT"})
    c.get_ticker_batch = AsyncMock(return_value={"BTCUSDT": {}})
    c.get_orderbook = AsyncMock(return_value={"bids": [], "asks": []})
    c.get_historical = AsyncMock(return_value={"candles": []})
    c.get_indicators = AsyncMock(return_value={"rsi": 50.0})
    c.get_funding_rate = AsyncMock(return_value={"funding_rate": 0.0001})
    c.get_funding_history = AsyncMock(return_value={"history": []})
    c.get_open_interest = AsyncMock(return_value={"points": []})
    c.get_instruments = AsyncMock(return_value={"instruments": []})
    c.get_option_chain = AsyncMock(return_value={"options": []})
    c.get_positions = AsyncMock(return_value=[])
    c.get_account_summary = AsyncMock(return_value={"equity": 0})
    c.get_trade_history = AsyncMock(return_value=[])
    c.get_open_orders = AsyncMock(return_value=[])
    c.get_basis_spot_perp = AsyncMock(return_value={"basis_pct": 0})
    c.place_order = AsyncMock(return_value={"order_id": "x"})
    c.amend_order = AsyncMock(return_value={"order_id": "x"})
    c.cancel_order = AsyncMock(return_value={"status": "cancelled"})
    c.cancel_all_orders = AsyncMock(return_value={"cancelled_ids": []})
    c.set_stop_loss = AsyncMock(return_value={"status": "stop_loss_set"})
    c.set_take_profit = AsyncMock(return_value={"status": "take_profit_set"})
    c.close_position = AsyncMock(return_value={"status": "submitted"})
    c.set_leverage = AsyncMock(return_value={"status": "leverage_set"})
    c.switch_position_mode = AsyncMock(return_value={"status": "mode_switched"})
    c.transfer_asset = AsyncMock(return_value={"transfer_id": "tx"})
    c.place_combo_order = AsyncMock(return_value={"orders": [{"order_id": "ord-1"}, {"order_id": "ord-2"}]})
    c.get_orderbook_imbalance = AsyncMock(return_value={"imbalance_ratio": 0.0, "microprice": 100.0})
    c.get_basis_term_structure = AsyncMock(return_value={"asset": "BTC", "term_structure": []})
    return c


@pytest.fixture
def http(mock_client, token_store):
    app = create_app(client=mock_client, token_store=token_store, creds={"max_leverage": 5})
    return TestClient(app)


CORE = {"Authorization": "Bearer core-tok"}
OBS = {"Authorization": "Bearer obs-tok"}

READ_ENDPOINTS = [
    ("/tools/get_ticker", {"symbol": "BTCUSDT"}),
    ("/tools/get_ticker_batch", {"symbols": ["BTCUSDT"]}),
    ("/tools/get_orderbook", {"symbol": "BTCUSDT"}),
    ("/tools/get_historical", {"symbol": "BTCUSDT"}),
    ("/tools/get_indicators", {"symbol": "BTCUSDT"}),
    ("/tools/get_funding_rate", {"symbol": "BTCUSDT"}),
    ("/tools/get_funding_history", {"symbol": "BTCUSDT"}),
    ("/tools/get_open_interest", {"symbol": "BTCUSDT"}),
    ("/tools/get_instruments", {}),
    ("/tools/get_option_chain", {"base_coin": "BTC"}),
    ("/tools/get_positions", {}),
    ("/tools/get_account_summary", {}),
    ("/tools/get_trade_history", {}),
    ("/tools/get_open_orders", {}),
    ("/tools/get_basis_spot_perp", {"asset": "BTC"}),
    ("/tools/get_orderbook_imbalance", {"symbol": "BTCUSDT"}),
    ("/tools/get_basis_term_structure", {"asset": "BTC"}),
]

WRITE_ENDPOINTS = [
    ("/tools/place_order", {"category": "linear", "symbol": "BTCUSDT", "side": "Buy", "qty": 0.01}),
    ("/tools/amend_order", {"category": "linear", "symbol": "BTCUSDT", "order_id": "o1"}),
    ("/tools/cancel_order", {"category": "linear", "symbol": "BTCUSDT", "order_id": "o1"}),
    ("/tools/cancel_all_orders", {"category": "linear"}),
    ("/tools/set_stop_loss", {"category": "linear", "symbol": "BTCUSDT", "stop_loss": 55000}),
    ("/tools/set_take_profit", {"category": "linear", "symbol": "BTCUSDT", "take_profit": 65000}),
    ("/tools/close_position", {"category": "linear", "symbol": "BTCUSDT"}),
    ("/tools/set_leverage", {"category": "linear", "symbol": "BTCUSDT", "leverage": 5}),
    ("/tools/switch_position_mode", {"category": "linear", "symbol": "BTCUSDT", "mode": "hedge"}),
    ("/tools/transfer_asset", {"coin": "USDT", "amount": 10.0, "from_type": "UNIFIED", "to_type": "FUND"}),
    ("/tools/place_combo_order", {
        "category": "option",
        "legs": [
            {"symbol": "BTC-30APR26-75000-C-USDT", "side": "Buy", "qty": 0.01, "order_type": "Limit", "price": 5.0},
            {"symbol": "BTC-30APR26-80000-C-USDT", "side": "Sell", "qty": 0.01, "order_type": "Limit", "price": 3.0},
        ],
    }),
]


def test_place_combo_order_min_legs(http):
    r = http.post(
        "/tools/place_combo_order",
        json={
            "category": "option",
            "legs": [{"symbol": "X", "side": "Buy", "qty": 1, "order_type": "Limit", "price": 1.0}],
        },
        headers=CORE,
    )
    assert r.status_code == 422


@pytest.mark.parametrize("path,payload", READ_ENDPOINTS)
def test_read_core_ok(http, path, payload):
    r = http.post(path, json=payload, headers=CORE)
    assert r.status_code == 200, (path, r.text)


@pytest.mark.parametrize("path,payload", READ_ENDPOINTS)
def test_read_observer_ok(http, path, payload):
    r = http.post(path, json=payload, headers=OBS)
    assert r.status_code == 200, (path, r.text)


@pytest.mark.parametrize("path,payload", READ_ENDPOINTS)
def test_read_no_auth_401(http, path, payload):
    r = http.post(path, json=payload)
    assert r.status_code == 401, (path, r.text)


@pytest.mark.parametrize("path,payload", WRITE_ENDPOINTS)
def test_write_core_ok(http, path, payload):
    r = http.post(path, json=payload, headers=CORE)
    assert r.status_code == 200, (path, r.text)


@pytest.mark.parametrize("path,payload", WRITE_ENDPOINTS)
def test_write_observer_403(http, path, payload):
    r = http.post(path, json=payload, headers=OBS)
    assert r.status_code == 403, (path, r.text)


@pytest.mark.parametrize("path,payload", WRITE_ENDPOINTS)
def test_write_no_auth_401(http, path, payload):
    r = http.post(path, json=payload)
    assert r.status_code == 401, (path, r.text)