124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.db import auth_db
|
|
from app.db.operations_dashboard import _display_error_summary, get_operations_dashboard
|
|
from app.web import web_server
|
|
|
|
|
|
def _login_user(email: str, admin: bool = False) -> str:
|
|
reg = auth_db.register_user(email, "StrongPass123")
|
|
auth_db.verify_email(email, reg["verification_code"])
|
|
user = auth_db.get_user_by_email(email)
|
|
auth_db.claim_free_trial(user["id"])
|
|
if admin:
|
|
auth_db.set_user_admin(email, True)
|
|
return auth_db.login_user(email, "StrongPass123")["token"]
|
|
|
|
|
|
def test_operations_page_and_api_require_admin():
|
|
token = _login_user("normal-ops@example.com")
|
|
client = TestClient(web_server.app)
|
|
client.cookies.set("altcoin_session", token)
|
|
|
|
page = client.get("/operations")
|
|
api = client.get("/api/admin/operations-dashboard")
|
|
|
|
assert page.status_code == 403
|
|
assert api.status_code == 403
|
|
|
|
|
|
def test_operations_admin_can_access_dashboard():
|
|
token = _login_user("admin-ops@example.com", admin=True)
|
|
client = TestClient(web_server.app)
|
|
client.cookies.set("altcoin_session", token)
|
|
|
|
page = client.get("/operations")
|
|
api = client.get("/api/admin/operations-dashboard?hours=24")
|
|
|
|
assert page.status_code == 200
|
|
assert "运行大屏" in page.text
|
|
assert api.status_code == 200
|
|
data = api.json()
|
|
assert "overall" in data
|
|
assert "scheduler" in data
|
|
assert "data_sources" in data
|
|
assert "funnel" in data
|
|
assert "trading" in data
|
|
assert "timeline" in data
|
|
|
|
|
|
def test_operations_dashboard_read_model_shape(pg_conn):
|
|
data = get_operations_dashboard(hours=24)
|
|
|
|
assert data["hours"] == 24
|
|
assert data["overall"]["status"] in {"ok", "warn", "danger"}
|
|
assert isinstance(data["scheduler"], list)
|
|
assert isinstance(data["data_sources"], list)
|
|
assert isinstance(data["funnel"], list)
|
|
assert isinstance(data["trading"], dict)
|
|
|
|
|
|
def test_operations_dashboard_hides_retired_onchain_runtime_data(pg_conn):
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO scheduler_job_config (job_name, command, every_seconds, enabled, lock_group, created_at, updated_at)
|
|
VALUES ('onchain', 'python -m app.cli onchain', 60, 1, 'onchain', NOW(), NOW())
|
|
ON CONFLICT(job_name) DO UPDATE SET enabled=1, updated_at=NOW()
|
|
"""
|
|
)
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO cron_run_log (job_name, script_name, run_status, result_status, started_at, finished_at, duration_ms)
|
|
VALUES ('onchain', 'onchain', 'success', 'no_onchain_data', NOW(), NOW(), 100)
|
|
"""
|
|
)
|
|
pg_conn.commit()
|
|
|
|
data = get_operations_dashboard(hours=24)
|
|
|
|
assert all(x.get("job_name") != "onchain" for x in data["scheduler"])
|
|
assert all("onchain" not in str(x.get("title") or "").lower() for x in data["timeline"])
|
|
|
|
|
|
def test_operations_dashboard_dedupes_repeated_gate_reject_timeline(pg_conn):
|
|
detail = {
|
|
"reason": "paper_order_gate_rejected",
|
|
"gate_reasons": ["rec_score_below_min"],
|
|
"strategy_code": "short_breakdown_retest_1h_v1",
|
|
}
|
|
for _ in range(3):
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO paper_trade_events (
|
|
trade_id, recommendation_id, symbol, event_type, event_time,
|
|
message, detail_json, strategy_code
|
|
)
|
|
VALUES (0, 9001, 'ONDO/USDT', 'paper_gate_reject', NOW(),
|
|
'日内策略挂单门禁拒绝', %s, 'short_breakdown_retest_1h_v1')
|
|
""",
|
|
(json.dumps(detail),),
|
|
)
|
|
pg_conn.commit()
|
|
|
|
data = get_operations_dashboard(hours=24)
|
|
gate_rows = [
|
|
x for x in data["timeline"]
|
|
if "ONDO/USDT" in str(x.get("title") or "") and "paper_gate_reject" in str(x.get("title") or "")
|
|
]
|
|
|
|
assert len(gate_rows) == 1
|
|
assert gate_rows[0]["status"] == "warn"
|
|
|
|
|
|
def test_operations_dashboard_sanitizes_external_provider_errors():
|
|
summary = _display_error_summary(
|
|
"market:HTTPSConnectionPool(host='api.binance.com', port=443): "
|
|
"Max retries exceeded with url: /v1/secret (Caused by NameResolutionError())"
|
|
)
|
|
|
|
assert summary == "外部服务连接异常"
|
|
assert "HTTPSConnectionPool" not in summary
|
|
assert "/v1/" not in summary
|