Files
xiaobaifupan/next/tests/test_screener_api.py
T

64 lines
2.2 KiB
Python

from __future__ import annotations
import httpx
from backend.bootstrap.application import create_application
from backend.bootstrap.settings import Settings
from tests.support import run_scenario
from tests.test_accounts import (
ADMIN_PASSWORD,
USER_PASSWORD,
csrf_headers,
register,
use_session,
)
def _formula() -> dict:
return {
"universe": {"exclude_st": True, "listed_days_min": 120},
"filters": [{"field": "close", "op": ">", "value": 1}],
"score": [{"field": "close", "weight": 1, "direction": "desc"}],
"limit": 10,
"min_score": 0.5,
}
def test_screener_catalog_lock_and_custom_strategy_boundary(tmp_path) -> None:
application = create_application(Settings.for_test(tmp_path))
async def scenario(client: httpx.AsyncClient) -> None:
_, admin = await register(client, "admin-screener", ADMIN_PASSWORD)
client.cookies.clear()
_, regular = await register(client, "regular-screener", USER_PASSWORD)
catalog = await client.get("/api/screener/catalog")
assert catalog.status_code == 200
assert len(catalog.json()["factors"]) == 109
locked = await client.get("/api/screener?date=2026-07-30")
assert locked.status_code == 403
assert locked.json()["error"]["code"] == "membership_required"
rejected = await client.put(
"/api/screener/custom",
headers=csrf_headers(regular),
json={"name": "越权策略", "formula": _formula()},
)
assert rejected.status_code == 403
use_session(client, admin)
workspace = await client.get("/api/screener?date=2026-07-30")
assert workspace.status_code == 200
assert workspace.json()["trade_date"] is None
saved = await client.put(
"/api/screener/custom",
headers=csrf_headers(admin),
json={"name": "我的策略", "formula": _formula()},
)
assert saved.status_code == 200
assert saved.json()["name"] == "我的策略"
refreshed = await client.get("/api/screener?date=2026-07-30")
assert [item["name"] for item in refreshed.json()["custom_strategies"]] == ["我的策略"]
run_scenario(application, scenario)