118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
"""使用线上 PostgreSQL 跑鉴权与核心 API 自检，临时用户会自动清理。
|
||
|
||
用法：uv run --directory backend python smoke_test.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import secrets
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import httpx
|
||
from sqlalchemy import delete, select
|
||
|
||
from app.auth import hash_password, token_digest
|
||
from app.db import async_session, engine
|
||
from app.main import app
|
||
from app.models import AuthSession, MarketDaily, User
|
||
|
||
|
||
async def main() -> None:
|
||
username = f"smoke_{secrets.token_hex(6)}"
|
||
password = secrets.token_urlsafe(24)
|
||
now = datetime.now(timezone.utc)
|
||
|
||
async with async_session() as db:
|
||
db.add(
|
||
User(
|
||
username=username,
|
||
password_hash=hash_password(password),
|
||
password_algo="argon2id",
|
||
is_active=True,
|
||
failed_login_count=0,
|
||
password_changed_at=now,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
try:
|
||
transport = httpx.ASGITransport(app=app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
||
health = await client.get("/api/health")
|
||
protected = await client.post("/api/backtest", json={})
|
||
bad_login = await client.post(
|
||
"/api/auth/login", json={"username": username, "password": "wrong-password"}
|
||
)
|
||
login = await client.post(
|
||
"/api/auth/login", json={"username": username, "password": password}
|
||
)
|
||
print("health:", health.status_code)
|
||
print("protected without login:", protected.status_code)
|
||
print("bad login:", bad_login.status_code)
|
||
print("login:", login.status_code)
|
||
assert health.status_code == 200
|
||
assert protected.status_code == 401
|
||
assert bad_login.status_code == 401
|
||
assert login.status_code == 200
|
||
assert client.cookies.get("stock_session")
|
||
|
||
me = await client.get("/api/auth/me")
|
||
assert me.status_code == 200 and me.json()["username"] == username
|
||
|
||
daily = await client.post(
|
||
"/api/backtest",
|
||
json={
|
||
"symbol": "DEMO",
|
||
"strategy": "macd_cross",
|
||
"params": {"fast": 12, "slow": 26, "signal": 9},
|
||
"initial_cash": 100000.0,
|
||
"fast_mode": False,
|
||
},
|
||
)
|
||
assert daily.status_code == 200, daily.text
|
||
result = daily.json()
|
||
assert len(result["candles"]) > 100
|
||
print("backtest metrics:", json.dumps(result["metrics"], ensure_ascii=False))
|
||
|
||
token = client.cookies.get("stock_session")
|
||
assert token
|
||
async with async_session() as db:
|
||
auth_session = (
|
||
await db.execute(
|
||
select(AuthSession).where(AuthSession.token_hash == token_digest(token))
|
||
)
|
||
).scalar_one()
|
||
auth_session.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
|
||
await db.commit()
|
||
assert (await client.get("/api/auth/me")).status_code == 401
|
||
print("expired session: 401")
|
||
|
||
login = await client.post(
|
||
"/api/auth/login", json={"username": username, "password": password}
|
||
)
|
||
assert login.status_code == 200
|
||
logout = await client.post("/api/auth/logout")
|
||
assert logout.status_code == 204
|
||
assert (await client.get("/api/auth/me")).status_code == 401
|
||
print("logout and revoke: ok")
|
||
|
||
async with async_session() as db:
|
||
has_market_data = bool(await db.scalar(select(MarketDaily.id).limit(1)))
|
||
print("market data present:", has_market_data)
|
||
finally:
|
||
async with async_session() as db:
|
||
user_id = await db.scalar(select(User.id).where(User.username == username))
|
||
if user_id is not None:
|
||
await db.execute(delete(AuthSession).where(AuthSession.user_id == user_id))
|
||
await db.execute(delete(User).where(User.id == user_id))
|
||
await db.commit()
|
||
await engine.dispose()
|
||
print("temporary user cleaned")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|