#!/usr/bin/env python3 """ 优化历史功能测试脚本 测试API接口和数据库操作 """ import requests import json import time from datetime import datetime # 测试配置 BASE_URL = "http://localhost:5002" TEST_USER_ID = 1 def test_api_endpoint(endpoint, method="GET", data=None, headers=None): """测试API接口""" url = f"{BASE_URL}{endpoint}" try: if method == "GET": response = requests.get(url, headers=headers) elif method == "POST": response = requests.post(url, json=data, headers=headers) elif method == "PUT": response = requests.put(url, json=data, headers=headers) elif method == "DELETE": response = requests.delete(url, headers=headers) print(f"🔍 {method} {endpoint}") print(f" 状态码: {response.status_code}") if response.status_code == 200: result = response.json() if result.get('success'): print(f" ✅ 成功: {result.get('message', '操作成功')}") return result else: print(f" ❌ 失败: {result.get('message', '未知错误')}") return None else: print(f" ❌ HTTP错误: {response.status_code}") return None except Exception as e: print(f" ❌ 请求失败: {str(e)}") return None def test_optimization_history_page(): """测试优化历史页面""" print("=" * 60) print("🧪 测试优化历史页面") print("=" * 60) # 测试页面访问 response = requests.get(f"{BASE_URL}/optimization-history") if response.status_code == 200: print("✅ 优化历史页面访问成功") return True else: print(f"❌ 优化历史页面访问失败: {response.status_code}") return False def test_optimization_history_api(): """测试优化历史API接口""" print("=" * 60) print("🧪 测试优化历史API接口") print("=" * 60) # 模拟用户认证头（实际应用中需要真实的token） headers = { 'Content-Type': 'application/json', 'Authorization': 'Bearer test_token' # 这里需要真实的认证token } # 1. 测试获取历史记录 print("\n📋 测试获取历史记录...") result = test_api_endpoint("/api/optimization-history", "GET", headers=headers) # 2. 测试添加历史记录 print("\n➕ 测试添加历史记录...") test_data = { "original_text": "测试原始文本", "optimized_text": "这是测试优化后的文本内容，包含了更详细和结构化的信息。", "optimization_type": "测试类型", "industry": "测试行业", "profession": "测试职业", "tags": ["测试标签1", "测试标签2"] } result = test_api_endpoint("/api/optimization-history", "POST", data=test_data, headers=headers) # 3. 测试获取统计信息 print("\n📊 测试获取统计信息...") result = test_api_endpoint("/api/optimization-history/stats", "GET", headers=headers) # 4. 测试获取分析数据 print("\n📈 测试获取分析数据...") result = test_api_endpoint("/api/optimization-history/analytics", "GET", headers=headers) # 5. 测试获取收藏记录 print("\n⭐ 测试获取收藏记录...") result = test_api_endpoint("/api/optimization-history/favorites", "GET", headers=headers) return True def test_database_connection(): """测试数据库连接""" print("=" * 60) print("🧪 测试数据库连接") print("=" * 60) try: import pymysql from config import Config # 解析数据库连接信息 db_uri = Config.TENCENT_SQLALCHEMY_DATABASE_URI if db_uri.startswith('mysql+pymysql://'): db_uri = db_uri.replace('mysql+pymysql://', '') if '?' in db_uri: db_uri, params = db_uri.split('?', 1) auth_host_db = db_uri.split('@') auth_part = auth_host_db[0] host_db_part = auth_host_db[1] username, password = auth_part.split(':', 1) host_port, database = host_db_part.split('/', 1) host, port = host_port.split(':') port = int(port) # 连接数据库 connection = pymysql.connect( host=host, port=port, user=username, password=password, database=database, charset='utf8mb4' ) cursor = connection.cursor() # 检查表是否存在 tables = ['optimization_history', 'optimization_tags', 'optimization_favorites', 'user_usage_stats', 'system_config'] for table in tables: cursor.execute(f"SHOW TABLES LIKE '{table}'") if cursor.fetchone(): cursor.execute(f"SELECT COUNT(*) FROM {table}") count = cursor.fetchone()[0] print(f"✅ 表 {table} 存在，记录数: {count}") else: print(f"❌ 表 {table} 不存在") # 检查示例数据 cursor.execute("SELECT id, original_text, optimization_type FROM optimization_history LIMIT 3") records = cursor.fetchall() print(f"\n📋 示例历史记录:") for record in records: print(f" ID: {record[0]}, 类型: {record[2]}, 内容: {record[1][:30]}...") cursor.close() connection.close() print("✅ 数据库连接测试成功") return True except Exception as e: print(f"❌ 数据库连接测试失败: {str(e)}") return False def test_frontend_integration(): """测试前端集成""" print("=" * 60) print("🧪 测试前端集成") print("=" * 60) # 检查优化历史页面模板是否存在 try: with open('src/flask_prompt_master/templates/optimization_history.html', 'r', encoding='utf-8') as f: content = f.read() # 检查关键元素 checks = [ ('历史记录列表', 'historyList' in content), ('搜索功能', 'historySearchInput' in content), ('筛选功能', 'typeFilter' in content), ('统计卡片', 'stats-card' in content), ('分页功能', 'pagination' in content), ('操作按钮', 'btn-copy' in content), ('收藏功能', 'btn-favorite' in content), ('评分功能', 'rating' in content) ] print("📋 前端元素检查:") for name, exists in checks: if exists: print(f" ✅ {name}") else: print(f" ❌ {name}") return all(exists for _, exists in checks) except FileNotFoundError: print("❌ 优化历史页面模板文件不存在") return False except Exception as e: print(f"❌ 前端集成测试失败: {str(e)}") return False def main(): """主测试函数""" print("🚀 优化历史功能测试开始") print("=" * 60) tests = [ ("数据库连接", test_database_connection), ("前端集成", test_frontend_integration), ("页面访问", test_optimization_history_page), ("API接口", test_optimization_history_api) ] results = [] for test_name, test_func in tests: print(f"\n🧪 开始测试: {test_name}") try: result = test_func() results.append((test_name, result)) except Exception as e: print(f"❌ 测试 {test_name} 异常: {str(e)}") results.append((test_name, False)) # 输出测试结果 print("\n" + "=" * 60) print("📊 测试结果汇总") print("=" * 60) passed = 0 total = len(results) for test_name, result in results: status = "✅ 通过" if result else "❌ 失败" print(f"{test_name}: {status}") if result: passed += 1 print(f"\n📈 总体结果: {passed}/{total} 测试通过") if passed == total: print("🎉 所有测试通过！优化历史功能已就绪！") else: print("⚠️ 部分测试失败，请检查相关功能") return passed == total if __name__ == '__main__': success = main() exit(0 if success else 1)