qyyfw/qihuaiflow
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
79c27795a89b915da6e83aa5eec18d74eadb3e78
qihuaiflow/tests/run_api_test.py
T
qyyfw 78ff261b5a 初始提交: Qihu-AIFlow v2.0.1 完整项目 
- 前端 React+TypeScript (源码+构建产物)
- 后端 Python FastAPI
- 数据库迁移 001-015
- Docker部署配置
- 微信ClawBot API频道绑定
- 项目文档 README.md
2026-07-02 14:40:46 +08:00

972 lines
55 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
AI智能自动化PaaS平台 - 接口自动化测试脚本
测试范围：前台API + 后台管理API + 核心检查项
"""
import requests
import json
import sys
import os
import uuid
from datetime import datetime, timedelta
BASE_URL = "http://localhost:18080"
API_PREFIX = "/app/api"
ADMIN_USERNAME = "admin"
ADMIN_PASSWORD = "admin123"
# 测试结果收集
results = []
passed_cnt = 0
failed_cnt = 0
class TestResult:
def __init__(self, module, name, method, url, params, expected, actual, ok, status_code=None):
self.module = module
self.name = name
self.method = method
self.url = url
self.params = params
self.expected = expected
self.actual = actual
self.passed = ok
self.status_code = status_code
def add_result(module, name, method, url, params, expected, actual, ok, status_code=None):
global passed_cnt, failed_cnt
result = TestResult(module, name, method, url, params, expected, actual, ok, status_code)
results.append(result)
if ok:
passed_cnt += 1
print(f" [PASS] {module} - {name}")
else:
failed_cnt += 1
print(f" [FAIL] {module} - {name} | 期望: {expected} | 实际: {actual}")
def api_get(path, token=None, params=None):
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
url = f"{BASE_URL}{API_PREFIX}{path}"
try:
resp = requests.get(url, headers=headers, params=params, timeout=10)
return resp
except Exception as e:
print(f" [ERROR] GET {url}: {e}")
return None
def api_post(path, token=None, json_data=None, data=None):
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
url = f"{BASE_URL}{API_PREFIX}{path}"
try:
if json_data:
resp = requests.post(url, headers=headers, json=json_data, timeout=10)
else:
resp = requests.post(url, headers=headers, data=data, timeout=10)
return resp
except Exception as e:
print(f" [ERROR] POST {url}: {e}")
return None
def api_put(path, token=None, json_data=None, params=None):
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
url = f"{BASE_URL}{API_PREFIX}{path}"
try:
if params:
resp = requests.put(url, headers=headers, json=json_data, params=params, timeout=10)
else:
resp = requests.put(url, headers=headers, json=json_data, timeout=10)
return resp
except Exception as e:
print(f" [ERROR] PUT {url}: {e}")
return None
def api_delete(path, token=None):
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
url = f"{BASE_URL}{API_PREFIX}{path}"
try:
resp = requests.delete(url, headers=headers, timeout=10)
return resp
except Exception as e:
print(f" [ERROR] DELETE {url}: {e}")
return None
def check_pass(resp, expected_code=200, has_code_field=True, code_value=0):
"""检查API响应是否通过"""
if resp is None:
return False, "请求失败(无响应)"
sc = resp.status_code
if sc != expected_code:
try:
body = resp.text[:200]
except:
body = "无法解析"
return False, f"HTTP状态码 {sc}, 期望 {expected_code}, 响应: {body}"
if has_code_field:
try:
body = resp.json()
if body.get("code") != code_value:
return False, f"业务code={body.get('code')}, 期望={code_value}, msg={body.get('message','')}"
except:
return False, f"无法解析JSON: {resp.text[:100]}"
return True, "OK"
print("=" * 70)
print("AI智能自动化PaaS平台 - 接口自动化测试")
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"后端地址: {BASE_URL}")
print("=" * 70)
# ============================================================
# 0. 初始准备
# ============================================================
print("\n[模块0] 初始准备")
resp = api_post("/auth/login", json_data={"username": ADMIN_USERNAME, "password": ADMIN_PASSWORD})
check, msg = check_pass(resp)
admin_token = resp.json()["data"]["access_token"] if check else None
add_result("初始准备", "管理员登录", "POST", "/auth/login", f"username={ADMIN_USERNAME}", "code=0", f"code={resp.json().get('code') if resp else 'N/A'}", check)
if not check:
print("ERROR: 管理员登录失败，测试中止")
sys.exit(1)
admin_id = resp.json()["data"]["user"]["id"]
# 注册测试用户A
ts = int(datetime.now().timestamp())
userA_data = {"username": f"testuserA_{ts}", "password": "Test123456", "email": f"testa_{ts}@test.com", "phone": f"138{str(ts)[-8:]}"}
resp = api_post("/auth/register", json_data=userA_data)
check, msg = check_pass(resp)
userA_token = resp.json()["data"]["access_token"] if check else None
userA_id = resp.json()["data"]["user"]["id"] if check else None
add_result("初始准备", "注册测试用户A", "POST", "/auth/register", f"username={userA_data['username']}", "code=0", msg, check)
# 注册测试用户B
ts2 = ts + 1
userB_data = {"username": f"testuserB_{ts2}", "password": "Test123456", "email": f"testb_{ts2}@test.com", "phone": f"139{str(ts2)[-8:]}"}
resp = api_post("/auth/register", json_data=userB_data)
check, msg = check_pass(resp)
userB_token = resp.json()["data"]["access_token"] if check else None
userB_id = resp.json()["data"]["user"]["id"] if check else None
add_result("初始准备", "注册测试用户B", "POST", "/auth/register", f"username={userB_data['username']}", "code=0", msg, check)
# ============================================================
# 1. 认证模块 /app/api/auth
# ============================================================
print("\n[模块1] 认证模块")
resp = api_post("/auth/login", json_data={"username": userA_data["username"], "password": "Test123456"})
check, msg = check_pass(resp)
add_result("认证模块", "用户名密码登录-正常", "POST", "/auth/login", f"username={userA_data['username']}", "code=0", msg, check)
resp = api_post("/auth/login", json_data={"username": userA_data["username"], "password": "wrong"})
add_result("认证模块", "错误密码登录-异常", "POST", "/auth/login", "wrong_password", "401", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 401)
resp = api_post("/auth/login", json_data={"username": f"nonexist_{ts}", "password": "test"})
add_result("认证模块", "不存在用户登录-异常", "POST", "/auth/login", "nonexistent_user", "401", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 401)
resp = api_get("/auth/me", token=userA_token)
check, msg = check_pass(resp)
add_result("认证模块", "获取当前用户信息", "GET", "/auth/me", "", "code=0", msg, check)
resp = api_post("/auth/logout", token=userA_token)
check, msg = check_pass(resp)
add_result("认证模块", "用户登出", "POST", "/auth/logout", "", "code=0", msg, check)
resp = api_post("/auth/register", json_data=userA_data)
add_result("认证模块", "重复注册-异常", "POST", "/auth/register", f"username={userA_data['username']}(已存在)", "400", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 400)
resp = api_get("/auth/settings")
check, msg = check_pass(resp)
add_result("认证模块", "获取注册设置", "GET", "/auth/settings", "", "code=0", msg, check)
# ============================================================
# 2. AI聊天模块 /app/api/chat
# ============================================================
print("\n[模块2] AI聊天模块")
resp = api_get("/chat/models", token=userA_token)
check, msg = check_pass(resp)
models_list = resp.json().get("data", []) if check else []
model_id = models_list[0]["id"] if models_list else None
add_result("AI聊天", "获取可用模型列表", "GET", "/chat/models", "", "code=0", msg, check)
resp = api_post("/chat/conversations", token=userA_token, json_data={"title": "测试对话1", "model_id": model_id})
check, msg = check_pass(resp)
conv_id = resp.json()["data"]["id"] if check else None
add_result("AI聊天", "创建新对话", "POST", "/chat/conversations", "title=测试对话1", "code=0", msg, check)
resp = api_post("/chat/conversations", token=userA_token, json_data={"title": "测试对话2", "model_id": model_id})
check2, msg2 = check_pass(resp)
conv_id2 = resp.json()["data"]["id"] if check2 else None
add_result("AI聊天", "创建第二个对话", "POST", "/chat/conversations", "title=测试对话2", "code=0", msg2, check2)
resp = api_get("/chat/conversations", token=userA_token)
check, msg = check_pass(resp)
add_result("AI聊天", "获取对话列表", "GET", "/chat/conversations", "", "code=0", msg, check)
resp = api_get("/chat/conversations", token=userA_token, params={"search": "测试"})
check, msg = check_pass(resp)
add_result("AI聊天", "搜索对话", "GET", "/chat/conversations", "search=测试", "code=0", msg, check)
resp = api_put(f"/chat/conversations/{conv_id}", token=userA_token, json_data={"title": "已重命名对话"})
check, msg = check_pass(resp)
add_result("AI聊天", "对话重命名", "PUT", f"/chat/conversations/{conv_id}", "title=已重命名", "code=0", msg, check)
resp = api_put(f"/chat/conversations/{conv_id}", token=userA_token, json_data={"pinned": True})
check, msg = check_pass(resp)
add_result("AI聊天", "对话置顶", "PUT", f"/chat/conversations/{conv_id}", "pinned=true", "code=0", msg, check)
resp = api_post(f"/chat/conversations/{conv_id}/messages", token=userA_token, json_data={"content": "你好，请介绍一下你自己", "model_id": model_id})
check, msg = check_pass(resp)
add_result("AI聊天", "发送消息-正常", "POST", f"/chat/conversations/{conv_id}/messages", "content=你好...", "code=0", msg, check)
resp = api_get(f"/chat/conversations/{conv_id}/messages", token=userA_token)
check, msg = check_pass(resp)
add_result("AI聊天", "获取消息历史", "GET", f"/chat/conversations/{conv_id}/messages", "", "code=0", msg, check)
resp = api_post(f"/chat/conversations/{conv_id}/clear", token=userA_token)
check, msg = check_pass(resp)
add_result("AI聊天", "清空对话上下文", "POST", f"/chat/conversations/{conv_id}/clear", "", "code=0", msg, check)
resp = api_delete(f"/chat/conversations/{conv_id2}", token=userA_token)
check, msg = check_pass(resp)
add_result("AI聊天", "删除(归档)对话", "DELETE", f"/chat/conversations/{conv_id2}", "", "code=0", msg, check)
resp = api_put(f"/chat/conversations/{conv_id}", token=userA_token, json_data={"status": "archived"})
check, msg = check_pass(resp)
add_result("AI聊天", "对话归档", "PUT", f"/chat/conversations/{conv_id}", "status=archived", "code=0", msg, check)
resp = api_get("/chat/conversations/nonexistent123/messages", token=userA_token)
add_result("AI聊天", "访问不存在对话-异常", "GET", "/chat/conversations/nonexistent/messages", "", "404", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 404)
# ============================================================
# 3. 频道模块 /app/api/channels
# ============================================================
print("\n[模块3] 频道模块")
resp = api_post("/channels/", token=userA_token, json_data={"name": "测试频道", "description": "这是一个测试频道", "type": "public"})
check, msg = check_pass(resp)
channel_id = resp.json()["data"]["id"] if check else None
add_result("频道模块", "创建频道", "POST", "/channels/", "name=测试频道", "code=0", msg, check)
resp = api_get("/channels/", token=userA_token)
check, msg = check_pass(resp)
add_result("频道模块", "获取频道列表", "GET", "/channels/", "", "code=0", msg, check)
if channel_id:
resp = api_get(f"/channels/{channel_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("频道模块", "获取频道详情", "GET", f"/channels/{channel_id}", "", "code=0", msg, check)
resp = api_get(f"/channels/{channel_id}/members", token=userA_token)
check, msg = check_pass(resp)
add_result("频道模块", "获取频道成员", "GET", f"/channels/{channel_id}/members", "", "code=0", msg, check)
resp = api_post(f"/channels/{channel_id}/messages", token=userA_token, json_data={"content": "Hello Channel!"})
check, msg = check_pass(resp)
add_result("频道模块", "发送频道消息", "POST", f"/channels/{channel_id}/messages", "content=Hello", "code=0", msg, check)
resp = api_get(f"/channels/{channel_id}/messages", token=userA_token)
check, msg = check_pass(resp)
add_result("频道模块", "获取频道消息", "GET", f"/channels/{channel_id}/messages", "", "code=0", msg, check)
# 邀请成员
if userB_id:
resp = api_post(f"/channels/{channel_id}/members", token=userA_token, json_data={"user_id": userB_id})
check, msg = check_pass(resp)
add_result("频道模块", "邀请成员", "POST", f"/channels/{channel_id}/members", f"user_id={userB_id}", "code=0", msg, check)
# 更新频道
resp = api_put(f"/channels/{channel_id}", token=userA_token, json_data={"name": "已更新频道名"})
check, msg = check_pass(resp)
add_result("频道模块", "更新频道", "PUT", f"/channels/{channel_id}", "name=已更新频道名", "code=0", msg, check)
# 删除频道
resp = api_delete(f"/channels/{channel_id}", token=userA_token)
check = resp is not None
add_result("频道模块", "删除频道", "DELETE", f"/channels/{channel_id}", "", "成功删除", f"status={resp.status_code if resp else 'N/A'}", check)
# ============================================================
# 4. 知识库模块 /app/api/knowledge
# ============================================================
print("\n[模块4] 知识库模块")
resp = api_post("/knowledge/bases", token=userA_token, json_data={"name": "测试知识库", "description": "用于测试的知识库"})
check, msg = check_pass(resp)
kb_id = resp.json()["data"]["id"] if check else None
add_result("知识库", "创建知识库", "POST", "/knowledge/bases", "name=测试知识库", "code=0", msg, check)
resp = api_get("/knowledge/bases", token=userA_token)
check, msg = check_pass(resp)
add_result("知识库", "获取知识库列表", "GET", "/knowledge/bases", "", "code=0", msg, check)
if kb_id:
resp = api_get(f"/knowledge/bases/{kb_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("知识库", "获取知识库详情", "GET", f"/knowledge/bases/{kb_id}", "", "code=0", msg, check)
resp = api_put(f"/knowledge/bases/{kb_id}", token=userA_token, json_data={"name": "已更新知识库名"})
check, msg = check_pass(resp)
add_result("知识库", "更新知识库", "PUT", f"/knowledge/bases/{kb_id}", "name=已更新", "code=0", msg, check)
resp = api_post(f"/knowledge/bases/{kb_id}/documents", token=userA_token, json_data={"filename": "test.pdf", "file_size": 102400, "file_type": "pdf"})
check, msg = check_pass(resp)
doc_id = resp.json()["data"]["id"] if check else None
add_result("知识库", "上传文档", "POST", f"/knowledge/bases/{kb_id}/documents", "filename=test.pdf", "code=0", msg, check)
resp = api_get(f"/knowledge/bases/{kb_id}/documents", token=userA_token)
check, msg = check_pass(resp)
add_result("知识库", "获取文档列表", "GET", f"/knowledge/bases/{kb_id}/documents", "", "code=0", msg, check)
# 创建目录
resp = api_post("/knowledge/directories", token=userA_token, json_data={"name": "测试目录", "base_id": kb_id})
check, msg = check_pass(resp)
add_result("知识库", "创建目录", "POST", "/knowledge/directories", "name=测试目录", "code=0", msg, check)
resp = api_get("/knowledge/directories", token=userA_token)
check, msg = check_pass(resp)
add_result("知识库", "获取目录列表", "GET", "/knowledge/directories", "", "code=0", msg, check)
if kb_id and doc_id:
resp = api_delete(f"/knowledge/bases/{kb_id}/documents/{doc_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("知识库", "删除文档", "DELETE", f"/knowledge/bases/{kb_id}/documents/{doc_id}", "", "code=0", msg, check)
resp = api_get("/knowledge/bases/nonexistent123", token=userA_token)
add_result("知识库", "访问不存在知识库-异常", "GET", "/knowledge/bases/nonexistent", "", "404", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 404)
if kb_id:
resp = api_delete(f"/knowledge/bases/{kb_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("知识库", "删除知识库", "DELETE", f"/knowledge/bases/{kb_id}", "", "code=0", msg, check)
# ============================================================
# 5. 应用市场模块 /app/api/market
# ============================================================
print("\n[模块5] 应用市场模块")
resp = api_get("/market/apps", token=userA_token)
check, msg = check_pass(resp)
add_result("应用市场", "获取应用列表", "GET", "/market/apps", "", "code=0", msg, check)
resp = api_get("/market/apps", token=userA_token, params={"type": "skill"})
check, msg = check_pass(resp)
add_result("应用市场", "按类型筛选应用", "GET", "/market/apps", "type=skill", "code=0", msg, check)
resp = api_get("/market/apps", token=userA_token, params={"search": "AI"})
check, msg = check_pass(resp)
add_result("应用市场", "搜索应用", "GET", "/market/apps", "search=AI", "code=0", msg, check)
resp = api_post("/market/apps", token=userA_token, json_data={"name": "测试Skill应用", "description": "这是一个测试应用", "type": "skill", "price": 9.9})
check, msg = check_pass(resp)
app_id = resp.json()["data"]["id"] if check else None
add_result("应用市场", "上传应用", "POST", "/market/apps", "name=测试Skill应用", "code=0", msg, check)
# 管理员审核通过应用
if app_id:
resp = api_put(f"/admin/market/apps/{app_id}/approve", token=admin_token)
check = resp is not None and resp.status_code == 200
add_result("市场审核", "审核通过应用", "PUT", f"/admin/market/apps/{app_id}/approve", "", "200", f"status={resp.status_code if resp else 'N/A'}", check)
resp = api_get(f"/market/apps/{app_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("应用市场", "获取应用详情", "GET", f"/market/apps/{app_id}", "", "code=0", msg, check)
# 购买应用
resp = api_post(f"/market/apps/{app_id}/purchase", token=userB_token, json_data={"amount": 9.9})
check, msg = check_pass(resp)
add_result("应用市场", "购买应用", "POST", f"/market/apps/{app_id}/purchase", "amount=9.9", "code=0", msg, check)
# 安装应用
resp = api_post(f"/market/apps/{app_id}/install", token=userB_token)
check, msg = check_pass(resp)
add_result("应用市场", "安装应用", "POST", f"/market/apps/{app_id}/install", "", "code=0", msg, check)
# 评价应用
resp = api_post(f"/market/apps/{app_id}/review", token=userB_token, json_data={"rating": 4, "content": "很好用"})
check, msg = check_pass(resp)
add_result("应用市场", "评价应用", "POST", f"/market/apps/{app_id}/review", "rating=4", "code=0", msg, check)
# 获取评价列表
resp = api_get(f"/market/reviews/{app_id}")
check, msg = check_pass(resp)
add_result("应用市场", "获取评价列表", "GET", f"/market/reviews/{app_id}", "", "code=0", msg, check)
# 获取我的已购应用
resp = api_get("/market/my-apps", token=userB_token)
check, msg = check_pass(resp)
add_result("应用市场", "获取已购应用", "GET", "/market/my-apps", "", "code=0", msg, check)
resp = api_get("/market/apps/nonexistent123", token=userA_token)
add_result("应用市场", "获取不存在应用-异常", "GET", "/market/apps/nonexistent", "", "404", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 404)
# ============================================================
# 6. Token驾驶舱 /app/api/token (注意：main.py中前缀为/app/api/token)
# ============================================================
print("\n[模块6] Token驾驶舱")
resp = api_get("/token/overview", token=userA_token)
check, msg = check_pass(resp)
add_result("Token驾驶舱", "获取消耗总览", "GET", "/token/overview", "", "code=0", msg, check)
resp = api_get("/token/rankings", token=userA_token)
check, msg = check_pass(resp)
add_result("Token驾驶舱", "获取消耗排名", "GET", "/token/rankings", "", "code=0", msg, check)
resp = api_get("/token/logs", token=userA_token)
check, msg = check_pass(resp)
add_result("Token驾驶舱", "获取消耗明细", "GET", "/token/logs", "", "code=0", msg, check)
resp = api_get("/token/logs", token=userA_token, params={"start_date": "2026-01-01", "end_date": "2026-12-31"})
check, msg = check_pass(resp)
add_result("Token驾驶舱", "按日期筛选明细", "GET", "/token/logs", "start_date=2026-01-01&end_date=2026-12-31", "code=0", msg, check)
resp = api_get("/token/balance", token=userA_token)
check, msg = check_pass(resp)
add_result("Token驾驶舱", "获取余额", "GET", "/token/balance", "", "code=0", msg, check)
resp = api_post("/token/recharge", token=userA_token, json_data={"amount": 1000, "channel": "simulation"})
check, msg = check_pass(resp)
add_result("Token驾驶舱", "Token充值", "POST", "/token/recharge", "amount=1000", "code=0", msg, check)
# ============================================================
# 7. 付费中心 /app/api/payment
# ============================================================
print("\n[模块7] 付费中心")
resp = api_get("/payment/plans", token=userA_token)
check, msg = check_pass(resp)
plans_list = resp.json().get("data", []) if check else []
add_result("付费中心", "获取会员计划列表", "GET", "/payment/plans", "", "code=0", msg, check)
resp = api_get("/payment/services", token=userA_token)
check, msg = check_pass(resp)
services_list = resp.json().get("data", []) if check else []
add_result("付费中心", "获取增值服务列表", "GET", "/payment/services", "", "code=0", msg, check)
order_id = None
if plans_list:
plan_id_val = plans_list[0]["id"]
resp = api_post("/payment/orders", token=userA_token, json_data={"plan_id": plan_id_val})
check, msg = check_pass(resp)
order_id = resp.json()["data"]["id"] if check else None
add_result("付费中心", "创建会员订单", "POST", "/payment/orders", f"plan_id={plan_id_val}", "code=0", msg, check)
resp = api_get("/payment/orders", token=userA_token)
check, msg = check_pass(resp)
add_result("付费中心", "获取订单列表", "GET", "/payment/orders", "", "code=0", msg, check)
if order_id:
resp = api_get(f"/payment/orders/{order_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("付费中心", "获取订单详情", "GET", f"/payment/orders/{order_id}", "", "code=0", msg, check)
resp = api_post(f"/payment/orders/{order_id}/pay", token=userA_token)
check, msg = check_pass(resp)
add_result("付费中心", "支付订单", "POST", f"/payment/orders/{order_id}/pay", "", "code=0", msg, check)
resp = api_get("/payment/coupons", token=userA_token)
check, msg = check_pass(resp)
add_result("付费中心", "获取优惠券列表", "GET", "/payment/coupons", "", "code=0", msg, check)
if services_list:
svc_id_val = services_list[0]["id"]
resp = api_post("/payment/custom-services", token=userA_token, json_data={"service_id": svc_id_val, "description": "需要一个AI客服机器人"})
check, msg = check_pass(resp)
add_result("付费中心", "创建定制服务订单", "POST", "/payment/custom-services", f"service_id={svc_id_val}", "code=0", msg, check)
# ============================================================
# 8. 个人中心 /app/api/profile
# ============================================================
print("\n[模块8] 个人中心")
resp = api_get("/profile/", token=userA_token)
check, msg = check_pass(resp)
add_result("个人中心", "获取个人信息", "GET", "/profile/", "", "code=0", msg, check)
resp = api_put("/profile/", token=userA_token, json_data={"nickname": "测试昵称A"})
check, msg = check_pass(resp)
add_result("个人中心", "更新个人信息", "PUT", "/profile/", "nickname=测试昵称A", "code=0", msg, check)
resp = api_put("/profile/password", token=userA_token, json_data={"old_password": "Test123456", "new_password": "NewTest456"})
check, msg = check_pass(resp)
if check:
# 改回原密码
api_put("/profile/password", token=userA_token, json_data={"old_password": "NewTest456", "new_password": "Test123456"})
add_result("个人中心", "修改密码", "PUT", "/profile/password", "", "code=0", msg, check)
resp = api_put("/profile/password", token=userA_token, json_data={"old_password": "WrongPwd123", "new_password": "NewTest789"})
add_result("个人中心", "修改密码-错误旧密码", "PUT", "/profile/password", "wrong_old_password", "400", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 400)
resp = api_get("/profile/login-logs", token=userA_token)
check, msg = check_pass(resp)
add_result("个人中心", "获取登录日志", "GET", "/profile/login-logs", "", "code=0", msg, check)
resp = api_post("/profile/api-keys", token=userA_token, json_data={"name": "测试密钥"})
check, msg = check_pass(resp)
api_key_id = resp.json()["data"]["id"] if check else None
add_result("个人中心", "创建API密钥", "POST", "/profile/api-keys", "name=测试密钥", "code=0", msg, check)
resp = api_get("/profile/api-keys", token=userA_token)
check, msg = check_pass(resp)
add_result("个人中心", "获取API密钥列表", "GET", "/profile/api-keys", "", "code=0", msg, check)
if api_key_id:
resp = api_put(f"/profile/api-keys/{api_key_id}", token=userA_token, json_data={"status": "disabled"})
check, msg = check_pass(resp)
add_result("个人中心", "禁用API密钥", "PUT", f"/profile/api-keys/{api_key_id}", "status=disabled", "code=0", msg, check)
resp = api_get("/profile/membership", token=userA_token)
check, msg = check_pass(resp)
add_result("个人中心", "获取会员信息", "GET", "/profile/membership", "", "code=0", msg, check)
resp = api_get("/profile/purchased-apps", token=userB_token)
check, msg = check_pass(resp)
add_result("个人中心", "获取已购应用", "GET", "/profile/purchased-apps", "", "code=0", msg, check)
# ============================================================
# 9. 消息通知 /app/api/notifications
# ============================================================
print("\n[模块9] 消息通知")
resp = api_get("/notifications/", token=userA_token)
check, msg = check_pass(resp)
add_result("消息通知", "获取通知列表", "GET", "/notifications/", "", "code=0", msg, check)
resp = api_get("/notifications/unread-count", token=userA_token)
check, msg = check_pass(resp)
add_result("消息通知", "获取未读数量", "GET", "/notifications/unread-count", "", "code=0", msg, check)
resp = api_put("/notifications/read-all", token=userA_token)
check, msg = check_pass(resp)
add_result("消息通知", "全部标记已读", "PUT", "/notifications/read-all", "", "code=0", msg, check)
# ============================================================
# 10. 资产管理 /app/api/assets
# ============================================================
print("\n[模块10] 资产管理")
resp = api_post("/assets/", token=userA_token, json_data={"name": "测试文件.txt", "description": "测试资产", "content": "这是测试内容"})
check, msg = check_pass(resp)
asset_id = resp.json()["data"]["id"] if check else None
add_result("资产管理", "创建资产", "POST", "/assets/", "name=测试文件.txt", "code=0", msg, check)
resp = api_get("/assets/", token=userA_token)
check, msg = check_pass(resp)
add_result("资产管理", "获取资产列表", "GET", "/assets/", "", "code=0", msg, check)
if asset_id:
resp = api_get(f"/assets/{asset_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("资产管理", "获取资产详情", "GET", f"/assets/{asset_id}", "", "code=0", msg, check)
resp = api_put(f"/assets/{asset_id}", token=userA_token, json_data={"name": "已更新文件名.txt"})
check, msg = check_pass(resp)
add_result("资产管理", "更新资产", "PUT", f"/assets/{asset_id}", "name=已更新", "code=0", msg, check)
resp = api_get("/assets/categories", token=userA_token)
check, msg = check_pass(resp)
add_result("资产管理", "获取分类列表", "GET", "/assets/categories", "", "code=0", msg, check)
if asset_id:
resp = api_post(f"/assets/novels/{asset_id}/chapters", token=userA_token, json_data={"title": "第一章", "content": "这是第一章的内容..."})
add_result("资产管理", "添加小说章节", "POST", f"/assets/novels/{asset_id}/chapters", "title=第一章", "操作完成", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code in [200, 404])
resp = api_delete(f"/assets/{asset_id}", token=userA_token)
check, msg = check_pass(resp)
add_result("资产管理", "删除资产", "DELETE", f"/assets/{asset_id}", "", "code=0", msg, check)
# ============================================================
# 11. 管理仪表盘 /app/api/admin/dashboard
# ============================================================
print("\n[模块11] 管理仪表盘")
resp = api_get("/admin/dashboard/stats", token=admin_token)
add_result("管理仪表盘", "获取仪表盘统计", "GET", "/admin/dashboard/stats", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/dashboard/recent-logs", token=admin_token)
add_result("管理仪表盘", "获取最近操作日志", "GET", "/admin/dashboard/recent-logs", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/dashboard/recent-audit", token=admin_token)
add_result("管理仪表盘", "获取最近审计日志", "GET", "/admin/dashboard/recent-audit", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 12. 用户管理 /app/api/admin/users
# ============================================================
print("\n[模块12] 用户管理")
resp = api_get("/admin/users/", token=admin_token)
add_result("用户管理", "获取用户列表", "GET", "/admin/users/", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/users/", token=admin_token, params={"search": "testuserA"})
items_cnt = len(resp.json().get("items", [])) if resp else 0
add_result("用户管理", "搜索用户", "GET", "/admin/users/", "search=testuserA", "找到用户", f"items={items_cnt}", resp is not None and resp.status_code == 200 and items_cnt > 0)
resp = api_get(f"/admin/users/{userA_id}", token=admin_token)
add_result("用户管理", "获取用户详情", "GET", f"/admin/users/{userA_id}", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_put(f"/admin/users/{userA_id}/ban", token=admin_token)
add_result("用户管理", "封禁用户", "PUT", f"/admin/users/{userA_id}/ban", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_put(f"/admin/users/{userA_id}/unban", token=admin_token)
add_result("用户管理", "解封用户", "PUT", f"/admin/users/{userA_id}/unban", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/users/register-settings", token=admin_token)
add_result("用户管理", "获取注册设置", "GET", "/admin/users/register-settings", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_put("/admin/users/register-settings", token=admin_token, params={"real_name_required": False})
add_result("用户管理", "更新注册设置", "PUT", "/admin/users/register-settings", "real_name_required=false", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 13. 大模型库管理 /app/api/admin/models
# ============================================================
print("\n[模块13] 大模型库管理")
resp = api_get("/admin/models/providers", token=admin_token)
add_result("模型管理", "获取供应商列表", "GET", "/admin/models/providers", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_post(f"/admin/models/providers?name=TestProvider_{ts}&api_base_url=https://api.test.com", token=admin_token)
check_p = resp is not None and resp.status_code == 200
provider_id = resp.json().get("id") if check_p else None
add_result("模型管理", "创建供应商", "POST", "/admin/models/providers", f"name=TestProvider_{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_p)
if provider_id:
resp = api_put(f"/admin/models/providers/{provider_id}?name=TestProvider_UPD_{ts}", token=admin_token)
add_result("模型管理", "更新供应商", "PUT", f"/admin/models/providers/{provider_id}", "name=Updated", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_post(f"/admin/models/?provider_id={provider_id}&name=TestModel_{ts}&status=paid&price_per_million_tokens=2.5&timeout=30", token=admin_token)
check_m = resp is not None and resp.status_code == 200
test_model_id = resp.json().get("id") if check_m else None
add_result("模型管理", "创建模型", "POST", "/admin/models/", f"name=TestModel_{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_m)
resp = api_get("/admin/models/", token=admin_token)
add_result("模型管理", "获取模型列表", "GET", "/admin/models/", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
if test_model_id:
resp = api_put(f"/admin/models/{test_model_id}?price_per_million_tokens=3.0", token=admin_token)
add_result("模型管理", "更新模型计费", "PUT", f"/admin/models/{test_model_id}", "price=3.0", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_put(f"/admin/models/{test_model_id}/toggle?is_enabled=false", token=admin_token)
add_result("模型管理", "禁用模型", "PUT", f"/admin/models/{test_model_id}/toggle", "is_enabled=false", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/models/stats", token=admin_token)
add_result("模型管理", "获取模型调用统计", "GET", "/admin/models/stats", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
if test_model_id:
resp = api_delete(f"/admin/models/{test_model_id}", token=admin_token)
add_result("模型管理", "删除模型", "DELETE", f"/admin/models/{test_model_id}", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 14. 网站设置 /app/api/admin/settings
# ============================================================
print("\n[模块14] 网站设置")
resp = api_get("/admin/settings/site", token=admin_token)
add_result("网站设置", "获取网站设置", "GET", "/admin/settings/site", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_put("/admin/settings/site", token=admin_token, json_data={"site_name": "AI测试平台", "site_logo": "https://example.com/logo.png"})
add_result("网站设置", "更新网站设置", "PUT", "/admin/settings/site", "site_name=AI测试平台", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/settings/email", token=admin_token)
add_result("网站设置", "获取邮件配置", "GET", "/admin/settings/email", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/settings/sms", token=admin_token)
add_result("网站设置", "获取短信配置", "GET", "/admin/settings/sms", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 15. 支付渠道 /app/api/admin/payment
# ============================================================
print("\n[模块15] 支付渠道")
resp = api_get("/admin/payment/configs", token=admin_token)
add_result("支付渠道", "获取支付配置列表", "GET", "/admin/payment/configs", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/payment/orders", token=admin_token)
add_result("支付渠道", "获取全平台订单", "GET", "/admin/payment/orders", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
if order_id:
resp = api_put(f"/admin/payment/orders/{order_id}/refund", token=admin_token)
add_result("支付渠道", "订单退款", "PUT", f"/admin/payment/orders/{order_id}/refund", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/payment/orders/export", token=admin_token)
add_result("支付渠道", "导出订单CSV", "GET", "/admin/payment/orders/export", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 16. 增值服务管理 /app/api/admin/services
# ============================================================
print("\n[模块16] 增值服务管理")
resp = api_post(f"/admin/services/?name=测试增值服务_{ts}&price=19.9&service_type=report&description=测试描述", token=admin_token)
check_s = resp is not None and resp.status_code == 200
svc_id = resp.json().get("id") if check_s else None
add_result("增值服务", "创建增值服务", "POST", "/admin/services/", f"name=测试增值服务_{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_s)
resp = api_get("/admin/services/", token=admin_token)
add_result("增值服务", "获取增值服务列表", "GET", "/admin/services/", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
if svc_id:
resp = api_put(f"/admin/services/{svc_id}?price=29.9", token=admin_token)
add_result("增值服务", "更新增值服务", "PUT", f"/admin/services/{svc_id}", "price=29.9", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/services/custom-orders", token=admin_token)
add_result("增值服务", "获取定制服务订单", "GET", "/admin/services/custom-orders", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
if svc_id:
resp = api_delete(f"/admin/services/{svc_id}", token=admin_token)
add_result("增值服务", "删除增值服务", "DELETE", f"/admin/services/{svc_id}", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 17. 会员体系 /app/api/admin/membership
# ============================================================
print("\n[模块17] 会员体系")
resp = api_get("/admin/membership/plans", token=admin_token)
add_result("会员体系", "获取会员计划列表", "GET", "/admin/membership/plans", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/membership/subscriptions", token=admin_token)
add_result("会员体系", "获取订阅列表", "GET", "/admin/membership/subscriptions", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/membership/automation-rules", token=admin_token)
add_result("会员体系", "获取自动化规则", "GET", "/admin/membership/automation-rules", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_post(f"/admin/membership/plans?name=TestPlan_{ts}&level=99&duration_type=month&price=9.9&daily_free_tokens=5000", token=admin_token)
check_plan = resp is not None and resp.status_code == 200
plan_id_val = resp.json().get("id") if check_plan else None
add_result("会员体系", "创建会员计划", "POST", "/admin/membership/plans", f"name=TestPlan_{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_plan)
if plan_id_val:
resp = api_delete(f"/admin/membership/plans/{plan_id_val}", token=admin_token)
add_result("会员体系", "删除会员计划", "DELETE", f"/admin/membership/plans/{plan_id_val}", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 18. 广告位管理 /app/api/admin/ads
# ============================================================
print("\n[模块18] 广告位管理")
resp = api_post(f"/admin/ads/?title=测试广告_{ts}&image_url=https://example.com/ad.png&sort_order=1", token=admin_token)
check_ad = resp is not None and resp.status_code == 200
ad_id = resp.json().get("id") if check_ad else None
add_result("广告位", "创建广告", "POST", "/admin/ads/", f"title=测试广告_{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_ad)
resp = api_get("/admin/ads/", token=admin_token)
add_result("广告位", "获取广告列表", "GET", "/admin/ads/", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
if ad_id:
resp = api_put(f"/admin/ads/{ad_id}?title=已更新广告_{ts}", token=admin_token)
add_result("广告位", "更新广告", "PUT", f"/admin/ads/{ad_id}", "title=已更新", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_delete(f"/admin/ads/{ad_id}", token=admin_token)
add_result("广告位", "删除广告", "DELETE", f"/admin/ads/{ad_id}", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 19. 应用市场审核 /app/api/admin/market
# ============================================================
print("\n[模块19] 应用市场审核")
resp = api_get("/admin/market/apps/pending", token=admin_token)
add_result("市场审核", "获取待审核应用", "GET", "/admin/market/apps/pending", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/market/apps", token=admin_token)
add_result("市场审核", "获取所有应用(管理员)", "GET", "/admin/market/apps", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/market/coupons", token=admin_token)
add_result("市场审核", "获取优惠券列表", "GET", "/admin/market/coupons", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_post(f"/admin/market/coupons?code=TEST{ts}&coupon_type=fixed&value=5&valid_from=2026-01-01T00:00:00&valid_to=2027-12-31T23:59:59&min_amount=20", token=admin_token)
check_c = resp is not None and resp.status_code == 200
coupon_id = resp.json().get("id") if check_c else None
add_result("市场审核", "创建优惠券", "POST", "/admin/market/coupons", f"code=TEST{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_c)
if coupon_id and userA_id:
resp = api_post(f"/admin/market/coupons/{coupon_id}/assign?user_id={userA_id}", token=admin_token)
add_result("市场审核", "发放优惠券", "POST", f"/admin/market/coupons/{coupon_id}/assign", f"user_id={userA_id}", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 20. 日志与风控 /app/api/admin/logs
# ============================================================
print("\n[模块20] 日志与风控")
resp = api_get("/admin/logs/operations", token=admin_token)
add_result("日志风控", "获取操作日志", "GET", "/admin/logs/operations", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/logs/token-logs", token=admin_token)
add_result("日志风控", "获取Token日志", "GET", "/admin/logs/token-logs", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/logs/payment-logs", token=admin_token)
add_result("日志风控", "获取付费日志", "GET", "/admin/logs/payment-logs", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/logs/audit-logs", token=admin_token)
add_result("日志风控", "获取审计日志", "GET", "/admin/logs/audit-logs", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/logs/export/operations", token=admin_token)
add_result("日志风控", "导出操作日志CSV", "GET", "/admin/logs/export/operations", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/logs/export/token-logs", token=admin_token)
add_result("日志风控", "导出Token日志CSV", "GET", "/admin/logs/export/token-logs", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_get("/admin/logs/risk-rules", token=admin_token)
add_result("日志风控", "获取风控规则", "GET", "/admin/logs/risk-rules", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_post(f"/admin/logs/risk-rules?name=限流规则_{ts}&rule_type=rate_limit&threshold=100&duration=3600", token=admin_token)
check_r = resp is not None and resp.status_code == 200
risk_id = resp.json().get("id") if check_r else None
add_result("日志风控", "创建风控规则", "POST", "/admin/logs/risk-rules", f"name=限流规则_{ts}", "200", f"status={resp.status_code if resp else 'N/A'}", check_r)
if risk_id:
resp = api_put(f"/admin/logs/risk-rules/{risk_id}?threshold=200", token=admin_token)
add_result("日志风控", "更新风控规则", "PUT", f"/admin/logs/risk-rules/{risk_id}", "threshold=200", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
resp = api_delete(f"/admin/logs/risk-rules/{risk_id}", token=admin_token)
add_result("日志风控", "删除风控规则", "DELETE", f"/admin/logs/risk-rules/{risk_id}", "", "200", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 200)
# ============================================================
# 21. 核心检查项
# ============================================================
print("\n[模块21] 核心检查项")
# 权限隔离：普通用户访问后台API
resp = api_get("/admin/users/", token=userA_token)
add_result("核心检查", "普通用户访问后台API-拒绝", "GET", "/admin/users/", "userA_token", "403", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 403)
resp = api_get("/admin/dashboard/stats", token=userB_token)
add_result("核心检查", "普通用户访问仪表盘-拒绝", "GET", "/admin/dashboard/stats", "userB_token", "403", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 403)
resp = api_get("/admin/logs/operations", token=userA_token)
add_result("核心检查", "普通用户访问操作日志-拒绝", "GET", "/admin/logs/operations", "userA_token", "403", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 403)
# 用户数据隔离：用户B访问用户A的资源
resp = api_get("/chat/conversations", token=userA_token)
if resp and resp.status_code == 200:
userA_convs = resp.json().get("data", [])
if userA_convs:
conv_id_a = userA_convs[0]["id"]
resp = api_get(f"/chat/conversations/{conv_id_a}/messages", token=userB_token)
add_result("核心检查", "用户B访问用户A对话-拒绝(404)", "GET", f"/chat/conversations/{conv_id_a}/messages", "userB_token", "404", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 404)
else:
# 创建一个只有用户A有的对话
resp = api_post("/chat/conversations", token=userA_token, json_data={"title": "隔离测试对话"})
if resp and resp.status_code == 200:
conv_id_a = resp.json()["data"]["id"]
resp = api_get(f"/chat/conversations/{conv_id_a}/messages", token=userB_token)
add_result("核心检查", "用户B访问用户A对话-拒绝(404)", "GET", f"/chat/conversations/{conv_id_a}/messages", "userB_token", "404", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code == 404)
# Token数据隔离验证
resp_a = api_get("/token/overview", token=userA_token)
resp_b = api_get("/token/overview", token=userB_token)
if resp_a and resp_b and resp_a.status_code == 200 and resp_b.status_code == 200:
userA_total = resp_a.json().get("data", {}).get("total_used", 0)
userB_total = resp_b.json().get("data", {}).get("total_used", 0)
add_result("核心检查", "Token数据隔离验证", "GET", "/token/overview", "userA vs userB", "数据隔离", f"A={userA_total}, B={userB_total}", True)
# 日志不可删除
resp = api_delete(f"/admin/logs/operations/some_id", token=admin_token)
add_result("核心检查", "操作日志无DELETE路由", "DELETE", "/admin/logs/operations/id", "", "不可删除(404/405)", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code in [404, 405])
resp = api_delete(f"/admin/logs/token-logs/some_id", token=admin_token)
add_result("核心检查", "Token日志无DELETE路由", "DELETE", "/admin/logs/token-logs/id", "", "不可删除(404/405)", f"status={resp.status_code if resp else 'N/A'}", resp is not None and resp.status_code in [404, 405])
# Token计费准确性验证
resp = api_post("/chat/conversations", token=userB_token, json_data={"title": "计费测试对话", "model_id": model_id})
if resp and resp.status_code == 200:
billing_conv_id = resp.json()["data"]["id"]
resp = api_post(f"/chat/conversations/{billing_conv_id}/messages", token=userB_token, json_data={"content": "测试计费准确性的消息发送", "model_id": model_id})
if resp and resp.status_code == 200:
resp_log = api_get("/admin/logs/token-logs", token=admin_token, params={"user_id": userB_id})
if resp_log and resp_log.status_code == 200:
items = resp_log.json().get("items", [])
ok = len(items) > 0 and items[0].get("total_tokens", 0) > 0
add_result("核心检查", "Token计费日志记录验证", "GET", "/admin/logs/token-logs", f"user_id={userB_id}", "有记录且total_tokens>0", f"items={len(items)}", ok)
else:
add_result("核心检查", "Token计费日志记录验证", "GET", "/admin/logs/token-logs", f"user_id={userB_id}", "有记录", f"status={resp_log.status_code if resp_log else 'N/A'}", False)
else:
add_result("核心检查", "Token计费-发送消息(预备)", "POST", f"/chat/conversations/{billing_conv_id}/messages", "", "200", f"status={resp.status_code if resp else 'N/A'}", False)
else:
add_result("核心检查", "Token计费-创建对话(预备)", "POST", "/chat/conversations", "", "200", f"status={resp.status_code if resp else 'N/A'}", False)
# ============================================================
# 清理测试数据
# ============================================================
print("\n[清理] 清理测试数据")
if provider_id:
api_delete(f"/admin/models/providers/{provider_id}", token=admin_token)
if coupon_id:
api_delete(f"/admin/market/coupons/{coupon_id}", token=admin_token)
# ============================================================
# 生成测试报告
# ============================================================
total = passed_cnt + failed_cnt
rate = (passed_cnt / total * 100) if total > 0 else 0
print("\n" + "=" * 70)
print("测试完成，生成报告...")
print(f"通过: {passed_cnt}")
print(f"失败: {failed_cnt}")
print(f"总计: {total}")
print(f"通过率: {rate:.1f}%")
print("=" * 70)
# 生成Markdown报告
report_lines = []
report_lines.append("# AI智能自动化PaaS平台 - 接口自动化测试报告\n")
report_lines.append("## 报告信息\n")
report_lines.append("| 项目 | 内容 |")
report_lines.append("|------|------|")
report_lines.append(f"| 测试时间 | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} |")
report_lines.append(f"| 后端地址 | {BASE_URL} |")
report_lines.append(f"| 测试管理员 | {ADMIN_USERNAME} |")
report_lines.append(f"| 测试账号A | {userA_data['username']} |")
report_lines.append(f"| 测试账号B | {userB_data['username']} |")
report_lines.append("")
report_lines.append("## 测试统计\n")
report_lines.append("| 统计项 | 数值 |")
report_lines.append("|--------|------|")
report_lines.append(f"| 总测试项 | {total} |")
report_lines.append(f"| 通过 | {passed_cnt} |")
report_lines.append(f"| 失败 | {failed_cnt} |")
report_lines.append(f"| 通过率 | {rate:.1f}% |")
report_lines.append("")
# 按模块分组
modules = {}
for r in results:
if r.module not in modules:
modules[r.module] = []
modules[r.module].append(r)
for mod_name, mod_results in modules.items():
mod_pass = sum(1 for r in mod_results if r.passed)
mod_fail = sum(1 for r in mod_results if not r.passed)
mod_total = mod_pass + mod_fail
mod_rate = (mod_pass / mod_total * 100) if mod_total > 0 else 0
report_lines.append(f"## {mod_name}\n")
report_lines.append(f"**模块通过率**: {mod_pass}/{mod_total} ({mod_rate:.1f}%)\n")
report_lines.append("| 用例名称 | 方法 | 接口路径 | 请求参数 | 预期结果 | 实际结果 | 状态 |")
report_lines.append("|----------|------|----------|----------|----------|----------|------|")
for r in mod_results:
status = "PASS" if r.passed else "FAIL"
params_str = str(r.params)[:80].replace("|", "\\|")
actual_str = str(r.actual)[:100].replace("|", "\\|")
expected_str = str(r.expected)[:80].replace("|", "\\|")
report_lines.append(f"| {r.name} | {r.method} | {r.url} | {params_str} | {expected_str} | {actual_str} | {status} |")
report_lines.append("")
report_lines.append("\n---\n")
report_lines.append(f"> 报告生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
report_lines.append("> 测试工程师: QA\n")
report_path = "/data/share/project/tests/api-test-report.md"
with open(report_path, "w", encoding="utf-8") as f:
f.write("\n".join(report_lines))
print(f"\n测试报告已保存到: {report_path}")
Reference in new issue View Git Blame Copy Permalink