|
|
#!/usr/bin/env python3
|
|
|
"""
|
|
|
Django后端综合测试套件 - 专业级测试框架
|
|
|
测试目标：检测并验证所有API接口、数据库操作、业务逻辑的正确性
|
|
|
"""
|
|
|
|
|
|
import os
|
|
|
import sys
|
|
|
import django
|
|
|
import unittest
|
|
|
import json
|
|
|
import tempfile
|
|
|
import io
|
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
|
from django.test import TestCase, Client
|
|
|
from django.db import connection
|
|
|
from django.conf import settings
|
|
|
import pandas as pd
|
|
|
|
|
|
# 设置Django环境
|
|
|
sys.path.append('/home/hzk/项目/moxun-1/信息抽取+数据检验/Django123/atc_extractor/backend')
|
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'backend.settings')
|
|
|
django.setup()
|
|
|
|
|
|
from extractor.views import *
|
|
|
from extractor.services import process_data
|
|
|
from extractor.infoextractor_czhwjq import CallSignExtractor
|
|
|
|
|
|
class DatabaseTestCase(TestCase):
|
|
|
"""数据库连接和表结构测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
"""测试前准备：创建测试数据"""
|
|
|
self.client = Client()
|
|
|
# 创建测试用的基础表
|
|
|
with connection.cursor() as cursor:
|
|
|
# 创建测试用的prewashed_table
|
|
|
cursor.execute("""
|
|
|
CREATE TABLE IF NOT EXISTS prewashed_table (
|
|
|
id VARCHAR(255) NOT NULL,
|
|
|
text TEXT,
|
|
|
PRIMARY KEY (id)
|
|
|
)
|
|
|
""")
|
|
|
|
|
|
# 插入测试数据
|
|
|
cursor.execute("""
|
|
|
INSERT IGNORE INTO prewashed_table (id, text) VALUES
|
|
|
('test1', 'CCA123 climb to flight level 350'),
|
|
|
('test2', 'CSN456 descend to flight level 280'),
|
|
|
('test3', 'CHH789 maintain heading 090')
|
|
|
""")
|
|
|
|
|
|
def test_database_connection(self):
|
|
|
"""测试数据库连接"""
|
|
|
try:
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute("SELECT 1")
|
|
|
result = cursor.fetchone()
|
|
|
self.assertEqual(result[0], 1)
|
|
|
except Exception as e:
|
|
|
self.fail(f"数据库连接失败: {e}")
|
|
|
|
|
|
def test_table_existence(self):
|
|
|
"""测试必要的表是否存在"""
|
|
|
required_tables = ['prewashed_table', 'air_company']
|
|
|
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute("SHOW TABLES")
|
|
|
existing_tables = [table[0] for table in cursor.fetchall()]
|
|
|
|
|
|
for table in required_tables:
|
|
|
if table in existing_tables:
|
|
|
print(f"✓ 表 {table} 存在")
|
|
|
else:
|
|
|
print(f"✗ 表 {table} 不存在")
|
|
|
|
|
|
def test_table_structure(self):
|
|
|
"""测试表结构的完整性"""
|
|
|
with connection.cursor() as cursor:
|
|
|
# 测试prewashed_table结构
|
|
|
cursor.execute("DESCRIBE prewashed_table")
|
|
|
columns = [col[0] for col in cursor.fetchall()]
|
|
|
self.assertIn('id', columns)
|
|
|
self.assertIn('text', columns)
|
|
|
|
|
|
|
|
|
class APIEndpointTestCase(TestCase):
|
|
|
"""API接口功能测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = Client()
|
|
|
# 准备测试数据
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute("""
|
|
|
CREATE TABLE IF NOT EXISTS prewashed_table (
|
|
|
id VARCHAR(255) NOT NULL,
|
|
|
text TEXT,
|
|
|
PRIMARY KEY (id)
|
|
|
)
|
|
|
""")
|
|
|
cursor.execute("""
|
|
|
INSERT IGNORE INTO prewashed_table (id, text) VALUES
|
|
|
('api_test1', 'CCA123 climb to flight level 350'),
|
|
|
('api_test2', 'CSN456 descend to flight level 280')
|
|
|
""")
|
|
|
|
|
|
def test_health_check_endpoint(self):
|
|
|
"""测试健康检查接口"""
|
|
|
response = self.client.get('/api/health/')
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'ok')
|
|
|
print("✓ 健康检查接口正常")
|
|
|
|
|
|
def test_original_data_endpoint(self):
|
|
|
"""测试原始数据获取接口"""
|
|
|
response = self.client.get('/api/original-data/')
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'success')
|
|
|
self.assertIsInstance(data['data'], list)
|
|
|
self.assertGreater(data['count'], 0)
|
|
|
print("✓ 原始数据接口正常")
|
|
|
|
|
|
def test_statistics_endpoint(self):
|
|
|
"""测试统计信息接口"""
|
|
|
response = self.client.get('/api/statistics/')
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'success')
|
|
|
self.assertIn('statistics', data)
|
|
|
print("✓ 统计接口正常")
|
|
|
|
|
|
def test_preprocess_endpoint(self):
|
|
|
"""测试预处理接口"""
|
|
|
response = self.client.post('/api/preprocess/',
|
|
|
json.dumps({'test': 'data'}),
|
|
|
content_type='application/json')
|
|
|
self.assertIn(response.status_code, [200, 400]) # 200成功或400数据相关错误
|
|
|
data = response.json()
|
|
|
self.assertIn(data['status'], ['success', 'error'])
|
|
|
print("✓ 预处理接口响应正常")
|
|
|
|
|
|
def test_merge_endpoint(self):
|
|
|
"""测试格式合并接口"""
|
|
|
response = self.client.post('/api/merge/',
|
|
|
json.dumps({'test': 'data'}),
|
|
|
content_type='application/json')
|
|
|
self.assertIn(response.status_code, [200, 400])
|
|
|
data = response.json()
|
|
|
self.assertIn(data['status'], ['success', 'error'])
|
|
|
print("✓ 格式合并接口响应正常")
|
|
|
|
|
|
def test_correct_endpoint(self):
|
|
|
"""测试单词纠错接口"""
|
|
|
response = self.client.post('/api/correct/',
|
|
|
json.dumps({'test': 'data'}),
|
|
|
content_type='application/json')
|
|
|
self.assertIn(response.status_code, [200, 400])
|
|
|
data = response.json()
|
|
|
self.assertIn(data['status'], ['success', 'error'])
|
|
|
print("✓ 单词纠错接口响应正常")
|
|
|
|
|
|
|
|
|
class FileUploadTestCase(TestCase):
|
|
|
"""文件上传功能测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = Client()
|
|
|
|
|
|
def test_csv_file_upload(self):
|
|
|
"""测试CSV文件上传"""
|
|
|
# 创建测试CSV文件
|
|
|
csv_content = "id,text\ntest1,CCA123 climb to FL350\ntest2,CSN456 descend to FL280"
|
|
|
csv_file = io.StringIO(csv_content)
|
|
|
csv_file.name = 'test_data.csv'
|
|
|
|
|
|
response = self.client.post('/api/upload/', {
|
|
|
'file': io.BytesIO(csv_content.encode('utf-8'))
|
|
|
})
|
|
|
|
|
|
# 检查响应状态
|
|
|
self.assertIn(response.status_code, [200, 400, 500])
|
|
|
print(f"✓ CSV上传接口响应状态: {response.status_code}")
|
|
|
|
|
|
def test_invalid_file_upload(self):
|
|
|
"""测试无效文件上传"""
|
|
|
response = self.client.post('/api/upload/', {})
|
|
|
self.assertEqual(response.status_code, 400)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'error')
|
|
|
print("✓ 无效文件上传处理正确")
|
|
|
|
|
|
|
|
|
class DataProcessingTestCase(TestCase):
|
|
|
"""数据处理业务逻辑测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = Client()
|
|
|
# 准备测试数据
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute("""
|
|
|
CREATE TABLE IF NOT EXISTS prewashed_table (
|
|
|
id VARCHAR(255) NOT NULL,
|
|
|
text TEXT,
|
|
|
PRIMARY KEY (id)
|
|
|
)
|
|
|
""")
|
|
|
cursor.execute("DELETE FROM prewashed_table")
|
|
|
cursor.execute("""
|
|
|
INSERT INTO prewashed_table (id, text) VALUES
|
|
|
('dp_test1', 'CCA123 climb to flight level 350'),
|
|
|
('dp_test2', 'CSN456 descend to flight level 280'),
|
|
|
('dp_test3', 'CHH789 maintain heading 090')
|
|
|
""")
|
|
|
|
|
|
@patch('extractor.services.ai_process_data')
|
|
|
def test_data_processing_with_mock(self, mock_ai_process):
|
|
|
"""测试数据处理流程（使用Mock）"""
|
|
|
# Mock AI处理结果
|
|
|
mock_ai_process.return_value = {
|
|
|
'status': 'success',
|
|
|
'processed_count': 3,
|
|
|
'valid_count': 2,
|
|
|
'invalid_count': 1
|
|
|
}
|
|
|
|
|
|
response = self.client.post('/api/process-data/')
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'success')
|
|
|
print("✓ 数据处理流程测试通过")
|
|
|
|
|
|
def test_empty_data_processing(self):
|
|
|
"""测试空数据处理"""
|
|
|
# 清空数据
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute("DELETE FROM prewashed_table")
|
|
|
|
|
|
response = self.client.post('/api/preprocess/')
|
|
|
self.assertEqual(response.status_code, 400)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'error')
|
|
|
self.assertIn('没有找到需要预处理的原始数据', data['message'])
|
|
|
print("✓ 空数据处理逻辑正确")
|
|
|
|
|
|
|
|
|
class ErrorHandlingTestCase(TestCase):
|
|
|
"""错误处理测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = Client()
|
|
|
|
|
|
@patch('extractor.views.connection')
|
|
|
def test_database_error_handling(self, mock_connection):
|
|
|
"""测试数据库连接错误处理"""
|
|
|
# 模拟数据库连接错误
|
|
|
mock_cursor = Mock()
|
|
|
mock_cursor.execute.side_effect = Exception("Database connection failed")
|
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
|
|
response = self.client.get('/api/statistics/')
|
|
|
self.assertEqual(response.status_code, 500)
|
|
|
data = response.json()
|
|
|
self.assertEqual(data['status'], 'error')
|
|
|
print("✓ 数据库错误处理正确")
|
|
|
|
|
|
def test_invalid_json_handling(self):
|
|
|
"""测试无效JSON处理"""
|
|
|
response = self.client.post('/api/preprocess/',
|
|
|
'invalid json',
|
|
|
content_type='application/json')
|
|
|
# Django会自动处理JSON解析错误，返回400
|
|
|
self.assertIn(response.status_code, [400, 500])
|
|
|
print("✓ 无效JSON处理正确")
|
|
|
|
|
|
|
|
|
class SecurityTestCase(TestCase):
|
|
|
"""安全性测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = Client()
|
|
|
|
|
|
def test_sql_injection_protection(self):
|
|
|
"""测试SQL注入防护"""
|
|
|
# 尝试SQL注入攻击
|
|
|
malicious_data = {"id": "1'; DROP TABLE prewashed_table; --"}
|
|
|
response = self.client.post('/api/preprocess/',
|
|
|
json.dumps(malicious_data),
|
|
|
content_type='application/json')
|
|
|
|
|
|
# 确保表仍然存在
|
|
|
with connection.cursor() as cursor:
|
|
|
cursor.execute("SHOW TABLES LIKE 'prewashed_table'")
|
|
|
result = cursor.fetchone()
|
|
|
self.assertIsNotNone(result)
|
|
|
print("✓ SQL注入防护有效")
|
|
|
|
|
|
def test_settings_security(self):
|
|
|
"""测试Django设置安全性"""
|
|
|
# 检查调试模式（生产环境应该关闭）
|
|
|
if settings.DEBUG:
|
|
|
print("⚠️ WARNING: DEBUG模式在生产环境中应该关闭")
|
|
|
|
|
|
# 检查SECRET_KEY
|
|
|
if 'django-insecure' in settings.SECRET_KEY:
|
|
|
print("⚠️ WARNING: 使用默认SECRET_KEY，生产环境需要更换")
|
|
|
|
|
|
print("✓ 安全设置检查完成")
|
|
|
|
|
|
|
|
|
class IntegrationTestCase(TestCase):
|
|
|
"""集成测试"""
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = Client()
|
|
|
|
|
|
def test_complete_workflow(self):
|
|
|
"""测试完整的工作流程"""
|
|
|
# 1. 检查健康状态
|
|
|
health_response = self.client.get('/api/health/')
|
|
|
self.assertEqual(health_response.status_code, 200)
|
|
|
|
|
|
# 2. 检查原始数据
|
|
|
data_response = self.client.get('/api/original-data/')
|
|
|
self.assertEqual(data_response.status_code, 200)
|
|
|
|
|
|
# 3. 执行预处理（如果有数据的话）
|
|
|
preprocess_response = self.client.post('/api/preprocess/',
|
|
|
json.dumps({}),
|
|
|
content_type='application/json')
|
|
|
self.assertIn(preprocess_response.status_code, [200, 400])
|
|
|
|
|
|
# 4. 获取统计信息
|
|
|
stats_response = self.client.get('/api/statistics/')
|
|
|
self.assertEqual(stats_response.status_code, 200)
|
|
|
|
|
|
print("✓ 完整工作流程测试通过")
|
|
|
|
|
|
|
|
|
def run_comprehensive_tests():
|
|
|
"""运行全面测试套件"""
|
|
|
print("🚀 开始运行Django后端综合测试...")
|
|
|
print("=" * 60)
|
|
|
|
|
|
# 创建测试套件
|
|
|
test_suite = unittest.TestSuite()
|
|
|
|
|
|
# 添加所有测试类
|
|
|
test_classes = [
|
|
|
DatabaseTestCase,
|
|
|
APIEndpointTestCase,
|
|
|
FileUploadTestCase,
|
|
|
DataProcessingTestCase,
|
|
|
ErrorHandlingTestCase,
|
|
|
SecurityTestCase,
|
|
|
IntegrationTestCase
|
|
|
]
|
|
|
|
|
|
for test_class in test_classes:
|
|
|
tests = unittest.TestLoader().loadTestsFromTestCase(test_class)
|
|
|
test_suite.addTests(tests)
|
|
|
|
|
|
# 运行测试
|
|
|
runner = unittest.TextTestRunner(verbosity=2)
|
|
|
result = runner.run(test_suite)
|
|
|
|
|
|
print("=" * 60)
|
|
|
print(f"测试完成: {result.testsRun} 个测试")
|
|
|
print(f"成功: {result.testsRun - len(result.failures) - len(result.errors)}")
|
|
|
print(f"失败: {len(result.failures)}")
|
|
|
print(f"错误: {len(result.errors)}")
|
|
|
|
|
|
if result.failures:
|
|
|
print("\n失败的测试:")
|
|
|
for test, traceback in result.failures:
|
|
|
print(f"- {test}: {traceback}")
|
|
|
|
|
|
if result.errors:
|
|
|
print("\n错误的测试:")
|
|
|
for test, traceback in result.errors:
|
|
|
print(f"- {test}: {traceback}")
|
|
|
|
|
|
return result.wasSuccessful()
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
success = run_comprehensive_tests()
|
|
|
if success:
|
|
|
print("\n🎉 所有测试通过！")
|
|
|
else:
|
|
|
print("\n❌ 部分测试失败，需要修复问题")
|
|
|
|
|
|
sys.exit(0 if success else 1) |
