yuxuanhui
93 lines
4.1 KiB
Python
93 lines
4.1 KiB
Python
"""Verify distribution through MCP authorization, validation and audit boundaries."""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.worldquant import WqClient
|
|
from tests.test_mcp import credentials, invoke
|
|
from tests.test_mcp import mcp_app as mcp_app
|
|
|
|
|
|
def row(name, count, region="USA", delay=1):
|
|
return {"category": {"id": name, "name": name}, "alphaCount": count,
|
|
"region": region, "delay": delay}
|
|
|
|
|
|
async def test_distribution(mcp_app):
|
|
calls = []
|
|
|
|
def platform(request):
|
|
calls.append(request)
|
|
assert request.method == "GET"
|
|
assert request.url.path == "/users/self/activities/pyramid-alphas"
|
|
return httpx.Response(200, json={"pyramids": [row("zero", 0), row("one", 1),
|
|
row("two", 2), row("three", 3), row("four", 4), row("other", 5, "GLB"),
|
|
row("delay_zero", 5, delay=0)]})
|
|
|
|
client = WqClient(mcp_app.state.settings, transport=httpx.MockTransport(platform))
|
|
client.credentials, client.authenticated = ("test", "test"), True
|
|
original = mcp_app.state.runner.client
|
|
mcp_app.state.runner.client = client
|
|
try:
|
|
reader, _ = await credentials(mcp_app, {"research:read"})
|
|
result = await invoke(mcp_app, reader, "get_pyramid_distribution", {"region": "USA", "delay": 1, "current_date": "2026-09-13"})
|
|
assert [r["alpha_count"] for r in result["lit"]] == [4, 3]
|
|
assert [r["remaining"] for r in result["in_progress"]] == [2, 1]
|
|
assert result["unlit"][0]["category"]["id"] == "zero"
|
|
assert dict(calls[0].url.params) == {"startDate": "2026-07-01", "endDate": "2026-09-30"}
|
|
assert result["period"] == {"quarter": "2026-Q3", "start_date": "2026-07-01", "end_date": "2026-09-30"}
|
|
for args in ({"region": "USA", "delay": 2}, {"region": "USA", "delay": True},
|
|
{"region": "../", "delay": 1}):
|
|
response = await mcp_app.state.mcp.invoke(reader, "get_pyramid_distribution", args | {"current_date": "2026-09-13"})
|
|
assert response.structured_content["error"]["code"] == "INVALID_INPUT"
|
|
assert len(calls) == 1
|
|
client.credentials, client.authenticated = None, False
|
|
response = await mcp_app.state.mcp.invoke(reader, "get_pyramid_distribution", {"region": "USA", "delay": 1, "current_date": "2026-09-13"})
|
|
assert response.structured_content["error"]["code"] == "DISCONNECTED"
|
|
finally:
|
|
mcp_app.state.runner.client = original
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.parametrize("rows", [[], [row("x", None)], [row("x", -1)],
|
|
[row("x", True)], [row("x", 1), row("x", 2)]])
|
|
def test_missing_evidence_is_not_zero(rows):
|
|
from app.research_access.pyramids import distribution
|
|
|
|
with pytest.raises(ValueError):
|
|
distribution({"pyramids": rows}, "USA", 1)
|
|
|
|
|
|
@pytest.mark.parametrize(('value', 'quarter', 'start', 'end'), [
|
|
('2026-01-01', '2026-Q1', '2026-01-01', '2026-03-31'),
|
|
('2026-03-31', '2026-Q1', '2026-01-01', '2026-03-31'),
|
|
('2026-04-01', '2026-Q2', '2026-04-01', '2026-06-30'),
|
|
('2026-06-30', '2026-Q2', '2026-04-01', '2026-06-30'),
|
|
('2026-07-01', '2026-Q3', '2026-07-01', '2026-09-30'),
|
|
('2026-09-30', '2026-Q3', '2026-07-01', '2026-09-30'),
|
|
('2026-10-01', '2026-Q4', '2026-10-01', '2026-12-31'),
|
|
('2026-12-31', '2026-Q4', '2026-10-01', '2026-12-31'),
|
|
('2027-01-01', '2027-Q1', '2027-01-01', '2027-03-31'),
|
|
('2024-02-29', '2024-Q1', '2024-01-01', '2024-03-31'),
|
|
])
|
|
def test_quarter_boundaries(value, quarter, start, end):
|
|
from datetime import date
|
|
|
|
from app.research_access.pyramids import quarter_period
|
|
|
|
assert quarter_period(date.fromisoformat(value)) == {
|
|
'quarter': quarter, 'start_date': start, 'end_date': end}
|
|
|
|
|
|
@pytest.mark.parametrize('value', [None, '2026-02-30', 'not-a-date'])
|
|
def test_date_required_and_valid(value):
|
|
from pydantic import ValidationError
|
|
|
|
from app.research_access.contracts import PyramidQuery
|
|
|
|
args = {'region': 'USA', 'delay': 1}
|
|
if value is not None:
|
|
args['current_date'] = value
|
|
with pytest.raises(ValidationError):
|
|
PyramidQuery.model_validate(args)
|