feat(mcp-macro): expose COT report tools via MCP endpoint
This commit is contained in:
@@ -125,3 +125,79 @@ def test_get_market_overview_observer_ok(http):
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user