xiaoxia-saas/tests/unit/test_clip_operations.py

"""clip_operations 单元测试 — 片段分割/合并纯逻辑层。

覆盖：
- validate_split_time: 分割时间校验
- calculate_split: 分割参数计算
- validate_merge_clips: 合并校验
- calculate_merge: 合并参数计算
- calculate_reorder_new_orders: 重排 order 计算
- calculate_shift_orders: order 偏移计算
"""

from __future__ import annotations

import unittest
from dataclasses import FrozenInstanceError, dataclass

from packages.domain.clip_operations import (
    ROUND_PRECISION,
    MergeResult,
    SplitResult,
    calculate_merge,
    calculate_reorder_new_orders,
    calculate_shift_orders,
    calculate_split,
    validate_merge_clips,
    validate_split_time,
)

# ── Mock Clip ────────────────────────────────────────────────────────────────


@dataclass
class _MockClip:
    id: str = "c1"
    plan_id: str = "p1"
    order: int = 0
    clip_type: str = "main"
    duration: float = 5.0
    start_time: float = 0.0
    text_content: str = ""
    config: dict | None = None


# ── validate_split_time 测试 ─────────────────────────────────────────────────


class TestValidateSplitTime(unittest.TestCase):
    """validate_split_time 分割时间校验测试。"""

    def test_valid_split(self):
        """合法分割时间不报错。"""
        validate_split_time(2.5, 5.0)  # 不抛异常

    def test_split_at_zero(self):
        """分割时间为 0 时报错。"""
        with self.assertRaises(ValueError) as ctx:
            validate_split_time(0.0, 5.0)
        self.assertIn("分割时间", str(ctx.exception))

    def test_split_negative(self):
        """分割时间为负数时报错。"""
        with self.assertRaises(ValueError):
            validate_split_time(-1.0, 5.0)

    def test_split_at_duration(self):
        """分割时间等于 duration 时报错。"""
        with self.assertRaises(ValueError):
            validate_split_time(5.0, 5.0)

    def test_split_over_duration(self):
        """分割时间超过 duration 时报错。"""
        with self.assertRaises(ValueError):
            validate_split_time(6.0, 5.0)

    def test_split_very_small(self):
        """很小的正数是合法的。"""
        validate_split_time(0.001, 5.0)  # 不抛异常

    def test_split_just_below_duration(self):
        """略小于 duration 是合法的。"""
        validate_split_time(4.999, 5.0)  # 不抛异常

    def test_error_message_contains_duration(self):
        """错误消息包含 duration 值。"""
        with self.assertRaises(ValueError) as ctx:
            validate_split_time(6.0, 5.0)
        self.assertIn("5.000", str(ctx.exception))


# ── calculate_split 测试 ─────────────────────────────────────────────────────


class TestCalculateSplit(unittest.TestCase):
    """calculate_split 分割计算测试。"""

    def test_middle_split(self):
        """从中间分割。"""
        result = calculate_split(duration=10.0, split_time=5.0)
        self.assertIsInstance(result, SplitResult)
        self.assertEqual(result.left_duration, 5.0)
        self.assertEqual(result.right_duration, 5.0)
        self.assertEqual(result.right_start_time, 5.0)
        self.assertEqual(result.left_trim_end, 5.0)
        self.assertEqual(result.right_trim_start, 5.0)

    def test_early_split(self):
        """从开头附近分割。"""
        result = calculate_split(duration=10.0, split_time=2.0)
        self.assertEqual(result.left_duration, 2.0)
        self.assertEqual(result.right_duration, 8.0)
        self.assertEqual(result.right_start_time, 2.0)

    def test_late_split(self):
        """从结尾附近分割。"""
        result = calculate_split(duration=10.0, split_time=8.0)
        self.assertEqual(result.left_duration, 8.0)
        self.assertEqual(result.right_duration, 2.0)

    def test_with_start_time_offset(self):
        """带 start_time 偏移。"""
        result = calculate_split(duration=5.0, split_time=2.0, start_time=10.0)
        self.assertEqual(result.left_duration, 2.0)
        self.assertEqual(result.right_duration, 3.0)
        self.assertEqual(result.right_start_time, 12.0)

    def test_zero_start_time(self):
        """start_time 为 0 时 right_start_time 等于 left_duration。"""
        result = calculate_split(duration=5.0, split_time=2.0, start_time=0.0)
        self.assertEqual(result.right_start_time, result.left_duration)

    def test_round_to_precision(self):
        """结果精度符合设置。"""
        result = calculate_split(duration=1.0, split_time=1 / 3, precision=3)
        # 1/3 ≈ 0.333（3位精度）
        self.assertAlmostEqual(result.left_duration, 0.333, places=3)
        self.assertAlmostEqual(result.right_duration, 0.667, places=3)

    def test_default_precision_3(self):
        """默认精度为 3 位小数。"""
        self.assertEqual(ROUND_PRECISION, 3)
        result = calculate_split(duration=1.0, split_time=0.333333)
        # 默认用 ROUND_PRECISION = 3
        self.assertEqual(result.left_duration, 0.333)

    def test_custom_precision(self):
        """自定义精度。"""
        result = calculate_split(duration=1.0, split_time=0.123456, precision=5)
        self.assertEqual(result.left_duration, 0.12346)
        self.assertEqual(result.right_duration, 0.87654)

    def test_invalid_split_time_raises(self):
        """不合法的分割时间抛出 ValueError。"""
        with self.assertRaises(ValueError):
            calculate_split(duration=5.0, split_time=0.0)
        with self.assertRaises(ValueError):
            calculate_split(duration=5.0, split_time=5.0)
        with self.assertRaises(ValueError):
            calculate_split(duration=5.0, split_time=-1.0)

    def test_trim_values_match_durations(self):
        """trim 值与对应时长一致。"""
        result = calculate_split(duration=7.5, split_time=3.0)
        self.assertEqual(result.right_trim_start, result.left_duration)
        self.assertEqual(result.left_trim_end, result.right_duration)

    def test_frozen_result(self):
        """SplitResult 是 frozen dataclass。"""
        result = calculate_split(duration=5.0, split_time=2.0)
        with self.assertRaises(FrozenInstanceError):
            result.left_duration = 3.0  # type: ignore[misc]


# ── validate_merge_clips 测试 ────────────────────────────────────────────────


class TestValidateMergeClips(unittest.TestCase):
    """validate_merge_clips 合并校验测试。"""

    def test_two_consecutive_clips(self):
        """两个连续片段：校验通过。"""
        clips = [
            _MockClip(id="c1", plan_id="p1", order=0, clip_type="main"),
            _MockClip(id="c2", plan_id="p1", order=1, clip_type="main"),
        ]
        plan_id, first_order = validate_merge_clips(clips)
        self.assertEqual(plan_id, "p1")
        self.assertEqual(first_order, 0)

    def test_three_consecutive_clips(self):
        """三个连续片段：校验通过。"""
        clips = [
            _MockClip(id="c1", plan_id="p1", order=2, clip_type="main"),
            _MockClip(id="c2", plan_id="p1", order=3, clip_type="main"),
            _MockClip(id="c3", plan_id="p1", order=4, clip_type="main"),
        ]
        plan_id, first_order = validate_merge_clips(clips)
        self.assertEqual(plan_id, "p1")
        self.assertEqual(first_order, 2)

    def test_unordered_input(self):
        """输入顺序不影响校验（自动排序）。"""
        clips = [
            _MockClip(id="c2", plan_id="p1", order=1, clip_type="main"),
            _MockClip(id="c1", plan_id="p1", order=0, clip_type="main"),
        ]
        plan_id, first_order = validate_merge_clips(clips)
        self.assertEqual(plan_id, "p1")
        self.assertEqual(first_order, 0)

    def test_single_clip_raises(self):
        """只有一个片段报错。"""
        clips = [_MockClip(id="c1", plan_id="p1", order=0)]
        with self.assertRaises(ValueError) as ctx:
            validate_merge_clips(clips)
        self.assertIn("至少需要 2 个", str(ctx.exception))

    def test_empty_clips_raises(self):
        """空列表报错。"""
        with self.assertRaises(ValueError):
            validate_merge_clips([])

    def test_different_plan_raises(self):
        """不同计划的片段报错。"""
        clips = [
            _MockClip(id="c1", plan_id="p1", order=0, clip_type="main"),
            _MockClip(id="c2", plan_id="p2", order=1, clip_type="main"),
        ]
        with self.assertRaises(ValueError) as ctx:
            validate_merge_clips(clips)
        self.assertIn("同一计划", str(ctx.exception))

    def test_non_consecutive_raises(self):
        """不连续的片段报错。"""
        clips = [
            _MockClip(id="c1", plan_id="p1", order=0, clip_type="main"),
            _MockClip(id="c2", plan_id="p1", order=2, clip_type="main"),
        ]
        with self.assertRaises(ValueError) as ctx:
            validate_merge_clips(clips)
        self.assertIn("不连续", str(ctx.exception))

    def test_different_type_raises(self):
        """类型不同报错。"""
        clips = [
            _MockClip(id="c1", plan_id="p1", order=0, clip_type="main"),
            _MockClip(id="c2", plan_id="p1", order=1, clip_type="title"),
        ]
        with self.assertRaises(ValueError) as ctx:
            validate_merge_clips(clips)
        self.assertIn("相同类型", str(ctx.exception))

    def test_gap_in_middle_raises(self):
        """中间有间隔报错（三个片段中间缺一个）。"""
        clips = [
            _MockClip(id="c1", plan_id="p1", order=0, clip_type="main"),
            _MockClip(id="c2", plan_id="p1", order=1, clip_type="main"),
            _MockClip(id="c3", plan_id="p1", order=3, clip_type="main"),
        ]
        with self.assertRaises(ValueError):
            validate_merge_clips(clips)


# ── calculate_merge 测试 ─────────────────────────────────────────────────────


class TestCalculateMerge(unittest.TestCase):
    """calculate_merge 合并计算测试。"""

    def test_two_clips_total_duration(self):
        """两个片段：总时长相加。"""
        clips = [
            _MockClip(id="c1", order=0, duration=5.0),
            _MockClip(id="c2", order=1, duration=3.0),
        ]
        result = calculate_merge(clips)
        self.assertIsInstance(result, MergeResult)
        self.assertEqual(result.total_duration, 8.0)
        self.assertEqual(result.first_order, 0)
        self.assertEqual(result.shift_amount, 1)

    def test_three_clips_total_duration(self):
        """三个片段：总时长相加。"""
        clips = [
            _MockClip(id="c1", order=2, duration=2.0),
            _MockClip(id="c2", order=3, duration=3.5),
            _MockClip(id="c3", order=4, duration=4.5),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.total_duration, 10.0)
        self.assertEqual(result.first_order, 2)
        self.assertEqual(result.shift_amount, 2)

    def test_unordered_input(self):
        """输入顺序不影响结果（自动按 order 排序）。"""
        clips = [
            _MockClip(id="c2", order=1, duration=3.0),
            _MockClip(id="c1", order=0, duration=5.0),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.total_duration, 8.0)
        self.assertEqual(result.first_order, 0)

    def test_merge_text_newline_join(self):
        """文案用换行连接，跳过空字符串。"""
        clips = [
            _MockClip(id="c1", order=0, duration=2.0, text_content="第一段"),
            _MockClip(id="c2", order=1, duration=3.0, text_content="第二段"),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.merged_text, "第一段\n第二段")

    def test_merge_text_skip_empty(self):
        """空文案或纯空格被跳过。"""
        clips = [
            _MockClip(id="c1", order=0, duration=2.0, text_content=""),
            _MockClip(id="c2", order=1, duration=3.0, text_content="有内容"),
            _MockClip(id="c3", order=2, duration=1.0, text_content="   "),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.merged_text, "有内容")

    def test_all_empty_text(self):
        """所有文案都空时合并结果为空字符串。"""
        clips = [
            _MockClip(id="c1", order=0, duration=2.0, text_content=""),
            _MockClip(id="c2", order=1, duration=3.0, text_content=" "),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.merged_text, "")

    def test_merge_config_later_overwrites(self):
        """后面的 config 覆盖前面的。"""
        clips = [
            _MockClip(
                id="c1",
                order=0,
                duration=2.0,
                config={"color": "red", "speed": 1.0},
            ),
            _MockClip(
                id="c2",
                order=1,
                duration=3.0,
                config={"color": "blue", "filter": "vintage"},
            ),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.merged_config["color"], "blue")  # 后面的覆盖
        self.assertEqual(result.merged_config["speed"], 1.0)  # 保留前面的
        self.assertEqual(result.merged_config["filter"], "vintage")  # 新增的

    def test_merge_config_none_handled(self):
        """config 为 None 时正常处理。"""
        clips = [
            _MockClip(id="c1", order=0, duration=2.0, config=None),
            _MockClip(id="c2", order=1, duration=3.0, config={"key": "value"}),
        ]
        result = calculate_merge(clips)
        self.assertEqual(result.merged_config["key"], "value")

    def test_trim_fields_removed(self):
        """trim_start 和 trim_end 被移除（合并后是完整片段）。"""
        clips = [
            _MockClip(
                id="c1",
                order=0,
                duration=2.0,
                config={"trim_end": 2.0, "color": "red"},
            ),
            _MockClip(
                id="c2",
                order=1,
                duration=3.0,
                config={"trim_start": 1.0, "speed": 1.5},
            ),
        ]
        result = calculate_merge(clips)
        self.assertNotIn("trim_start", result.merged_config)
        self.assertNotIn("trim_end", result.merged_config)
        self.assertEqual(result.merged_config["color"], "red")
        self.assertEqual(result.merged_config["speed"], 1.5)

    def test_empty_clips_raises(self):
        """空列表报错。"""
        with self.assertRaises(ValueError):
            calculate_merge([])

    def test_single_clip_merge(self):
        """单个片段也能计算（虽然 validate 会限制，但函数本身支持）。"""
        clips = [_MockClip(id="c1", order=5, duration=5.0, text_content="唯一")]
        result = calculate_merge(clips)
        self.assertEqual(result.total_duration, 5.0)
        self.assertEqual(result.first_order, 5)
        self.assertEqual(result.shift_amount, 0)
        self.assertEqual(result.merged_text, "唯一")

    def test_round_precision(self):
        """时长精度符合设置。"""
        clips = [
            _MockClip(id="c1", order=0, duration=1 / 3),
            _MockClip(id="c2", order=1, duration=1 / 3),
        ]
        result = calculate_merge(clips, precision=3)
        self.assertEqual(result.total_duration, 0.667)

    def test_frozen_result(self):
        """MergeResult 是 frozen dataclass。"""
        clips = [
            _MockClip(id="c1", order=0, duration=2.0),
            _MockClip(id="c2", order=1, duration=3.0),
        ]
        result = calculate_merge(clips)
        with self.assertRaises(FrozenInstanceError):
            result.total_duration = 10.0  # type: ignore[misc]


# ── calculate_reorder_new_orders 测试 ────────────────────────────────────────


class TestCalculateReorderNewOrders(unittest.TestCase):
    """calculate_reorder_new_orders 重排计算测试。"""

    def test_two_items_swap(self):
        """两个元素交换顺序。"""
        items = [
            _MockClip(id="a", order=0),
            _MockClip(id="b", order=1),
        ]
        result = calculate_reorder_new_orders(["b", "a"], items)
        self.assertEqual(result["b"], 0)
        self.assertEqual(result["a"], 1)

    def test_three_items_reorder(self):
        """三个元素重新排序。"""
        items = [
            _MockClip(id="a", order=0),
            _MockClip(id="b", order=1),
            _MockClip(id="c", order=2),
        ]
        result = calculate_reorder_new_orders(["c", "a", "b"], items)
        self.assertEqual(result["c"], 0)
        self.assertEqual(result["a"], 1)
        self.assertEqual(result["b"], 2)

    def test_same_order(self):
        """顺序不变。"""
        items = [
            _MockClip(id="a", order=0),
            _MockClip(id="b", order=1),
        ]
        result = calculate_reorder_new_orders(["a", "b"], items)
        self.assertEqual(result["a"], 0)
        self.assertEqual(result["b"], 1)

    def test_mismatched_ids_raises(self):
        """ID 不匹配报错。"""
        items = [
            _MockClip(id="a", order=0),
            _MockClip(id="b", order=1),
        ]
        with self.assertRaises(ValueError) as ctx:
            calculate_reorder_new_orders(["a", "c"], items)
        self.assertIn("不匹配", str(ctx.exception))

    def test_extra_id_in_list_raises(self):
        """有序列表多出 ID 报错。"""
        items = [_MockClip(id="a", order=0)]
        with self.assertRaises(ValueError):
            calculate_reorder_new_orders(["a", "b"], items)

    def test_missing_id_raises(self):
        """有序列表缺少 ID 报错。"""
        items = [
            _MockClip(id="a", order=0),
            _MockClip(id="b", order=1),
        ]
        with self.assertRaises(ValueError):
            calculate_reorder_new_orders(["a"], items)

    def test_custom_attr_names(self):
        """自定义属性名。"""

        @dataclass
        class _Item:
            key: str
            pos: int

        items = [_Item(key="x", pos=0), _Item(key="y", pos=1)]
        result = calculate_reorder_new_orders(["y", "x"], items, id_attr="key", order_attr="pos")
        self.assertEqual(result["y"], 0)
        self.assertEqual(result["x"], 1)


# ── calculate_shift_orders 测试 ──────────────────────────────────────────────


class TestCalculateShiftOrders(unittest.TestCase):
    """calculate_shift_orders order 偏移计算测试。"""

    def test_shift_positive(self):
        """正偏移：order 增加。"""
        items = [
            _MockClip(id="c1", order=0),
            _MockClip(id="c2", order=1),
            _MockClip(id="c3", order=2),
        ]
        result = calculate_shift_orders(items, threshold_order=0, shift=1)
        shifted = {c.id: new_order for c, new_order in result}
        # order > 0 的才会被偏移
        self.assertEqual(len(shifted), 2)
        self.assertEqual(shifted["c2"], 2)
        self.assertEqual(shifted["c3"], 3)
        self.assertNotIn("c1", shifted)

    def test_shift_negative(self):
        """负偏移：order 减少。"""
        items = [
            _MockClip(id="c1", order=0),
            _MockClip(id="c2", order=2),
            _MockClip(id="c3", order=3),
        ]
        result = calculate_shift_orders(items, threshold_order=1, shift=-1)
        shifted = {c.id: new_order for c, new_order in result}
        self.assertEqual(shifted["c2"], 1)
        self.assertEqual(shifted["c3"], 2)
        self.assertNotIn("c1", shifted)

    def test_excluded_ids(self):
        """排除指定 ID。"""
        items = [
            _MockClip(id="c1", order=0),
            _MockClip(id="c2", order=1),
            _MockClip(id="c3", order=2),
        ]
        result = calculate_shift_orders(
            items,
            threshold_order=0,
            shift=1,
            excluded_ids={"c2"},
        )
        shifted = {c.id: new_order for c, new_order in result}
        self.assertNotIn("c1", shifted)
        self.assertNotIn("c2", shifted)  # 被排除
        self.assertEqual(shifted["c3"], 3)

    def test_no_items_above_threshold(self):
        """没有元素高于阈值时返回空。"""
        items = [
            _MockClip(id="c1", order=0),
            _MockClip(id="c2", order=1),
        ]
        result = calculate_shift_orders(items, threshold_order=5, shift=1)
        self.assertEqual(result, [])

    def test_zero_shift(self):
        """偏移量为 0 时仍返回（虽然 order 不变）。"""
        items = [
            _MockClip(id="c1", order=0),
            _MockClip(id="c2", order=1),
        ]
        result = calculate_shift_orders(items, threshold_order=0, shift=0)
        shifted = {c.id: new_order for c, new_order in result}
        self.assertEqual(shifted["c2"], 1)  # 1 + 0 = 1

    def test_threshold_exclusive(self):
        """阈值是严格大于（不包含等于）。"""
        items = [
            _MockClip(id="c1", order=5),  # 等于 threshold，不偏移
            _MockClip(id="c2", order=6),  # 大于 threshold，偏移
        ]
        result = calculate_shift_orders(items, threshold_order=5, shift=2)
        shifted = {c.id: new_order for c, new_order in result}
        self.assertNotIn("c1", shifted)
        self.assertEqual(shifted["c2"], 8)

    def test_empty_items(self):
        """空 items 返回空列表。"""
        result = calculate_shift_orders([], threshold_order=0, shift=1)
        self.assertEqual(result, [])


# ── 集成测试 ────────────────────────────────────────────────────────────────


class TestEndToEndClipOperations(unittest.TestCase):
    """端到端集成测试：完整分割+重排流程。"""

    def test_split_then_shift(self):
        """分割一个片段后，后面的片段 order +1。"""
        # 模拟：有3个片段，分割第1个（order=1），后面的+1
        clips = [
            _MockClip(id="c1", order=0, duration=3.0),
            _MockClip(id="c2", order=1, duration=5.0),
            _MockClip(id="c3", order=2, duration=4.0),
        ]

        # 计算分割
        split = calculate_split(duration=5.0, split_time=2.0)

        # 后面的片段 order +1
        shift_result = calculate_shift_orders(
            clips,
            threshold_order=1,
            shift=1,
            excluded_ids={"c2"},
        )

        shifted = {c.id: new_order for c, new_order in shift_result}
        self.assertEqual(shifted["c3"], 3)  # 2+1
        self.assertNotIn("c1", shifted)  # order <= 1
        self.assertNotIn("c2", shifted)  # 被排除

        # 左半部分保留 order=1，右半部分 order=2
        self.assertEqual(split.left_duration, 2.0)
        self.assertEqual(split.right_duration, 3.0)

    def test_merge_then_shift(self):
        """合并两个片段后，后面的片段 order -1。"""
        clips = [
            _MockClip(id="c1", order=0, duration=3.0, text_content="A"),
            _MockClip(id="c2", order=1, duration=2.0, text_content="B"),
            _MockClip(id="c3", order=2, duration=4.0),
            _MockClip(id="c4", order=3, duration=1.0),
        ]

        # 校验合并
        validate_merge_clips(clips[:2])

        # 计算合并
        merge = calculate_merge(clips[:2])
        self.assertEqual(merge.total_duration, 5.0)
        self.assertEqual(merge.merged_text, "A\nB")
        self.assertEqual(merge.shift_amount, 1)

        # 后面的片段前移 1 位
        shift_result = calculate_shift_orders(
            clips,
            threshold_order=0,  # order > 0 的
            shift=-merge.shift_amount,
            excluded_ids={"c1", "c2"},
        )

        shifted = {c.id: new_order for c, new_order in shift_result}
        self.assertEqual(shifted["c3"], 1)  # 2-1
        self.assertEqual(shifted["c4"], 2)  # 3-1


if __name__ == "__main__":
    unittest.main()