xiaoxia-saas/tests/unit/test_video_concat.py

"""video_concat 视频拼接领域模型单元测试."""

from __future__ import annotations

import pytest

from packages.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_not_empty(self):
        """支持的视频格式不为空."""
        assert len(ALLOWED_VIDEO_EXTENSIONS) > 0

    def test_common_formats_supported(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

    def test_demuxer_params_not_empty(self):
        """demuxer必需参数不为空."""
        assert len(CONCAT_DEMUXER_REQUIRED_PARAMS) > 0

    def test_demuxer_params_include_codec(self):
        """包含编解码相关参数."""
        assert "codec_name" in CONCAT_DEMUXER_REQUIRED_PARAMS
        assert "width" in CONCAT_DEMUXER_REQUIRED_PARAMS
        assert "height" in CONCAT_DEMUXER_REQUIRED_PARAMS
        assert "r_frame_rate" in CONCAT_DEMUXER_REQUIRED_PARAMS


# ── ConcatSegment 测试 ───────────────────────────────────────────────────────


class TestConcatSegment:
    """ConcatSegment 测试."""

    def test_basic_creation(self):
        """基础创建."""
        seg = ConcatSegment(video_path="/tmp/video.mp4")
        assert seg.video_path == "/tmp/video.mp4"
        assert seg.start_time == 0.0
        assert seg.duration == 0.0
        assert seg.has_audio is True

    def test_full_creation(self):
        """完整字段创建."""
        seg = ConcatSegment(
            video_path="/tmp/v.mov",
            start_time=5.5,
            duration=10.0,
            has_audio=False,
        )
        assert seg.video_path == "/tmp/v.mov"
        assert seg.start_time == 5.5
        assert seg.duration == 10.0
        assert seg.has_audio is False

    # is_valid 属性

    def test_is_valid_with_path(self):
        """有视频路径是有效的."""
        seg = ConcatSegment(video_path="/tmp/v.mp4")
        assert seg.is_valid is True

    def test_is_valid_empty_path(self):
        """空路径无效."""
        seg = ConcatSegment(video_path="")
        assert seg.is_valid is False

    def test_effective_duration_positive(self):
        """正的effective_duration."""
        seg = ConcatSegment(video_path="v.mp4", duration=10.5)
        assert seg.effective_duration == 10.5

    def test_effective_duration_zero(self):
        """duration为0时effective_duration为0."""
        seg = ConcatSegment(video_path="v.mp4", duration=0)
        assert seg.effective_duration == 0.0

    def test_effective_duration_negative(self):
        """负的duration被钳制到0."""
        seg = ConcatSegment(video_path="v.mp4", duration=-5)
        assert seg.effective_duration == 0.0


# ── ConcatSegment.from_dict 测试 ─────────────────────────────────────────────


class TestConcatSegmentFromDict:
    """ConcatSegment.from_dict 工厂方法测试."""

    def test_none_input(self):
        """None输入返回默认片段（空路径）."""
        seg = ConcatSegment.from_dict(None)
        assert seg.video_path == ""
        assert seg.is_valid is False

    def test_empty_dict(self):
        """空dict返回默认."""
        seg = ConcatSegment.from_dict({})
        assert seg.video_path == ""

    def test_not_dict_input(self):
        """非dict输入安全处理."""
        seg = ConcatSegment.from_dict("not_a_dict")
        assert seg.video_path == ""

    def test_list_input(self):
        """list输入安全处理."""
        seg = ConcatSegment.from_dict([1, 2, 3])
        assert seg.video_path == ""

    def test_with_video_path(self):
        """带视频路径."""
        seg = ConcatSegment.from_dict({"video_path": "/tmp/v.mp4"})
        assert seg.video_path == "/tmp/v.mp4"
        assert seg.is_valid is True

    def test_with_start_time(self):
        """带start_time."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "start_time": 3.5})
        assert seg.start_time == 3.5

    def test_start_time_negative_clamped(self):
        """负的start_time钳制到0."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "start_time": -5})
        assert seg.start_time == 0.0

    def test_start_time_invalid_string(self):
        """无效start_time字符串回退到0."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "start_time": "abc"})
        assert seg.start_time == 0.0

    def test_with_duration(self):
        """带duration."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "duration": 8.5})
        assert seg.duration == 8.5

    def test_duration_negative_clamped(self):
        """负的duration钳制到0."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "duration": -10})
        assert seg.duration == 0.0

    def test_duration_invalid_string(self):
        """无效duration回退到0."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "duration": "xyz"})
        assert seg.duration == 0.0

    def test_has_audio_true(self):
        """has_audio为True."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "has_audio": True})
        assert seg.has_audio is True

    def test_has_audio_false(self):
        """has_audio为False."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "has_audio": False})
        assert seg.has_audio is False

    def test_has_audio_default_true(self):
        """has_audio默认True."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4"})
        assert seg.has_audio is True

    def test_string_start_time(self):
        """字符串形式的start_time."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "start_time": "2.5"})
        assert seg.start_time == 2.5

    def test_int_duration(self):
        """整数duration."""
        seg = ConcatSegment.from_dict({"video_path": "v.mp4", "duration": 10})
        assert seg.duration == 10.0


# ── ConcatConfig 基础测试 ────────────────────────────────────────────────────


class TestConcatConfig:
    """ConcatConfig 基础测试."""

    def test_default_creation(self):
        """默认创建."""
        cfg = ConcatConfig()
        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
        assert cfg.transition == "none"
        assert cfg.transition_duration == 0.3

    def test_with_segments(self):
        """带片段创建."""
        segs = [ConcatSegment(video_path="v1.mp4"), ConcatSegment(video_path="v2.mp4")]
        cfg = ConcatConfig(segments=segs, output_width=1920, output_height=1080)
        assert len(cfg.segments) == 2
        assert cfg.output_width == 1920
        assert cfg.output_height == 1080

    # 属性测试

    def test_has_effect_with_two_segments(self):
        """2个以上有效片段has_effect为True."""
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="v1.mp4"),
                ConcatSegment(video_path="v2.mp4"),
            ]
        )
        assert cfg.has_effect is True

    def test_has_effect_with_one_segment(self):
        """只有1个有效片段has_effect为False."""
        cfg = ConcatConfig(segments=[ConcatSegment(video_path="v1.mp4")])
        assert cfg.has_effect is False

    def test_has_effect_with_no_segments(self):
        """空片段has_effect为False."""
        cfg = ConcatConfig()
        assert cfg.has_effect is False

    def test_valid_segment_count(self):
        """有效片段计数."""
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="v1.mp4"),
                ConcatSegment(video_path=""),  # 无效
                ConcatSegment(video_path="v2.mp4"),
            ]
        )
        assert cfg.valid_segment_count == 2

    def test_total_segments_alias(self):
        """total_segments是valid_segment_count的别名."""
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="v1.mp4"),
                ConcatSegment(video_path="v2.mp4"),
            ]
        )
        assert cfg.total_segments == cfg.valid_segment_count
        assert cfg.total_segments == 2

    def test_first_valid_segment(self):
        """第一个有效片段."""
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path=""),  # 无效
                ConcatSegment(video_path="first_valid.mp4"),
                ConcatSegment(video_path="second.mp4"),
            ]
        )
        first = cfg.first_valid_segment
        assert first is not None
        assert first.video_path == "first_valid.mp4"

    def test_first_valid_segment_none(self):
        """无有效片段时first_valid_segment为None."""
        cfg = ConcatConfig(segments=[ConcatSegment(video_path="")])
        assert cfg.first_valid_segment is None

    def test_first_valid_segment_empty(self):
        """空列表时为None."""
        cfg = ConcatConfig()
        assert cfg.first_valid_segment is None

    def test_estimated_total_duration(self):
        """估算总时长."""
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="v1.mp4", duration=10.0),
                ConcatSegment(video_path="v2.mp4", duration=5.5),
                ConcatSegment(video_path="v3.mp4", duration=0),  # 不计入
            ]
        )
        assert cfg.estimated_total_duration == pytest.approx(15.5)

    def test_estimated_total_duration_empty(self):
        """空片段时长为0."""
        cfg = ConcatConfig()
        assert cfg.estimated_total_duration == 0.0

    def test_estimated_total_duration_skips_invalid(self):
        """跳过无效片段."""
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="", duration=100),  # 无效，跳过
                ConcatSegment(video_path="v1.mp4", duration=5.0),
            ]
        )
        assert cfg.estimated_total_duration == 5.0


# ── ConcatConfig.from_config_dict 测试 ───────────────────────────────────────


class TestConcatConfigFromConfigDict:
    """ConcatConfig.from_config_dict 工厂方法测试."""

    def test_none_config(self):
        """None返回默认配置."""
        cfg = ConcatConfig.from_config_dict(None)
        assert cfg.segments == []
        assert cfg.output_width == 0

    def test_empty_config(self):
        """空dict返回默认."""
        cfg = ConcatConfig.from_config_dict({})
        assert cfg.segments == []

    def test_not_dict(self):
        """非dict安全处理."""
        cfg = ConcatConfig.from_config_dict("not_dict")
        assert cfg.segments == []

    def test_with_segments(self):
        """带片段列表."""
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "v1.mp4", "duration": 10},
                    {"video_path": "v2.mp4", "start_time": 2},
                ]
            }
        )
        assert len(cfg.segments) == 2
        assert cfg.segments[0].video_path == "v1.mp4"
        assert cfg.segments[0].duration == 10.0
        assert cfg.segments[1].start_time == 2.0

    def test_segments_not_list(self):
        """segments不是list时忽略."""
        cfg = ConcatConfig.from_config_dict({"segments": "not_a_list"})
        assert cfg.segments == []

    def test_skips_segments_without_path(self):
        """跳过没有video_path的片段."""
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "v1.mp4"},
                    {"duration": 5},  # 没有video_path
                    {"video_path": ""},  # 空path
                ]
            }
        )
        assert len(cfg.segments) == 1
        assert cfg.segments[0].video_path == "v1.mp4"

    def test_skips_non_dict_segments(self):
        """跳过非dict片段."""
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "v1.mp4"},
                    "not_a_dict",
                    None,
                    123,
                ]
            }
        )
        assert len(cfg.segments) == 1

    def test_output_dimensions(self):
        """输出尺寸."""
        cfg = ConcatConfig.from_config_dict(
            {
                "output_width": 1920,
                "output_height": 1080,
            }
        )
        assert cfg.output_width == 1920
        assert cfg.output_height == 1080

    def test_output_dimensions_negative_clamped(self):
        """负的尺寸钳制到0."""
        cfg = ConcatConfig.from_config_dict(
            {
                "output_width": -100,
                "output_height": -50,
            }
        )
        assert cfg.output_width == 0
        assert cfg.output_height == 0

    def test_output_dimensions_invalid(self):
        """无效尺寸回退到0."""
        cfg = ConcatConfig.from_config_dict(
            {
                "output_width": "abc",
                "output_height": "xyz",
            }
        )
        assert cfg.output_width == 0
        assert cfg.output_height == 0

    def test_output_fps(self):
        """输出帧率."""
        cfg = ConcatConfig.from_config_dict({"output_fps": 30.0})
        assert cfg.output_fps == 30.0

    def test_output_fps_negative_clamped(self):
        """负帧率钳制到0."""
        cfg = ConcatConfig.from_config_dict({"output_fps": -5})
        assert cfg.output_fps == 0.0

    def test_force_reencode(self):
        """强制重新编码."""
        cfg = ConcatConfig.from_config_dict({"force_reencode": True})
        assert cfg.force_reencode is True

    def test_force_reencode_default_false(self):
        """默认不强制重编码."""
        cfg = ConcatConfig.from_config_dict({})
        assert cfg.force_reencode is False

    def test_transition(self):
        """转场效果."""
        cfg = ConcatConfig.from_config_dict({"transition": "crossfade"})
        assert cfg.transition == "crossfade"

    def test_transition_default_none(self):
        """默认转场none."""
        cfg = ConcatConfig.from_config_dict({})
        assert cfg.transition == "none"

    def test_transition_duration(self):
        """转场时长."""
        cfg = ConcatConfig.from_config_dict({"transition_duration": 1.0})
        assert cfg.transition_duration == 1.0

    def test_transition_duration_min(self):
        """转场时长最小值0.1."""
        cfg = ConcatConfig.from_config_dict({"transition_duration": 0.01})
        assert cfg.transition_duration == 0.1

    def test_transition_duration_invalid(self):
        """无效转场时长回退到默认."""
        cfg = ConcatConfig.from_config_dict({"transition_duration": "invalid"})
        assert cfg.transition_duration == 0.3


# ── clamp_segments 测试 ──────────────────────────────────────────────────────


class TestClampSegments:
    """clamp_segments 截断测试."""

    def test_under_limit_no_change(self):
        """低于上限时不截断."""
        segs = [ConcatSegment(video_path=f"v{i}.mp4") for i in range(10)]
        cfg = ConcatConfig(segments=segs)
        cfg.clamp_segments(max_segments=50)
        assert len(cfg.segments) == 10

    def test_over_limit_truncated(self):
        """超过上限时截断."""
        segs = [ConcatSegment(video_path=f"v{i}.mp4") for i in range(100)]
        cfg = ConcatConfig(segments=segs)
        cfg.clamp_segments(max_segments=30)
        assert len(cfg.segments) == 30
        assert cfg.segments[0].video_path == "v0.mp4"
        assert cfg.segments[-1].video_path == "v29.mp4"

    def test_default_max_uses_constant(self):
        """默认max_segments使用常量."""
        segs = [ConcatSegment(video_path=f"v{i}.mp4") for i in range(100)]
        cfg = ConcatConfig(segments=segs)
        cfg.clamp_segments()  # 默认MAX_CONCAT_SEGMENTS
        assert len(cfg.segments) == MAX_CONCAT_SEGMENTS

    def test_empty_segments(self):
        """空列表不报错."""
        cfg = ConcatConfig()
        cfg.clamp_segments()
        assert cfg.segments == []

    def test_at_limit_stays(self):
        """刚好在上限时不变."""
        segs = [ConcatSegment(video_path=f"v{i}.mp4") for i in range(50)]
        cfg = ConcatConfig(segments=segs)
        cfg.clamp_segments(max_segments=50)
        assert len(cfg.segments) == 50