xiaoxia-saas/tests/unit/test_concat_engine.py

"""
视频拼接引擎配置与纯逻辑测试.

覆盖 ConcatSegment.from_dict / ConcatConfig.from_config_dict / has_effect / total_segments 等纯逻辑.
引擎核心 render 方法依赖 FFmpeg，由集成测试覆盖.
"""

from __future__ import annotations

import pytest
from video_processing.concat_engine import ConcatConfig, ConcatSegment


class TestConcatSegmentFromDict:
    """ConcatSegment.from_dict 构造逻辑."""

    def test_basic(self):
        seg = ConcatSegment.from_dict({"video_path": "/tmp/a.mp4"})
        assert seg.video_path == "/tmp/a.mp4"
        assert seg.start_time == 0.0
        assert seg.duration == 0.0
        assert seg.has_audio is True

    def test_full_fields(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "/tmp/b.mp4",
                "start_time": 5.5,
                "duration": 10.0,
                "has_audio": False,
            }
        )
        assert seg.video_path == "/tmp/b.mp4"
        assert seg.start_time == 5.5
        assert seg.duration == 10.0
        assert seg.has_audio is False

    def test_negative_start_time_clamped(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "/tmp/a.mp4",
                "start_time": -1.0,
            }
        )
        assert seg.start_time == 0.0

    def test_negative_duration_clamped(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "/tmp/a.mp4",
                "duration": -5.0,
            }
        )
        assert seg.duration == 0.0

    def test_invalid_start_time_type_falls_back(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "/tmp/a.mp4",
                "start_time": "not_a_number",
            }
        )
        assert seg.start_time == 0.0

    def test_invalid_duration_type_falls_back(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "/tmp/a.mp4",
                "duration": "abc",
            }
        )
        assert seg.duration == 0.0

    def test_start_time_none_falls_back(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "/tmp/a.mp4",
                "start_time": None,
            }
        )
        assert seg.start_time == 0.0

    def test_empty_video_path_stored(self):
        seg = ConcatSegment.from_dict({"video_path": ""})
        assert seg.video_path == ""


class TestConcatConfigFromConfigDict:
    """ConcatConfig.from_config_dict 构造逻辑."""

    def test_none_returns_default(self):
        cfg = ConcatConfig.from_config_dict(None)
        assert cfg.segments == []
        assert cfg.output_width == 0
        assert cfg.output_height == 0
        assert cfg.output_fps == 0.0
        assert cfg.force_reencode is False

    def test_empty_dict_returns_default(self):
        cfg = ConcatConfig.from_config_dict({})
        assert cfg.segments == []

    def test_non_dict_returns_default(self):
        cfg = ConcatConfig.from_config_dict("not a dict")
        assert cfg.segments == []

    def test_single_segment(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "/tmp/a.mp4", "duration": 5.0},
                ],
            }
        )
        assert len(cfg.segments) == 1
        assert cfg.segments[0].video_path == "/tmp/a.mp4"
        assert cfg.segments[0].duration == 5.0

    def test_multiple_segments(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "/tmp/a.mp4"},
                    {"video_path": "/tmp/b.mp4", "start_time": 2.0},
                    {"video_path": "/tmp/c.mp4", "duration": 3.0, "has_audio": False},
                ],
            }
        )
        assert len(cfg.segments) == 3
        assert cfg.segments[0].video_path == "/tmp/a.mp4"
        assert cfg.segments[1].start_time == 2.0
        assert cfg.segments[2].has_audio is False

    def test_invalid_segments_filtered(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "/tmp/valid.mp4"},
                    {"video_path": ""},  # 空路径被过滤
                    {"not_video_path": "xxx"},  # 没有video_path被过滤
                    "not_a_dict",  # 不是dict被过滤
                    None,  # None被过滤
                ],
            }
        )
        assert len(cfg.segments) == 1
        assert cfg.segments[0].video_path == "/tmp/valid.mp4"

    def test_segments_not_a_list(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": "not_a_list",
            }
        )
        assert cfg.segments == []

    def test_output_params(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [],
                "output_width": 1920,
                "output_height": 1080,
                "output_fps": 30.0,
                "force_reencode": True,
            }
        )
        assert cfg.output_width == 1920
        assert cfg.output_height == 1080
        assert cfg.output_fps == 30.0
        assert cfg.force_reencode is True

    def test_negative_output_params_clamped(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [],
                "output_width": -100,
                "output_height": -50,
                "output_fps": -1.0,
            }
        )
        assert cfg.output_width == 0
        assert cfg.output_height == 0
        assert cfg.output_fps == 0.0

    def test_invalid_output_params_fall_back(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [],
                "output_width": "abc",
                "output_height": None,
                "output_fps": "xyz",
            }
        )
        assert cfg.output_width == 0
        assert cfg.output_height == 0
        assert cfg.output_fps == 0.0

    def test_transition_config(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [],
                "transition": "crossfade",
                "transition_duration": 1.0,
            }
        )
        assert cfg.transition == "crossfade"
        assert cfg.transition_duration == 1.0

    def test_transition_duration_minimum(self):
        """transition_duration 不能小于 0.1."""
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [],
                "transition_duration": 0.01,
            }
        )
        assert cfg.transition_duration >= 0.1

    def test_default_values(self):
        cfg = ConcatConfig.from_config_dict({"segments": []})
        assert cfg.transition == "none"
        assert cfg.transition_duration == 0.3
        assert cfg.force_reencode is False


class TestConcatConfigProperties:
    """has_effect / total_segments 属性."""

    def test_has_effect_two_or_more_valid(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="/tmp/a.mp4"),
                ConcatSegment(video_path="/tmp/b.mp4"),
            ]
        )
        assert cfg.has_effect is True

    def test_no_effect_one_segment(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="/tmp/a.mp4"),
            ]
        )
        assert cfg.has_effect is False

    def test_no_effect_zero_segments(self):
        cfg = ConcatConfig(segments=[])
        assert cfg.has_effect is False

    def test_no_effect_empty_paths(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path=""),
                ConcatSegment(video_path=""),
            ]
        )
        assert cfg.has_effect is False

    def test_total_segments(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="/tmp/a.mp4"),
                ConcatSegment(video_path=""),
                ConcatSegment(video_path="/tmp/b.mp4"),
            ]
        )
        assert cfg.total_segments == 2

    def test_total_segments_empty(self):
        cfg = ConcatConfig(segments=[])
        assert cfg.total_segments == 0