46 lines
2.6 KiB
Python
46 lines
2.6 KiB
Python
|
"""Submission limits filter cached unsubmitted Alpha snapshots before pagination."""
|
||
|
|
|
||
|
|
import csv
|
||
|
|
import io
|
||
|
|
|
||
|
|
from app.alphas import snapshot_columns, upsert_alpha
|
||
|
|
from tests.conftest import alpha
|
||
|
|
|
||
|
|
|
||
|
|
async def test_blocked_filter_intersects_filters_and_export_and_refresh(app, logged_in):
|
||
|
|
async with app.state.sessions.begin() as db:
|
||
|
|
for name, result, status, region in [
|
||
|
|
("blocked1", "FAIL", "UNSUBMITTED", "USA"),
|
||
|
|
("blocked2", "fail", "UNSUBMITTED", "USA"),
|
||
|
|
("other_region", "FAIL", "UNSUBMITTED", "CHN"),
|
||
|
|
("submitted", "FAIL", "ACTIVE", "USA"),
|
||
|
|
("passed", "PASS", "UNSUBMITTED", "USA"),
|
||
|
|
("pending", "PENDING", "UNSUBMITTED", "USA"),
|
||
|
|
("missing", None, "UNSUBMITTED", "USA"),
|
||
|
|
]:
|
||
|
|
checks = [{"name": "LOW_SHARPE", "result": "PASS"}]
|
||
|
|
if result:
|
||
|
|
checks.append({"name": "REGULAR_SUBMISSION", "result": result})
|
||
|
|
await upsert_alpha(db, alpha(name, status=status, settings={"region": region}, **{"is": {"checks": checks}}))
|
||
|
|
query = "submission_blocked=true®ion=USA&check_type=PRE_CHECK&sort=id&direction=asc&limit=1&offset=1"
|
||
|
|
response = await logged_in.get(f"/api/v1/alphas?{query}")
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json()["total"] == 2
|
||
|
|
assert [row["id"] for row in response.json()["items"]] == ["blocked2"]
|
||
|
|
export = await logged_in.get(f"/api/v1/alphas/export?{query}")
|
||
|
|
assert [row["id"] for row in csv.DictReader(io.StringIO(export.text.lstrip("\ufeff")))] == ["blocked1", "blocked2"]
|
||
|
|
assert (await logged_in.get("/api/v1/alphas?submission_blocked=true&submission=SUBMITTED")).json()["total"] == 0
|
||
|
|
assert (await logged_in.get("/api/v1/alphas?submission_blocked=true&q=blocked1")).json()["total"] == 1
|
||
|
|
async with app.state.sessions.begin() as db:
|
||
|
|
await upsert_alpha(db, alpha("blocked1", **{"is": {"checks": [{"name": "REGULAR_SUBMISSION", "result": "PASS"}]}}))
|
||
|
|
assert (await logged_in.get("/api/v1/alphas?submission_blocked=true&q=blocked1")).json()["total"] == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_explicit_check_refreshes_limit_independently_of_quality():
|
||
|
|
for checked in (False, True):
|
||
|
|
for result, blocked in [("FAIL", True), ("PASS", False), ("PENDING", False)]:
|
||
|
|
columns = snapshot_columns({}, {}, [{"name": "LOW_SHARPE", "result": "PASS"},
|
||
|
|
{"name": "REGULAR_SUBMISSION", "result": result}], checked=checked)
|
||
|
|
assert columns["submission_blocked"] is blocked
|
||
|
|
assert columns["check_type"] == ("PASS" if checked else "PRE_CHECK")
|