#!/usr/bin/env python3 """ 测试运行器 - 统一管理所有测试 包含问题检测、修复建议和测试报告 """ import os import sys import subprocess import json import requests import time from pathlib import Path # 项目路径配置 PROJECT_ROOT = Path(__file__).parent.parent DJANGO_ROOT = PROJECT_ROOT / "信息抽取+数据检验/Django123/atc_extractor/backend" BACKEND_API_URL = "http://127.0.0.1:8000/api" class TestRunner: def __init__(self): self.test_results = { 'static_analysis': {}, 'api_tests': {}, 'database_tests': {}, 'security_tests': {}, 'integration_tests': {} } self.issues_found = [] def run_static_analysis(self): """静态代码分析""" print("🔍 进行静态代码分析...") issues = [] # 1. 检查数据库凭据安全性 settings_file = DJANGO_ROOT / "backend/settings.py" if settings_file.exists(): with open(settings_file, 'r', encoding='utf-8') as f: content = f.read() if 'hzk200407140238' in content: issues.append({ 'type': 'SECURITY', 'severity': 'HIGH', 'message': '数据库密码硬编码在源代码中', 'file': str(settings_file), 'recommendation': '使用环境变量或配置文件管理敏感信息' }) if 'DEBUG = True' in content: issues.append({ 'type': 'SECURITY', 'severity': 'MEDIUM', 'message': 'DEBUG模式在生产环境中应该关闭', 'file': str(settings_file), 'recommendation': '在生产环境中设置DEBUG = False' }) if 'django-insecure' in content: issues.append({ 'type': 'SECURITY', 'severity': 'HIGH', 'message': '使用默认的不安全SECRET_KEY', 'file': str(settings_file), 'recommendation': '生成新的安全SECRET_KEY' }) # 2. 检查表名拼写错误 views_file = DJANGO_ROOT / "extractor/views.py" if views_file.exists(): with open(views_file, 'r', encoding='utf-8') as f: content = f.read() if 'precessed_table' in content: issues.append({ 'type': 'BUG', 'severity': 'HIGH', 'message': '表名拼写错误: precessed_table 应该是 processed_table', 'file': str(views_file), 'recommendation': '修正表名拼写或确保数据库中表名一致' }) # 3. 检查模型文件是否为空 models_file = DJANGO_ROOT / "extractor/models.py" if models_file.exists(): with open(models_file, 'r', encoding='utf-8') as f: content = f.read().strip() if len(content) < 100: # 基本上是空文件 issues.append({ 'type': 'ARCHITECTURE', 'severity': 'MEDIUM', 'message': 'models.py文件为空，未使用Django ORM', 'file': str(models_file), 'recommendation': '考虑使用Django ORM管理数据模型，或添加注释说明原因' }) # 4. 检查依赖管理 requirements_file = DJANGO_ROOT / "requirements.txt" if not requirements_file.exists(): issues.append({ 'type': 'DEPLOYMENT', 'severity': 'MEDIUM', 'message': '缺少requirements.txt文件', 'file': str(DJANGO_ROOT), 'recommendation': '创建requirements.txt管理项目依赖' }) self.test_results['static_analysis'] = { 'issues_count': len(issues), 'issues': issues } self.issues_found.extend(issues) print(f" 发现 {len(issues)} 个静态分析问题") return issues def test_api_endpoints(self): """测试API接口""" print("🌐 测试API接口...") endpoints = [ ('GET', '/health/', '健康检查'), ('GET', '/original-data/', '原始数据获取'), ('GET', '/statistics/', '统计信息'), ('GET', '/processed-data/', '处理后数据'), ('POST', '/preprocess/', '数据预处理'), ('POST', '/merge/', '格式合并'), ('POST', '/correct/', '单词纠错'), ('POST', '/analyze/', '大模型分析'), ('POST', '/upload/', '文件上传'), ] results = {} for method, endpoint, description in endpoints: try: url = f"{BACKEND_API_URL}{endpoint}" if method == 'GET': response = requests.get(url, timeout=5) else: # POST请求发送空JSON response = requests.post(url, json={}, timeout=5) results[endpoint] = { 'status_code': response.status_code, 'success': response.status_code in [200, 400], # 400也可能是正常的业务错误 'description': description, 'response_size': len(response.content) } if response.status_code == 200: print(f" ✓ {description} ({endpoint}): {response.status_code}") elif response.status_code == 400: print(f" ⚠ {description} ({endpoint}): {response.status_code} (业务错误)") else: print(f" ✗ {description} ({endpoint}): {response.status_code}") except requests.RequestException as e: results[endpoint] = { 'error': str(e), 'success': False, 'description': description } print(f" ✗ {description} ({endpoint}): 连接失败 - {e}") self.test_results['api_tests'] = results return results def test_database_connectivity(self): """测试数据库连接""" print("🗄️ 测试数据库连接...") # 通过API测试数据库连接 try: response = requests.get(f"{BACKEND_API_URL}/statistics/", timeout=5) if response.status_code == 200: data = response.json() if data.get('status') == 'success': print(" ✓ 数据库连接正常") return True print(" ✗ 数据库连接异常") return False except Exception as e: print(f" ✗ 数据库连接测试失败: {e}") return False def check_server_status(self): """检查Django服务器状态""" print("🚀 检查Django服务器状态...") try: response = requests.get(f"{BACKEND_API_URL}/health/", timeout=3) if response.status_code == 200: print(" ✓ Django服务器运行正常") return True else: print(f" ✗ Django服务器响应异常: {response.status_code}") return False except requests.RequestException: print(" ✗ Django服务器未运行或无法连接") print(" 💡 请先启动Django服务器: python manage.py runserver") return False def generate_test_report(self): """生成测试报告""" print("\n" + "="*60) print("📊 测试报告汇总") print("="*60) # 静态分析结果 static_issues = self.test_results['static_analysis'].get('issues', []) print(f"\n🔍 静态分析: 发现 {len(static_issues)} 个问题") for issue in static_issues: severity_emoji = { 'HIGH': '🔴', 'MEDIUM': '🟡', 'LOW': '🟢' }.get(issue['severity'], '⚪') print(f" {severity_emoji} [{issue['type']}] {issue['message']}") print(f" 文件: {issue['file']}") print(f" 建议: {issue['recommendation']}") print() # API测试结果 api_results = self.test_results['api_tests'] successful_apis = sum(1 for r in api_results.values() if r.get('success', False)) print(f"\n🌐 API测试: {successful_apis}/{len(api_results)} 个接口正常") for endpoint, result in api_results.items(): if result.get('success'): print(f" ✓ {result['description']}: {result.get('status_code', 'OK')}") else: print(f" ✗ {result['description']}: {result.get('error', result.get('status_code', '未知错误'))}") # 生成修复建议 self.generate_fix_recommendations() # 保存详细报告 report_file = PROJECT_ROOT / "test" / "test_report.json" with open(report_file, 'w', encoding='utf-8') as f: json.dump(self.test_results, f, indent=2, ensure_ascii=False) print(f"\n📄 详细报告已保存到: {report_file}") def generate_fix_recommendations(self): """生成修复建议""" print("\n" + "="*60) print("🔧 修复建议") print("="*60) high_priority_issues = [ issue for issue in self.issues_found if issue.get('severity') == 'HIGH' ] if high_priority_issues: print("\n🔴 高优先级问题（需要立即修复）:") for i, issue in enumerate(high_priority_issues, 1): print(f"\n{i}. {issue['message']}") print(f" 类型: {issue['type']}") print(f" 文件: {issue['file']}") print(f" 建议: {issue['recommendation']}") # 通用修复建议 print("\n💡 通用优化建议:") print("1. 创建环境变量配置文件 (.env)") print("2. 添加单元测试覆盖所有API接口") print("3. 实现日志记录和监控") print("4. 添加数据验证和错误处理") print("5. 考虑使用Django ORM替代原生SQL") print("6. 添加API文档（Swagger/OpenAPI）") print("7. 实现缓存机制提高性能") print("8. 添加数据库迁移管理") def run_all_tests(self): """运行所有测试""" print("🚀 开始综合测试...") # 1. 检查服务器状态 if not self.check_server_status(): print("\n❌ 服务器未运行，跳过API测试") print("请先启动Django服务器:") print(f"cd {DJANGO_ROOT}") print("python manage.py runserver") # 只运行静态分析 self.run_static_analysis() else: # 2. 静态分析 self.run_static_analysis() # 3. API测试 self.test_api_endpoints() # 4. 数据库测试 self.test_database_connectivity() # 5. 生成报告 self.generate_test_report() def main(): """主函数""" print("🎯 Django后端测试系统") print("=" * 60) runner = TestRunner() runner.run_all_tests() print("\n✅ 测试完成！") # 检查是否有严重问题 high_issues = sum(1 for issue in runner.issues_found if issue.get('severity') == 'HIGH') if high_issues > 0: print(f"⚠️ 发现 {high_issues} 个高优先级问题需要修复") return 1 else: print("🎉 未发现严重问题") return 0 if __name__ == "__main__": sys.exit(main())