xiaoxia
3ff041440b
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Failing after 2m37s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m20s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m14s
CI/CD Pipeline / Frontend Lint (push) Failing after 47s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 19m58s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m19s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 8m46s
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m15s
CI/CD Pipeline / Integration Tests (push) Successful in 2m34s
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 2m4s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 31s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 2m15s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m55s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Unit Tests (push) Failing after 43m35s
423 lines
15 KiB
Python
Executable File
423 lines
15 KiB
Python
Executable File
"""Duplication 领域模型单元测试。"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from packages.domain.duplication import DuplicateSegment, DuplicationRecord
|
|
|
|
|
|
class TestDuplicateSegmentCreate:
|
|
def test_create_success(self):
|
|
seg = DuplicateSegment.create(
|
|
source_start=10.0,
|
|
source_end=20.0,
|
|
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 # uuid4 hex
|
|
assert seg.source_start == 10.0
|
|
assert seg.source_end == 20.0
|
|
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_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="v",
|
|
matched_video_name="n",
|
|
matched_start=0.0,
|
|
matched_end=5.0,
|
|
similarity=50.0,
|
|
)
|
|
|
|
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="v",
|
|
matched_video_name="n",
|
|
matched_start=0.0,
|
|
matched_end=5.0,
|
|
similarity=50.0,
|
|
)
|
|
|
|
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.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_create_matched_end_equals_start(self):
|
|
with pytest.raises(ValueError, match="invalid matched segment range"):
|
|
DuplicateSegment.create(
|
|
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_create_similarity_negative(self):
|
|
with pytest.raises(ValueError, match="similarity must be between 0 and 100"):
|
|
DuplicateSegment.create(
|
|
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_create_similarity_over_100(self):
|
|
with pytest.raises(ValueError, match="similarity must be between 0 and 100"):
|
|
DuplicateSegment.create(
|
|
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_create_similarity_zero(self):
|
|
seg = DuplicateSegment.create(
|
|
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.0
|
|
|
|
def test_create_similarity_100(self):
|
|
seg = DuplicateSegment.create(
|
|
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.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:
|
|
def test_create_success_defaults(self):
|
|
record = DuplicationRecord.create(
|
|
user_id="user123",
|
|
filename="my_video.mp4",
|
|
file_size=1024000,
|
|
storage_key="videos/vid1.mp4",
|
|
)
|
|
assert record.id is not None
|
|
assert len(record.id) == 32
|
|
assert record.user_id == "user123"
|
|
assert record.filename == "my_video.mp4"
|
|
assert record.file_size == 1024000
|
|
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.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="v.mp4",
|
|
file_size=100,
|
|
storage_key="k",
|
|
duration_seconds=120.5,
|
|
)
|
|
assert record.duration_seconds == 120.5
|
|
|
|
def test_create_strips_user_id(self):
|
|
record = DuplicationRecord.create(
|
|
user_id=" user_trimmed ",
|
|
filename="v.mp4",
|
|
file_size=100,
|
|
storage_key="k",
|
|
)
|
|
assert record.user_id == "user_trimmed"
|
|
|
|
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="v.mp4",
|
|
file_size=100,
|
|
storage_key="k",
|
|
)
|
|
|
|
def test_create_empty_filename(self):
|
|
with pytest.raises(ValueError, match="filename cannot be empty"):
|
|
DuplicationRecord.create(
|
|
user_id="u1",
|
|
filename="",
|
|
file_size=100,
|
|
storage_key="k",
|
|
)
|
|
|
|
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="v.mp4",
|
|
file_size=0,
|
|
storage_key="k",
|
|
)
|
|
|
|
def test_create_negative_file_size(self):
|
|
with pytest.raises(ValueError, match="file_size must be positive"):
|
|
DuplicationRecord.create(
|
|
user_id="u1",
|
|
filename="v.mp4",
|
|
file_size=-100,
|
|
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 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 >= before
|
|
|
|
def test_mark_processing_updates_timestamp(self):
|
|
record = DuplicationRecord.create("u", "f", 100, "k")
|
|
old_time = record.updated_at
|
|
# 确保时间戳会变（datetime.now 精度问题，直接赋值模拟）
|
|
record.mark_processing()
|
|
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 == 75.5
|
|
assert record.duplicate_count == 3
|
|
assert len(record.segments) == 1
|
|
assert record.segments[0].matched_video_id == "v"
|
|
|
|
def test_mark_completed_zero_rate(self):
|
|
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
|
|
assert record.duplicate_count == 0
|
|
assert record.segments == []
|
|
|
|
def test_mark_completed_100_rate(self):
|
|
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_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.0, duplicate_count=0, segments=[])
|
|
|
|
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.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("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("something went wrong")
|
|
assert record.status == "failed"
|
|
assert record.error_message == "something went wrong"
|
|
|
|
|
|
class TestCanRetry:
|
|
def test_can_retry_failed(self):
|
|
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("u", "f", 100, "k")
|
|
assert record.can_retry() is False
|
|
|
|
def test_cannot_retry_processing(self):
|
|
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("u", "f", 100, "k")
|
|
record.mark_completed(50.0, 1, [])
|
|
assert record.can_retry() is False
|
|
|
|
|
|
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.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
|
|
|
|
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.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__")
|