""" 集成测试 测试整个系统的协同工作 """ import pytest from pathlib import Path import tempfile import sys import os project_root = Path(__file__).parent.parent.absolute() sys.path.insert(0, str(project_root)) os.chdir(project_root) from main import create_app class TestAppIntegration: """测试应用集成""" def test_full_app_startup(self): """测试完整应用启动""" app = create_app() assert app is not None # 验证所有路由都已注册 rules = [rule.rule for rule in app.url_map.iter_rules()] # 根路由 assert '/' in rules # ASR 路由 assert '/api/recognize' in rules assert '/api/result' in rules # 转码路由 assert '/api/convert' in rules assert '/api/getVidUrl' in rules # CORS 路由 assert '/config' in rules def test_all_routes_respond(self): """测试所有路由都能响应""" app = create_app() app.config['TESTING'] = True with app.test_client() as client: # 测试根路由 response = client.get('/') assert response.status_code == 200 # 测试 CORS OPTIONS response = client.options('/config') assert response.status_code == 200 def test_error_handling_consistency(self): """测试错误处理一致性""" app = create_app() app.config['TESTING'] = True with app.test_client() as client: # 测试 404 路由 response = client.get('/nonexistent') assert response.status_code == 404 # 测试错误响应格式 response = client.get('/api/recognize') # 缺少参数 assert response.status_code == 400 data = response.get_json() assert 'status' in data assert 'message' in data assert data['status'] == 'error' class TestModuleIsolation: """测试模块隔离""" def test_asr_module_independent(self): """测试 ASR 模块独立性""" # 只导入 ASR 路由,不应该影响其他模块 from app.asr.routes import register_asr_routes from flask import Flask app = Flask(__name__) register_asr_routes(app) rules = [rule.rule for rule in app.url_map.iter_rules()] assert '/api/recognize' in rules assert '/api/result' in rules # 不应该有转码路由 assert '/api/convert' not in rules def test_transcode_module_independent(self): """测试转码模块独立性""" # 只导入转码路由,不应该影响其他模块 from app.transcode.routes import register_transcode_routes from flask import Flask import tempfile app = Flask(__name__) app.config['OUTPUT_DIR'] = tempfile.mkdtemp() register_transcode_routes(app) rules = [rule.rule for rule in app.url_map.iter_rules()] assert '/api/convert' in rules assert '/api/getVidUrl' in rules # 不应该有 ASR 路由 assert '/api/recognize' not in rules class TestConfigIsolation: """测试配置隔离""" def test_test_config_isolation(self): """测试测试配置隔离""" app = create_app() app.config['TESTING'] = True # 修改测试配置不应该影响全局配置 from app.settings import config original_port = config['API_PORT'] app.config['API_PORT'] = 9999 assert app.config['API_PORT'] == 9999 assert config['API_PORT'] == original_port class TestConcurrentRequests: """测试并发请求""" def test_multiple_requests(self): """测试多个请求""" app = create_app() app.config['TESTING'] = True with app.test_client() as client: # 发送多个请求 for i in range(5): response = client.get('/') assert response.status_code == 200 def test_request_isolation(self): """测试请求隔离""" app = create_app() app.config['TESTING'] = True with app.test_client() as client: # 第一个请求 response1 = client.get('/') data1 = response1.get_json() # 第二个请求 response2 = client.get('/') data2 = response2.get_json() # 时间戳应该不同(或至少请求是独立的) assert data1['timestamp'] is not None assert data2['timestamp'] is not None class TestGlobalState: """测试全局状态""" def test_task_running_is_dict(self): """测试任务运行状态是字典类型""" from app.asr.routes import task_running # 验证是字典类型 assert isinstance(task_running, dict) def test_global_services_initial_state(self): """测试全局服务初始状态""" from app.asr.routes import GLOBAL_ASR_SERVICE, GLOBAL_DIAR_SERVICE # 初始应该是 None assert GLOBAL_ASR_SERVICE is None assert GLOBAL_DIAR_SERVICE is None class TestPathHandling: """测试路径处理""" def test_relative_path_handling(self, temp_dirs): """测试相对路径处理""" app = create_app() app.config['TESTING'] = True app.config['INPUT_DIR'] = str(temp_dirs['input']) app.config['OUTPUT_DIR'] = str(temp_dirs['output']) with app.test_client() as client: # 使用相对路径 response = client.get('/api/convert?path=test.mp4') # 应该能处理(可能返回 404,但不会崩溃) assert response.status_code in [404, 500] def test_absolute_path_handling(self, temp_dirs): """测试绝对路径处理""" app = create_app() app.config['TESTING'] = True app.config['INPUT_DIR'] = str(temp_dirs['input']) app.config['OUTPUT_DIR'] = str(temp_dirs['output']) with app.test_client() as client: # 使用绝对路径 abs_path = str(temp_dirs['input'] / 'test.mp4') response = client.get(f'/api/convert?path={abs_path}') # 应该能处理(可能返回 404,但不会崩溃) assert response.status_code in [404, 500] class TestResourceCleanup: """测试资源清理""" def test_temp_dir_cleanup(self, temp_dirs): """测试临时目录清理""" # 创建临时文件 temp_file = temp_dirs['temp'] / 'test.txt' temp_file.touch() assert temp_file.exists() # fixture 会自动清理,这里验证清理机制 # 实际清理由 conftest.py 中的 fixture 处理 def test_output_dir_creation(self, temp_dirs): """测试输出目录创建""" output_dir = temp_dirs['output'] / 'test_subdir' output_dir.mkdir(parents=True, exist_ok=True) assert output_dir.exists() if __name__ == '__main__': pytest.main([__file__, '-v'])