test: P3-1 第49波单元测试(generated_video/duplication/tts_config/classification,+88) #837

Merged
xiaoxia merged 1 commits from test/wave49-domain-generatedvideo-duplication-ttsconfig-classification into develop 2026-07-24 18:58:18 +08:00
4 changed files with 703 additions and 0 deletions
+115
View File
@@ -0,0 +1,115 @@
"""classification 分类领域模型单测."""
import pytest
from packages.domain.classification import (
AssetClassification,
AssetLibraryKind,
ClassificationJob,
ClassificationJobStatus,
ClassificationStatus,
IngestJobStatus,
)
class TestAssetLibraryKind:
def test_values(self):
assert AssetLibraryKind.VIDEO.value == "video"
assert AssetLibraryKind.VOICE.value == "voice"
assert AssetLibraryKind.IMAGE.value == "image"
def test_is_str(self):
assert isinstance(AssetLibraryKind.VIDEO, str)
class TestIngestJobStatus:
def test_values(self):
assert IngestJobStatus.PENDING.value == "pending"
assert IngestJobStatus.PROCESSING.value == "processing"
assert IngestJobStatus.COMPLETED.value == "completed"
assert IngestJobStatus.FAILED.value == "failed"
class TestClassificationJobStatusMissing:
"""ClassificationJobStatus._missing_ 兼容性测试."""
def test_normal_values(self):
assert ClassificationJobStatus("pending") == ClassificationJobStatus.PENDING
assert ClassificationJobStatus("processing") == ClassificationJobStatus.PROCESSING
assert ClassificationJobStatus("completed") == ClassificationJobStatus.COMPLETED
assert ClassificationJobStatus("failed") == ClassificationJobStatus.FAILED
@pytest.mark.parametrize("value", ["done", "success", "finished", "complete"])
def test_completed_aliases(self, value):
assert ClassificationJobStatus(value) == ClassificationJobStatus.COMPLETED
@pytest.mark.parametrize("value", ["fail", "error", "err"])
def test_failed_aliases(self, value):
assert ClassificationJobStatus(value) == ClassificationJobStatus.FAILED
@pytest.mark.parametrize("value", ["process", "processing", "running", "run"])
def test_processing_aliases(self, value):
assert ClassificationJobStatus(value) == ClassificationJobStatus.PROCESSING
@pytest.mark.parametrize("value", ["unknown", "foobar", ""])
def test_unknown_fallback_to_pending(self, value):
assert ClassificationJobStatus(value) == ClassificationJobStatus.PENDING
def test_none_fallback_to_pending(self):
assert ClassificationJobStatus(None) == ClassificationJobStatus.PENDING # type: ignore[arg-type]
def test_case_insensitive_with_strip(self):
assert ClassificationJobStatus(" DONE ") == ClassificationJobStatus.COMPLETED
assert ClassificationJobStatus("ERROR") == ClassificationJobStatus.FAILED
def test_backward_compat_alias(self):
"""ClassificationStatus 是 ClassificationJobStatus 的别名."""
assert ClassificationStatus is ClassificationJobStatus
assert ClassificationStatus("done") == ClassificationJobStatus.COMPLETED
class TestAssetClassification:
def test_values(self):
assert AssetClassification.SCENIC.value == "scenic"
assert AssetClassification.PRODUCT.value == "product"
assert AssetClassification.PERSON.value == "person"
assert AssetClassification.ANIMAL.value == "animal"
assert AssetClassification.FOOD.value == "food"
assert AssetClassification.TECH.value == "tech"
assert AssetClassification.SPORT.value == "sport"
assert AssetClassification.MUSIC.value == "music"
assert AssetClassification.OTHER.value == "other"
class TestClassificationJobCreate:
def test_create_normal(self):
job = ClassificationJob.create(project_id="proj1", asset_id="asset1")
assert job.id
assert job.project_id == "proj1"
assert job.asset_id == "asset1"
assert job.status == ClassificationJobStatus.PENDING
assert job.classification == ""
assert job.confidence == 0.0
assert job.error_message == ""
def test_create_strips_whitespace(self):
job = ClassificationJob.create(project_id=" proj1 ", asset_id=" asset1 ")
assert job.project_id == "proj1"
assert job.asset_id == "asset1"
def test_create_empty_project_id_raises(self):
with pytest.raises(ValueError, match="project_id 不能为空"):
ClassificationJob.create(project_id="", asset_id="a1")
def test_create_empty_asset_id_raises(self):
with pytest.raises(ValueError, match="asset_id 不能为空"):
ClassificationJob.create(project_id="p1", asset_id="")
def test_create_whitespace_project_id_raises(self):
with pytest.raises(ValueError, match="project_id 不能为空"):
ClassificationJob.create(project_id=" ", asset_id="a1")
def test_create_unique_ids(self):
job1 = ClassificationJob.create(project_id="p1", asset_id="a1")
job2 = ClassificationJob.create(project_id="p1", asset_id="a2")
assert job1.id != job2.id
+298
View File
@@ -0,0 +1,298 @@
"""Duplication 查重记录领域实体单测."""
import pytest
from packages.domain.duplication import DuplicateSegment, DuplicationRecord
class TestDuplicateSegmentCreate:
def test_create_normal(self):
seg = DuplicateSegment.create(
source_start=10.0,
source_end=20.0,
matched_video_id="vid123",
matched_video_name="测试视频",
matched_start=5.0,
matched_end=15.0,
similarity=85.5,
)
assert seg.id
assert seg.source_start == 10.0
assert seg.source_end == 20.0
assert seg.matched_video_id == "vid123"
assert seg.matched_video_name == "测试视频"
assert seg.matched_start == 5.0
assert seg.matched_end == 15.0
assert seg.similarity == 85.5
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=10.0,
matched_video_id="v1",
matched_video_name="v",
matched_start=0.0,
matched_end=10.0,
similarity=50.0,
)
def test_create_zero_duration_source_raises(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="v",
matched_start=0.0,
matched_end=10.0,
similarity=50.0,
)
def test_create_reversed_source_range_raises(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="v",
matched_start=0.0,
matched_end=10.0,
similarity=50.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=10.0,
matched_video_id="v1",
matched_video_name="v",
matched_start=-1.0,
matched_end=10.0,
similarity=50.0,
)
def test_create_zero_duration_matched_raises(self):
with pytest.raises(ValueError, match="invalid matched segment range"):
DuplicateSegment.create(
source_start=0.0,
source_end=10.0,
matched_video_id="v1",
matched_video_name="v",
matched_start=5.0,
matched_end=5.0,
similarity=50.0,
)
def test_create_similarity_negative_raises(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="v1",
matched_video_name="v",
matched_start=0.0,
matched_end=10.0,
similarity=-1.0,
)
def test_create_similarity_over_100_raises(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="v1",
matched_video_name="v",
matched_start=0.0,
matched_end=10.0,
similarity=101.0,
)
def test_create_similarity_boundary_values(self):
# 0 和 100 都是合法的
seg0 = DuplicateSegment.create(
source_start=0.0, source_end=1.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=1.0,
similarity=0.0,
)
assert seg0.similarity == 0.0
seg100 = DuplicateSegment.create(
source_start=0.0, source_end=1.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=1.0,
similarity=100.0,
)
assert seg100.similarity == 100.0
def test_create_unique_ids(self):
seg1 = DuplicateSegment.create(
source_start=0.0, source_end=1.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=1.0,
similarity=50.0,
)
seg2 = DuplicateSegment.create(
source_start=0.0, source_end=1.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=1.0,
similarity=50.0,
)
assert seg1.id != seg2.id
class TestDuplicationRecordCreate:
def test_create_normal(self):
record = DuplicationRecord.create(
user_id="user1",
filename="test.mp4",
file_size=1024000,
storage_key="videos/test.mp4",
duration_seconds=30.5,
)
assert record.id
assert record.user_id == "user1"
assert record.filename == "test.mp4"
assert record.file_size == 1024000
assert record.storage_key == "videos/test.mp4"
assert record.duration_seconds == 30.5
assert record.status == "pending"
assert record.duplicate_rate is None
assert record.duplicate_count == 0
assert record.segments == []
assert record.error_message == ""
def test_create_strips_whitespace(self):
record = DuplicationRecord.create(
user_id=" user1 ",
filename=" test.mp4 ",
file_size=100,
storage_key="key1",
)
assert record.user_id == "user1"
assert record.filename == "test.mp4"
def test_create_empty_user_id_raises(self):
with pytest.raises(ValueError, match="user_id cannot be empty"):
DuplicationRecord.create(
user_id="",
filename="test.mp4",
file_size=100,
storage_key="key1",
)
def test_create_empty_filename_raises(self):
with pytest.raises(ValueError, match="filename cannot be empty"):
DuplicationRecord.create(
user_id="u1",
filename="",
file_size=100,
storage_key="key1",
)
def test_create_zero_file_size_raises(self):
with pytest.raises(ValueError, match="file_size must be positive"):
DuplicationRecord.create(
user_id="u1",
filename="test.mp4",
file_size=0,
storage_key="key1",
)
def test_create_negative_file_size_raises(self):
with pytest.raises(ValueError, match="file_size must be positive"):
DuplicationRecord.create(
user_id="u1",
filename="test.mp4",
file_size=-100,
storage_key="key1",
)
class TestDuplicationRecordStatus:
def test_mark_processing(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
record.mark_processing()
assert record.status == "processing"
def test_mark_completed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
seg = DuplicateSegment.create(
source_start=0.0, source_end=5.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=5.0,
similarity=90.0,
)
record.mark_completed(duplicate_rate=25.5, duplicate_count=3, segments=[seg])
assert record.status == "completed"
assert record.duplicate_rate == 25.5
assert record.duplicate_count == 3
assert len(record.segments) == 1
def test_mark_completed_invalid_rate_raises(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
record.mark_completed(duplicate_rate=-1, duplicate_count=0, segments=[])
with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"):
record.mark_completed(duplicate_rate=101, duplicate_count=0, segments=[])
def test_mark_failed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
record.mark_failed("网络超时")
assert record.status == "failed"
assert record.error_message == "网络超时"
def test_can_retry_only_failed(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
assert record.can_retry() is False # pending
record.mark_processing()
assert record.can_retry() is False # processing
record.mark_failed("error")
assert record.can_retry() is True # failed
seg = DuplicateSegment.create(
source_start=0.0, source_end=1.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=1.0,
similarity=50.0,
)
record2 = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
record2.mark_completed(10.0, 1, [seg])
assert record2.can_retry() is False # completed
def test_reset_for_retry(self):
record = DuplicationRecord.create(
user_id="u1", filename="t.mp4", file_size=100, storage_key="k1"
)
seg = DuplicateSegment.create(
source_start=0.0, source_end=1.0,
matched_video_id="v1", matched_video_name="v",
matched_start=0.0, matched_end=1.0,
similarity=50.0,
)
record.mark_completed(50.0, 2, [seg])
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
+139
View File
@@ -0,0 +1,139 @@
"""GeneratedVideo 生成视频领域实体单测."""
import pytest
from packages.domain.generated_video import GeneratedVideo
class TestGeneratedVideoCreate:
def test_create_normal(self):
video = GeneratedVideo.create(
project_id="proj1",
generation_task_id="task1",
name="我的视频",
file_url="https://example.com/video.mp4",
file_size=1024000,
duration=30.5,
width=1920,
height=1080,
fps=30.0,
)
assert video.id
assert video.project_id == "proj1"
assert video.generation_task_id == "task1"
assert video.name == "我的视频"
assert video.file_url == "https://example.com/video.mp4"
assert video.file_size == 1024000
assert video.duration == 30.5
assert video.width == 1920
assert video.height == 1080
assert video.fps == 30.0
assert video.status == "completed"
assert video.review_status == "pending_review"
assert video.is_duplicate is False
assert video.generation_params == {}
def test_create_strips_whitespace(self):
video = GeneratedVideo.create(
project_id=" proj1 ",
generation_task_id=" task1 ",
name=" 我的视频 ",
file_url=" https://example.com/video.mp4 ",
user_id=" user1 ",
)
assert video.project_id == "proj1"
assert video.generation_task_id == "task1"
assert video.name == "我的视频"
assert video.file_url == "https://example.com/video.mp4"
assert video.user_id == "user1"
def test_create_empty_project_id_raises(self):
with pytest.raises(ValueError, match="project_id cannot be empty"):
GeneratedVideo.create(
project_id="",
generation_task_id="task1",
name="视频",
file_url="https://example.com/v.mp4",
)
def test_create_whitespace_project_id_raises(self):
with pytest.raises(ValueError, match="project_id cannot be empty"):
GeneratedVideo.create(
project_id=" ",
generation_task_id="task1",
name="视频",
file_url="https://example.com/v.mp4",
)
def test_create_empty_generation_task_id_raises(self):
with pytest.raises(ValueError, match="generation_task_id cannot be empty"):
GeneratedVideo.create(
project_id="proj1",
generation_task_id="",
name="视频",
file_url="https://example.com/v.mp4",
)
def test_create_empty_name_raises(self):
with pytest.raises(ValueError, match="name cannot be empty"):
GeneratedVideo.create(
project_id="proj1",
generation_task_id="task1",
name="",
file_url="https://example.com/v.mp4",
)
def test_create_empty_file_url_raises(self):
with pytest.raises(ValueError, match="file_url cannot be empty"):
GeneratedVideo.create(
project_id="proj1",
generation_task_id="task1",
name="视频",
file_url="",
)
def test_create_default_values(self):
video = GeneratedVideo.create(
project_id="proj1",
generation_task_id="task1",
name="视频",
file_url="https://example.com/v.mp4",
)
assert video.file_size == 0
assert video.duration == 0.0
assert video.width == 0
assert video.height == 0
assert video.fps == 0.0
assert video.thumbnail_url is None
assert video.user_id == ""
assert video.generation_params == {}
def test_create_with_generation_params(self):
params = {"mode": "pip", "resolution": "1080p"}
video = GeneratedVideo.create(
project_id="proj1",
generation_task_id="task1",
name="视频",
file_url="https://example.com/v.mp4",
generation_params=params,
)
assert video.generation_params == params
def test_create_none_generation_params(self):
video = GeneratedVideo.create(
project_id="proj1",
generation_task_id="task1",
name="视频",
file_url="https://example.com/v.mp4",
generation_params=None,
)
assert video.generation_params == {}
def test_create_unique_ids(self):
v1 = GeneratedVideo.create(
project_id="proj1", generation_task_id="t1", name="v1", file_url="https://a.com/1.mp4"
)
v2 = GeneratedVideo.create(
project_id="proj1", generation_task_id="t2", name="v2", file_url="https://a.com/2.mp4"
)
assert v1.id != v2.id
+151
View File
@@ -0,0 +1,151 @@
"""TtsConfig 配音配置模型单测."""
import pytest
from packages.domain.tts_config import TtsConfig
class TestTtsConfigDefaults:
def test_default_values(self):
config = TtsConfig()
assert config.enabled is False
assert config.voice_id == ""
assert config.speed == 1.0
assert config.pitch == 0.0
assert config.volume == 0.8
assert config.text == ""
assert config.align_mode == "full"
assert config.overlap_mode == "replace"
class TestTtsConfigParse:
def test_parse_none(self):
config = TtsConfig.parse(None)
assert config.enabled is False
assert isinstance(config, TtsConfig)
def test_parse_empty_dict(self):
config = TtsConfig.parse({})
assert config.enabled is False
def test_parse_not_dict(self):
config = TtsConfig.parse("not a dict")
assert config.enabled is False
def test_parse_enabled_false_returns_disabled(self):
# 即使传了其他参数,enabled=False 就直接返回禁用
config = TtsConfig.parse({"enabled": False, "voice_id": "v1", "speed": 1.5})
assert config.enabled is False
assert config.voice_id == ""
assert config.speed == 1.0
def test_parse_enabled_true_with_all_fields(self):
config = TtsConfig.parse({
"enabled": True,
"voice_id": "female_warm",
"speed": 1.5,
"pitch": 2.0,
"volume": 0.9,
"text": "你好世界",
"align_mode": "subtitle",
"overlap_mode": "mix",
})
assert config.enabled is True
assert config.voice_id == "female_warm"
assert config.speed == 1.5
assert config.pitch == 2.0
assert config.volume == 0.9
assert config.text == "你好世界"
assert config.align_mode == "subtitle"
assert config.overlap_mode == "mix"
def test_parse_enabled_not_bool(self):
config = TtsConfig.parse({"enabled": "true", "voice_id": "v1"})
assert config.enabled is False # 非 bool 值视为 False
def test_parse_voice_id_not_string(self):
config = TtsConfig.parse({"enabled": True, "voice_id": 123})
assert config.voice_id == ""
def test_parse_speed_not_number(self):
config = TtsConfig.parse({"enabled": True, "speed": "fast"})
assert config.speed == 1.0
def test_parse_pitch_not_number(self):
config = TtsConfig.parse({"enabled": True, "pitch": "high"})
assert config.pitch == 0.0
def test_parse_volume_not_number(self):
config = TtsConfig.parse({"enabled": True, "volume": "loud"})
assert config.volume == 0.8
def test_parse_text_not_string(self):
config = TtsConfig.parse({"enabled": True, "text": 12345})
assert config.text == ""
def test_parse_invalid_align_mode(self):
config = TtsConfig.parse({"enabled": True, "align_mode": "invalid"})
assert config.align_mode == "full"
def test_parse_invalid_overlap_mode(self):
config = TtsConfig.parse({"enabled": True, "overlap_mode": "invalid"})
assert config.overlap_mode == "replace"
class TestTtsConfigClamp:
def test_speed_below_min(self):
config = TtsConfig.parse({"enabled": True, "speed": 0.1})
assert config.speed == 0.5
def test_speed_above_max(self):
config = TtsConfig.parse({"enabled": True, "speed": 3.0})
assert config.speed == 2.0
def test_speed_within_range(self):
config = TtsConfig.parse({"enabled": True, "speed": 1.2})
assert config.speed == 1.2
def test_speed_boundary_values(self):
config_low = TtsConfig.parse({"enabled": True, "speed": 0.5})
assert config_low.speed == 0.5
config_high = TtsConfig.parse({"enabled": True, "speed": 2.0})
assert config_high.speed == 2.0
def test_pitch_below_min(self):
config = TtsConfig.parse({"enabled": True, "pitch": -20})
assert config.pitch == -12
def test_pitch_above_max(self):
config = TtsConfig.parse({"enabled": True, "pitch": 20})
assert config.pitch == 12
def test_pitch_within_range(self):
config = TtsConfig.parse({"enabled": True, "pitch": -3.5})
assert config.pitch == -3.5
def test_volume_below_min(self):
config = TtsConfig.parse({"enabled": True, "volume": -0.5})
assert config.volume == 0.0
def test_volume_above_max(self):
config = TtsConfig.parse({"enabled": True, "volume": 2.0})
assert config.volume == 1.0
def test_volume_within_range(self):
config = TtsConfig.parse({"enabled": True, "volume": 0.5})
assert config.volume == 0.5
def test_int_speed_converted_to_float(self):
config = TtsConfig.parse({"enabled": True, "speed": 1})
assert isinstance(config.speed, float)
assert config.speed == 1.0
def test_int_pitch_converted_to_float(self):
config = TtsConfig.parse({"enabled": True, "pitch": 2})
assert isinstance(config.pitch, float)
assert config.pitch == 2.0
def test_int_volume_converted_to_float(self):
config = TtsConfig.parse({"enabled": True, "volume": 1})
assert isinstance(config.volume, float)
assert config.volume == 1.0