xiaoxia-saas/tests/unit/domain/test_video_concat.py

"""video_concat 视频拼接配置单测."""

import pytest
from domain.video_concat import (
    ALLOWED_VIDEO_EXTENSIONS,
    CONCAT_DEMUXER_REQUIRED_PARAMS,
    MAX_CONCAT_SEGMENTS,
    ConcatConfig,
    ConcatSegment,
)

# ── 常量测试 ─────────────────────────────────────────────────────────────────


class TestConstants:
    """模块常量"""

    def test_max_concat_segments(self):
        assert MAX_CONCAT_SEGMENTS == 50

    def test_allowed_extensions(self):
        assert ".mp4" in ALLOWED_VIDEO_EXTENSIONS
        assert ".mov" in ALLOWED_VIDEO_EXTENSIONS
        assert ".avi" in ALLOWED_VIDEO_EXTENSIONS
        assert ".mkv" in ALLOWED_VIDEO_EXTENSIONS
        assert ".webm" in ALLOWED_VIDEO_EXTENSIONS
        assert ".flv" in ALLOWED_VIDEO_EXTENSIONS
        assert ".wmv" in ALLOWED_VIDEO_EXTENSIONS

    def test_concat_demuxer_params(self):
        params = CONCAT_DEMUXER_REQUIRED_PARAMS
        assert "codec_name" in params
        assert "width" in params
        assert "height" in params
        assert "r_frame_rate" in params
        assert "pix_fmt" in params
        assert "sample_rate" in params
        assert "channels" in params
        assert "audio_codec" in params
        assert len(params) == 8


# ── ConcatSegment ────────────────────────────────────────────────────────────


class TestConcatSegmentDefaults:
    """ConcatSegment 默认值"""

    def test_required_path(self):
        s = ConcatSegment(video_path="/video.mp4")
        assert s.video_path == "/video.mp4"
        assert s.start_time == 0.0
        assert s.duration == 0.0
        assert s.has_audio is True

    def test_all_custom(self):
        s = ConcatSegment(
            video_path="/clip.mp4",
            start_time=5.0,
            duration=10.0,
            has_audio=False,
        )
        assert s.video_path == "/clip.mp4"
        assert s.start_time == 5.0
        assert s.duration == 10.0
        assert s.has_audio is False


class TestConcatSegmentFromDict:
    """ConcatSegment.from_dict"""

    def test_none_returns_empty_path(self):
        s = ConcatSegment.from_dict(None)
        assert s.video_path == ""
        assert s.is_valid is False

    def test_empty_dict(self):
        s = ConcatSegment.from_dict({})
        assert s.video_path == ""

    def test_not_dict(self):
        s = ConcatSegment.from_dict("not a dict")
        assert s.video_path == ""

    def test_full_dict(self):
        s = ConcatSegment.from_dict(
            {
                "video_path": "/clip.mp4",
                "start_time": 2.5,
                "duration": 15.0,
                "has_audio": False,
            }
        )
        assert s.video_path == "/clip.mp4"
        assert s.start_time == 2.5
        assert s.duration == 15.0
        assert s.has_audio is False

    def test_invalid_start_time_falls_back(self):
        s = ConcatSegment.from_dict({"video_path": "/a.mp4", "start_time": "bad"})
        assert s.start_time == 0.0

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

    def test_invalid_duration_falls_back(self):
        s = ConcatSegment.from_dict({"video_path": "/a.mp4", "duration": None})
        assert s.duration == 0.0

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

    def test_has_audio_default_true(self):
        s = ConcatSegment.from_dict({"video_path": "/a.mp4"})
        assert s.has_audio is True

    def test_has_audio_false(self):
        s = ConcatSegment.from_dict({"video_path": "/a.mp4", "has_audio": False})
        assert s.has_audio is False

    def test_path_is_string(self):
        s = ConcatSegment.from_dict({"video_path": 123})
        assert s.video_path == "123"


class TestConcatSegmentProperties:
    """ConcatSegment 属性方法"""

    def test_is_valid_true(self):
        s = ConcatSegment(video_path="/a.mp4")
        assert s.is_valid is True

    def test_is_valid_false_empty(self):
        s = ConcatSegment(video_path="")
        assert s.is_valid is False

    def test_effective_duration_positive(self):
        s = ConcatSegment(video_path="/a.mp4", duration=10.0)
        assert s.effective_duration == 10.0

    def test_effective_duration_zero(self):
        s = ConcatSegment(video_path="/a.mp4", duration=0.0)
        assert s.effective_duration == 0.0

    def test_effective_duration_negative(self):
        s = ConcatSegment(video_path="/a.mp4", duration=-5.0)
        assert s.effective_duration == 0.0


# ── ConcatConfig ─────────────────────────────────────────────────────────────


class TestConcatConfigDefaults:
    """ConcatConfig 默认值"""

    def test_default_values(self):
        c = ConcatConfig()
        assert c.segments == []
        assert c.output_width == 0
        assert c.output_height == 0
        assert c.output_fps == 0.0
        assert c.force_reencode is False
        assert c.transition == "none"
        assert c.transition_duration == 0.3


class TestConcatConfigFromDict:
    """ConcatConfig.from_config_dict"""

    def test_none_returns_default(self):
        c = ConcatConfig.from_config_dict(None)
        assert c.segments == []

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

    def test_not_dict_returns_default(self):
        c = ConcatConfig.from_config_dict("config")
        assert c.segments == []

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

    def test_multiple_segments(self):
        c = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "/a.mp4", "duration": 10.0},
                    {"video_path": "/b.mp4", "duration": 20.0},
                    {"video_path": "/c.mp4", "duration": 15.0},
                ],
            }
        )
        assert len(c.segments) == 3

    def test_skip_no_path_segments(self):
        c = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "/a.mp4"},
                    {"video_path": ""},
                    {"duration": 5.0},  # 没有 video_path
                    {"video_path": "/b.mp4"},
                ],
            }
        )
        assert len(c.segments) == 2

    def test_skip_non_dict_segments(self):
        c = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "/a.mp4"},
                    "not a dict",
                    123,
                    None,
                    {"video_path": "/b.mp4"},
                ],
            }
        )
        assert len(c.segments) == 2

    def test_output_resolution(self):
        c = ConcatConfig.from_config_dict(
            {
                "segments": [{"video_path": "/a.mp4"}],
                "output_width": 1920,
                "output_height": 1080,
            }
        )
        assert c.output_width == 1920
        assert c.output_height == 1080

    def test_output_width_clamped(self):
        c = ConcatConfig.from_config_dict({"output_width": -100})
        assert c.output_width == 0

    def test_invalid_output_width_falls_back(self):
        c = ConcatConfig.from_config_dict({"output_width": "wide"})
        assert c.output_width == 0

    def test_output_fps(self):
        c = ConcatConfig.from_config_dict({"output_fps": 30.0})
        assert c.output_fps == 30.0

    def test_output_fps_clamped(self):
        c = ConcatConfig.from_config_dict({"output_fps": -1.0})
        assert c.output_fps == 0.0

    def test_invalid_output_fps_falls_back(self):
        c = ConcatConfig.from_config_dict({"output_fps": "fast"})
        assert c.output_fps == 0.0

    def test_force_reencode_true(self):
        c = ConcatConfig.from_config_dict({"force_reencode": True})
        assert c.force_reencode is True

    def test_transition_crossfade(self):
        c = ConcatConfig.from_config_dict({"transition": "crossfade"})
        assert c.transition == "crossfade"

    def test_transition_duration(self):
        c = ConcatConfig.from_config_dict({"transition_duration": 1.0})
        assert c.transition_duration == 1.0

    def test_transition_duration_min_clamped(self):
        c = ConcatConfig.from_config_dict({"transition_duration": 0.01})
        # max(0.1, 0.01) = 0.1
        assert c.transition_duration == 0.1
        # 代码里 transition_duration = max(0.1, ...)，默认 0.3
        # 0.01 < 0.1 ，所以被钳制到 0.1

    def test_invalid_transition_duration_falls_back(self):
        c = ConcatConfig.from_config_dict({"transition_duration": "long"})
        assert c.transition_duration == 0.3

    def test_segments_not_list_ignored(self):
        c = ConcatConfig.from_config_dict({"segments": "not a list"})
        assert c.segments == []


class TestConcatConfigProperties:
    """ConcatConfig 属性方法"""

    def _make_config(self, n=3):
        return ConcatConfig.from_config_dict(
            {
                "segments": [{"video_path": f"/s{i}.mp4", "duration": 10.0 + i} for i in range(n)],
            }
        )

    def test_has_effect_true(self):
        c = self._make_config(3)
        assert c.has_effect is True

    def test_has_effect_false_one_segment(self):
        c = self._make_config(1)
        assert c.has_effect is False

    def test_has_effect_false_empty(self):
        c = ConcatConfig()
        assert c.has_effect is False

    def test_valid_segment_count(self):
        c = self._make_config(5)
        assert c.valid_segment_count == 5

    def test_total_segments_alias(self):
        c = self._make_config(4)
        assert c.total_segments == 4
        assert c.total_segments == c.valid_segment_count

    def test_first_valid_segment(self):
        c = self._make_config(3)
        first = c.first_valid_segment
        assert first is not None
        assert first.video_path == "/s0.mp4"

    def test_first_valid_segment_empty(self):
        c = ConcatConfig()
        assert c.first_valid_segment is None

    def test_estimated_total_duration(self):
        c = ConcatConfig(
            segments=[
                ConcatSegment("/a.mp4", duration=10.0),
                ConcatSegment("/b.mp4", duration=20.0),
                ConcatSegment("/c.mp4", duration=0.0),  # 不计入
            ]
        )
        assert c.estimated_total_duration == 30.0

    def test_estimated_total_duration_empty(self):
        c = ConcatConfig()
        assert c.estimated_total_duration == 0.0

    def test_clamp_segments_within_limit(self):
        c = self._make_config(10)
        original = len(c.segments)
        c.clamp_segments(max_segments=50)
        assert len(c.segments) == original

    def test_clamp_segments_over_limit(self):
        c = self._make_config(10)
        c.clamp_segments(max_segments=3)
        assert len(c.segments) == 3
        assert c.segments[0].video_path == "/s0.mp4"
        assert c.segments[2].video_path == "/s2.mp4"

    def test_clamp_segments_default_max(self):
        # 默认应该是 MAX_CONCAT_SEGMENTS
        c = ConcatConfig(segments=[ConcatSegment(f"/s{i}.mp4") for i in range(100)])
        c.clamp_segments()
        assert len(c.segments) == MAX_CONCAT_SEGMENTS


class TestTransitionDurationClamp:
    """transition_duration 钳制边界"""

    def test_min_boundary_01(self):
        c = ConcatConfig.from_config_dict({"transition_duration": 0.1})
        assert c.transition_duration == 0.1

    def test_below_min_clamped(self):
        c = ConcatConfig.from_config_dict({"transition_duration": 0.05})
        # max(0.1, 0.05) = 0.1
        assert c.transition_duration == 0.1

    def test_large_duration_ok(self):
        c = ConcatConfig.from_config_dict({"transition_duration": 5.0})
        assert c.transition_duration == 5.0

    def test_zero_clamped(self):
        c = ConcatConfig.from_config_dict({"transition_duration": 0.0})
        # max(0.1, 0.0) = 0.1
        assert c.transition_duration == 0.1

    def test_negative_clamped(self):
        c = ConcatConfig.from_config_dict({"transition_duration": -1.0})
        # max(0.1, -1.0) = 0.1
        assert c.transition_duration == 0.1