#!/usr/bin/env python # -*- coding: utf-8 -*- """ 整合后端接口性能测试工具 整合了所有10个模块的接口测试： 1. 基础接口 (flask_wht_server_api.py) 2. 认证接口 (auth_routes.py) 3. 工作台接口 (dashboard_routes.py) 4. 企业接口 (enterprise_routes.py) 5. 图片接口 (image_routes.py) 6. 日志接口 (log_routes.py) 7. 文章接口 (article_routes.py) 8. 员工接口 (employee_routes.py) 9. 作者接口 (author_routes.py) 10. 搜索服务接口 (flask_wht_server_search.py) """ import requests import time import csv from datetime import datetime import io from PIL import Image import sys # API配置 API_BASE_URL = "http://127.0.0.1:8216" # 主服务 SEARCH_API_BASE_URL = "http://127.0.0.1:8321" # 搜索服务 USERNAME = "13621242430" PASSWORD = "admin123" # 测试结果存储 test_results = [] api_docs = [] # 全局token global_token = None # 测试模块选择（可配置） TEST_MODULES = { 'basic': True, # 基础接口 'auth': True, # 认证接口 'dashboard': True, # 工作台 'enterprise': True, # 企业 'image': True, # 图片 'log': True, # 日志 'article': True, # 文章 'employee': True, # 员工 'author': True, # 作者 'search': False, # 搜索服务（默认关闭，需要单独服务） } def print_module_header(module_name, module_num, total_modules): """打印模块测试头部""" print("\n" + "="*80) print(f"[{module_num}/{total_modules}] {module_name}") print("="*80) def login(): """登录获取token""" global global_token print("\n" + "="*80) print("正在登录...") url = f"{API_BASE_URL}/api/auth/login" data = {"username": USERNAME, "password": PASSWORD} start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 if response.status_code == 200: global_token = response.json()['data']['token'] print(f"✓ 登录成功，耗时: {elapsed_time:.2f}ms") return global_token else: print(f"✗ 登录失败: {response.status_code}") return None # ============================================================================ # 模块1: 基础接口测试 # ============================================================================ def test_basic_apis(): """测试基础接口""" if not TEST_MODULES['basic']: return print_module_header("基础接口测试", 1, 10) # 根路径 print("\n[1/3] 测试根路径...") url = f"{API_BASE_URL}/" for i in range(3): try: start_time = time.time() response = requests.get(url, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '基础接口', 'endpoint': '/', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '基础接口', 'endpoint': '/', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 健康检查 print("[2/3] 测试健康检查...") url = f"{API_BASE_URL}/health" for i in range(3): try: start_time = time.time() response = requests.get(url, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '基础接口', 'endpoint': '/health', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '基础接口', 'endpoint': '/health', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 404测试 print("[3/3] 测试404错误处理...") url = f"{API_BASE_URL}/nonexistent-test" try: start_time = time.time() response = requests.get(url, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '基础接口', 'endpoint': '/nonexistent-test', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 404, 'error': '' }) except Exception as e: test_results.append({ 'module': '基础接口', 'endpoint': '/nonexistent-test', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 添加API文档 api_docs.extend([ {'module': '基础接口', 'endpoint': '/', 'method': 'GET', 'description': '根路径', 'auth': '无'}, {'module': '基础接口', 'endpoint': '/health', 'method': 'GET', 'description': '健康检查', 'auth': '无'}, ]) # ============================================================================ # 模块2: 认证接口测试 (原脚本测试数: 11个) # ============================================================================ def test_auth_apis(token): """测试认证接口 - 包含统一登录、员工登录、登出、错误场景等""" if not TEST_MODULES['auth']: return print_module_header("认证接口测试", 2, 10) # 1. 统一登录 - 用户名 print("\n[1/11] 测试统一登录(用户名)...") url = f"{API_BASE_URL}/api/auth/login" data = {"username": USERNAME, "password": PASSWORD} try: start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (用户名)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (用户名)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 2. 统一登录 - 手机号 print("[2/11] 测试统一登录(手机号)...") data = {"phone": USERNAME, "password": PASSWORD} try: start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (手机号)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (手机号)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 3. 统一登录 - 错误密码 print("[3/11] 测试统一登录(错误密码)...") data = {"username": USERNAME, "password": "wrong_password"} try: start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (错误密码)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 401, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (错误密码)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 4. 统一登录 - 缺少参数 print("[4/11] 测试统一登录(缺少参数)...") data = {"username": USERNAME} try: start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (缺少参数)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 400, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/login (缺少参数)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 5. 员工登录 - 正常 print("[5/11] 测试员工登录...") url = f"{API_BASE_URL}/api/auth/employee/login" data = {"phone": USERNAME, "password": PASSWORD} try: start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/employee/login', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/employee/login', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6. 员工登录 - 错误密码 print("[6/11] 测试员工登录(错误密码)...") data = {"phone": USERNAME, "password": "wrong_password"} try: start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/employee/login (错误密码)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 401, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/employee/login (错误密码)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 7. 登出 - 带token print("[7/11] 测试用户登出(带token)...") url = f"{API_BASE_URL}/api/auth/logout" headers = {"Authorization": f"Bearer {token}"} try: start_time = time.time() response = requests.post(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/logout (带token)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/logout (带token)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 8. 登出 - 不带token print("[8/11] 测试用户登出(不带token)...") try: start_time = time.time() response = requests.post(url, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/logout (不带token)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '认证接口', 'endpoint': '/api/auth/logout (不带token)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 9-11. 并发登录测试 print("[9-11/11] 测试并发登录(3次)...") url = f"{API_BASE_URL}/api/auth/login" data = {"username": USERNAME, "password": PASSWORD} import concurrent.futures def single_login(): start_time = time.time() response = requests.post(url, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 return response.status_code, elapsed_time try: with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: futures = [executor.submit(single_login) for _ in range(3)] results = [f.result() for f in concurrent.futures.as_completed(futures)] for i, (status, elapsed) in enumerate(results, 1): test_results.append({ 'module': '认证接口', 'endpoint': f'/api/auth/login (并发{i})', 'method': 'POST', 'status_code': status, 'elapsed_time_ms': f"{elapsed:.2f}", 'success': status == 200, 'error': '' }) except Exception as e: for i in range(3): test_results.append({ 'module': '认证接口', 'endpoint': f'/api/auth/login (并发{i+1})', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '认证接口', 'endpoint': '/api/auth/login', 'method': 'POST', 'description': '统一登录(支持用户名/手机号)', 'auth': '无'}, {'module': '认证接口', 'endpoint': '/api/auth/employee/login', 'method': 'POST', 'description': '员工登录', 'auth': '无'}, {'module': '认证接口', 'endpoint': '/api/auth/logout', 'method': 'POST', 'description': '用户登出', 'auth': '可选'}, ]) # ============================================================================ # 模块3: 工作台接口测试 (原脚本测试数: 7个) # ============================================================================ def test_dashboard_apis(token): """测试工作台接口 - 概览、最近发布、热门产品""" if not TEST_MODULES['dashboard']: return print_module_header("工作台接口测试", 3, 10) headers = {"Authorization": f"Bearer {token}"} # 1-5. 工作台概览 (多次测试获取平均性能) print("\n[1-5/9] 测试工作台概览(5次)...") url = f"{API_BASE_URL}/api/dashboard/overview" for i in range(5): try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '工作台', 'endpoint': f'/api/dashboard/overview (第{i+1}次)', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '工作台', 'endpoint': f'/api/dashboard/overview (第{i+1}次)', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6-8. 最近发布 (不同limit参数) print("[6-8/9] 测试最近发布(不同参数)...") url = f"{API_BASE_URL}/api/dashboard/recent-publishes" test_cases = [ {"limit": 5, "desc": "5条记录"}, {"limit": 10, "desc": "10条记录"}, {"limit": 20, "desc": "20条记录"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '工作台', 'endpoint': f'/api/dashboard/recent-publishes ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '工作台', 'endpoint': f'/api/dashboard/recent-publishes ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 9. 热门产品 print("[9/9] 测试热门产品...") url = f"{API_BASE_URL}/api/dashboard/hot-products" test_cases = [ {"limit": 4, "desc": "4个产品"}, {"limit": 8, "desc": "8个产品"}, {"limit": 12, "desc": "12个产品"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '工作台', 'endpoint': f'/api/dashboard/hot-products ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '工作台', 'endpoint': f'/api/dashboard/hot-products ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '工作台', 'endpoint': '/api/dashboard/overview', 'method': 'GET', 'description': '工作台概览(统计数据)', 'auth': '需要'}, {'module': '工作台', 'endpoint': '/api/dashboard/recent-publishes', 'method': 'GET', 'description': '最近发布(可指定条数)', 'auth': '需要'}, {'module': '工作台', 'endpoint': '/api/dashboard/hot-products', 'method': 'GET', 'description': '热门产品(按发布次数)', 'auth': '需要'}, ]) # ============================================================================ # 模块4: 企业接口测试 (原脚本测试数: 15个) # ============================================================================ def test_enterprise_apis(token): """测试企业接口 - 列表、统计、信息、CRUD操作""" if not TEST_MODULES['enterprise']: return print_module_header("企业接口测试", 4, 10) headers = {"Authorization": f"Bearer {token}"} # 1-5. 企业列表 (不同查询参数) print("\n[1-5/17] 测试企业列表(不同参数)...") url = f"{API_BASE_URL}/api/enterprises/list" test_cases = [ {"page": 1, "pageSize": 20, "desc": "基本查询"}, {"page": 1, "pageSize": 10, "desc": "每页10条"}, {"page": 1, "pageSize": 20, "keyword": "企业", "desc": "关键词搜索"}, {"page": 1, "pageSize": 20, "status": "active", "desc": "状态筛选"}, {"page": 2, "pageSize": 20, "desc": "第2页"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '企业接口', 'endpoint': f'/api/enterprises/list ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '企业接口', 'endpoint': f'/api/enterprises/list ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6-10. 企业统计 (多次测试) print("[6-10/17] 测试企业统计(5次)...") url = f"{API_BASE_URL}/api/enterprises/stats" for i in range(5): try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '企业接口', 'endpoint': f'/api/enterprises/stats (第{i+1}次)', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '企业接口', 'endpoint': f'/api/enterprises/stats (第{i+1}次)', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 11-15. 企业信息 (多次测试) print("[11-15/17] 测试企业信息(5次)...") url = f"{API_BASE_URL}/api/enterprises/info" for i in range(5): try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '企业接口', 'endpoint': f'/api/enterprises/info (第{i+1}次)', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '企业接口', 'endpoint': f'/api/enterprises/info (第{i+1}次)', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 16. 更新企业信息 print("[16/17] 测试更新企业信息...") url = f"{API_BASE_URL}/api/enterprises/info" data = {"short_name": f"测试更新{int(time.time())}", "email": "test@example.com"} try: start_time = time.time() response = requests.put(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '企业接口', 'endpoint': '/api/enterprises/info (PUT)', 'method': 'PUT', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '企业接口', 'endpoint': '/api/enterprises/info (PUT)', 'method': 'PUT', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 17. 修改密码 print("[17/17] 测试修改密码...") url = f"{API_BASE_URL}/api/enterprises/change-password" data = {"old_password": "wrong", "new_password": "test123"} try: start_time = time.time() response = requests.put(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '企业接口', 'endpoint': '/api/enterprises/change-password', 'method': 'PUT', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '企业接口', 'endpoint': '/api/enterprises/change-password', 'method': 'PUT', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '企业接口', 'endpoint': '/api/enterprises/list', 'method': 'GET', 'description': '企业列表(支持分页、搜索、筛选)', 'auth': '需要'}, {'module': '企业接口', 'endpoint': '/api/enterprises/stats', 'method': 'GET', 'description': '企业统计(总数、激活数等)', 'auth': '需要'}, {'module': '企业接口', 'endpoint': '/api/enterprises/info', 'method': 'GET', 'description': '企业信息', 'auth': '需要'}, {'module': '企业接口', 'endpoint': '/api/enterprises/info', 'method': 'PUT', 'description': '更新企业信息', 'auth': '需要'}, {'module': '企业接口', 'endpoint': '/api/enterprises/change-password', 'method': 'PUT', 'description': '修改密码', 'auth': '需要'}, ]) # ============================================================================ # 模块5: 图片接口测试 (原脚本测试数: 17个) # ============================================================================ def test_image_apis(token): """测试图片接口 - 图片、标签、类型管理""" if not TEST_MODULES['image']: return print_module_header("图片接口测试", 5, 10) headers = {"Authorization": f"Bearer {token}"} # 1-2. 图片列表 print("\n[1-2/17] 测试图片列表(不同参数)...") url = f"{API_BASE_URL}/api/images/list" test_cases = [ {"page": 1, "pageSize": 20, "desc": "第1页"}, {"page": 1, "pageSize": 10, "desc": "每页10条"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': f'/api/images/list ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': f'/api/images/list ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 3. 图片仪表盘 print("[3/17] 测试图片仪表盘...") url = f"{API_BASE_URL}/api/images/list_dashboard" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/list_dashboard', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/list_dashboard', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 4. 上传图片 print("[4/17] 测试上传图片...") url = f"{API_BASE_URL}/api/images/upload" try: img = Image.new('RGB', (100, 100)) img_bytes = io.BytesIO() img.save(img_bytes, format='PNG') img_bytes.seek(0) files = {'image': ('test.png', img_bytes, 'image/png')} data = {'image_name': '测试图片', 'image_type_name': '测试类型', 'tag_keywords': '测试,性能'} start_time = time.time() response = requests.post(url, headers=headers, files=files, data=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/upload', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/upload', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 5. 删除图片 print("[5/17] 测试删除图片...") url = f"{API_BASE_URL}/api/images/999999" try: start_time = time.time() response = requests.delete(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/', 'method': 'DELETE', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 404], 'error': '' }) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/', 'method': 'DELETE', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6. 标签名称列表 print("[6/17] 测试标签名称列表...") url = f"{API_BASE_URL}/api/images/tags/names/list" try: start_time = time.time() response = requests.get(url, headers=headers, params={"page": 1, "pageSize": 20}, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/tags/names/list', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/tags/names/list', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 7. 创建标签名称 print("[7/17] 测试创建标签名称...") url = f"{API_BASE_URL}/api/images/tags/names/create" data = {"tag_name": f"测试标签{int(time.time())}", "description": "测试"} try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/tags/names/create', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': '/api/images/tags/names/create', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 8-11. 标签名称CRUD操作 print("[8-11/17] 测试标签名称CRUD...") endpoints = [ ('/api/images/tags/names/999999', 'PUT', '更新标签名称'), ('/api/images/tags/names/999999', 'DELETE', '删除标签名称'), ('/api/images/tags/relations/list', 'GET', '标签关系列表'), ('/api/images/types/list', 'GET', '图片类型列表'), ] for endpoint, method, desc in endpoints: url = f"{API_BASE_URL}{endpoint}" try: start_time = time.time() if method == 'GET': response = requests.get(url, headers=headers, params={"page": 1, "pageSize": 20}, timeout=30) elif method == 'PUT': response = requests.put(url, headers=headers, json={"description": "更新"}, timeout=30) else: response = requests.delete(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': f'{endpoint} ({desc})', 'method': method, 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 404], 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': f'{endpoint} ({desc})', 'method': method, 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 12-17. 图片类型和标签管理 print("[12-17/17] 测试类型和标签管理...") test_cases = [ ('/api/images/types/create', 'POST', {"type_name": f"测试类型{int(time.time())}"}, '创建类型'), ('/api/images/types/999999', 'PUT', {"type_name": "更新类型"}, '更新类型'), ('/api/images/types/999999', 'DELETE', None, '删除类型'), ('/api/images/tags/list', 'GET', None, '标签列表'), ('/api/images/tags/create', 'POST', {"image_id": 1, "tag_id": 1}, '创建标签'), ('/api/images/tags/999999', 'DELETE', None, '删除标签'), ] for endpoint, method, data, desc in test_cases: url = f"{API_BASE_URL}{endpoint}" try: start_time = time.time() if method == 'GET': response = requests.get(url, headers=headers, params={"page": 1, "pageSize": 20}, timeout=30) elif method == 'POST': response = requests.post(url, headers=headers, json=data, timeout=30) elif method == 'PUT': response = requests.put(url, headers=headers, json=data, timeout=30) else: response = requests.delete(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '图片接口', 'endpoint': f'{endpoint} ({desc})', 'method': method, 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400, 404, 409], 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '图片接口', 'endpoint': f'{endpoint} ({desc})', 'method': method, 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '图片接口', 'endpoint': '/api/images/list', 'method': 'GET', 'description': '图片列表', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/list_dashboard', 'method': 'GET', 'description': '图片仪表盘', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/upload', 'method': 'POST', 'description': '上传图片', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/', 'method': 'DELETE', 'description': '删除图片', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/names/list', 'method': 'GET', 'description': '标签名称列表', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/names/create', 'method': 'POST', 'description': '创建标签名称', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/names/', 'method': 'PUT', 'description': '更新标签名称', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/names/', 'method': 'DELETE', 'description': '删除标签名称', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/relations/list', 'method': 'GET', 'description': '标签关系列表', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/types/list', 'method': 'GET', 'description': '图片类型列表', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/types/create', 'method': 'POST', 'description': '创建类型', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/types/', 'method': 'PUT', 'description': '更新类型', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/types/', 'method': 'DELETE', 'description': '删除类型', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/list', 'method': 'GET', 'description': '标签列表', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/create', 'method': 'POST', 'description': '创建标签', 'auth': '需要'}, {'module': '图片接口', 'endpoint': '/api/images/tags/', 'method': 'DELETE', 'description': '删除标签', 'auth': '需要'}, ]) # ============================================================================ # 模块6: 日志接口测试 (原脚本测试数: 7个) # ============================================================================ def test_log_apis(token): """测试日志接口 - 操作日志、文件日志""" if not TEST_MODULES['log']: return print_module_header("日志接口测试", 6, 10) headers = {"Authorization": f"Bearer {token}"} # 1-3. 获取操作日志 (不同参数) print("\n[1-3/7] 测试获取操作日志...") url = f"{API_BASE_URL}/api/log/logs" test_cases = [ {"page": 1, "size": 10, "desc": "第1页10条"}, {"page": 1, "size": 20, "desc": "第1页20条"}, {"page": 2, "size": 10, "desc": "第2页10条"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '日志接口', 'endpoint': f'/api/log/logs ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 500], 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '日志接口', 'endpoint': f'/api/log/logs ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 4. 日志文件列表 print("[4/7] 测试日志文件列表...") url = f"{API_BASE_URL}/api/log/logs/files" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '日志接口', 'endpoint': '/api/log/logs/files', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '日志接口', 'endpoint': '/api/log/logs/files', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 5-7. 文件日志内容 (不同类型和行数) print("[5-7/7] 测试文件日志内容...") url = f"{API_BASE_URL}/api/log/logs/file" test_cases = [ {"type": "article", "lines": 50, "desc": "article日志50行"}, {"type": "error", "lines": 50, "desc": "error日志50行"}, {"type": "article", "lines": 100, "desc": "article日志100行"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '日志接口', 'endpoint': f'/api/log/logs/file ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 404], 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '日志接口', 'endpoint': f'/api/log/logs/file ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '日志接口', 'endpoint': '/api/log/logs', 'method': 'GET', 'description': '获取操作日志(分页)', 'auth': '需要'}, {'module': '日志接口', 'endpoint': '/api/log/logs/files', 'method': 'GET', 'description': '日志文件列表', 'auth': '需要'}, {'module': '日志接口', 'endpoint': '/api/log/logs/file', 'method': 'GET', 'description': '文件日志内容', 'auth': '需要'}, ]) # ============================================================================ # 模块7: 文章接口测试 (原脚本测试数: 10个) # ============================================================================ def test_article_apis(token): """测试文章接口 - 列表、详情、发布""" if not TEST_MODULES['article']: return print_module_header("文章接口测试", 7, 10) headers = {"Authorization": f"Bearer {token}"} # 1-3. 文章列表 (不同参数) print("\n[1-3/10] 测试文章列表...") url = f"{API_BASE_URL}/api/articles/list" test_cases = [ {"page": 1, "pageSize": 20, "desc": "基本查询"}, {"page": 1, "pageSize": 10, "keyword": "文章", "desc": "关键词搜索"}, {"page": 2, "pageSize": 20, "desc": "第2页"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '文章接口', 'endpoint': f'/api/articles/list ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '文章接口', 'endpoint': f'/api/articles/list ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 4. 文章仪表盘 print("[4/10] 测试文章仪表盘...") url = f"{API_BASE_URL}/api/articles/list_dashboard" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '文章接口', 'endpoint': '/api/articles/list_dashboard', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '文章接口', 'endpoint': '/api/articles/list_dashboard', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 5. 生成文章 print("[5/10] 测试生成文章...") url = f"{API_BASE_URL}/api/articles/generate" data = {"product_id": 1, "prompt_workflow_id": 1, "topics": ["测试主题1", "测试主题2"], "count": 2} try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '文章接口', 'endpoint': '/api/articles/generate', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '文章接口', 'endpoint': '/api/articles/generate', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6-10. 文章详情和更新（如果有文章） print("[6-10/10] 测试文章详情和操作...") list_response = requests.get(f"{API_BASE_URL}/api/articles/list", headers=headers, params={"page": 1, "pageSize": 1}, timeout=30) if list_response.status_code == 200: articles = list_response.json().get('data', {}).get('list', []) if articles: article_id = articles[0]['id'] # 文章详情 url = f"{API_BASE_URL}/api/articles/{article_id}" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '文章接口', 'endpoint': f'/api/articles/{article_id}', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '文章接口', 'endpoint': f'/api/articles/{article_id}', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 更新文章 data = {"title": f"测试更新标题_{int(time.time())}", "status": "draft"} try: start_time = time.time() response = requests.put(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '文章接口', 'endpoint': f'/api/articles/{article_id} (PUT)', 'method': 'PUT', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '文章接口', 'endpoint': f'/api/articles/{article_id} (PUT)', 'method': 'PUT', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 批量发布测试 print(" 测试批量发布接口...") endpoints = [ ('/api/articles/batch-published-account-cycle', {"article_ids": [1, 2], "author_ids": [1, 2]}, '账号循环'), ('/api/articles/batch-published-review', {"article_ids": [1, 2], "author_id": 1}, '指定作者'), ] for endpoint, data, desc in endpoints: url = f"{API_BASE_URL}{endpoint}" try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '文章接口', 'endpoint': f'{endpoint} ({desc})', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '文章接口', 'endpoint': f'{endpoint} ({desc})', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '文章接口', 'endpoint': '/api/articles/list', 'method': 'GET', 'description': '文章列表', 'auth': '需要'}, {'module': '文章接口', 'endpoint': '/api/articles/list_dashboard', 'method': 'GET', 'description': '文章仪表盘', 'auth': '需要'}, {'module': '文章接口', 'endpoint': '/api/articles/generate', 'method': 'POST', 'description': '生成文章', 'auth': '需要'}, {'module': '文章接口', 'endpoint': '/api/articles/', 'method': 'GET', 'description': '文章详情', 'auth': '需要'}, {'module': '文章接口', 'endpoint': '/api/articles/', 'method': 'PUT', 'description': '更新文章', 'auth': '需要'}, {'module': '文章接口', 'endpoint': '/api/articles/batch-published-account-cycle', 'method': 'POST', 'description': '批量发布(账号循环)', 'auth': '需要'}, {'module': '文章接口', 'endpoint': '/api/articles/batch-published-review', 'method': 'POST', 'description': '批量发布(指定作者)', 'auth': '需要'}, ]) # ============================================================================ # 模块8: 员工接口测试 (原脚本测试数: 8个) # ============================================================================ def test_employee_apis(token): """测试员工接口 - 列表、统计、CRUD""" if not TEST_MODULES['employee']: return print_module_header("员工接口测试", 8, 10) headers = {"Authorization": f"Bearer {token}"} # 1-5. 员工列表 (不同参数) print("\n[1-5/8] 测试员工列表(不同参数)...") url = f"{API_BASE_URL}/api/employees/list" test_cases = [ {"page": 1, "pageSize": 20, "desc": "基本查询"}, {"page": 1, "pageSize": 10, "desc": "每页10条"}, {"page": 1, "pageSize": 20, "keyword": "员工", "desc": "关键词搜索"}, {"page": 1, "pageSize": 20, "status": "active", "desc": "状态筛选"}, {"page": 2, "pageSize": 20, "desc": "第2页"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '员工接口', 'endpoint': f'/api/employees/list ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '员工接口', 'endpoint': f'/api/employees/list ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6. 员工统计 print("[6/8] 测试员工统计...") url = f"{API_BASE_URL}/api/employees/stats" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '员工接口', 'endpoint': '/api/employees/stats', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '员工接口', 'endpoint': '/api/employees/stats', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 7. 添加员工 print("[7/8] 测试添加员工...") url = f"{API_BASE_URL}/api/employees/add" timestamp = int(time.time()) data = { "name": f"测试员工{timestamp}", "phone": f"1382{timestamp % 10000000:07d}", "password": "test123456", "role": "employee", "department": "测试部门" } try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '员工接口', 'endpoint': '/api/employees/add', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '员工接口', 'endpoint': '/api/employees/add', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 8. 删除员工 print("[8/8] 测试删除员工...") url = f"{API_BASE_URL}/api/employees/999999" try: start_time = time.time() response = requests.delete(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '员工接口', 'endpoint': '/api/employees/', 'method': 'DELETE', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 404], 'error': '' }) except Exception as e: test_results.append({ 'module': '员工接口', 'endpoint': '/api/employees/', 'method': 'DELETE', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '员工接口', 'endpoint': '/api/employees/list', 'method': 'GET', 'description': '员工列表(分页、搜索)', 'auth': '需要'}, {'module': '员工接口', 'endpoint': '/api/employees/stats', 'method': 'GET', 'description': '员工统计', 'auth': '需要'}, {'module': '员工接口', 'endpoint': '/api/employees/add', 'method': 'POST', 'description': '添加员工', 'auth': '需要'}, {'module': '员工接口', 'endpoint': '/api/employees/', 'method': 'DELETE', 'description': '删除员工', 'auth': '需要'}, ]) # ============================================================================ # 模块9: 作者接口测试 (原脚本测试数: 20个) # ============================================================================ def test_author_apis(token): """测试作者接口 - 列表、搜索、CRUD""" if not TEST_MODULES['author']: return print_module_header("作者接口测试", 9, 10) headers = {"Authorization": f"Bearer {token}"} # 1-3. 作者列表 (不同参数) print("\n[1-3/12] 测试作者列表...") url = f"{API_BASE_URL}/api/authors" test_cases = [ {"page": 1, "size": 10, "desc": "第1页"}, {"page": 1, "size": 5, "desc": "每页5条"}, {"page": 1, "size": 10, "search": "医生", "desc": "搜索关键词"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': f'/api/authors ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': f'/api/authors ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 4. 作者列表(用户关联) print("[4/12] 测试作者列表(用户关联)...") url = f"{API_BASE_URL}/api/authors/list" try: start_time = time.time() response = requests.get(url, headers=headers, params={"page": 1, "size": 10}, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': '/api/authors/list', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': '/api/authors/list', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 5-6. 作者详细列表 print("[5-6/12] 测试作者详细列表...") url = f"{API_BASE_URL}/api/authors/detail_list" test_cases = [ {"page": 1, "size": 10, "desc": "基本查询"}, {"page": 1, "size": 20, "search": "医生", "desc": "搜索查询"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': f'/api/authors/detail_list ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': f'/api/authors/detail_list ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 7. 科室列表 print("[7/12] 测试科室列表...") url = f"{API_BASE_URL}/api/authors/departments" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': '/api/authors/departments', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': '/api/authors/departments', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 8-9. 搜索作者 print("[8-9/12] 测试搜索作者...") url = f"{API_BASE_URL}/api/authors/search" test_cases = [ {"keyword": "医生", "limit": 10, "desc": "搜索医生"}, {"keyword": "科室", "limit": 5, "desc": "搜索科室"}, ] for i, params in enumerate(test_cases, 1): desc = params.pop('desc') try: start_time = time.time() response = requests.get(url, headers=headers, params=params, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': f'/api/authors/search ({desc})', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': f'/api/authors/search ({desc})', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 10-12. 作者CRUD操作 print("[10-12/12] 测试作者CRUD...") # 创建作者 url = f"{API_BASE_URL}/api/authors" timestamp = int(time.time()) data = { "author_name": f"测试医生{timestamp}", "app_id": f"test_app_{timestamp}", "app_token": f"test_token_{timestamp}", "department_name": "测试科室", "title": "主任医师", "hospital": "测试医院", "channel": 1 } try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': '/api/authors (POST)', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 201, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': '/api/authors (POST)', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 更新和删除 for endpoint, method, desc in [ ('/api/authors/999999', 'PUT', '更新作者'), ('/api/authors/999999', 'DELETE', '删除作者'), ]: url = f"{API_BASE_URL}{endpoint}" try: start_time = time.time() if method == 'PUT': response = requests.put(url, headers=headers, json={"author_name": "更新测试"}, timeout=30) else: response = requests.delete(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '作者接口', 'endpoint': f'{endpoint} ({desc})', 'method': method, 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 404], 'error': '' }) time.sleep(0.1) except Exception as e: test_results.append({ 'module': '作者接口', 'endpoint': f'{endpoint} ({desc})', 'method': method, 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '作者接口', 'endpoint': '/api/authors', 'method': 'GET', 'description': '作者列表(按科室分组)', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors', 'method': 'POST', 'description': '创建作者', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors/', 'method': 'PUT', 'description': '更新作者', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors/', 'method': 'DELETE', 'description': '删除作者', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors/list', 'method': 'GET', 'description': '作者列表(用户关联)', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors/detail_list', 'method': 'GET', 'description': '作者详细列表', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors/departments', 'method': 'GET', 'description': '科室列表', 'auth': '需要'}, {'module': '作者接口', 'endpoint': '/api/authors/search', 'method': 'GET', 'description': '搜索作者', 'auth': '需要'}, ]) # ============================================================================ # 模块10: 搜索服务接口测试 (原脚本测试数: 6个) # ============================================================================ def test_search_apis(token): """测试搜索服务接口 - 独立服务（8321端口）""" if not TEST_MODULES['search']: return print_module_header("搜索服务接口测试", 10, 10) print("注意: 搜索服务使用独立端口 8321") headers = {"Authorization": f"Bearer {token}"} # 1. 健康检查 print("\n[1/6] 测试健康检查...") url = f"{SEARCH_API_BASE_URL}/health" try: start_time = time.time() response = requests.get(url, timeout=5) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '搜索服务', 'endpoint': '/health', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '搜索服务', 'endpoint': '/health', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 2. 查询日志 print("[2/6] 测试查询日志...") url = f"{SEARCH_API_BASE_URL}/api/search-service/logs" try: start_time = time.time() response = requests.get(url, headers=headers, params={"page": 1, "size": 10}, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/logs', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 500], 'error': '' }) except Exception as e: test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/logs', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 3. 日志文件列表 print("[3/6] 测试日志文件列表...") url = f"{SEARCH_API_BASE_URL}/api/search-service/logs/files" try: start_time = time.time() response = requests.get(url, headers=headers, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/logs/files', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code == 200, 'error': '' }) except Exception as e: test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/logs/files', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 4. 文件日志内容 print("[4/6] 测试文件日志内容...") url = f"{SEARCH_API_BASE_URL}/api/search-service/logs/file" try: start_time = time.time() response = requests.get(url, headers=headers, params={"type": "article", "lines": 50}, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/logs/file', 'method': 'GET', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 404], 'error': '' }) except Exception as e: test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/logs/file', 'method': 'GET', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 5. 生成文章 print("[5/6] 测试生成文章...") url = f"{SEARCH_API_BASE_URL}/api/search-service/generate_article" data = {"query": "测试查询", "max_results": 5} try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/generate_article', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/generate_article', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) # 6. 关键词搜索 print("[6/6] 测试关键词搜索...") url = f"{SEARCH_API_BASE_URL}/api/search-service/search" data = {"keyword": "测试", "limit": 10} try: start_time = time.time() response = requests.post(url, headers=headers, json=data, timeout=30) elapsed_time = (time.time() - start_time) * 1000 test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/search', 'method': 'POST', 'status_code': response.status_code, 'elapsed_time_ms': f"{elapsed_time:.2f}", 'success': response.status_code in [200, 400], 'error': '' }) except Exception as e: test_results.append({ 'module': '搜索服务', 'endpoint': '/api/search-service/search', 'method': 'POST', 'status_code': 0, 'elapsed_time_ms': '0', 'success': False, 'error': str(e) }) api_docs.extend([ {'module': '搜索服务', 'endpoint': '/health', 'method': 'GET', 'description': '健康检查', 'auth': '不需要'}, {'module': '搜索服务', 'endpoint': '/api/search-service/logs', 'method': 'GET', 'description': '查询日志', 'auth': '需要'}, {'module': '搜索服务', 'endpoint': '/api/search-service/logs/files', 'method': 'GET', 'description': '日志文件列表', 'auth': '需要'}, {'module': '搜索服务', 'endpoint': '/api/search-service/logs/file', 'method': 'GET', 'description': '文件日志内容', 'auth': '需要'}, {'module': '搜索服务', 'endpoint': '/api/search-service/generate_article', 'method': 'POST', 'description': '生成文章', 'auth': '需要'}, {'module': '搜索服务', 'endpoint': '/api/search-service/search', 'method': 'POST', 'description': '关键词搜索', 'auth': '需要'}, ]) # ============================================================================ # 结果保存和输出 # ============================================================================ def save_to_csv(): """保存测试结果到CSV文件""" timestamp = datetime.now().strftime('%Y%m%d') filename = f'整体后端接口性能测试_{timestamp}.csv' with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=[ 'module', 'endpoint', 'method', 'status_code', 'elapsed_time_ms', 'success', 'error' ]) writer.writeheader() writer.writerows(test_results) print(f"\n✓ 性能测试结果已保存到: {filename}") return filename def save_api_docs(): """保存接口文档到CSV文件""" timestamp = datetime.now().strftime('%Y%m%d') filename = f'整体后端接口文档_{timestamp}.csv' with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=[ 'module', 'endpoint', 'method', 'description', 'auth' ]) writer.writeheader() writer.writerows(api_docs) print(f"✓ 接口文档已保存到: {filename}") return filename def generate_markdown_report(): """生成Markdown格式的综合数据报表""" from collections import defaultdict timestamp = datetime.now().strftime('%Y%m%d') filename = f'整体后端接口测试报表_{timestamp}.md' # 总体统计 total = len(test_results) success = sum(1 for r in test_results if r['success']) failed = total - success times = [float(r['elapsed_time_ms']) for r in test_results if r['elapsed_time_ms'] != '0'] avg_time = sum(times) / len(times) if times else 0 max_time = max(times) if times else 0 min_time = min(times) if times else 0 # 找出最快和最慢的接口 max_item = max(test_results, key=lambda x: float(x['elapsed_time_ms']) if x['elapsed_time_ms'] != '0' else 0) min_item = min([r for r in test_results if r['elapsed_time_ms'] != '0'], key=lambda x: float(x['elapsed_time_ms'])) if times else None # 按模块统计 modules = {} for r in test_results: module = r['module'] if module not in modules: modules[module] = { 'total': 0, 'success': 0, 'times': [], 'endpoints': [], 'errors': [] } modules[module]['total'] += 1 if r['success']: modules[module]['success'] += 1 else: modules[module]['errors'].append(r['endpoint']) if r['elapsed_time_ms'] != '0': modules[module]['times'].append(float(r['elapsed_time_ms'])) modules[module]['endpoints'].append((r['endpoint'], float(r['elapsed_time_ms']))) # HTTP方法统计 methods = defaultdict(lambda: {'total': 0, 'success': 0}) for r in test_results: method = r['method'] methods[method]['total'] += 1 if r['success']: methods[method]['success'] += 1 # 性能分布 fast = sum(1 for t in times if t < 100) normal = sum(1 for t in times if 100 <= t < 500) slow = sum(1 for t in times if 500 <= t < 1000) very_slow = sum(1 for t in times if t >= 1000) # 状态码分布 status_codes = defaultdict(int) for r in test_results: status_codes[r['status_code']] += 1 # 接口文档统计 doc_modules = defaultdict(int) doc_methods = defaultdict(int) auth_required = sum(1 for d in api_docs if d['auth'] == '需要') for doc in api_docs: doc_modules[doc['module']] += 1 doc_methods[doc['method']] += 1 # 开始写入Markdown文件 with open(filename, 'w', encoding='utf-8') as f: # 标题和概览 f.write("# 整体后端接口性能测试报表\n\n") f.write(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") f.write("---\n\n") # 总体统计 f.write("## 📈 总体测试统计\n\n") f.write("| 指标 | 数值 |\n") f.write("|------|------|\n") f.write(f"| 测试总数 | {total} 个 |\n") f.write(f"| 成功数量 | {success} 个 ✓ |\n") f.write(f"| 失败数量 | {failed} 个 ✗ |\n") f.write(f"| **成功率** | **{success/total*100:.2f}%** |\n") f.write(f"| 平均耗时 | {avg_time:.2f}ms |\n") f.write(f"| 最大耗时 | {max_time:.2f}ms |\n") f.write(f"| 最小耗时 | {min_time:.2f}ms |\n\n") if max_item: f.write("**最慢接口**:\n") f.write(f"- `{max_item['endpoint']}` ({max_item['module']}) - {max_item['elapsed_time_ms']}ms\n\n") if min_item: f.write("**最快接口**:\n") f.write(f"- `{min_item['endpoint']}` ({min_item['module']}) - {min_item['elapsed_time_ms']}ms\n\n") f.write("---\n\n") # 各模块性能分析 f.write("## 📊 各模块性能分析\n\n") sorted_modules = sorted(modules.items(), key=lambda x: (x[1]['success']/x[1]['total'] if x[1]['total'] > 0 else 0), reverse=True) f.write("| 模块 | 测试数 | 成功 | 失败 | 成功率 | 平均耗时 | 最大耗时 | 最小耗时 | 状态 |\n") f.write("|------|--------|------|------|--------|----------|----------|----------|------|\n") for module, stats in sorted_modules: success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 avg_time = (sum(stats['times']) / len(stats['times'])) if stats['times'] else 0 max_time = max(stats['times']) if stats['times'] else 0 min_time = min(stats['times']) if stats['times'] else 0 if success_rate >= 90: status = "🟢 优秀" elif success_rate >= 70: status = "🟡 良好" elif success_rate >= 50: status = "🟠 一般" else: status = "🔴 需优化" f.write(f"| {module} | {stats['total']} | {stats['success']} | {stats['total']-stats['success']} | " f"{success_rate:.1f}% | {avg_time:.2f}ms | {max_time:.2f}ms | {min_time:.2f}ms | {status} |\n") f.write("\n") # 失败接口详情 f.write("### ⚠️ 失败接口详情\n\n") has_errors = False for module, stats in sorted_modules: if stats['errors']: has_errors = True f.write(f"**{module}** ({len(stats['errors'])}个失败): \n") for err_endpoint in stats['errors']: f.write(f"- `{err_endpoint}`\n") f.write("\n") if not has_errors: f.write("✅ 所有接口测试通过！\n\n") f.write("---\n\n") # HTTP方法分布 f.write("## 🔧 HTTP方法分布\n\n") f.write("| 方法 | 总数 | 成功 | 失败 | 成功率 |\n") f.write("|------|------|------|------|--------|\n") for method, stats in sorted(methods.items()): success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 f.write(f"| {method} | {stats['total']} | {stats['success']} | " f"{stats['total']-stats['success']} | {success_rate:.1f}% |\n") f.write("\n---\n\n") # 性能分布 f.write("## ⚡ 性能分布分析\n\n") f.write("| 性能等级 | 数量 | 占比 |\n") f.write("|----------|------|------|\n") f.write(f"| 🚀 极快 (<100ms) | {fast} | {fast/len(times)*100:.1f}% |\n") f.write(f"| ✅ 正常 (100-500ms) | {normal} | {normal/len(times)*100:.1f}% |\n") f.write(f"| ⚠️ 较慢 (500-1000ms) | {slow} | {slow/len(times)*100:.1f}% |\n") f.write(f"| 🐌 很慢 (>1000ms) | {very_slow} | {very_slow/len(times)*100:.1f}% |\n") f.write("\n") # 性能分布图（文本版） f.write("### 性能分布可视化\n\n") f.write("```\n") max_bar_len = 50 if times: fast_bar = '█' * int((fast/len(times)) * max_bar_len) normal_bar = '█' * int((normal/len(times)) * max_bar_len) slow_bar = '█' * int((slow/len(times)) * max_bar_len) very_slow_bar = '█' * int((very_slow/len(times)) * max_bar_len) f.write(f"极快 (<100ms): {fast_bar} {fast}个\n") f.write(f"正常 (100-500ms): {normal_bar} {normal}个\n") f.write(f"较慢 (500-1000ms):{slow_bar} {slow}个\n") f.write(f"很慢 (>1000ms): {very_slow_bar} {very_slow}个\n") f.write("```\n\n") f.write("---\n\n") # HTTP状态码分布 f.write("## 📋 HTTP状态码分布\n\n") f.write("| 状态码 | 数量 | 占比 | 说明 |\n") f.write("|--------|------|------|------|\n") for code, count in sorted(status_codes.items()): percentage = (count / total * 100) if code == 200: desc = "✅ 成功" elif code >= 500: desc = "❌ 服务器错误" elif code >= 400: desc = "⚠️ 客户端错误" elif code == 0: desc = "❌ 请求失败" else: desc = "ℹ️ 其他" f.write(f"| {code} | {count} | {percentage:.1f}% | {desc} |\n") f.write("\n---\n\n") # 接口文档统计 f.write("## 📚 接口文档统计\n\n") f.write(f"**已记录接口总数**: {len(api_docs)} 个 \n") f.write(f"**需要认证**: {auth_required} 个 ({auth_required/len(api_docs)*100:.1f}%)\n\n") f.write("### 按模块分类\n\n") f.write("| 模块 | 接口数 |\n") f.write("|------|--------|\n") for module, count in sorted(doc_modules.items(), key=lambda x: x[1], reverse=True): f.write(f"| {module} | {count} |\n") f.write("\n") f.write("### 按方法分类\n\n") f.write("| 方法 | 接口数 |\n") f.write("|------|--------|\n") for method, count in sorted(doc_methods.items()): f.write(f"| {method} | {count} |\n") f.write("\n") f.write("---\n\n") # 优化建议 f.write("## 💡 优化建议\n\n") suggestions = [] # 检查失败率高的模块 for module, stats in modules.items(): success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 if success_rate < 70: suggestions.append(f"🔴 **{module}** 模块成功率仅 {success_rate:.1f}%，建议优先排查认证、参数和数据库连接问题") # 检查慢接口 slow_endpoints = [(r['module'], r['endpoint'], float(r['elapsed_time_ms'])) for r in test_results if r['elapsed_time_ms'] != '0' and float(r['elapsed_time_ms']) > 500] if slow_endpoints: slow_endpoints.sort(key=lambda x: x[2], reverse=True) suggestions.append(f"⚠️ 发现 {len(slow_endpoints)} 个耗时超过500ms的接口，建议添加缓存或优化查询") suggestions.append(" 慢接口TOP 3:") for module, endpoint, time in slow_endpoints[:3]: suggestions.append(f" - `{endpoint}` ({module}) - {time:.2f}ms") # 检查错误率 if failed > 0: error_rate = failed / total * 100 if error_rate > 20: suggestions.append(f"❌ 总体错误率达到 {error_rate:.1f}%，建议进行全面的接口测试和修复") if suggestions: for suggestion in suggestions: f.write(f"{suggestion}\n") else: f.write("✅ 所有接口性能良好，无需特别优化\n") f.write("\n---\n\n") # 测试配置信息 f.write("## ⚙️ 测试配置\n\n") f.write(f"- **主服务地址**: `{API_BASE_URL}`\n") f.write(f"- **搜索服务地址**: `{SEARCH_API_BASE_URL}`\n") f.write(f"- **测试账号**: `{USERNAME}`\n") enabled_modules = [name for name, enabled in TEST_MODULES.items() if enabled] disabled_modules = [name for name, enabled in TEST_MODULES.items() if not enabled] f.write(f"- **启用模块** ({len(enabled_modules)}个): {', '.join(enabled_modules)}\n") if disabled_modules: f.write(f"- **禁用模块** ({len(disabled_modules)}个): {', '.join(disabled_modules)}\n") f.write("\n---\n\n") f.write("*报表由整体后端接口性能测试工具自动生成*\n") print(f"\n✅ Markdown报表已生成: {filename}") return filename def generate_data_report(): """生成综合数据报表""" from collections import defaultdict print("\n" + "█"*80) print("📊 综合数据报表") print("█"*80) # 总体统计 total = len(test_results) success = sum(1 for r in test_results if r['success']) failed = total - success times = [float(r['elapsed_time_ms']) for r in test_results if r['elapsed_time_ms'] != '0'] avg_time = sum(times) / len(times) if times else 0 max_time = max(times) if times else 0 min_time = min(times) if times else 0 # 找出最快和最慢的接口 max_item = min(test_results, key=lambda x: float(x['elapsed_time_ms']) if x['elapsed_time_ms'] != '0' else float('inf')) min_item = max(test_results, key=lambda x: float(x['elapsed_time_ms']) if x['elapsed_time_ms'] != '0' else 0) print("\n" + "="*80) print("📈 总体测试统计") print("="*80) print(f" 测试总数: {total} 个") print(f" 成功数量: {success} 个 ✓") print(f" 失败数量: {failed} 个 ✗") print(f" 成功率: {success/total*100:.2f}%") print(f" 平均耗时: {avg_time:.2f}ms") print(f" 最大耗时: {max_time:.2f}ms") print(f" └─ 接口: {max_item['endpoint']} ({max_item['module']})") print(f" 最小耗时: {min_time:.2f}ms") print(f" └─ 接口: {min_item['endpoint']} ({min_item['module']})") # 按模块统计 print("\n" + "="*80) print("📊 各模块性能分析") print("="*80) modules = {} for r in test_results: module = r['module'] if module not in modules: modules[module] = { 'total': 0, 'success': 0, 'times': [], 'endpoints': [], 'errors': [] } modules[module]['total'] += 1 if r['success']: modules[module]['success'] += 1 else: modules[module]['errors'].append(r['endpoint']) if r['elapsed_time_ms'] != '0': modules[module]['times'].append(float(r['elapsed_time_ms'])) modules[module]['endpoints'].append((r['endpoint'], float(r['elapsed_time_ms']))) # 按成功率排序 sorted_modules = sorted(modules.items(), key=lambda x: (x[1]['success']/x[1]['total'] if x[1]['total'] > 0 else 0), reverse=True) for module, stats in sorted_modules: success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 avg_time = (sum(stats['times']) / len(stats['times'])) if stats['times'] else 0 max_time = max(stats['times']) if stats['times'] else 0 min_time = min(stats['times']) if stats['times'] else 0 # 确定性能等级 if success_rate >= 90: status = "🟢 优秀" elif success_rate >= 70: status = "🟡 良好" elif success_rate >= 50: status = "🟠 一般" else: status = "🔴 需优化" print(f"\n{status} {module}") print(f" 测试数: {stats['total']} | 成功: {stats['success']} | 失败: {stats['total']-stats['success']}") print(f" 成功率: {success_rate:.2f}%") print(f" 平均耗时: {avg_time:.2f}ms | 最大: {max_time:.2f}ms | 最小: {min_time:.2f}ms") # 显示失败的接口 if stats['errors']: print(f" ⚠️ 失败接口 ({len(stats['errors'])}个):") for err_endpoint in stats['errors'][:3]: # 最多显示3个 print(f" • {err_endpoint}") if len(stats['errors']) > 3: print(f" • ... 还有 {len(stats['errors'])-3} 个") # HTTP方法统计 print("\n" + "="*80) print("🔧 HTTP方法分布") print("="*80) methods = defaultdict(lambda: {'total': 0, 'success': 0}) for r in test_results: method = r['method'] methods[method]['total'] += 1 if r['success']: methods[method]['success'] += 1 for method, stats in sorted(methods.items()): success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 print(f" {method:6} | 总数: {stats['total']:3} | 成功: {stats['success']:3} | 成功率: {success_rate:5.1f}%") # 性能分布 print("\n" + "="*80) print("⚡ 性能分布分析") print("="*80) fast = sum(1 for t in times if t < 100) normal = sum(1 for t in times if 100 <= t < 500) slow = sum(1 for t in times if 500 <= t < 1000) very_slow = sum(1 for t in times if t >= 1000) print(f" 🚀 极快 (<100ms): {fast:3} 个 ({fast/len(times)*100:.1f}%)") print(f" ✅ 正常 (100-500ms): {normal:3} 个 ({normal/len(times)*100:.1f}%)") print(f" ⚠️ 较慢 (500-1000ms): {slow:3} 个 ({slow/len(times)*100:.1f}%)") print(f" 🐌 很慢 (>1000ms): {very_slow:3} 个 ({very_slow/len(times)*100:.1f}%)") # 状态码分布 print("\n" + "="*80) print("📋 HTTP状态码分布") print("="*80) status_codes = defaultdict(int) for r in test_results: status_codes[r['status_code']] += 1 for code, count in sorted(status_codes.items()): percentage = (count / total * 100) if code == 200: print(f" ✅ {code}: {count:3} 个 ({percentage:.1f}%)") elif code >= 400: print(f" ❌ {code}: {count:3} 个 ({percentage:.1f}%)") else: print(f" ℹ️ {code}: {count:3} 个 ({percentage:.1f}%)") # 接口文档统计 print("\n" + "="*80) print("📚 接口文档统计") print("="*80) print(f" 已记录接口: {len(api_docs)} 个") doc_modules = defaultdict(int) doc_methods = defaultdict(int) auth_required = sum(1 for d in api_docs if d['auth'] == '需要') for doc in api_docs: doc_modules[doc['module']] += 1 doc_methods[doc['method']] += 1 print(f" 需要认证: {auth_required} 个") print(f"\n 按模块分类:") for module, count in sorted(doc_modules.items(), key=lambda x: x[1], reverse=True): print(f" • {module}: {count} 个") print(f"\n 按方法分类:") for method, count in sorted(doc_methods.items()): print(f" • {method}: {count} 个") print("\n" + "█"*80) def print_summary(): """打印测试摘要（兼容旧版）""" generate_data_report() def print_config(): """打印测试配置""" print("测试配置:") print("-"*80) enabled_modules = [name for name, enabled in TEST_MODULES.items() if enabled] disabled_modules = [name for name, enabled in TEST_MODULES.items() if not enabled] print(f"启用的模块 ({len(enabled_modules)}个): {', '.join(enabled_modules)}") if disabled_modules: print(f"禁用的模块 ({len(disabled_modules)}个): {', '.join(disabled_modules)}") print("-"*80) def main(): """主函数""" print("="*80) print("整合后端接口性能测试工具") print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"主服务地址: {API_BASE_URL}") print(f"搜索服务地址: {SEARCH_API_BASE_URL}") print("="*80) # 打印配置 print_config() # 登录 token = login() if not token: print("\n✗ 登录失败，无法继续测试") return 1 # 执行所有测试 try: test_basic_apis() test_auth_apis(token) test_dashboard_apis(token) test_enterprise_apis(token) test_image_apis(token) test_log_apis(token) test_article_apis(token) test_employee_apis(token) test_author_apis(token) test_search_apis(token) except Exception as e: print(f"\n✗ 测试过程中出现异常: {e}") import traceback traceback.print_exc() # 生成综合数据报表（控制台输出） generate_data_report() # 保存结果 csv_file = save_to_csv() doc_file = save_api_docs() md_report_file = generate_markdown_report() # 生成Markdown报表 print("\n" + "="*80) print("✅ 测试完成！") print("="*80) print(f"📁 输出文件:") print(f" • 性能数据: {csv_file}") print(f" • 接口文档: {doc_file}") print(f" • 测试报表: {md_report_file}") print(f"\n💡 提示:") print(f" • 查看Markdown报表: {md_report_file}") print(f" • 使用 'python 查看综合测试报表.py' 查看控制台版详细报表") print(f" • 成功率低于70%的模块需要优化") print(f" • 耗时超过500ms的接口建议添加缓存或优化查询") print("="*80) return 0 if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: print("\n\n✗ 测试被用户中断") sys.exit(1) except Exception as e: print(f"\n✗ 测试异常: {e}") import traceback traceback.print_exc() sys.exit(1)