#!/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}")