FrameTour-RenderWorker/tests/unit/test_render_video_effects.py

# -*- coding: utf-8 -*-

import pytest

from domain.config import WorkerConfig
from domain.task import Effect, OutputSpec, RenderSpec
from handlers.render_video import RenderSegmentTsHandler


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 RenderSegmentTsHandler(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-STARTPTS)' in vf_value


def test_default_speed_always_normalizes_pts(tmp_path):
    handler = _create_handler(tmp_path)
    render_spec = RenderSpec()
    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=PTS-STARTPTS' in filters


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-STARTPTS)' 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-STARTPTS)' 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-STARTPTS)' in filters
    assert 'setpts=5.0*(PTS-STARTPTS)' 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