AdrianoDev
128 lines
3.4 KiB
Python
128 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from mcp_macro.server import create_app
|
|
from option_mcp_common.auth import Principal, TokenStore
|
|
|
|
|
|
@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
|