377 lines
12 KiB
Python
377 lines
12 KiB
Python
"""duplication 单测.
|
||
|
||
domain 层查重记录纯逻辑模块，0 外部依赖。
|
||
覆盖：DuplicateSegment 工厂/校验、DuplicationRecord 创建/状态流转/重试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from packages.domain.duplication import DuplicateSegment, DuplicationRecord
|
||
|
||
|
||
class TestDuplicateSegmentCreate:
|
||
"""DuplicateSegment.create 工厂方法测试."""
|
||
|
||
def test_create_valid(self):
|
||
"""正常创建."""
|
||
seg = DuplicateSegment.create(
|
||
source_start=1.0,
|
||
source_end=5.0,
|
||
matched_video_id="vid123",
|
||
matched_video_name="测试视频",
|
||
matched_start=10.0,
|
||
matched_end=14.0,
|
||
similarity=85.5,
|
||
)
|
||
assert seg.source_start == 1.0
|
||
assert seg.source_end == 5.0
|
||
assert seg.matched_video_id == "vid123"
|
||
assert seg.matched_video_name == "测试视频"
|
||
assert seg.matched_start == 10.0
|
||
assert seg.matched_end == 14.0
|
||
assert seg.similarity == 85.5
|
||
assert isinstance(seg.id, str)
|
||
assert len(seg.id) > 0
|
||
|
||
def test_create_generates_unique_id(self):
|
||
"""每次创建生成不同的 id."""
|
||
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
|
||
|
||
def test_create_negative_source_start(self):
|
||
"""source_start 为负抛出 ValueError."""
|
||
try:
|
||
DuplicateSegment.create(-1, 5, "v", "n", 0, 1, 50.0)
|
||
assert False, "应该抛出 ValueError"
|
||
except ValueError as e:
|
||
assert "source" in str(e).lower()
|
||
|
||
def test_create_source_end_equals_start(self):
|
||
"""source_end 等于 source_start 无效."""
|
||
try:
|
||
DuplicateSegment.create(5, 5, "v", "n", 0, 1, 50.0)
|
||
assert False
|
||
except ValueError as e:
|
||
assert "source" in str(e).lower()
|
||
|
||
def test_create_source_end_less_than_start(self):
|
||
"""source_end 小于 source_start 无效."""
|
||
try:
|
||
DuplicateSegment.create(5, 3, "v", "n", 0, 1, 50.0)
|
||
assert False
|
||
except ValueError as e:
|
||
assert "source" in str(e).lower()
|
||
|
||
def test_create_negative_matched_start(self):
|
||
"""matched_start 为负无效."""
|
||
try:
|
||
DuplicateSegment.create(0, 5, "v", "n", -1, 1, 50.0)
|
||
assert False
|
||
except ValueError as e:
|
||
assert "matched" in str(e).lower()
|
||
|
||
def test_create_matched_end_invalid(self):
|
||
"""matched_end <= matched_start 无效."""
|
||
try:
|
||
DuplicateSegment.create(0, 5, "v", "n", 5, 5, 50.0)
|
||
assert False
|
||
except ValueError as e:
|
||
assert "matched" in str(e).lower()
|
||
|
||
def test_create_similarity_zero(self):
|
||
"""similarity = 0 是合法的."""
|
||
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 0.0)
|
||
assert seg.similarity == 0.0
|
||
|
||
def test_create_similarity_100(self):
|
||
"""similarity = 100 是合法的."""
|
||
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 100.0)
|
||
assert seg.similarity == 100.0
|
||
|
||
def test_create_similarity_negative(self):
|
||
"""similarity < 0 无效."""
|
||
try:
|
||
DuplicateSegment.create(0, 1, "v", "n", 0, 1, -1.0)
|
||
assert False
|
||
except ValueError as e:
|
||
assert "similarity" in str(e).lower()
|
||
|
||
def test_create_similarity_over_100(self):
|
||
"""similarity > 100 无效."""
|
||
try:
|
||
DuplicateSegment.create(0, 1, "v", "n", 0, 1, 101.0)
|
||
assert False
|
||
except ValueError as e:
|
||
assert "similarity" in str(e).lower()
|
||
|
||
|
||
class TestDuplicationRecordCreate:
|
||
"""DuplicationRecord.create 工厂方法测试."""
|
||
|
||
def test_create_minimal(self):
|
||
"""最简创建."""
|
||
rec = DuplicationRecord.create(
|
||
user_id="user1",
|
||
filename="test.mp4",
|
||
file_size=1024,
|
||
storage_key="oss://bucket/test.mp4",
|
||
)
|
||
assert rec.user_id == "user1"
|
||
assert rec.filename == "test.mp4"
|
||
assert rec.file_size == 1024
|
||
assert rec.storage_key == "oss://bucket/test.mp4"
|
||
assert rec.duration_seconds == 0.0
|
||
assert rec.status == "pending"
|
||
assert rec.duplicate_rate is None
|
||
assert rec.duplicate_count == 0
|
||
assert rec.segments == []
|
||
assert rec.error_message == ""
|
||
assert isinstance(rec.id, str)
|
||
assert len(rec.id) > 0
|
||
|
||
def test_create_with_duration(self):
|
||
"""带时长创建."""
|
||
rec = DuplicationRecord.create(
|
||
user_id="user1",
|
||
filename="test.mp4",
|
||
file_size=1024,
|
||
storage_key="oss://key",
|
||
duration_seconds=120.5,
|
||
)
|
||
assert rec.duration_seconds == 120.5
|
||
|
||
def test_create_strips_whitespace(self):
|
||
"""user_id 和 filename 会 strip."""
|
||
rec = DuplicationRecord.create(
|
||
user_id=" user1 ",
|
||
filename=" test.mp4 ",
|
||
file_size=1024,
|
||
storage_key="oss://key",
|
||
)
|
||
assert rec.user_id == "user1"
|
||
assert rec.filename == "test.mp4"
|
||
|
||
def test_create_empty_user_id(self):
|
||
"""空 user_id 无效."""
|
||
try:
|
||
DuplicationRecord.create("", "test.mp4", 1024, "oss://key")
|
||
assert False
|
||
except ValueError as e:
|
||
assert "user_id" in str(e)
|
||
|
||
def test_create_whitespace_user_id(self):
|
||
"""纯空白 user_id 无效."""
|
||
try:
|
||
DuplicationRecord.create(" ", "test.mp4", 1024, "oss://key")
|
||
assert False
|
||
except ValueError as e:
|
||
assert "user_id" in str(e)
|
||
|
||
def test_create_empty_filename(self):
|
||
"""空 filename 无效."""
|
||
try:
|
||
DuplicationRecord.create("user1", "", 1024, "oss://key")
|
||
assert False
|
||
except ValueError as e:
|
||
assert "filename" in str(e)
|
||
|
||
def test_create_whitespace_filename(self):
|
||
"""纯空白 filename 无效."""
|
||
try:
|
||
DuplicationRecord.create("user1", " ", 1024, "oss://key")
|
||
assert False
|
||
except ValueError as e:
|
||
assert "filename" in str(e)
|
||
|
||
def test_create_zero_file_size(self):
|
||
"""file_size = 0 无效."""
|
||
try:
|
||
DuplicationRecord.create("user1", "test.mp4", 0, "oss://key")
|
||
assert False
|
||
except ValueError as e:
|
||
assert "file_size" in str(e)
|
||
|
||
def test_create_negative_file_size(self):
|
||
"""file_size 为负无效."""
|
||
try:
|
||
DuplicationRecord.create("user1", "test.mp4", -1, "oss://key")
|
||
assert False
|
||
except ValueError as e:
|
||
assert "file_size" in str(e)
|
||
|
||
def test_create_unique_id(self):
|
||
"""不同记录 id 不同."""
|
||
r1 = DuplicationRecord.create("u", "f", 1, "k")
|
||
r2 = DuplicationRecord.create("u", "f", 1, "k")
|
||
assert r1.id != r2.id
|
||
|
||
def test_create_has_timestamps(self):
|
||
"""有创建和更新时间."""
|
||
rec = DuplicationRecord.create("u", "f", 1, "k")
|
||
assert rec.created_at is not None
|
||
assert rec.updated_at is not None
|
||
# 两者应该很接近（都是 now）
|
||
delta = (rec.updated_at - rec.created_at).total_seconds()
|
||
assert abs(delta) < 1.0
|
||
|
||
|
||
class TestDuplicationRecordStatusFlow:
|
||
"""状态流转测试."""
|
||
|
||
def _make_record(self):
|
||
return DuplicationRecord.create("user1", "test.mp4", 1024, "oss://key")
|
||
|
||
def test_initial_status_pending(self):
|
||
"""初始状态 pending."""
|
||
rec = self._make_record()
|
||
assert rec.status == "pending"
|
||
|
||
def test_mark_processing(self):
|
||
"""标记为处理中."""
|
||
rec = self._make_record()
|
||
old_updated = rec.updated_at
|
||
rec.mark_processing()
|
||
assert rec.status == "processing"
|
||
assert rec.updated_at >= old_updated
|
||
|
||
def test_mark_completed(self):
|
||
"""标记为完成."""
|
||
rec = self._make_record()
|
||
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 80.0)
|
||
rec.mark_completed(duplicate_rate=45.5, duplicate_count=3, segments=[seg])
|
||
assert rec.status == "completed"
|
||
assert rec.duplicate_rate == 45.5
|
||
assert rec.duplicate_count == 3
|
||
assert len(rec.segments) == 1
|
||
assert rec.segments[0].similarity == 80.0
|
||
|
||
def test_mark_completed_zero_rate(self):
|
||
"""重复率为 0 合法."""
|
||
rec = self._make_record()
|
||
rec.mark_completed(0.0, 0, [])
|
||
assert rec.status == "completed"
|
||
assert rec.duplicate_rate == 0.0
|
||
assert rec.duplicate_count == 0
|
||
assert rec.segments == []
|
||
|
||
def test_mark_completed_full_rate(self):
|
||
"""重复率 100 合法."""
|
||
rec = self._make_record()
|
||
rec.mark_completed(100.0, 1, [])
|
||
assert rec.duplicate_rate == 100.0
|
||
|
||
def test_mark_completed_negative_rate(self):
|
||
"""重复率为负无效."""
|
||
rec = self._make_record()
|
||
try:
|
||
rec.mark_completed(-1, 0, [])
|
||
assert False
|
||
except ValueError as e:
|
||
assert "duplicate_rate" in str(e)
|
||
|
||
def test_mark_completed_over_100(self):
|
||
"""重复率超过 100 无效."""
|
||
rec = self._make_record()
|
||
try:
|
||
rec.mark_completed(101, 0, [])
|
||
assert False
|
||
except ValueError as e:
|
||
assert "duplicate_rate" in str(e)
|
||
|
||
def test_mark_failed(self):
|
||
"""标记为失败."""
|
||
rec = self._make_record()
|
||
rec.mark_failed("网络超时")
|
||
assert rec.status == "failed"
|
||
assert rec.error_message == "网络超时"
|
||
|
||
def test_mark_failed_empty_message(self):
|
||
"""失败信息可以为空字符串."""
|
||
rec = self._make_record()
|
||
rec.mark_failed("")
|
||
assert rec.status == "failed"
|
||
assert rec.error_message == ""
|
||
|
||
def test_can_retry_failed(self):
|
||
"""failed 状态可以重试."""
|
||
rec = self._make_record()
|
||
rec.mark_failed("error")
|
||
assert rec.can_retry() is True
|
||
|
||
def test_cannot_retry_pending(self):
|
||
"""pending 状态不可重试."""
|
||
rec = self._make_record()
|
||
assert rec.can_retry() is False
|
||
|
||
def test_cannot_retry_processing(self):
|
||
"""processing 状态不可重试."""
|
||
rec = self._make_record()
|
||
rec.mark_processing()
|
||
assert rec.can_retry() is False
|
||
|
||
def test_cannot_retry_completed(self):
|
||
"""completed 状态不可重试."""
|
||
rec = self._make_record()
|
||
rec.mark_completed(50, 1, [])
|
||
assert rec.can_retry() is False
|
||
|
||
def test_reset_for_retry(self):
|
||
"""重置回 pending."""
|
||
rec = self._make_record()
|
||
rec.mark_failed("error")
|
||
seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
|
||
rec.segments = [seg]
|
||
rec.video_fingerprint = {"hash": "abc"}
|
||
rec.duplicate_rate = 50.0
|
||
rec.duplicate_count = 5
|
||
|
||
rec.reset_for_retry()
|
||
assert rec.status == "pending"
|
||
assert rec.duplicate_rate is None
|
||
assert rec.duplicate_count == 0
|
||
assert rec.error_message == ""
|
||
assert rec.segments == []
|
||
assert rec.video_fingerprint is None
|
||
|
||
def test_reset_updates_timestamp(self):
|
||
"""重置更新 updated_at."""
|
||
rec = self._make_record()
|
||
rec.mark_failed("error")
|
||
old_updated = rec.updated_at
|
||
rec.reset_for_retry()
|
||
assert rec.updated_at >= old_updated
|
||
|
||
|
||
class TestDuplicationRecordSegments:
|
||
"""segments 列表相关测试."""
|
||
|
||
def _make_record(self):
|
||
return DuplicationRecord.create("user1", "test.mp4", 1024, "oss://key")
|
||
|
||
def test_segments_default_empty(self):
|
||
"""初始 segments 为空列表."""
|
||
rec = self._make_record()
|
||
assert rec.segments == []
|
||
|
||
def test_segments_independent_list(self):
|
||
"""不同记录的 segments 是独立列表."""
|
||
r1 = self._make_record()
|
||
r2 = self._make_record()
|
||
r1.segments.append("fake")
|
||
assert len(r2.segments) == 0
|
||
|
||
def test_completed_with_multiple_segments(self):
|
||
"""完成时带多个片段."""
|
||
rec = self._make_record()
|
||
segs = [
|
||
DuplicateSegment.create(0, 1, "v1", "n1", 0, 1, 90.0),
|
||
DuplicateSegment.create(2, 3, "v2", "n2", 5, 6, 70.0),
|
||
DuplicateSegment.create(4, 5, "v3", "n3", 10, 11, 85.0),
|
||
]
|
||
rec.mark_completed(60.0, 3, segs)
|
||
assert len(rec.segments) == 3
|
||
assert rec.segments[0].similarity == 90.0
|
||
assert rec.segments[1].matched_video_id == "v2"
|
||
assert rec.segments[2].matched_video_name == "n3"
|