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"| {r.get('name')} | {r.get('status')} | {r.get('message','')} | {json.dumps(r.get('detail', {}), ensure_ascii=False, indent=2)} |
"
)
html = f"""
{title}
{title}
生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}
"""
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())