xiaoxia-saas/tests/unit/test_duplication_domain.py

"""查重域模型单元测试。

覆盖：
- DuplicationRecord.create() 工厂方法及验证
- DuplicationRecord 状态转换（mark_processing / mark_completed / mark_failed）
- DuplicationRecord.can_retry() / reset_for_retry()
- DuplicateSegment.create() 工厂方法及验证
"""

from __future__ import annotations

import pytest

from packages.domain.duplication import DuplicateSegment, DuplicationRecord


class TestDuplicationRecordCreate:
    """DuplicationRecord.create() 工厂方法测试。"""

    def test_create_success(self):
        record = DuplicationRecord.create(
            user_id="user-1",
            filename="test.mp4",
            file_size=1024,
            storage_key="oss/key/test.mp4",
            duration_seconds=30.0,
        )
        assert record.user_id == "user-1"
        assert record.filename == "test.mp4"
        assert record.file_size == 1024
        assert record.storage_key == "oss/key/test.mp4"
        assert record.duration_seconds == 30.0
        assert record.status == "pending"
        assert record.duplicate_rate is None
        assert record.duplicate_count == 0
        assert record.error_message == ""
        assert record.segments == []
        assert record.video_fingerprint is None
        assert record.id  # 自动生成 ID

    def test_create_with_default_duration(self):
        record = DuplicationRecord.create(
            user_id="user-1",
            filename="test.mp4",
            file_size=1024,
            storage_key="oss/key",
        )
        assert record.duration_seconds == 0.0

    def test_create_empty_user_id_raises(self):
        with pytest.raises(ValueError, match="user_id"):
            DuplicationRecord.create(
                user_id="",
                filename="test.mp4",
                file_size=1024,
                storage_key="oss/key",
            )

    def test_create_whitespace_user_id_raises(self):
        with pytest.raises(ValueError, match="user_id"):
            DuplicationRecord.create(
                user_id="   ",
                filename="test.mp4",
                file_size=1024,
                storage_key="oss/key",
            )

    def test_create_empty_filename_raises(self):
        with pytest.raises(ValueError, match="filename"):
            DuplicationRecord.create(
                user_id="user-1",
                filename="",
                file_size=1024,
                storage_key="oss/key",
            )

    def test_create_zero_file_size_raises(self):
        with pytest.raises(ValueError, match="file_size"):
            DuplicationRecord.create(
                user_id="user-1",
                filename="test.mp4",
                file_size=0,
                storage_key="oss/key",
            )

    def test_create_negative_file_size_raises(self):
        with pytest.raises(ValueError, match="file_size"):
            DuplicationRecord.create(
                user_id="user-1",
                filename="test.mp4",
                file_size=-100,
                storage_key="oss/key",
            )


class TestDuplicationRecordStateTransitions:
    """状态转换测试。"""

    @pytest.fixture
    def record(self):
        return DuplicationRecord.create(
            user_id="user-1",
            filename="test.mp4",
            file_size=1024,
            storage_key="oss/key",
        )

    def test_mark_processing(self, record):
        record.mark_processing()
        assert record.status == "processing"

    def test_mark_completed_success(self, record):
        record.mark_processing()
        segments = [
            DuplicateSegment.create(
                source_start=0.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="existing.mp4",
                matched_start=0.0,
                matched_end=5.0,
                similarity=92.5,
            )
        ]
        record.mark_completed(duplicate_rate=15.0, duplicate_count=1, segments=segments)
        assert record.status == "completed"
        assert record.duplicate_rate == 15.0
        assert record.duplicate_count == 1
        assert len(record.segments) == 1

    def test_mark_completed_invalid_rate_raises(self, record):
        record.mark_processing()
        with pytest.raises(ValueError, match="duplicate_rate"):
            record.mark_completed(duplicate_rate=101.0, duplicate_count=0, segments=[])

    def test_mark_completed_negative_rate_raises(self, record):
        record.mark_processing()
        with pytest.raises(ValueError, match="duplicate_rate"):
            record.mark_completed(duplicate_rate=-1.0, duplicate_count=0, segments=[])

    def test_mark_failed(self, record):
        record.mark_processing()
        record.mark_failed("处理超时")
        assert record.status == "failed"
        assert record.error_message == "处理超时"


class TestDuplicationRecordRetry:
    """can_retry() 和 reset_for_retry() 测试。"""

    @pytest.fixture
    def record(self):
        return DuplicationRecord.create(
            user_id="user-1",
            filename="test.mp4",
            file_size=1024,
            storage_key="oss/key",
        )

    def test_mark_failed_sets_status_and_error(self, record):
        record.mark_processing()
        record.mark_failed("处理失败")
        assert record.status == "failed"
        assert record.error_message == "处理失败"

    def test_mark_failed_updates_timestamp(self, record):
        old_updated = record.updated_at
        record.mark_processing()
        record.mark_failed("错误")
        assert record.updated_at >= old_updated

    def test_failed_record_preserves_result_fields(self, record):
        """mark_failed 不改变 duplicate_rate 等结果字段（由 use case 层重置）。"""
        record.mark_processing()
        record.mark_completed(duplicate_rate=10.0, duplicate_count=1, segments=[])
        record.mark_failed("重试失败")
        assert record.status == "failed"
        assert record.error_message == "重试失败"
        assert record.duplicate_rate == 10.0


class TestDuplicateSegmentCreate:
    """DuplicateSegment.create() 工厂方法测试。"""

    def test_create_success(self):
        seg = DuplicateSegment.create(
            source_start=1.0,
            source_end=5.0,
            matched_video_id="vid-1",
            matched_video_name="existing.mp4",
            matched_start=2.0,
            matched_end=6.0,
            similarity=85.5,
        )
        assert seg.source_start == 1.0
        assert seg.source_end == 5.0
        assert seg.matched_video_id == "vid-1"
        assert seg.matched_video_name == "existing.mp4"
        assert seg.matched_start == 2.0
        assert seg.matched_end == 6.0
        assert seg.similarity == 85.5
        assert seg.id  # 自动生成 ID

    def test_create_negative_source_start_raises(self):
        with pytest.raises(ValueError, match="invalid source segment range"):
            DuplicateSegment.create(
                source_start=-1.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="v.mp4",
                matched_start=0.0,
                matched_end=5.0,
                similarity=80.0,
            )

    def test_create_source_end_le_start_raises(self):
        with pytest.raises(ValueError, match="invalid source segment range"):
            DuplicateSegment.create(
                source_start=5.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="v.mp4",
                matched_start=0.0,
                matched_end=5.0,
                similarity=80.0,
            )

    def test_create_negative_matched_start_raises(self):
        with pytest.raises(ValueError, match="invalid matched segment range"):
            DuplicateSegment.create(
                source_start=0.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="v.mp4",
                matched_start=-1.0,
                matched_end=5.0,
                similarity=80.0,
            )

    def test_create_matched_end_le_start_raises(self):
        with pytest.raises(ValueError, match="invalid matched segment range"):
            DuplicateSegment.create(
                source_start=0.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="v.mp4",
                matched_start=2.0,
                matched_end=1.0,
                similarity=80.0,
            )

    def test_create_similarity_out_of_range_raises(self):
        with pytest.raises(ValueError, match="similarity"):
            DuplicateSegment.create(
                source_start=0.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="v.mp4",
                matched_start=0.0,
                matched_end=5.0,
                similarity=101.0,
            )

    def test_create_negative_similarity_raises(self):
        with pytest.raises(ValueError, match="similarity"):
            DuplicateSegment.create(
                source_start=0.0,
                source_end=5.0,
                matched_video_id="vid-1",
                matched_video_name="v.mp4",
                matched_start=0.0,
                matched_end=5.0,
                similarity=-1.0,
            )