# -*- coding: utf-8 -*- import pytest from domain.config import WorkerConfig from domain.task import Effect, OutputSpec, RenderSpec from handlers.render_video import RenderSegmentVideoHandler class _DummyApiClient: pass def _create_handler(tmp_path): config = WorkerConfig( api_endpoint='http://127.0.0.1:18084/api', access_key='TEST_ACCESS_KEY', worker_id='test-worker', temp_dir=str(tmp_path), cache_enabled=False, cache_dir=str(tmp_path / 'cache') ) return RenderSegmentVideoHandler(config, _DummyApiClient()) def test_get_zoom_params_with_valid_values(): effect = Effect.from_string('zoom:1.5,1.35,2') assert effect is not None start_sec, scale_factor, duration_sec = effect.get_zoom_params() assert start_sec == pytest.approx(1.5) assert scale_factor == pytest.approx(1.35) assert duration_sec == pytest.approx(2.0) @pytest.mark.parametrize( 'effect_str', [ 'zoom:-1,0.9,-2', 'zoom:nan,inf,0', 'zoom:bad,value,data', 'zoom:,,', ], ) def test_get_zoom_params_invalid_values_fallback_to_default(effect_str): effect = Effect.from_string(effect_str) assert effect is not None assert effect.get_zoom_params() == (0.0, 1.2, 1.0) def test_build_command_with_zoom_uses_filter_complex(tmp_path): handler = _create_handler(tmp_path) render_spec = RenderSpec(effects='zoom:1.5,1.4,2') output_spec = OutputSpec(width=1080, height=1920, fps=30) command = handler._build_command( input_file='input.mp4', output_file='output.mp4', render_spec=render_spec, output_spec=output_spec, duration_ms=6000, ) assert '-filter_complex' in command assert '-vf' not in command def test_build_video_filters_zoom_and_camera_shot_stack_in_order(tmp_path): handler = _create_handler(tmp_path) render_spec = RenderSpec(effects='cameraShot:3,1|zoom:1,1.2,2') output_spec = OutputSpec(width=1080, height=1920, fps=30) filters = handler._build_video_filters( render_spec=render_spec, output_spec=output_spec, duration_ms=8000, source_duration_sec=10.0, ) camera_shot_marker = 'concat=n=2:v=1:a=0' zoom_marker = "overlay=0:0:enable='between(t,1.0,3.0)'" assert camera_shot_marker in filters assert zoom_marker in filters assert filters.index(camera_shot_marker) < filters.index(zoom_marker) # ---------- ospeed 测试 ---------- def test_get_ospeed_params_with_valid_values(): effect = Effect.from_string('ospeed:2') assert effect is not None assert effect.get_ospeed_params() == pytest.approx(2.0) effect2 = Effect.from_string('ospeed:0.5') assert effect2 is not None assert effect2.get_ospeed_params() == pytest.approx(0.5) @pytest.mark.parametrize( 'effect_str', [ 'ospeed:0', 'ospeed:-1', 'ospeed:nan', 'ospeed:inf', 'ospeed:abc', 'ospeed:', ], ) def test_get_ospeed_params_invalid_values_fallback(effect_str): effect = Effect.from_string(effect_str) assert effect is not None assert effect.get_ospeed_params() == 1.0 def test_get_ospeed_params_no_params(): effect = Effect.from_string('ospeed') assert effect is not None assert effect.get_ospeed_params() == 1.0 def test_ospeed_does_not_trigger_filter_complex(tmp_path): handler = _create_handler(tmp_path) render_spec = RenderSpec(effects='ospeed:2') output_spec = OutputSpec(width=1080, height=1920, fps=30) command = handler._build_command( input_file='input.mp4', output_file='output.mp4', render_spec=render_spec, output_spec=output_spec, duration_ms=6000, ) assert '-vf' in command assert '-filter_complex' not in command # 验证 setpts 滤镜 vf_idx = command.index('-vf') vf_value = command[vf_idx + 1] assert 'setpts=2.0*PTS' in vf_value def test_ospeed_combined_with_speed(tmp_path): handler = _create_handler(tmp_path) # speed=2 → 1/2=0.5, ospeed=3 → 0.5*3=1.5 render_spec = RenderSpec(speed='2', effects='ospeed:3') output_spec = OutputSpec(width=1080, height=1920, fps=30) filters = handler._build_video_filters( render_spec=render_spec, output_spec=output_spec, duration_ms=6000, source_duration_sec=10.0, ) assert 'setpts=1.5*PTS' in filters def test_ospeed_with_complex_effects(tmp_path): handler = _create_handler(tmp_path) render_spec = RenderSpec(effects='ospeed:2|zoom:1,1.2,2') output_spec = OutputSpec(width=1080, height=1920, fps=30) filters = handler._build_video_filters( render_spec=render_spec, output_spec=output_spec, duration_ms=8000, source_duration_sec=10.0, ) # zoom 触发 filter_complex assert '[v_base]' in filters # setpts 在 base_chain 中 assert 'setpts=2.0*PTS' in filters # zoom 正常处理 assert "overlay=0:0:enable='between(t,1.0,3.0)'" in filters def test_only_first_ospeed_is_used(tmp_path): handler = _create_handler(tmp_path) render_spec = RenderSpec(effects='ospeed:2|ospeed:5') output_spec = OutputSpec(width=1080, height=1920, fps=30) filters = handler._build_video_filters( render_spec=render_spec, output_spec=output_spec, duration_ms=6000, source_duration_sec=10.0, ) assert 'setpts=2.0*PTS' in filters assert 'setpts=5.0*PTS' not in filters def test_ospeed_affects_tpad_calculation(tmp_path): handler = _create_handler(tmp_path) # ospeed:2 使 10s 视频变为 20s 有效时长,目标 6s → 无需 tpad render_spec = RenderSpec(effects='ospeed:2') output_spec = OutputSpec(width=1080, height=1920, fps=30) filters = handler._build_video_filters( render_spec=render_spec, output_spec=output_spec, duration_ms=6000, source_duration_sec=10.0, ) assert 'tpad' not in filters # 对比:无 ospeed 时 10s 视频 → 目标 15s → 需要 5s tpad render_spec_no_ospeed = RenderSpec() filters_no_ospeed = handler._build_video_filters( render_spec=render_spec_no_ospeed, output_spec=output_spec, duration_ms=15000, source_duration_sec=10.0, ) assert 'tpad' in filters_no_ospeed assert 'stop_duration=5.0' in filters_no_ospeed