yuxuanhui
50 lines
2.0 KiB
Python
50 lines
2.0 KiB
Python
"""Exercise OPTIONS through the real authenticated adapter, without live account traffic."""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.worldquant import WqClient, WqError
|
|
from tests.catalog_fake import platform_response
|
|
|
|
|
|
async def test_options_refreshes_auth_and_keeps_new_choices(settings):
|
|
calls = []
|
|
auth = 0
|
|
|
|
def handler(request):
|
|
nonlocal auth
|
|
calls.append((request.method, request.url.path))
|
|
if request.url.path == "/authentication":
|
|
auth += 1
|
|
return httpx.Response(201, json={})
|
|
assert request.method == "OPTIONS" and request.url.path == "/simulations"
|
|
if auth == 1:
|
|
return httpx.Response(401)
|
|
data = platform_response()
|
|
children = data["actions"]["POST"]["settings"]["children"]
|
|
children["delay"]["choices"]["instrumentType"]["EQUITY"]["region"]["IND"] = [{"value": 2}]
|
|
return httpx.Response(200, json=data)
|
|
|
|
client = WqClient(settings, transport=httpx.MockTransport(handler))
|
|
try:
|
|
await client.authenticate("test@example.com", "test-only")
|
|
data = await client.get_platform_setting_options()
|
|
assert auth == 2
|
|
assert {"instrument_type": "EQUITY", "region": "IND", "delay": 2,
|
|
"universes": ["TOP500"]} in data["instrument_options"]
|
|
assert all(method == "OPTIONS" or path == "/authentication" for method, path in calls)
|
|
finally:
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.parametrize("body", [{}, {"actions": None}, {"actions": {"POST": {"settings": {"children": []}}}}])
|
|
async def test_malformed_options_fail_without_static_fallback(settings, body):
|
|
client = WqClient(settings, transport=httpx.MockTransport(lambda _: httpx.Response(200, json=body)))
|
|
client.credentials, client.authenticated = ("test", "test"), True
|
|
try:
|
|
with pytest.raises(WqError) as error:
|
|
await client.get_platform_setting_options()
|
|
assert error.value.code == "invalid_response"
|
|
finally:
|
|
await client.close()
|