68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
import io
|
|
import json
|
|
import zipfile
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.db import altcoin_db, auth_db
|
|
from app.db.data_export import build_data_export_bundle
|
|
from app.web import web_server
|
|
|
|
|
|
def _login_admin(email: str = "admin-export@example.com") -> str:
|
|
reg = auth_db.register_user(email, "StrongPass123")
|
|
auth_db.verify_email(email, reg["verification_code"])
|
|
auth_db.claim_free_trial(auth_db.get_user_by_email(email)["id"])
|
|
auth_db.set_user_admin(email, True)
|
|
return auth_db.login_user(email, "StrongPass123")["token"]
|
|
|
|
|
|
def test_build_data_export_bundle_contains_manifest_and_recent_tables():
|
|
altcoin_db.create_recommendation(
|
|
symbol="EXPORT/USDT",
|
|
rec_state="爆发",
|
|
rec_score=30,
|
|
entry_price=100,
|
|
stop_loss=95,
|
|
tp1=110,
|
|
signals=["导出测试"],
|
|
entry_plan={"entry_action": "可即刻买入", "entry_price": 100},
|
|
)
|
|
|
|
filename, content, manifest = build_data_export_bundle(hours=24)
|
|
|
|
assert filename.endswith("_24h.zip")
|
|
assert manifest["window_hours"] == 24
|
|
with zipfile.ZipFile(io.BytesIO(content)) as zf:
|
|
names = set(zf.namelist())
|
|
assert "manifest.json" in names
|
|
assert "tables/recommendation.json" in names
|
|
assert "tables/paper_trades.json" in names
|
|
loaded_manifest = json.loads(zf.read("manifest.json").decode("utf-8"))
|
|
rows = json.loads(zf.read("tables/recommendation.json").decode("utf-8"))
|
|
|
|
assert loaded_manifest["tables"]["recommendation"]["rows"] >= 1
|
|
assert any(row["symbol"] == "EXPORT/USDT" for row in rows)
|
|
|
|
|
|
def test_admin_data_export_api_downloads_zip():
|
|
token = _login_admin()
|
|
client = TestClient(web_server.app)
|
|
client.cookies.set("altcoin_session", token)
|
|
|
|
resp = client.get("/api/admin/data-export?hours=24")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"] == "application/zip"
|
|
assert "attachment" in resp.headers["content-disposition"]
|
|
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
|
|
assert "manifest.json" in zf.namelist()
|
|
|
|
|
|
def test_data_export_page_requires_admin():
|
|
client = TestClient(web_server.app)
|
|
resp = client.get("/data-export")
|
|
|
|
assert resp.status_code == 200
|
|
assert "登录" in resp.text or "会员" in resp.text
|