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

from __future__ import annotations

from unittest.mock import AsyncMock, patch

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


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


# --- Health ---

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


# --- get_economic_indicators ---

def test_get_economic_indicators_core_ok(http):
    with patch(
        "mcp_macro.server.fetch_economic_indicators",
        new=AsyncMock(return_value={"fed_rate": 5.25, "updated_at": "2024-01-01T00:00:00+00:00"}),
    ):
        r = http.post(
            "/tools/get_economic_indicators",
            headers={"Authorization": "Bearer ct"},
            json={},
        )
    assert r.status_code == 200
    assert r.json()["fed_rate"] == 5.25


def test_get_economic_indicators_observer_ok(http):
    with patch(
        "mcp_macro.server.fetch_economic_indicators",
        new=AsyncMock(return_value={"fed_rate": 5.25}),
    ):
        r = http.post(
            "/tools/get_economic_indicators",
            headers={"Authorization": "Bearer ot"},
            json={},
        )
    assert r.status_code == 200


def test_get_economic_indicators_no_auth_401(http):
    r = http.post("/tools/get_economic_indicators", json={})
    assert r.status_code == 401


# --- get_macro_calendar ---

def test_get_macro_calendar_core_ok(http):
    with patch(
        "mcp_macro.server.fetch_macro_calendar",
        new=AsyncMock(return_value={"events": []}),
    ):
        r = http.post(
            "/tools/get_macro_calendar",
            headers={"Authorization": "Bearer ct"},
            json={"days": 7},
        )
    assert r.status_code == 200


def test_get_macro_calendar_observer_ok(http):
    with patch(
        "mcp_macro.server.fetch_macro_calendar",
        new=AsyncMock(return_value={"events": []}),
    ):
        r = http.post(
            "/tools/get_macro_calendar",
            headers={"Authorization": "Bearer ot"},
            json={},
        )
    assert r.status_code == 200


def test_get_macro_calendar_no_auth_401(http):
    r = http.post("/tools/get_macro_calendar", json={})
    assert r.status_code == 401


# --- get_market_overview ---

def test_get_market_overview_core_ok(http):
    with patch(
        "mcp_macro.server.fetch_market_overview",
        new=AsyncMock(return_value={"btc_dominance": 52.0, "btc_price": 65000}),
    ):
        r = http.post(
            "/tools/get_market_overview",
            headers={"Authorization": "Bearer ct"},
            json={},
        )
    assert r.status_code == 200
    assert r.json()["btc_price"] == 65000


def test_get_market_overview_observer_ok(http):
    with patch(
        "mcp_macro.server.fetch_market_overview",
        new=AsyncMock(return_value={"btc_dominance": 52.0}),
    ):
        r = http.post(
            "/tools/get_market_overview",
            headers={"Authorization": "Bearer ot"},
            json={},
        )
    assert r.status_code == 200


def test_get_market_overview_no_auth_401(http):
    r = http.post("/tools/get_market_overview", json={})
    assert r.status_code == 401


from unittest.mock import AsyncMock, patch


def test_get_cot_tff_core_ok(http):
    with patch(
        "mcp_macro.server.fetch_cot_tff",
        new=AsyncMock(return_value={"symbol": "ES", "rows": []}),
    ):
        r = http.post(
            "/tools/get_cot_tff",
            headers={"Authorization": "Bearer ct"},
            json={"symbol": "ES"},
        )
    assert r.status_code == 200
    assert r.json()["symbol"] == "ES"


def test_get_cot_tff_observer_ok(http):
    with patch(
        "mcp_macro.server.fetch_cot_tff",
        new=AsyncMock(return_value={"symbol": "ES", "rows": []}),
    ):
        r = http.post(
            "/tools/get_cot_tff",
            headers={"Authorization": "Bearer ot"},
            json={"symbol": "ES"},
        )
    assert r.status_code == 200


def test_get_cot_tff_no_auth_401(http):
    r = http.post("/tools/get_cot_tff", json={"symbol": "ES"})
    assert r.status_code == 401


def test_get_cot_disagg_observer_ok(http):
    with patch(
        "mcp_macro.server.fetch_cot_disaggregated",
        new=AsyncMock(return_value={"symbol": "CL", "rows": []}),
    ):
        r = http.post(
            "/tools/get_cot_disaggregated",
            headers={"Authorization": "Bearer ot"},
            json={"symbol": "CL"},
        )
    assert r.status_code == 200


def test_get_cot_disagg_no_auth_401(http):
    r = http.post("/tools/get_cot_disaggregated", json={"symbol": "CL"})
    assert r.status_code == 401


def test_get_cot_extreme_positioning_ok(http):
    with patch(
        "mcp_macro.server.fetch_cot_extreme_positioning",
        new=AsyncMock(return_value={"extremes": []}),
    ):
        r = http.post(
            "/tools/get_cot_extreme_positioning",
            headers={"Authorization": "Bearer ot"},
            json={},
        )
    assert r.status_code == 200


def test_get_cot_extreme_positioning_lookback_too_short(http):
    """Pydantic validation: lookback_weeks < 4 → 422."""
    r = http.post(
        "/tools/get_cot_extreme_positioning",
        headers={"Authorization": "Bearer ct"},
        json={"lookback_weeks": 2},
    )
    assert r.status_code == 422