xiaoxia-saas/tests/unit/test_video_concat.py

"""video_concat 领域模型单测 — 纯逻辑，48个测试用例."""

from __future__ import annotations

import pytest

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

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


class TestConcatSegmentBasics:
    def test_default_values(self):
        seg = ConcatSegment(video_path="test.mp4")
        assert seg.video_path == "test.mp4"
        assert seg.start_time == 0.0
        assert seg.duration == 0.0
        assert seg.has_audio is True

    def test_full_params(self):
        seg = ConcatSegment(
            video_path="video.mp4",
            start_time=5.5,
            duration=10.0,
            has_audio=False,
        )
        assert seg.video_path == "video.mp4"
        assert seg.start_time == 5.5
        assert seg.duration == 10.0
        assert seg.has_audio is False


class TestConcatSegmentFromDict:
    def test_normal_dict(self):
        seg = ConcatSegment.from_dict(
            {
                "video_path": "test.mp4",
                "start_time": 2.0,
                "duration": 5.0,
                "has_audio": False,
            }
        )
        assert seg.video_path == "test.mp4"
        assert seg.start_time == 2.0
        assert seg.duration == 5.0
        assert seg.has_audio is False

    def test_empty_dict(self):
        seg = ConcatSegment.from_dict({})
        assert seg.video_path == ""
        assert seg.start_time == 0.0
        assert seg.duration == 0.0
        assert seg.has_audio is True

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

    def test_non_dict_input(self):
        seg = ConcatSegment.from_dict("not a dict")
        assert seg.video_path == ""

    def test_start_time_negative_clamped(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "start_time": -5})
        assert seg.start_time == 0.0

    def test_duration_negative_clamped(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "duration": -10})
        assert seg.duration == 0.0

    def test_start_time_invalid_string(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "start_time": "abc"})
        assert seg.start_time == 0.0

    def test_duration_invalid_string(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "duration": "xyz"})
        assert seg.duration == 0.0

    def test_start_time_int_casted(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "start_time": 3})
        assert seg.start_time == 3.0

    def test_duration_int_casted(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "duration": 7})
        assert seg.duration == 7.0

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

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

    def test_has_audio_truthy_value(self):
        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "has_audio": 1})
        assert seg.has_audio is True


class TestConcatSegmentProperties:
    def test_is_valid_with_path(self):
        seg = ConcatSegment(video_path="test.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):
        seg = ConcatSegment(video_path="a.mp4", duration=10.5)
        assert seg.effective_duration == 10.5

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

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


# ── ConcatConfig 测试 ────────────────────────────────────────────────────────


class TestConcatConfigBasics:
    def test_default_values(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="a.mp4")]
        cfg = ConcatConfig(segments=segs)
        assert len(cfg.segments) == 1
        assert cfg.segments[0].video_path == "a.mp4"


class TestConcatConfigFromDict:
    def test_none_config(self):
        cfg = ConcatConfig.from_config_dict(None)
        assert cfg.segments == []
        assert cfg.output_width == 0

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

    def test_non_dict_input(self):
        cfg = ConcatConfig.from_config_dict("config")
        assert cfg.segments == []

    def test_with_valid_segments(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "a.mp4", "duration": 10},
                    {"video_path": "b.mp4", "duration": 20},
                ],
            }
        )
        assert len(cfg.segments) == 2
        assert cfg.segments[0].video_path == "a.mp4"
        assert cfg.segments[1].video_path == "b.mp4"

    def test_skips_empty_video_path(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "a.mp4"},
                    {"video_path": ""},
                    {"video_path": "b.mp4"},
                ],
            }
        )
        assert len(cfg.segments) == 2

    def test_skips_invalid_segment_dict(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "segments": [
                    {"video_path": "a.mp4"},
                    "not a dict",
                    {"video_path": "b.mp4"},
                ],
            }
        )
        assert len(cfg.segments) == 2

    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(
            {
                "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_output_width_negative_clamped(self):
        cfg = ConcatConfig.from_config_dict({"output_width": -100})
        assert cfg.output_width == 0

    def test_output_height_invalid_string(self):
        cfg = ConcatConfig.from_config_dict({"output_height": "abc"})
        assert cfg.output_height == 0

    def test_output_fps_invalid_string(self):
        cfg = ConcatConfig.from_config_dict({"output_fps": "xyz"})
        assert cfg.output_fps == 0.0

    def test_transition_params(self):
        cfg = ConcatConfig.from_config_dict(
            {
                "transition": "crossfade",
                "transition_duration": 1.0,
            }
        )
        assert cfg.transition == "crossfade"
        assert cfg.transition_duration == 1.0

    def test_transition_duration_minimum(self):
        cfg = ConcatConfig.from_config_dict({"transition_duration": 0.01})
        assert cfg.transition_duration == 0.1

    def test_transition_duration_negative(self):
        cfg = ConcatConfig.from_config_dict({"transition_duration": -1})
        assert cfg.transition_duration == 0.1

    def test_force_reencode_false_by_default(self):
        cfg = ConcatConfig.from_config_dict({})
        assert cfg.force_reencode is False


class TestConcatConfigProperties:
    def test_has_effect_two_segments(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="a.mp4"),
                ConcatSegment(video_path="b.mp4"),
            ]
        )
        assert cfg.has_effect is True

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

    def test_has_effect_empty(self):
        cfg = ConcatConfig()
        assert cfg.has_effect is False

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

    def test_valid_segment_count(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="a.mp4"),
                ConcatSegment(video_path=""),
                ConcatSegment(video_path="b.mp4"),
            ]
        )
        assert cfg.valid_segment_count == 2

    def test_first_valid_segment(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path=""),
                ConcatSegment(video_path="first.mp4"),
                ConcatSegment(video_path="second.mp4"),
            ]
        )
        assert cfg.first_valid_segment is not None
        assert cfg.first_valid_segment.video_path == "first.mp4"

    def test_first_valid_segment_none_when_all_empty(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path=""),
                ConcatSegment(video_path=""),
            ]
        )
        assert cfg.first_valid_segment is None

    def test_first_valid_segment_empty_list(self):
        cfg = ConcatConfig()
        assert cfg.first_valid_segment is None

    def test_estimated_total_duration(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="a.mp4", duration=10.0),
                ConcatSegment(video_path="b.mp4", duration=20.0),
                ConcatSegment(video_path="c.mp4", duration=0.0),
            ]
        )
        assert cfg.estimated_total_duration == 30.0

    def test_estimated_total_duration_skips_invalid(self):
        cfg = ConcatConfig(
            segments=[
                ConcatSegment(video_path="", duration=10.0),
                ConcatSegment(video_path="a.mp4", duration=5.0),
            ]
        )
        assert cfg.estimated_total_duration == 5.0


class TestConcatConfigClampSegments:
    def test_clamp_when_over_max(self):
        segs = [ConcatSegment(video_path=f"s{i}.mp4") for i in range(100)]
        cfg = ConcatConfig(segments=segs)
        cfg.clamp_segments(50)
        assert len(cfg.segments) == 50
        assert cfg.segments[0].video_path == "s0.mp4"
        assert cfg.segments[-1].video_path == "s49.mp4"

    def test_no_clamp_when_under_max(self):
        segs = [ConcatSegment(video_path=f"s{i}.mp4") for i in range(10)]
        cfg = ConcatConfig(segments=segs)
        cfg.clamp_segments(50)
        assert len(cfg.segments) == 10

    def test_default_max_constant(self):
        assert MAX_CONCAT_SEGMENTS == 50


class TestConstants:
    def test_allowed_extensions(self):
        assert ".mp4" in ALLOWED_VIDEO_EXTENSIONS
        assert ".mov" in ALLOWED_VIDEO_EXTENSIONS
        assert ".webm" in ALLOWED_VIDEO_EXTENSIONS

    def test_demuxer_params(self):
        assert "codec_name" in CONCAT_DEMUXER_REQUIRED_PARAMS
        assert "width" in CONCAT_DEMUXER_REQUIRED_PARAMS
        assert "r_frame_rate" in CONCAT_DEMUXER_REQUIRED_PARAMS