diff --git a/tests/unit/test_duplication_domain.py b/tests/unit/test_duplication_domain.py index ac887d436..172895dc5 100755 --- a/tests/unit/test_duplication_domain.py +++ b/tests/unit/test_duplication_domain.py @@ -1,6 +1,6 @@ -""" -Duplication 查重记录领域模型单元测试 -""" +"""Duplication 领域模型单元测试。""" + +from datetime import datetime, timezone import pytest @@ -8,255 +8,281 @@ from packages.domain.duplication import DuplicateSegment, DuplicationRecord class TestDuplicateSegmentCreate: - """DuplicateSegment.create 测试""" - def test_create_success(self): seg = DuplicateSegment.create( source_start=10.0, source_end=20.0, - matched_video_id="vid_123", - matched_video_name="测试视频", + matched_video_id="vid123", + matched_video_name="test.mp4", matched_start=5.0, matched_end=15.0, similarity=85.5, ) assert seg.id is not None - assert len(seg.id) == 32 + assert len(seg.id) == 32 # uuid4 hex assert seg.source_start == 10.0 assert seg.source_end == 20.0 - assert seg.matched_video_id == "vid_123" - assert seg.matched_video_name == "测试视频" + assert seg.matched_video_id == "vid123" + assert seg.matched_video_name == "test.mp4" assert seg.matched_start == 5.0 assert seg.matched_end == 15.0 assert seg.similarity == 85.5 - def test_invalid_source_negative_start(self): + def test_create_negative_source_start(self): with pytest.raises(ValueError, match="invalid source segment range"): DuplicateSegment.create( source_start=-1.0, source_end=10.0, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=50, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=50.0, ) - def test_invalid_source_end_before_start(self): - with pytest.raises(ValueError, match="invalid source segment range"): - DuplicateSegment.create( - source_start=20.0, - source_end=10.0, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=50, - ) - - def test_invalid_source_end_equals_start(self): + def test_create_source_end_equals_start(self): with pytest.raises(ValueError, match="invalid source segment range"): DuplicateSegment.create( source_start=10.0, source_end=10.0, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=50, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=50.0, ) - def test_invalid_matched_negative_start(self): + def test_create_source_end_less_than_start(self): + with pytest.raises(ValueError, match="invalid source segment range"): + DuplicateSegment.create( + source_start=20.0, + source_end=10.0, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=50.0, + ) + + def test_create_negative_matched_start(self): with pytest.raises(ValueError, match="invalid matched segment range"): DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=-5, - matched_end=10, - similarity=50, + source_start=0.0, + source_end=10.0, + matched_video_id="v", + matched_video_name="n", + matched_start=-1.0, + matched_end=5.0, + similarity=50.0, ) - def test_invalid_matched_end_before_start(self): + def test_create_matched_end_equals_start(self): with pytest.raises(ValueError, match="invalid matched segment range"): DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=15, - matched_end=10, - similarity=50, + source_start=0.0, + source_end=10.0, + matched_video_id="v", + matched_video_name="n", + matched_start=5.0, + matched_end=5.0, + similarity=50.0, ) - def test_invalid_similarity_negative(self): + def test_create_similarity_negative(self): with pytest.raises(ValueError, match="similarity must be between 0 and 100"): DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=-1, + source_start=0.0, + source_end=10.0, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=-1.0, ) - def test_invalid_similarity_over_100(self): + def test_create_similarity_over_100(self): with pytest.raises(ValueError, match="similarity must be between 0 and 100"): DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=101, + source_start=0.0, + source_end=10.0, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=101.0, ) - def test_similarity_boundary_zero(self): + def test_create_similarity_zero(self): seg = DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=0, + source_start=0.0, + source_end=5.0, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=0.0, ) - assert seg.similarity == 0 + assert seg.similarity == 0.0 - def test_similarity_boundary_100(self): + def test_create_similarity_100(self): seg = DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=100, + source_start=0.0, + source_end=5.0, + matched_video_id="v", + matched_video_name="n", + matched_start=0.0, + matched_end=5.0, + similarity=100.0, ) - assert seg.similarity == 100 + assert seg.similarity == 100.0 + + def test_create_unique_ids(self): + seg1 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0) + seg2 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0) + assert seg1.id != seg2.id class TestDuplicationRecordCreate: - """DuplicationRecord.create 测试""" - - def test_create_minimal(self): + def test_create_success_defaults(self): record = DuplicationRecord.create( user_id="user123", - filename="test.mp4", + filename="my_video.mp4", file_size=1024000, - storage_key="oss://bucket/test.mp4", + storage_key="videos/vid1.mp4", ) assert record.id is not None assert len(record.id) == 32 assert record.user_id == "user123" - assert record.filename == "test.mp4" + assert record.filename == "my_video.mp4" assert record.file_size == 1024000 - assert record.storage_key == "oss://bucket/test.mp4" + assert record.storage_key == "videos/vid1.mp4" + assert record.duration_seconds == 0.0 assert record.status == "pending" assert record.duplicate_rate is None assert record.duplicate_count == 0 assert record.segments == [] - assert record.duration_seconds == 0.0 - assert record.created_at is not None - assert record.updated_at is not None + assert record.error_message == "" + assert isinstance(record.created_at, datetime) + assert isinstance(record.updated_at, datetime) def test_create_with_duration(self): record = DuplicationRecord.create( user_id="u1", - filename="video.mp4", - file_size=5000, - storage_key="key", + filename="v.mp4", + file_size=100, + storage_key="k", duration_seconds=120.5, ) assert record.duration_seconds == 120.5 - def test_create_strips_whitespace(self): + def test_create_strips_user_id(self): record = DuplicationRecord.create( - user_id=" user456 ", - filename=" my video.mp4 ", + user_id=" user_trimmed ", + filename="v.mp4", file_size=100, - storage_key="key", + storage_key="k", ) - assert record.user_id == "user456" - assert record.filename == "my video.mp4" + assert record.user_id == "user_trimmed" - def test_empty_user_id_raises(self): + def test_create_strips_filename(self): + record = DuplicationRecord.create( + user_id="u1", + filename=" trimmed.mp4 ", + file_size=100, + storage_key="k", + ) + assert record.filename == "trimmed.mp4" + + def test_create_empty_user_id(self): + with pytest.raises(ValueError, match="user_id cannot be empty"): + DuplicationRecord.create( + user_id="", + filename="v.mp4", + file_size=100, + storage_key="k", + ) + + def test_create_whitespace_user_id(self): with pytest.raises(ValueError, match="user_id cannot be empty"): DuplicationRecord.create( user_id=" ", - filename="test.mp4", + filename="v.mp4", file_size=100, - storage_key="key", + storage_key="k", ) - def test_empty_filename_raises(self): + def test_create_empty_filename(self): with pytest.raises(ValueError, match="filename cannot be empty"): DuplicationRecord.create( user_id="u1", - filename=" ", + filename="", file_size=100, - storage_key="key", + storage_key="k", ) - def test_zero_file_size_raises(self): + def test_create_whitespace_filename(self): + with pytest.raises(ValueError, match="filename cannot be empty"): + DuplicationRecord.create( + user_id="u1", + filename=" \t ", + file_size=100, + storage_key="k", + ) + + def test_create_zero_file_size(self): with pytest.raises(ValueError, match="file_size must be positive"): DuplicationRecord.create( user_id="u1", - filename="test.mp4", + filename="v.mp4", file_size=0, - storage_key="key", + storage_key="k", ) - def test_negative_file_size_raises(self): + def test_create_negative_file_size(self): with pytest.raises(ValueError, match="file_size must be positive"): DuplicationRecord.create( user_id="u1", - filename="test.mp4", + filename="v.mp4", file_size=-100, - storage_key="key", + storage_key="k", ) + def test_create_unique_ids(self): + r1 = DuplicationRecord.create("u", "f", 100, "k") + r2 = DuplicationRecord.create("u", "f", 100, "k") + assert r1.id != r2.id -class TestDuplicationRecordLifecycle: - """生命周期状态转换测试""" - def test_mark_processing(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") - old_updated = record.updated_at +class TestMarkProcessing: + def test_mark_processing_from_pending(self): + record = DuplicationRecord.create("u", "f", 100, "k") + before = record.updated_at record.mark_processing() assert record.status == "processing" - assert record.updated_at >= old_updated + assert record.updated_at >= before - def test_mark_completed(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + def test_mark_processing_updates_timestamp(self): + record = DuplicationRecord.create("u", "f", 100, "k") + old_time = record.updated_at + # 确保时间戳会变（datetime.now 精度问题，直接赋值模拟） record.mark_processing() - segments = [ - DuplicateSegment.create( - source_start=0, - source_end=10, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=10, - similarity=90, - ) - ] - record.mark_completed( - duplicate_rate=25.5, - duplicate_count=1, - segments=segments, - ) + assert record.status == "processing" + assert record.updated_at.tzinfo == timezone.utc + + +class TestMarkCompleted: + def test_mark_completed_success(self): + record = DuplicationRecord.create("u", "f", 100, "k") + seg = DuplicateSegment.create(0, 5, "v", "n", 0, 5, 80.0) + record.mark_completed(duplicate_rate=75.5, duplicate_count=3, segments=[seg]) assert record.status == "completed" - assert record.duplicate_rate == 25.5 - assert record.duplicate_count == 1 + assert record.duplicate_rate == 75.5 + assert record.duplicate_count == 3 assert len(record.segments) == 1 - assert record.error_message == "" + assert record.segments[0].matched_video_id == "v" def test_mark_completed_zero_rate(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + record = DuplicationRecord.create("u", "f", 100, "k") record.mark_completed(duplicate_rate=0.0, duplicate_count=0, segments=[]) assert record.status == "completed" assert record.duplicate_rate == 0.0 @@ -264,88 +290,133 @@ class TestDuplicationRecordLifecycle: assert record.segments == [] def test_mark_completed_100_rate(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + record = DuplicationRecord.create("u", "f", 100, "k") record.mark_completed(duplicate_rate=100.0, duplicate_count=5, segments=[]) assert record.duplicate_rate == 100.0 - def test_mark_completed_invalid_rate_negative(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + def test_mark_completed_negative_rate(self): + record = DuplicationRecord.create("u", "f", 100, "k") with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"): - record.mark_completed(duplicate_rate=-1, duplicate_count=0, segments=[]) + record.mark_completed(duplicate_rate=-1.0, duplicate_count=0, segments=[]) - def test_mark_completed_invalid_rate_over_100(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + def test_mark_completed_over_100_rate(self): + record = DuplicationRecord.create("u", "f", 100, "k") with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"): - record.mark_completed(duplicate_rate=101, duplicate_count=0, segments=[]) + record.mark_completed(duplicate_rate=101.0, duplicate_count=0, segments=[]) + def test_mark_completed_updates_timestamp(self): + record = DuplicationRecord.create("u", "f", 100, "k") + record.mark_completed(50.0, 1, []) + assert record.updated_at.tzinfo == timezone.utc + + +class TestMarkFailed: def test_mark_failed(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + record = DuplicationRecord.create("u", "f", 100, "k") + record.mark_failed("network timeout") + assert record.status == "failed" + assert record.error_message == "network timeout" + + def test_mark_failed_empty_message(self): + record = DuplicationRecord.create("u", "f", 100, "k") + record.mark_failed("") + assert record.status == "failed" + assert record.error_message == "" + + def test_mark_failed_from_processing(self): + record = DuplicationRecord.create("u", "f", 100, "k") record.mark_processing() - record.mark_failed("网络超时") + record.mark_failed("something went wrong") assert record.status == "failed" - assert record.error_message == "网络超时" - assert record.duplicate_rate is None - - def test_mark_failed_from_pending(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") - record.mark_failed("文件损坏") - assert record.status == "failed" - assert record.error_message == "文件损坏" + assert record.error_message == "something went wrong" -class TestDuplicationRecordRetry: - """重试逻辑测试""" - +class TestCanRetry: def test_can_retry_failed(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + record = DuplicationRecord.create("u", "f", 100, "k") record.mark_failed("error") assert record.can_retry() is True def test_cannot_retry_pending(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + record = DuplicationRecord.create("u", "f", 100, "k") assert record.can_retry() is False def test_cannot_retry_processing(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + record = DuplicationRecord.create("u", "f", 100, "k") record.mark_processing() assert record.can_retry() is False def test_cannot_retry_completed(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") - record.mark_completed(duplicate_rate=10, duplicate_count=1, segments=[]) + record = DuplicationRecord.create("u", "f", 100, "k") + record.mark_completed(50.0, 1, []) assert record.can_retry() is False - def test_reset_for_retry(self): - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") - record.mark_processing() - segments = [ - DuplicateSegment.create( - source_start=0, - source_end=5, - matched_video_id="v1", - matched_video_name="n1", - matched_start=0, - matched_end=5, - similarity=80, - ) - ] - record.mark_completed(duplicate_rate=30, duplicate_count=1, segments=segments) + +class TestResetForRetry: + def test_reset_from_failed(self): + record = DuplicationRecord.create("u", "f", 100, "k") + seg = DuplicateSegment.create(0, 5, "v", "n", 0, 5, 80.0) + record.mark_completed(80.0, 2, [seg]) + record.mark_failed("error") # 模拟先完成再失败的场景不成立，直接从 failed 重置 + # 直接设置到 failed 状态 record.status = "failed" - record.error_message = "something wrong" + record.duplicate_rate = 50.0 + record.duplicate_count = 3 + record.error_message = "old error" record.video_fingerprint = {"hash": "abc"} record.reset_for_retry() + 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.updated_at is not None - def test_reset_for_retry_from_pending(self): - """即使从 pending 也能重置（调用方负责判断 can_retry）""" - record = DuplicationRecord.create(user_id="u1", filename="t.mp4", file_size=100, storage_key="k") + def test_reset_clears_segments(self): + record = DuplicationRecord.create("u", "f", 100, "k") + record.status = "failed" + record.segments = [ + DuplicateSegment.create(0, 1, "v1", "n1", 0, 1, 50.0), + DuplicateSegment.create(2, 3, "v2", "n2", 0, 1, 60.0), + ] record.reset_for_retry() - assert record.status == "pending" - assert record.duplicate_count == 0 + assert record.segments == [] + + def test_reset_preserves_identity(self): + record = DuplicationRecord.create("u", "f", 100, "k", duration_seconds=120.0) + record.status = "failed" + orig_id = record.id + orig_user = record.user_id + orig_filename = record.filename + orig_size = record.file_size + orig_storage = record.storage_key + orig_duration = record.duration_seconds + + record.reset_for_retry() + + assert record.id == orig_id + assert record.user_id == orig_user + assert record.filename == orig_filename + assert record.file_size == orig_size + assert record.storage_key == orig_storage + assert record.duration_seconds == orig_duration + + def test_reset_updates_timestamp(self): + record = DuplicationRecord.create("u", "f", 100, "k") + record.status = "failed" + old_time = record.updated_at + record.reset_for_retry() + assert record.updated_at >= old_time + + +class TestDataclassSlots: + def test_duplicate_segment_slots(self): + seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0) + # slots=True 时没有 __dict__ + assert not hasattr(seg, "__dict__") or hasattr(seg, "__slots__") + + def test_duplication_record_slots(self): + record = DuplicationRecord.create("u", "f", 100, "k") + assert hasattr(record, "__slots__") or hasattr(record, "__dict__")