148 lines
5.5 KiB
Python
148 lines
5.5 KiB
Python
"""
|
|
本地 mqsrv (http://localhost:3001) 接口测试脚本
|
|
|
|
用法:
|
|
python test_local.py
|
|
python test_local.py --user sixteenth --pass "@WT65eijh10"
|
|
"""
|
|
import json
|
|
import sys
|
|
import argparse
|
|
from datetime import datetime
|
|
import requests
|
|
|
|
LOCAL_BASE = "http://localhost:3001"
|
|
|
|
# 复用 origet 的 AES 解密
|
|
import os, sys as _sys
|
|
_origet_dir = os.path.dirname(os.path.abspath(__file__))
|
|
if _origet_dir not in _sys.path:
|
|
_sys.path.insert(0, _origet_dir)
|
|
from decrypt import try_decrypt
|
|
|
|
|
|
def test(name, fn, *args, **kwargs):
|
|
print(f"\n--- {name} ---")
|
|
try:
|
|
result = fn(*args, **kwargs)
|
|
decrypted = try_decrypt(result)
|
|
if isinstance(decrypted, dict):
|
|
keys = list(decrypted.keys())
|
|
print(f" OK keys: {keys}")
|
|
if "data" in decrypted and isinstance(decrypted["data"], list):
|
|
print(f" data_len={len(decrypted['data'])}")
|
|
if decrypted["data"]:
|
|
print(f" first: {json.dumps(decrypted['data'][0], ensure_ascii=False)[:300]}")
|
|
if "pagination" in decrypted:
|
|
print(f" pagination: {decrypted['pagination']}")
|
|
elif isinstance(decrypted, list):
|
|
print(f" OK list_len={len(decrypted)}")
|
|
if decrypted:
|
|
print(f" first: {json.dumps(decrypted[0], ensure_ascii=False)[:200]}")
|
|
return decrypted
|
|
except requests.exceptions.ConnectionError:
|
|
print(f" [ERROR] 连接失败 - 请确保服务端已启动 (npm run dev)")
|
|
return None
|
|
except Exception as e:
|
|
print(f" [ERROR] {e}")
|
|
return None
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--user", default="sixteenth")
|
|
parser.add_argument("--pass", dest="password", default="")
|
|
args = parser.parse_args()
|
|
|
|
s = requests.Session()
|
|
|
|
def get(path, params=None, auth=False, token=None):
|
|
h = {}
|
|
if auth and token:
|
|
h["Authorization"] = f"Bearer {token}"
|
|
r = s.get(f"{LOCAL_BASE}{path}", params=params, headers=h, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def post(path, body, auth=False, token=None, extra_headers=None):
|
|
h = {"Content-Type": "application/json"}
|
|
if auth and token:
|
|
h["Authorization"] = f"Bearer {token}"
|
|
if extra_headers:
|
|
h.update(extra_headers)
|
|
r = s.post(f"{LOCAL_BASE}{path}", json=body, headers=h, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def delete(path, token=None):
|
|
h = {"Authorization": f"Bearer {token}"} if token else {}
|
|
r = s.delete(f"{LOCAL_BASE}{path}", headers=h, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
print(f"[{datetime.now().isoformat()}] 测试 {LOCAL_BASE}")
|
|
|
|
# 1. 公开接口
|
|
test("武器分类", get, "/api/weapon-categories")
|
|
test("武器列表 (全部)", get, "/api/weapons")
|
|
test("武器列表 (突击步枪)", get, "/api/weapons", params={"category": "突击步枪"})
|
|
test("方案列表 (schemes)", get, "/api/schemes", params={"sort": "hot", "page": "1", "limit": "6"})
|
|
test("方案列表 (schemes_aob)", get, "/api/schemes_aob", params={"sort": "hot", "page": "1", "limit": "6"})
|
|
test("广告列表", get, "/api/adverts/list")
|
|
|
|
# 2. 注册 + 登录
|
|
if args.password:
|
|
test("注册 (可能已存在)", post, "/api/register", body={
|
|
"username": args.user, "password": args.password, "email": f"{args.user}@test.com"
|
|
})
|
|
login_resp = test("登录", post, "/api/login", body={
|
|
"username": args.user, "password": args.password
|
|
})
|
|
token = login_resp.get("token") if login_resp else None
|
|
|
|
if token:
|
|
print(f"\n Token: {token[:50]}...")
|
|
|
|
# 3. 认证接口
|
|
test("会话状态", get, "/api/session-status", auth=True, token=token)
|
|
test("VIP状态", get, "/api/vip-status", auth=True, token=token)
|
|
|
|
# 4. 创建方案
|
|
scheme_resp = test("创建方案", post, "/api/schemes", body={
|
|
"description": "测试M4A1方案",
|
|
"category": "突击步枪",
|
|
"weaponName": "M4A1突击步枪",
|
|
"scheme": "M4A1突击步枪-烽火地带-TESTCODE123",
|
|
"price": 30,
|
|
}, auth=True, token=token,
|
|
extra_headers={
|
|
"x-user-info": "eyJpZCI6NDc5MTcsInVzZXJuYW1lIjoic2l4dGVlbnRoIn0=" # test userinfo base64
|
|
})
|
|
scheme_id = scheme_resp.get("data", {}).get("id") if scheme_resp else None
|
|
|
|
if scheme_id:
|
|
# 5. 收藏
|
|
test("添加收藏", post, "/api/favorites", body={
|
|
"schemeId": scheme_id, "source": "烽火地带"
|
|
}, auth=True, token=token)
|
|
test("检查收藏", get, "/api/favorites/check", params={
|
|
"schemeId": scheme_id, "source": "烽火地带"
|
|
}, auth=True, token=token)
|
|
test("收藏列表", get, "/api/favorites", auth=True, token=token)
|
|
test("收藏数量", get, "/api/favorites/count")
|
|
test("取消收藏", delete, f"/api/favorites/{scheme_id}?source=烽火地带", token=token)
|
|
|
|
# 6. 使用记录
|
|
test("记录使用", post, f"/api/schemes/{scheme_id}/use", body={}, token=token)
|
|
else:
|
|
print("\n 登录失败（可能未注册）")
|
|
else:
|
|
print("\n 跳过登录测试 (未提供 --pass)")
|
|
|
|
print(f"\n{'='*60}")
|
|
print(" DONE")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|