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