test

tests/test_integration/test_ffmpeg_execution.py

@@ -0,0 +1,282 @@
+"""FFmpeg执行集成测试"""
+import pytest
+import subprocess
+import tempfile
+import os
+from pathlib import Path
+
+from entity.ffmpeg_command_builder import FFmpegCommandBuilder
+from entity.render_task import RenderTask
+from config.settings import Config
+from services.render_service import DefaultRenderService
+from tests.utils.test_helpers import MockRenderTask, create_test_video_file
+
+
+@pytest.mark.integration
+class TestFFmpegExecution:
+    """FFmpeg执行集成测试"""
+
+    @pytest.fixture
+    def test_config(self, temp_dir):
+        """测试配置"""
+        return Config(
+            encoder_args="-c:v libx264",
+            video_args="-preset ultrafast -crf 23",
+            template_dir=temp_dir,
+            api_endpoint="http://test.local",
+            access_key="test_key"
+        )
+
+    @pytest.fixture
+    def sample_video(self, temp_dir):
+        """创建测试视频文件"""
+        video_path = os.path.join(temp_dir, "test_input.mp4")
+        success = create_test_video_file(video_path, duration=3, resolution="320x240")
+        if not success:
+            pytest.skip("Cannot create test video file")
+        return video_path
+
+    @pytest.fixture
+    def render_service(self, test_config):
+        """渲染服务实例"""
+        return DefaultRenderService()
+
+    def test_simple_copy_execution(self, sample_video, temp_dir, test_config):
+        """测试简单复制执行"""
+        output_path = os.path.join(temp_dir, "copy_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        task.effects = []
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        # 执行命令
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=30)
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+            assert os.path.getsize(output_path) > 0, "Output file is empty"
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    def test_zoom_effect_execution(self, sample_video, temp_dir, test_config):
+        """测试缩放特效执行"""
+        output_path = os.path.join(temp_dir, "zoom_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        task.effects = ["zoom:0,2.0,2.0"]  # 2倍缩放，持续2秒
+        task.ext_data = {"posJson": "{}"}
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=60)
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+            assert os.path.getsize(output_path) > 0, "Output file is empty"
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    def test_speed_effect_execution(self, sample_video, temp_dir, test_config):
+        """测试变速特效执行"""
+        output_path = os.path.join(temp_dir, "speed_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        task.effects = ["ospeed:2.0"]  # 2倍速
+        task.ext_data = {}
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=60)
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+            assert os.path.getsize(output_path) > 0, "Output file is empty"
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    def test_multiple_effects_execution(self, sample_video, temp_dir, test_config):
+        """测试多特效组合执行"""
+        output_path = os.path.join(temp_dir, "multi_effects_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        task.effects = ["zoom:0,1.5,1.0", "ospeed:1.5"]  # 缩放+变速
+        task.ext_data = {"posJson": "{}"}
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=60)
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+            assert os.path.getsize(output_path) > 0, "Output file is empty"
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    def test_concat_execution(self, temp_dir, test_config):
+        """测试视频拼接执行"""
+        # 创建两个测试视频
+        video1_path = os.path.join(temp_dir, "video1.mp4")
+        video2_path = os.path.join(temp_dir, "video2.mp4")
+        output_path = os.path.join(temp_dir, "concat_output.mp4")
+
+        success1 = create_test_video_file(video1_path, duration=2, resolution="320x240")
+        success2 = create_test_video_file(video2_path, duration=2, resolution="320x240")
+
+        if not (success1 and success2):
+            pytest.skip("Cannot create test video files")
+
+        task = MockRenderTask()
+        task.input_files = [video1_path, video2_path]
+        task.output_path = output_path
+        task.effects = []
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=60)
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+            assert os.path.getsize(output_path) > 0, "Output file is empty"
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    def test_invalid_effect_execution(self, sample_video, temp_dir, test_config):
+        """测试无效特效的执行处理"""
+        output_path = os.path.join(temp_dir, "invalid_effect_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        task.effects = ["invalid_effect:params", "zoom:0,2.0,1.0"]  # 混合有效和无效特效
+        task.ext_data = {"posJson": "{}"}
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        # 应该忽略无效特效，继续处理有效特效
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=60)
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    def test_render_service_integration(self, sample_video, temp_dir, test_config, render_service):
+        """测试渲染服务集成"""
+        output_path = os.path.join(temp_dir, "service_output.mp4")
+
+        # 创建真实的RenderTask（不是Mock）
+        task_data = {
+            "task_id": "integration_test",
+            "template_id": "test_template",
+            "input_files": [sample_video],
+            "output_path": output_path,
+            "effects": ["zoom:0,1.8,2.0"],
+            "ext_data": {"posJson": "{}"},
+            "frame_rate": 25
+        }
+
+        # 这里需要根据实际的RenderTask构造方法调整
+        task = MockRenderTask(**task_data)
+
+        # 使用渲染服务执行
+        try:
+            # 这里的方法调用需要根据实际的渲染服务接口调整
+            # success = render_service.render(task, test_config)
+            # assert success, "Render service failed"
+
+            # 临时直接使用FFmpegCommandBuilder测试
+            builder = FFmpegCommandBuilder(task, test_config)
+            command = builder.build_command()
+            result = subprocess.run(command, capture_output=True, text=True, timeout=60)
+
+            assert result.returncode == 0, f"Render failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+
+        except Exception as e:
+            pytest.fail(f"Render service integration failed: {e}")
+
+    def test_error_handling_missing_input(self, temp_dir, test_config):
+        """测试缺失输入文件的错误处理"""
+        missing_file = os.path.join(temp_dir, "nonexistent.mp4")
+        output_path = os.path.join(temp_dir, "error_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [missing_file]
+        task.output_path = output_path
+        task.effects = []
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        # 应该失败，因为输入文件不存在
+        result = subprocess.run(command, capture_output=True, text=True, timeout=30)
+        assert result.returncode != 0, "FFmpeg should fail with missing input file"
+
+    def test_performance_multiple_effects(self, sample_video, temp_dir, test_config):
+        """测试多特效性能"""
+        output_path = os.path.join(temp_dir, "performance_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        # 多个特效组合
+        task.effects = [
+            "zoom:0,1.5,1.0",
+            "ospeed:1.2",
+            "zoom:1.5,2.0,1.0"
+        ]
+        task.ext_data = {"posJson": "{}"}
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        import time
+        start_time = time.time()
+
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=120)
+            execution_time = time.time() - start_time
+
+            assert result.returncode == 0, f"FFmpeg failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+            assert execution_time < 60, f"Execution took too long: {execution_time}s"
+
+        except subprocess.TimeoutExpired:
+            pytest.fail("FFmpeg execution timed out")
+
+    @pytest.mark.skipif(not os.environ.get('RUN_STRESS_TESTS'), reason="Stress tests disabled")
+    def test_stress_test_large_effects_chain(self, sample_video, temp_dir, test_config):
+        """压力测试：大量特效链"""
+        output_path = os.path.join(temp_dir, "stress_output.mp4")
+
+        task = MockRenderTask()
+        task.input_files = [sample_video]
+        task.output_path = output_path
+        # 创建大量特效
+        task.effects = [f"zoom:{i*0.5},1.{i+5},{i*0.2+0.5}" for i in range(10)]
+        task.ext_data = {"posJson": "{}"}
+
+        builder = FFmpegCommandBuilder(task, test_config)
+        command = builder.build_command()
+
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=300)
+            assert result.returncode == 0, f"Stress test failed: {result.stderr}"
+            assert os.path.exists(output_path), "Output file was not created"
+        except subprocess.TimeoutExpired:
+            pytest.fail("Stress test timed out")