import os
import sys
import json
import time
import uuid
import random
from typing import Dict, Any, List, Tuple, Optional
import httpx
def now_ms() -> int:
return int(time.time() * 1000)
def make_url(base_url: str, path: str) -> str:
if base_url.endswith("/"):
base_url = base_url[:-1]
return f"{base_url}{path}"
def write_html_report(filepath: str, title: str, results: List[Dict[str, Any]]) -> None:
"""
生成HTML测试报告
参数:
filepath: 报告输出文件路径
title: 报告标题
results: 测试结果列表，包含 name/status/message/detail
返回:
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)
def _ensure_dict(obj: Any) -> Dict[str, Any]:
if isinstance(obj, dict):
return obj
return {"raw": str(obj)}
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]]:
"""
发送POST JSON请求
参数:
client: httpx异步客户端
url: 完整URL
payload: 请求体JSON
headers: 请求头
返回:
(状态码, 响应JSON)
"""
r = await client.post(url, json=payload, headers=headers or {})
try:
parsed = r.json()
except Exception:
parsed = {"raw": r.text}
if parsed is None:
parsed = {"raw": r.text}
return r.status_code, _ensure_dict(parsed)
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]]:
"""
发送GET请求
参数:
client: httpx异步客户端
url: 完整URL
headers: 请求头
params: 查询参数
返回:
(状态码, 响应JSON)
"""
r = await client.get(url, headers=headers or {}, params=params or {})
try:
parsed = r.json()
except Exception:
parsed = {"raw": r.text}
if parsed is None:
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]]:
"""
发送PUT JSON请求
参数:
client: httpx异步客户端
url: 完整URL
payload: 请求体JSON
headers: 请求头
返回:
(状态码, 响应JSON)
"""
r = await client.put(url, json=payload, headers=headers or {})
try:
parsed = r.json()
except Exception:
parsed = {"raw": r.text}
if parsed is None:
parsed = {"raw": r.text}
return r.status_code, _ensure_dict(parsed)
async def user_register_flow(base_url: str, client: httpx.AsyncClient, phone: str, expect_success: bool = True) -> Dict[str, Any]:
"""
用户注册流程
参数:
base_url: 基础URL(含 /api/v1)
client: httpx客户端
phone: 手机号
返回:
测试结果字典
"""
url = make_url(base_url, "/app-user/register")
code, data = await api_post_json(client, url, {"phone": phone})
rs = {"name": f"注册-{phone}", "status": "FAIL", "message": "", "detail": {"http": code, "body": _ensure_dict(data)}}
body = _ensure_dict(data)
payload = _ensure_dict(body.get("data"))
ok = (body.get("code") == 200 and payload.get("phone") == phone)
# 期望失败场景：重复注册或无效格式
if not expect_success:
ok = (body.get("code") in (400, 422) or (isinstance(body.get("msg"), str) and "已存在" in body.get("msg")))
rs["message"] = "注册失败(符合预期)" if ok else "注册失败(不符合预期)"
else:
rs["message"] = "注册成功" if ok else "注册失败"
rs["status"] = "PASS" if ok else "FAIL"
return rs
async def user_login_flow(base_url: str, client: httpx.AsyncClient, phone: str, password: str, expect_success: bool = True) -> Tuple[Dict[str, Any], str]:
"""
用户登录流程
参数:
base_url: 基础URL(含 /api/v1)
client: httpx客户端
phone: 手机号
password: 密码
返回:
(测试结果字典, access_token字符串或空)
"""
url = make_url(base_url, "/app-user/login")
code, data = await api_post_json(client, url, {"phone": phone, "password": password})
token = ""
is_ok = (code == 200 and isinstance(data, dict) and data.get("access_token"))
if is_ok:
token = data.get("access_token", "")
if not expect_success:
ok = (code in (401, 403))
rs = {"name": f"登录-{phone}", "status": "PASS" if ok else "FAIL", "message": "登录失败(符合预期)" if ok else "登录失败(不符合预期)", "detail": {"http": code, "body": data}}
else:
rs = {"name": f"登录-{phone}", "status": "PASS" if is_ok else "FAIL", "message": "登录成功" if is_ok else "登录失败", "detail": {"http": code, "body": data}}
return rs, token
async def user_profile_flow(base_url: str, client: httpx.AsyncClient, token: str) -> Dict[str, Any]:
"""
用户资料查看与编辑
参数:
base_url: 基础URL(含 /api/v1)
client: httpx客户端
token: 用户JWT
返回:
测试结果字典
"""
headers = {"token": token}
view_url = make_url(base_url, "/app-user/profile")
v_code, v_data = await api_get(client, view_url, headers=headers)
ok_view = (v_code == 200 and isinstance(v_data, dict) and v_data.get("id"))
upd_url = make_url(base_url, "/app-user/profile")
nickname = "tester-" + uuid.uuid4().hex[:6]
u_code, u_data = await api_put_json(client, upd_url, {"nickname": nickname}, headers=headers)
ok_upd = (u_code == 200 and isinstance(u_data, dict) and u_data.get("nickname") == nickname)
is_ok = ok_view and ok_upd
return {"name": "资料查看与编辑", "status": "PASS" if is_ok else "FAIL", "message": "个人资料操作成功" if is_ok else "个人资料操作失败", "detail": {"view": {"http": v_code, "body": v_data}, "update": {"http": u_code, "body": u_data}}}
async def permission_flow(base_url: str, client: httpx.AsyncClient, admin_token: str) -> Dict[str, Any]:
"""
权限控制验证
参数:
base_url: 基础URL(含 /api/v1)
client: httpx客户端
admin_token: 管理端token头值
返回:
测试结果字典
"""
protected_url = make_url(base_url, "/user/list")
c1, d1 = await api_get(client, protected_url)
c2, d2 = await api_get(client, protected_url, headers={"token": admin_token})
ok1 = (c1 in (401, 403, 422))
ok2 = (c2 in (200, 403))
is_ok = ok1 and ok2
return {"name": "权限控制", "status": "PASS" if is_ok else "FAIL", "message": "权限校验完成", "detail": {"no_token": {"http": c1, "body": d1}, "with_token": {"http": c2, "body": d2}}}
async def main() -> None:
"""
主流程
参数:
无
返回:
None
"""
base = os.getenv("TEST_BASE_URL", "http://localhost:9991/api/v1")
admin_token = os.getenv("ADMIN_TOKEN", "dev")
results: List[Dict[str, Any]] = []
endpoint_list = [
{"path": "/app-user/register", "desc": "用户注册"},
{"path": "/app-user/login", "desc": "用户登录"},
{"path": "/app-user/profile", "desc": "获取用户信息(需token)"},
{"path": "/app-user/profile", "desc": "更新用户信息(需token) PUT"},
{"path": "/app-user/dashboard", "desc": "用户首页摘要(需token)"},
{"path": "/app-user/quota", "desc": "剩余估值次数(需token)"},
{"path": "/app-user/change-password", "desc": "修改密码(需token)"},
{"path": "/app-user/validate-token", "desc": "验证token(需token)"},
{"path": "/app-user/logout", "desc": "登出(需token)"},
{"path": "/upload/file", "desc": "上传文件"},
{"path": "/app-valuations/", "desc": "创建估值评估(需token)"},
{"path": "/app-valuations/", "desc": "获取我的估值评估列表(需token)"},
{"path": "/app-valuations/{id}", "desc": "获取估值评估详情(需token)"},
{"path": "/app-valuations/statistics/overview", "desc": "获取我的估值统计(需token)"},
{"path": "/app-valuations/{id}", "desc": "删除估值评估(需token) DELETE"},
]
async with httpx.AsyncClient(timeout=10) as client:
def gen_cn_phone() -> str:
second = str(random.choice([3,4,5,6,7,8,9]))
rest = "".join(random.choice("0123456789") for _ in range(9))
return "1" + second + rest
phone_ok = gen_cn_phone()
r1 = await user_register_flow(base, client, phone_ok, expect_success=True)
results.append(r1)
r2 = await user_register_flow(base, client, phone_ok, expect_success=False)
results.append(r2)
r3 = await user_register_flow(base, client, "abc", expect_success=False)
results.append(r3)
lr_ok, token = await user_login_flow(base, client, phone_ok, phone_ok[-6:], expect_success=True)
results.append(lr_ok)
lr_bad, _ = await user_login_flow(base, client, phone_ok, "wrong", expect_success=False)
results.append(lr_bad)
# token 场景：验证、资料、首页、配额
if token:
# 验证token
vt_code, vt_data = await api_get(client, make_url(base, "/app-user/validate-token"), headers={"token": token})
vt_ok = (vt_code == 200 and isinstance(vt_data, dict) and vt_data.get("data", {}).get("user_id"))
results.append({"name": "验证token", "status": "PASS" if vt_ok else "FAIL", "message": "token有效" if vt_ok else "token无效", "detail": {"http": vt_code, "body": vt_data}})
# 资料查看与编辑
pr = await user_profile_flow(base, client, token)
results.append(pr)
# 首页摘要
db_code, db_data = await api_get(client, make_url(base, "/app-user/dashboard"), headers={"token": token})
db_ok = (db_code == 200 and isinstance(db_data, dict))
results.append({"name": "用户首页摘要", "status": "PASS" if db_ok else "FAIL", "message": "获取成功" if db_ok else "获取失败", "detail": {"http": db_code, "body": db_data}})
# 剩余估值次数
qt_code, qt_data = await api_get(client, make_url(base, "/app-user/quota"), headers={"token": token})
qt_ok = (qt_code == 200 and isinstance(qt_data, dict))
results.append({"name": "剩余估值次数", "status": "PASS" if qt_ok else "FAIL", "message": "获取成功" if qt_ok else "获取失败", "detail": {"http": qt_code, "body": qt_data}})
# 修改密码并验证新旧密码
cp_code, cp_data = await api_post_json(client, make_url(base, "/app-user/change-password"), {"old_password": phone_ok[-6:], "new_password": "Npw" + phone_ok[-6:]}, headers={"token": token})
cp_ok = (cp_code == 200 and isinstance(cp_data, dict) and cp_data.get("code") == 200)
results.append({"name": "修改密码", "status": "PASS" if cp_ok else "FAIL", "message": "修改成功" if cp_ok else "修改失败", "detail": {"http": cp_code, "body": cp_data}})
# 旧密码登录应失败
lr_old, _ = await user_login_flow(base, client, phone_ok, phone_ok[-6:], expect_success=False)
results.append(lr_old)
# 新密码登录成功
lr_new, token2 = await user_login_flow(base, client, phone_ok, "Npw" + phone_ok[-6:], expect_success=True)
results.append(lr_new)
use_token = token2 or token
# 上传文件（pdf）
file_url = ""
try:
up_resp = await client.post(make_url(base, "/upload/file"), files={"file": ("demo.pdf", b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n", "application/pdf")})
u_code = up_resp.status_code
u_data = _ensure_dict(up_resp.json() if up_resp.headers.get("content-type", "").startswith("application/json") else {"raw": up_resp.text})
file_url = u_data.get("url", "")
u_ok = (u_code == 200 and file_url)
results.append({"name": "上传文件", "status": "PASS" if u_ok else "FAIL", "message": "上传成功" if u_ok else "上传失败", "detail": {"http": u_code, "body": u_data}})
except Exception as e:
results.append({"name": "上传文件", "status": "FAIL", "message": "上传异常", "detail": {"error": repr(e)}})
# 创建估值评估
create_payload = {
"asset_name": "测试资产",
"institution": "测试机构",
"industry": "测试行业",
"three_year_income": [100, 120, 140],
"application_coverage": "全国覆盖",
"rd_investment": "10",
"annual_revenue": "100",
"price_fluctuation": [10, 20],
"platform_accounts": {"douyin": {"likes": 1, "comments": 1, "shares": 1}},
"pattern_images": [],
"report_url": file_url or None,
"certificate_url": file_url or None,
}
cv_code, cv_data = await api_post_json(client, make_url(base, "/app-valuations/"), create_payload, headers={"token": use_token})
cv_ok = (cv_code == 200 and isinstance(cv_data, dict) and cv_data.get("data", {}).get("task_status") == "queued")
results.append({"name": "创建估值评估", "status": "PASS" if cv_ok else "FAIL", "message": "任务已提交" if cv_ok else "提交失败", "detail": {"http": cv_code, "body": cv_data}})
# 等待片刻后获取列表与详情
import asyncio
await asyncio.sleep(0.3)
gl_code, gl_data = await api_get(client, make_url(base, "/app-valuations/"), headers={"token": use_token}, params={"page": 1, "size": 10})
gl_ok = (gl_code == 200 and isinstance(gl_data, dict) and isinstance(gl_data.get("data"), list))
results.append({"name": "估值列表", "status": "PASS" if gl_ok else "FAIL", "message": "获取成功" if gl_ok else "获取失败", "detail": {"http": gl_code, "body": gl_data}})
vid = None
if gl_ok and gl_data.get("data"):
vid = gl_data["data"][0].get("id")
if vid:
gd_code, gd_data = await api_get(client, make_url(base, f"/app-valuations/{vid}"), headers={"token": use_token})
gd_ok = (gd_code == 200 and isinstance(gd_data, dict) and gd_data.get("data", {}).get("id") == vid)
results.append({"name": "估值详情", "status": "PASS" if gd_ok else "FAIL", "message": "获取成功" if gd_ok else "获取失败", "detail": {"http": gd_code, "body": gd_data}})
# 统计
st_code, st_data = await api_get(client, make_url(base, "/app-valuations/statistics/overview"), headers={"token": use_token})
st_ok = (st_code == 200 and isinstance(st_data, dict))
results.append({"name": "估值统计", "status": "PASS" if st_ok else "FAIL", "message": "获取成功" if st_ok else "获取失败", "detail": {"http": st_code, "body": st_data}})
# 删除
del_resp = await client.delete(make_url(base, f"/app-valuations/{vid}"), headers={"token": use_token})
d_code = del_resp.status_code
d_data = _ensure_dict(del_resp.json() if del_resp.headers.get("content-type", "").startswith("application/json") else {"raw": del_resp.text})
d_ok = (d_code == 200 and isinstance(d_data, dict) and d_data.get("data", {}).get("deleted"))
results.append({"name": "删除估值", "status": "PASS" if d_ok else "FAIL", "message": "删除成功" if d_ok else "删除失败", "detail": {"http": d_code, "body": d_data}})
# 登出
lo_code, lo_data = await api_post_json(client, make_url(base, "/app-user/logout"), {}, headers={"token": use_token})
lo_ok = (lo_code == 200)
results.append({"name": "登出", "status": "PASS" if lo_ok else "FAIL", "message": "登出成功" if lo_ok else "登出失败", "detail": {"http": lo_code, "body": lo_data}})
perm = await permission_flow(base, client, admin_token)
results.append(perm)
passes = sum(1 for r in results if r.get("status") == "PASS")
total = len(results)
print(json.dumps({"total": total, "passes": passes, "results": results, "endpoints": endpoint_list}, ensure_ascii=False, indent=2))
write_html_report("reports/user_flow_script_report.html", "用户维度功能测试报告（脚本）", results)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
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]]:
"""
发送PUT JSON请求
参数:
client: httpx异步客户端
url: 完整URL
payload: 请求体JSON
headers: 请求头
返回:
(状态码, 响应JSON)
"""
r = await client.put(url, json=payload, headers=headers or {})
data = {}
try:
data = r.json()
except Exception:
data = {"raw": r.text}
return r.status_code, data