233 lines
11 KiB
Python
233 lines
11 KiB
Python
"""API 全链路测试：认证、平台/账号 CRUD、设置。"""
|
|
|
|
import base64
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api import create_app
|
|
from app.config import Config
|
|
|
|
|
|
@pytest.fixture
|
|
def client(test_db, tmp_path, monkeypatch):
|
|
from app import config as cfgmod
|
|
|
|
monkeypatch.setattr(cfgmod, "CONFIG_PATH", tmp_path / "config.json")
|
|
cfg = Config(
|
|
port=8000, password="admin123", global_interval_seconds=60,
|
|
retry_count=2, timeout_seconds=10, max_workers=4,
|
|
)
|
|
app = create_app(cfg)
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
def _auth(client):
|
|
return {"Authorization": "Bearer " + client.post("/api/login", json={"password": "admin123"}).json()["token"]}
|
|
|
|
|
|
class TestAuth:
|
|
def test_unauthorized_blocked(self, client):
|
|
assert client.get("/api/platforms").status_code == 401
|
|
|
|
def test_wrong_password(self, client):
|
|
assert client.post("/api/login", json={"password": "nope"}).status_code == 401
|
|
|
|
def test_login_ok(self, client):
|
|
resp = client.post("/api/login", json={"password": "admin123"})
|
|
assert resp.status_code == 200 and resp.json()["token"]
|
|
|
|
def test_logout_invalidates(self, client):
|
|
token = _auth(client)["Authorization"].removeprefix("Bearer ")
|
|
assert client.post("/api/logout", headers={"Authorization": f"Bearer {token}"}).status_code == 200
|
|
assert client.get("/api/platforms", headers={"Authorization": f"Bearer {token}"}).status_code == 401
|
|
|
|
|
|
PLATFORM_PAYLOAD = {
|
|
"provider_id": "deepseek", "name": "DeepSeek-Test",
|
|
"interval_seconds": 120, "retry_count": 1, "timeout_seconds": 15,
|
|
"enabled": True, "note": "",
|
|
}
|
|
|
|
|
|
class TestPlatforms:
|
|
def test_builtin_platforms_auto_synced(self, client):
|
|
"""代码内置 provider 启动时自动同步为平台记录，无需手动添加。"""
|
|
h = _auth(client)
|
|
provs = client.get("/api/providers", headers=h).json()
|
|
plats = client.get("/api/platforms", headers=h).json()
|
|
for p in provs:
|
|
assert any(x["provider_id"] == p["id"] for x in plats), p["id"]
|
|
dp = next(x for x in plats if x["provider_id"] == "deepseek")
|
|
assert dp["name"] == "DeepSeek" and dp["currency"] == "CNY"
|
|
|
|
def test_crud_flow(self, client):
|
|
h = _auth(client)
|
|
pid = client.post("/api/platforms", json=PLATFORM_PAYLOAD, headers=h).json()["id"]
|
|
|
|
lst = client.get("/api/platforms", headers=h).json()
|
|
created = next(p for p in lst if p["id"] == pid)
|
|
assert created["provider_id"] == "deepseek"
|
|
assert created["currency"] == "CNY" # 默认取 provider 的货币
|
|
|
|
upd = client.put(f"/api/platforms/{pid}", json={"currency": "USD", "interval_seconds": 300}, headers=h)
|
|
assert upd.status_code == 200
|
|
assert next(p for p in client.get("/api/platforms", headers=h).json() if p["id"] == pid)["currency"] == "USD"
|
|
|
|
assert client.delete(f"/api/platforms/{pid}", headers=h).status_code == 200
|
|
assert all(p["id"] != pid for p in client.get("/api/platforms", headers=h).json())
|
|
|
|
def test_duplicate_name_409(self, client):
|
|
h = _auth(client)
|
|
client.post("/api/platforms", json=PLATFORM_PAYLOAD, headers=h)
|
|
assert client.post("/api/platforms", json=PLATFORM_PAYLOAD, headers=h).status_code == 409
|
|
|
|
def test_unknown_provider_400(self, client):
|
|
h = _auth(client)
|
|
assert client.post("/api/platforms", json=dict(PLATFORM_PAYLOAD, provider_id="nope"), headers=h).status_code == 400
|
|
|
|
def test_validation_errors(self, client):
|
|
h = _auth(client)
|
|
assert client.post("/api/platforms", json=dict(PLATFORM_PAYLOAD, name=""), headers=h).status_code == 422
|
|
assert client.post("/api/platforms", json=dict(PLATFORM_PAYLOAD, interval_seconds=5), headers=h).status_code == 422
|
|
|
|
def test_providers_list(self, client):
|
|
h = _auth(client)
|
|
provs = client.get("/api/providers", headers=h).json()
|
|
assert any(p["id"] == "deepseek" and p["currency"] == "CNY" for p in provs)
|
|
|
|
def test_delete_cascades_accounts(self, client, test_db):
|
|
from app import db
|
|
|
|
h = _auth(client)
|
|
pid = client.post("/api/platforms", json=PLATFORM_PAYLOAD, headers=h).json()["id"]
|
|
client.post("/api/accounts", json={"platform_id": pid, "name": "a1", "api_key": "sk-1"}, headers=h)
|
|
client.delete(f"/api/platforms/{pid}", headers=h)
|
|
with db.get_conn() as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM accounts").fetchone()[0] == 0
|
|
|
|
|
|
class TestAccounts:
|
|
def _make_platform(self, client, h):
|
|
return client.post("/api/platforms", json=PLATFORM_PAYLOAD, headers=h).json()["id"]
|
|
|
|
def test_create_returns_decoded_key(self, client):
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
aid = client.post("/api/accounts", json={"platform_id": pid, "name": "主", "api_key": "sk-secret", "threshold": 10}, headers=h).json()["id"]
|
|
acc = client.get("/api/accounts", headers=h).json()[0]
|
|
assert acc["id"] == aid and acc["api_key"] == "sk-secret"
|
|
assert acc["platform_name"] == "DeepSeek-Test" and acc["currency"] == "CNY"
|
|
|
|
def test_key_stored_base64(self, client, test_db):
|
|
from app import db
|
|
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
client.post("/api/accounts", json={"platform_id": pid, "name": "主", "api_key": "sk-secret"}, headers=h)
|
|
with db.get_conn() as conn:
|
|
raw = conn.execute("SELECT api_key FROM accounts").fetchone()[0]
|
|
assert raw == base64.b64encode(b"sk-secret").decode()
|
|
|
|
def test_update_and_delete(self, client):
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
aid = client.post("/api/accounts", json={"platform_id": pid, "name": "主", "api_key": "sk-1"}, headers=h).json()["id"]
|
|
assert client.put(f"/api/accounts/{aid}", json={"threshold": 5.5, "name": "新名"}, headers=h).status_code == 200
|
|
acc = client.get("/api/accounts", headers=h).json()[0]
|
|
assert acc["name"] == "新名" and acc["threshold"] == 5.5
|
|
assert client.delete(f"/api/accounts/{aid}", headers=h).status_code == 200
|
|
assert client.get("/api/accounts", headers=h).json() == []
|
|
|
|
def test_account_requires_platform(self, client):
|
|
h = _auth(client)
|
|
assert client.post("/api/accounts", json={"platform_id": 999, "name": "x", "api_key": "k"}, headers=h).status_code == 404
|
|
|
|
def test_check_now_disabled_account(self, client):
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
aid = client.post("/api/accounts", json={"platform_id": pid, "name": "x", "api_key": "k", "enabled": False}, headers=h).json()["id"]
|
|
assert client.post(f"/api/accounts/{aid}/check", headers=h).status_code == 409
|
|
|
|
def test_history_flow(self, client, test_db):
|
|
from app import db
|
|
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
aid = client.post("/api/accounts", json={"platform_id": pid, "name": "x", "api_key": "k"}, headers=h).json()["id"]
|
|
with db.get_conn() as conn:
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (aid, 1.5))
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (aid, 2.5))
|
|
hist = client.get(f"/api/accounts/{aid}/history", headers=h).json()
|
|
assert [x["balance"] for x in hist] == [1.5, 2.5]
|
|
|
|
def test_batch_history(self, client, test_db):
|
|
from app import db
|
|
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
a1 = client.post("/api/accounts", json={"platform_id": pid, "name": "x", "api_key": "k"}, headers=h).json()["id"]
|
|
a2 = client.post("/api/accounts", json={"platform_id": pid, "name": "y", "api_key": "k2"}, headers=h).json()["id"]
|
|
with db.get_conn() as conn:
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (a1, 1.0))
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (a1, 2.0))
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (a2, 9.0))
|
|
data = client.get("/api/history?limit=30", headers=h).json()
|
|
assert [x["balance"] for x in data[str(a1)]] == [1.0, 2.0]
|
|
assert [x["balance"] for x in data[str(a2)]] == [9.0]
|
|
assert len(data) == 2
|
|
|
|
def test_history_window_filter(self, client, test_db):
|
|
from app import db
|
|
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
aid = client.post("/api/accounts", json={"platform_id": pid, "name": "x", "api_key": "k"}, headers=h).json()["id"]
|
|
with db.get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO balance_history (account_id, balance, checked_at) VALUES (?, ?, datetime('now','localtime','-30 days'))",
|
|
(aid, 100.0),
|
|
)
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (aid, 1.0))
|
|
all_rows = client.get(f"/api/accounts/{aid}/history", headers=h).json()
|
|
assert len(all_rows) == 2
|
|
day_rows = client.get(f"/api/accounts/{aid}/history?window=1d", headers=h).json()
|
|
assert len(day_rows) == 1 and day_rows[0]["balance"] == 1.0
|
|
yr_rows = client.get(f"/api/accounts/{aid}/history?window=1yr", headers=h).json()
|
|
assert len(yr_rows) == 2
|
|
|
|
def test_history_downsample(self, client, test_db):
|
|
from app import db
|
|
|
|
h = _auth(client)
|
|
pid = self._make_platform(client, h)
|
|
aid = client.post("/api/accounts", json={"platform_id": pid, "name": "x", "api_key": "k"}, headers=h).json()["id"]
|
|
with db.get_conn() as conn:
|
|
for i in range(50):
|
|
conn.execute("INSERT INTO balance_history (account_id, balance) VALUES (?, ?)", (aid, i))
|
|
rows = client.get(f"/api/accounts/{aid}/history?limit=10", headers=h).json()
|
|
assert len(rows) == 10
|
|
assert rows[0]["balance"] == 0 and rows[-1]["balance"] == 49 # 首尾保留
|
|
# 非法窗口参数被拒
|
|
assert client.get(f"/api/accounts/{aid}/history?window=2h", headers=h).status_code == 422
|
|
|
|
|
|
class TestSettings:
|
|
def test_update_settings(self, client, tmp_path):
|
|
h = _auth(client)
|
|
resp = client.put("/api/settings", json={"global_interval_seconds": 120, "telegram_bot_token": "t:1"}, headers=h)
|
|
assert resp.status_code == 200
|
|
cfg = client.get("/api/settings", headers=h).json()
|
|
assert cfg["global_interval_seconds"] == 120 and cfg["telegram_bot_token"] == "t:1"
|
|
# 写回 config.json 且可重新加载
|
|
from app.config import load_config
|
|
assert load_config().global_interval_seconds == 120
|
|
|
|
def test_change_password(self, client):
|
|
h = _auth(client)
|
|
assert client.post("/api/settings/password", json={"old_password": "wrong", "new_password": "new123"}, headers=h).status_code == 401
|
|
assert client.post("/api/settings/password", json={"old_password": "admin123", "new_password": "new123"}, headers=h).status_code == 200
|
|
assert client.post("/api/login", json={"password": "admin123"}).status_code == 401
|
|
assert client.post("/api/login", json={"password": "new123"}).status_code == 200
|