邹方成
cc352d3184
refactor: 优化API路由和响应模型 feat(admin): 添加App用户管理接口 feat(sms): 实现阿里云短信服务集成 feat(email): 添加SMTP邮件发送功能 feat(upload): 支持文件上传接口 feat(rate-limiter): 实现手机号限流器 fix: 修复计算步骤入库问题 docs: 更新API文档和测试计划 chore: 更新依赖和配置
213 lines
13 KiB
Python
213 lines
13 KiB
Python
import os
|
|
import json
|
|
import time
|
|
import uuid
|
|
import random
|
|
from typing import Dict, Any, List, Optional, Tuple
|
|
|
|
import httpx
|
|
|
|
|
|
def make_url(base_url: str, path: str) -> str:
|
|
if base_url.endswith("/"):
|
|
base_url = base_url[:-1]
|
|
return f"{base_url}{path}"
|
|
|
|
|
|
def now_ms() -> int:
|
|
return int(time.time() * 1000)
|
|
|
|
|
|
def ensure_dict(obj: Any) -> Dict[str, Any]:
|
|
if isinstance(obj, dict):
|
|
return obj
|
|
return {"raw": str(obj)}
|
|
|
|
|
|
async def api_get(client: httpx.AsyncClient, url: str, headers: Optional[Dict[str, str]] = None, params: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]:
|
|
r = await client.get(url, headers=headers or {}, params=params or {})
|
|
try:
|
|
parsed = r.json()
|
|
except Exception:
|
|
parsed = {"raw": r.text}
|
|
return r.status_code, ensure_dict(parsed)
|
|
|
|
|
|
async def api_post_json(client: httpx.AsyncClient, url: str, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> Tuple[int, Dict[str, Any]]:
|
|
r = await client.post(url, json=payload, headers=headers or {})
|
|
try:
|
|
parsed = r.json()
|
|
except Exception:
|
|
parsed = {"raw": r.text}
|
|
return r.status_code, ensure_dict(parsed)
|
|
|
|
|
|
async def api_put_json(client: httpx.AsyncClient, url: str, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> Tuple[int, Dict[str, Any]]:
|
|
r = await client.put(url, json=payload, headers=headers or {})
|
|
try:
|
|
parsed = r.json()
|
|
except Exception:
|
|
parsed = {"raw": r.text}
|
|
return r.status_code, ensure_dict(parsed)
|
|
|
|
|
|
async def api_delete(client: httpx.AsyncClient, url: str, headers: Optional[Dict[str, str]] = None, params: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]:
|
|
r = await client.delete(url, headers=headers or {}, params=params or {})
|
|
try:
|
|
parsed = r.json()
|
|
except Exception:
|
|
parsed = {"raw": r.text}
|
|
return r.status_code, ensure_dict(parsed)
|
|
|
|
|
|
def write_html_report(filepath: str, title: str, results: List[Dict[str, Any]]) -> None:
|
|
rows = []
|
|
for r in results:
|
|
color = {"PASS": "#4caf50", "FAIL": "#f44336"}.get(r.get("status"), "#9e9e9e")
|
|
rows.append(
|
|
f"<tr><td>{r.get('name')}</td><td style='color:{color};font-weight:600'>{r.get('status')}</td><td>{r.get('message','')}</td><td><pre>{json.dumps(r.get('detail', {}), ensure_ascii=False, indent=2)}</pre></td></tr>"
|
|
)
|
|
html = f"""
|
|
<!doctype html>
|
|
<html><head><meta charset='utf-8'><title>{title}</title>
|
|
<style>body{{font-family:Arial;padding:12px}} table{{border-collapse:collapse;width:100%}} td,th{{border:1px solid #ddd;padding:8px}}</style>
|
|
</head><body>
|
|
<h2>{title}</h2>
|
|
<p>生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}</p>
|
|
<table><thead><tr><th>用例</th><th>结果</th><th>说明</th><th>详情</th></tr></thead><tbody>
|
|
{''.join(rows)}
|
|
</tbody></table>
|
|
</body></html>
|
|
"""
|
|
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write(html)
|
|
|
|
|
|
async def test_base(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
for path, name in [("/base/userinfo", "admin用户信息"), ("/base/userapi", "admin接口权限"), ("/base/usermenu", "admin菜单")]:
|
|
code, data = await api_get(client, make_url(base, path), headers={"token": token})
|
|
ok = (code == 200)
|
|
results.append({"name": name, "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
|
|
|
|
async def test_users_crud(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
email = f"admin_{uuid.uuid4().hex[:6]}@test.com"
|
|
username = "adm_" + uuid.uuid4().hex[:6]
|
|
code, data = await api_post_json(client, make_url(base, "/user/create"), {"email": email, "username": username, "password": "123456", "is_active": True, "is_superuser": False, "role_ids": [], "dept_id": 0}, headers={"token": token})
|
|
results.append({"name": "创建用户", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_get(client, make_url(base, "/user/list"), headers={"token": token}, params={"page": 1, "page_size": 10, "email": email})
|
|
ok = (code == 200 and isinstance(data.get("data"), list))
|
|
uid = None
|
|
if ok and data["data"]:
|
|
uid = data["data"][0].get("id")
|
|
results.append({"name": "查询用户", "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
if uid:
|
|
code, data = await api_post_json(client, make_url(base, "/user/update"), {"id": uid, "email": email, "username": username + "_u", "is_active": True, "is_superuser": False, "role_ids": [], "dept_id": 0}, headers={"token": token})
|
|
results.append({"name": "更新用户", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_delete(client, make_url(base, "/user/delete"), headers={"token": token}, params={"user_id": uid})
|
|
results.append({"name": "删除用户", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
|
|
|
|
async def test_roles_menus_apis(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
rname = "role_" + uuid.uuid4().hex[:6]
|
|
code, data = await api_post_json(client, make_url(base, "/role/create"), {"name": rname, "desc": "测试角色"}, headers={"token": token})
|
|
results.append({"name": "创建角色", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_get(client, make_url(base, "/role/list"), headers={"token": token}, params={"page": 1, "page_size": 10, "role_name": rname})
|
|
ok = (code == 200 and isinstance(data.get("data"), list))
|
|
rid = None
|
|
if ok and data["data"]:
|
|
rid = data["data"][0].get("id")
|
|
results.append({"name": "查询角色", "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
code, data = await api_post_json(client, make_url(base, "/api/refresh"), {}, headers={"token": token})
|
|
results.append({"name": "刷新API权限表", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_get(client, make_url(base, "/api/list"), headers={"token": token}, params={"page": 1, "page_size": 10})
|
|
ok_apis = (code == 200 and isinstance(data.get("data"), list))
|
|
results.append({"name": "API列表", "status": "PASS" if ok_apis else "FAIL", "message": "获取成功" if ok_apis else "获取失败", "detail": {"http": code, "body": data}})
|
|
if rid and ok_apis:
|
|
api_infos = []
|
|
if data["data"]:
|
|
first = data["data"][0]
|
|
api_infos = [{"path": first.get("path"), "method": first.get("method")}] if first.get("path") and first.get("method") else []
|
|
code, data = await api_post_json(client, make_url(base, "/role/authorized"), {"id": rid, "menu_ids": [], "api_infos": api_infos}, headers={"token": token})
|
|
results.append({"name": "角色授权", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_delete(client, make_url(base, "/role/delete"), headers={"token": token}, params={"role_id": rid})
|
|
results.append({"name": "删除角色", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
|
|
|
|
async def test_dept_crud(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
dname = "dept_" + uuid.uuid4().hex[:6]
|
|
code, data = await api_post_json(client, make_url(base, "/dept/create"), {"name": dname, "desc": "测试部门"}, headers={"token": token})
|
|
results.append({"name": "创建部门", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_get(client, make_url(base, "/dept/list"), headers={"token": token}, params={"page": 1, "page_size": 10})
|
|
ok = (code == 200 and isinstance(data.get("data"), list))
|
|
results.append({"name": "查询部门", "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
|
|
|
|
async def test_valuations_admin(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
payload = {"asset_name": "Admin资产", "institution": "Admin机构", "industry": "行业", "three_year_income": [10, 20, 30]}
|
|
code, data = await api_post_json(client, make_url(base, "/valuations/"), payload, headers={"token": token})
|
|
results.append({"name": "创建估值(管理员)", "status": "PASS" if code == 200 and data.get("code") == 200 else "FAIL", "message": data.get("msg"), "detail": {"http": code, "body": data}})
|
|
code, data = await api_get(client, make_url(base, "/valuations/"), headers={"token": token}, params={"page": 1, "size": 5})
|
|
ok = (code == 200 and isinstance(data.get("data"), list))
|
|
results.append({"name": "估值列表(管理员)", "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
|
|
|
|
async def test_invoice_transactions(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
code, data = await api_get(client, make_url(base, "/invoice/list"), headers={"token": token}, params={"page": 1, "page_size": 10})
|
|
ok = (code == 200 and isinstance(data.get("data"), list))
|
|
results.append({"name": "发票列表", "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
code, data = await api_get(client, make_url(base, "/transactions/receipts"), headers={"token": token}, params={"page": 1, "page_size": 10})
|
|
ok = (code == 200 and isinstance(data.get("data"), list))
|
|
results.append({"name": "对公转账列表", "status": "PASS" if ok else "FAIL", "message": "获取成功" if ok else "获取失败", "detail": {"http": code, "body": data}})
|
|
|
|
|
|
async def perf_benchmark(client: httpx.AsyncClient, base: str, token: str, results: List[Dict[str, Any]]):
|
|
endpoints = ["/user/list", "/valuations/", "/invoice/list"]
|
|
conc = 20
|
|
metrics = []
|
|
for ep in endpoints:
|
|
start = now_ms()
|
|
tasks = [api_get(client, make_url(base, ep), headers={"token": token}, params={"page": 1, "page_size": 10}) for _ in range(conc)]
|
|
rets = await httpx.AsyncClient.gather(*tasks) if hasattr(httpx.AsyncClient, "gather") else None
|
|
# 兼容：无 gather 则顺序执行
|
|
if rets is None:
|
|
rets = []
|
|
for _ in range(conc):
|
|
rets.append(await api_get(client, make_url(base, ep), headers={"token": token}, params={"page": 1, "page_size": 10}))
|
|
dur = now_ms() - start
|
|
ok = sum(1 for (code, _) in rets if code == 200)
|
|
metrics.append({"endpoint": ep, "concurrency": conc, "duration_ms": dur, "success": ok, "total": conc})
|
|
results.append({"name": "性能基准", "status": "PASS", "message": "并发测试完成", "detail": {"metrics": metrics}})
|
|
|
|
|
|
async def main() -> None:
|
|
base = os.getenv("ADMIN_BASE_URL", "http://localhost:9999/api/v1")
|
|
token = os.getenv("ADMIN_TOKEN", "dev")
|
|
results: List[Dict[str, Any]] = []
|
|
endpoint_list = [
|
|
{"path": "/base/userinfo", "desc": "管理员信息"},
|
|
{"path": "/user/*", "desc": "用户管理"},
|
|
{"path": "/role/*", "desc": "角色管理与授权"},
|
|
{"path": "/api/*", "desc": "API权限管理与刷新"},
|
|
{"path": "/dept/*", "desc": "部门管理"},
|
|
{"path": "/valuations/*", "desc": "估值评估管理"},
|
|
{"path": "/invoice/*", "desc": "发票与抬头"},
|
|
{"path": "/transactions/*", "desc": "对公转账记录"},
|
|
]
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
await test_base(client, base, token, results)
|
|
await test_users_crud(client, base, token, results)
|
|
await test_roles_menus_apis(client, base, token, results)
|
|
await test_dept_crud(client, base, token, results)
|
|
await test_valuations_admin(client, base, token, results)
|
|
await test_invoice_transactions(client, base, token, results)
|
|
await perf_benchmark(client, base, token, results)
|
|
passes = sum(1 for r in results if r.get("status") == "PASS")
|
|
print(json.dumps({"total": len(results), "passes": passes, "results": results, "endpoints": endpoint_list}, ensure_ascii=False, indent=2))
|
|
write_html_report("reports/admin_flow_script_report.html", "后台管理员维度接口全流程测试报告", results)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import asyncio
|
|
asyncio.run(main()) |