Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d8f0bda439 | |||
| 745b500810 |
@@ -1,181 +0,0 @@
|
||||
"""classification 单测.
|
||||
|
||||
domain 层素材分类模块纯逻辑,0 外部依赖。
|
||||
覆盖:4个枚举 + ClassificationJob 工厂/校验。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from packages.domain.classification import (
|
||||
AssetClassification,
|
||||
AssetLibraryKind,
|
||||
ClassificationJob,
|
||||
ClassificationJobStatus,
|
||||
IngestJobStatus,
|
||||
)
|
||||
|
||||
|
||||
class TestAssetLibraryKind:
|
||||
"""AssetLibraryKind 枚举测试."""
|
||||
|
||||
def test_two_values(self):
|
||||
"""视频和配音两类."""
|
||||
assert len(AssetLibraryKind) == 2
|
||||
|
||||
def test_video(self):
|
||||
assert AssetLibraryKind.VIDEO == "video"
|
||||
|
||||
def test_voice(self):
|
||||
assert AssetLibraryKind.VOICE == "voice"
|
||||
|
||||
def test_str_compatible(self):
|
||||
"""StrEnum 字符串兼容."""
|
||||
assert AssetLibraryKind.VIDEO == "video"
|
||||
|
||||
|
||||
class TestIngestJobStatus:
|
||||
"""IngestJobStatus 枚举测试."""
|
||||
|
||||
def test_four_statuses(self):
|
||||
assert len(IngestJobStatus) == 4
|
||||
|
||||
def test_pending(self):
|
||||
assert IngestJobStatus.PENDING == "pending"
|
||||
|
||||
def test_processing(self):
|
||||
assert IngestJobStatus.PROCESSING == "processing"
|
||||
|
||||
def test_completed(self):
|
||||
assert IngestJobStatus.COMPLETED == "completed"
|
||||
|
||||
def test_failed(self):
|
||||
assert IngestJobStatus.FAILED == "failed"
|
||||
|
||||
|
||||
class TestClassificationJobStatus:
|
||||
"""ClassificationJobStatus 枚举测试."""
|
||||
|
||||
def test_four_statuses(self):
|
||||
assert len(ClassificationJobStatus) == 4
|
||||
|
||||
def test_pending(self):
|
||||
assert ClassificationJobStatus.PENDING == "pending"
|
||||
|
||||
def test_processing(self):
|
||||
assert ClassificationJobStatus.PROCESSING == "processing"
|
||||
|
||||
def test_completed(self):
|
||||
assert ClassificationJobStatus.COMPLETED == "completed"
|
||||
|
||||
def test_failed(self):
|
||||
assert ClassificationJobStatus.FAILED == "failed"
|
||||
|
||||
def test_same_values_as_ingest(self):
|
||||
"""两种任务状态值相同."""
|
||||
assert set(ClassificationJobStatus) == set(IngestJobStatus)
|
||||
|
||||
|
||||
class TestAssetClassification:
|
||||
"""AssetClassification 枚举测试."""
|
||||
|
||||
def test_nine_categories(self):
|
||||
"""9个分类."""
|
||||
assert len(AssetClassification) == 9
|
||||
|
||||
def test_scenic(self):
|
||||
assert AssetClassification.SCENIC == "scenic"
|
||||
|
||||
def test_product(self):
|
||||
assert AssetClassification.PRODUCT == "product"
|
||||
|
||||
def test_person(self):
|
||||
assert AssetClassification.PERSON == "person"
|
||||
|
||||
def test_animal(self):
|
||||
assert AssetClassification.ANIMAL == "animal"
|
||||
|
||||
def test_food(self):
|
||||
assert AssetClassification.FOOD == "food"
|
||||
|
||||
def test_tech(self):
|
||||
assert AssetClassification.TECH == "tech"
|
||||
|
||||
def test_sport(self):
|
||||
assert AssetClassification.SPORT == "sport"
|
||||
|
||||
def test_music(self):
|
||||
assert AssetClassification.MUSIC == "music"
|
||||
|
||||
def test_other(self):
|
||||
assert AssetClassification.OTHER == "other"
|
||||
|
||||
def test_all_values_unique(self):
|
||||
"""所有分类值唯一."""
|
||||
values = [c.value for c in AssetClassification]
|
||||
assert len(values) == len(set(values))
|
||||
|
||||
|
||||
class TestClassificationJobCreate:
|
||||
"""ClassificationJob.create 测试."""
|
||||
|
||||
def test_create_valid(self):
|
||||
"""正常创建."""
|
||||
job = ClassificationJob.create(project_id="proj1", asset_id="asset1")
|
||||
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 == ""
|
||||
assert isinstance(job.id, str)
|
||||
assert len(job.id) > 0
|
||||
|
||||
def test_create_strips(self):
|
||||
"""project_id 和 asset_id 会 strip."""
|
||||
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(self):
|
||||
"""空 project_id 无效."""
|
||||
try:
|
||||
ClassificationJob.create(project_id="", asset_id="a1")
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "project_id" in str(e)
|
||||
|
||||
def test_create_whitespace_project_id(self):
|
||||
"""纯空白 project_id 无效."""
|
||||
try:
|
||||
ClassificationJob.create(project_id=" ", asset_id="a1")
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "project_id" in str(e)
|
||||
|
||||
def test_create_empty_asset_id(self):
|
||||
"""空 asset_id 无效."""
|
||||
try:
|
||||
ClassificationJob.create(project_id="p1", asset_id="")
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "asset_id" in str(e)
|
||||
|
||||
def test_create_whitespace_asset_id(self):
|
||||
"""纯空白 asset_id 无效."""
|
||||
try:
|
||||
ClassificationJob.create(project_id="p1", asset_id=" ")
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "asset_id" in str(e)
|
||||
|
||||
def test_create_unique_id(self):
|
||||
"""不同 job id 不同."""
|
||||
j1 = ClassificationJob.create("p", "a")
|
||||
j2 = ClassificationJob.create("p", "a")
|
||||
assert j1.id != j2.id
|
||||
|
||||
def test_create_has_timestamps(self):
|
||||
"""有创建和更新时间."""
|
||||
job = ClassificationJob.create("p", "a")
|
||||
assert job.created_at is not None
|
||||
assert job.updated_at is not None
|
||||
@@ -1,376 +0,0 @@
|
||||
"""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"
|
||||
@@ -1,362 +0,0 @@
|
||||
"""edit_plan_clip 单测.
|
||||
|
||||
domain 层剪辑计划片段纯逻辑模块,0 外部依赖。
|
||||
覆盖:枚举常量、create工厂/校验、素材分配、状态流转、属性计算。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||||
|
||||
|
||||
class TestEditPlanClipStatus:
|
||||
"""EditPlanClipStatus 枚举测试."""
|
||||
|
||||
def test_four_statuses(self):
|
||||
"""四种状态."""
|
||||
assert len(EditPlanClipStatus) == 4
|
||||
|
||||
def test_pending(self):
|
||||
"""pending 状态."""
|
||||
assert EditPlanClipStatus.PENDING == "pending"
|
||||
|
||||
def test_ready(self):
|
||||
"""ready 状态."""
|
||||
assert EditPlanClipStatus.READY == "ready"
|
||||
|
||||
def test_rendered(self):
|
||||
"""rendered 状态."""
|
||||
assert EditPlanClipStatus.RENDERED == "rendered"
|
||||
|
||||
def test_failed(self):
|
||||
"""failed 状态."""
|
||||
assert EditPlanClipStatus.FAILED == "failed"
|
||||
|
||||
def test_is_string(self):
|
||||
"""枚举值是字符串."""
|
||||
for status in EditPlanClipStatus:
|
||||
assert isinstance(status.value, str)
|
||||
assert len(status.value) > 0
|
||||
|
||||
def test_str_compatible(self):
|
||||
"""StrEnum 可与字符串比较."""
|
||||
assert EditPlanClipStatus.PENDING == "pending"
|
||||
assert EditPlanClipStatus.READY + "" == "ready"
|
||||
|
||||
|
||||
class TestEditPlanClipCreate:
|
||||
"""EditPlanClip.create 工厂方法测试."""
|
||||
|
||||
def test_create_minimal(self):
|
||||
"""最简创建."""
|
||||
clip = EditPlanClip.create(plan_id="plan1", clip_type="video", order=0)
|
||||
assert clip.plan_id == "plan1"
|
||||
assert clip.clip_type == "video"
|
||||
assert clip.order == 0
|
||||
assert clip.status == EditPlanClipStatus.PENDING
|
||||
assert clip.template_clip_config_id == ""
|
||||
assert clip.asset_id == ""
|
||||
assert clip.text_content == ""
|
||||
assert clip.start_time == 0.0
|
||||
assert clip.duration == 0.0
|
||||
assert clip.transition_effect == "cut"
|
||||
assert clip.config == {}
|
||||
assert isinstance(clip.id, str)
|
||||
assert len(clip.id) > 0
|
||||
|
||||
def test_create_with_all_fields(self):
|
||||
"""带全部字段创建."""
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="plan1",
|
||||
clip_type="video",
|
||||
order=2,
|
||||
template_clip_config_id="tpl1",
|
||||
asset_id="asset1",
|
||||
text_content=" 你好世界 ",
|
||||
start_time=10.5,
|
||||
duration=5.0,
|
||||
transition_effect="fade",
|
||||
config={"key": "value"},
|
||||
)
|
||||
assert clip.order == 2
|
||||
assert clip.template_clip_config_id == "tpl1"
|
||||
assert clip.asset_id == "asset1"
|
||||
assert clip.text_content == "你好世界" # strip了
|
||||
assert clip.start_time == 10.5
|
||||
assert clip.duration == 5.0
|
||||
assert clip.transition_effect == "fade"
|
||||
assert clip.config == {"key": "value"}
|
||||
|
||||
def test_create_strips_ids(self):
|
||||
"""plan_id 和 clip_type 会 strip."""
|
||||
clip = EditPlanClip.create(plan_id=" plan1 ", clip_type=" video ", order=0)
|
||||
assert clip.plan_id == "plan1"
|
||||
assert clip.clip_type == "video"
|
||||
|
||||
def test_create_empty_plan_id(self):
|
||||
"""空 plan_id 无效."""
|
||||
try:
|
||||
EditPlanClip.create(plan_id="", clip_type="video", order=0)
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "plan_id" in str(e)
|
||||
|
||||
def test_create_whitespace_plan_id(self):
|
||||
"""纯空白 plan_id 无效."""
|
||||
try:
|
||||
EditPlanClip.create(plan_id=" ", clip_type="video", order=0)
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "plan_id" in str(e)
|
||||
|
||||
def test_create_empty_clip_type(self):
|
||||
"""空 clip_type 无效."""
|
||||
try:
|
||||
EditPlanClip.create(plan_id="p1", clip_type="", order=0)
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "clip_type" in str(e)
|
||||
|
||||
def test_create_whitespace_clip_type(self):
|
||||
"""纯空白 clip_type 无效."""
|
||||
try:
|
||||
EditPlanClip.create(plan_id="p1", clip_type=" ", order=0)
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "clip_type" in str(e)
|
||||
|
||||
def test_create_negative_start_time(self):
|
||||
"""start_time 为负无效."""
|
||||
try:
|
||||
EditPlanClip.create(plan_id="p1", clip_type="v", order=0, start_time=-1.0)
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "start_time" in str(e)
|
||||
|
||||
def test_create_negative_duration(self):
|
||||
"""duration 为负无效."""
|
||||
try:
|
||||
EditPlanClip.create(plan_id="p1", clip_type="v", order=0, duration=-1.0)
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "duration" in str(e)
|
||||
|
||||
def test_create_zero_duration_valid(self):
|
||||
"""duration 为 0 合法."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, duration=0.0)
|
||||
assert clip.duration == 0.0
|
||||
|
||||
def test_create_empty_transition_defaults_to_cut(self):
|
||||
"""空 transition_effect 默认 cut."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, transition_effect="")
|
||||
assert clip.transition_effect == "cut"
|
||||
|
||||
def test_create_whitespace_transition_defaults_to_cut(self):
|
||||
"""空白 transition_effect 默认 cut."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, transition_effect=" ")
|
||||
assert clip.transition_effect == "cut"
|
||||
|
||||
def test_create_config_none_defaults_empty_dict(self):
|
||||
"""config=None 默认为空 dict."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, config=None)
|
||||
assert clip.config == {}
|
||||
|
||||
def test_create_unique_id(self):
|
||||
"""不同 clip id 不同."""
|
||||
c1 = EditPlanClip.create("p1", "v", 0)
|
||||
c2 = EditPlanClip.create("p1", "v", 0)
|
||||
assert c1.id != c2.id
|
||||
|
||||
def test_create_has_timestamps(self):
|
||||
"""有创建和更新时间."""
|
||||
clip = EditPlanClip.create("p1", "v", 0)
|
||||
assert clip.created_at is not None
|
||||
assert clip.updated_at is not None
|
||||
|
||||
def test_create_negative_order_valid(self):
|
||||
"""order 可以为负(表示排序位置)."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=-1)
|
||||
assert clip.order == -1
|
||||
|
||||
|
||||
class TestEditPlanClipAssignAsset:
|
||||
"""素材分配测试."""
|
||||
|
||||
def _make_clip(self):
|
||||
return EditPlanClip.create(plan_id="p1", clip_type="video", order=0)
|
||||
|
||||
def test_assign_asset(self):
|
||||
"""正常分配素材."""
|
||||
clip = self._make_clip()
|
||||
old_updated = clip.updated_at
|
||||
clip.assign_asset("asset123")
|
||||
assert clip.asset_id == "asset123"
|
||||
assert clip.updated_at >= old_updated
|
||||
|
||||
def test_assign_asset_strips(self):
|
||||
"""asset_id 会 strip."""
|
||||
clip = self._make_clip()
|
||||
clip.assign_asset(" asset123 ")
|
||||
assert clip.asset_id == "asset123"
|
||||
|
||||
def test_assign_asset_empty(self):
|
||||
"""空 asset_id 无效."""
|
||||
clip = self._make_clip()
|
||||
try:
|
||||
clip.assign_asset("")
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "asset_id" in str(e)
|
||||
|
||||
def test_assign_asset_whitespace(self):
|
||||
"""纯空白 asset_id 无效."""
|
||||
clip = self._make_clip()
|
||||
try:
|
||||
clip.assign_asset(" ")
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "asset_id" in str(e)
|
||||
|
||||
def test_has_asset_false_initially(self):
|
||||
"""初始无素材."""
|
||||
clip = self._make_clip()
|
||||
assert clip.has_asset is False
|
||||
|
||||
def test_has_asset_true_after_assign(self):
|
||||
"""分配后有素材."""
|
||||
clip = self._make_clip()
|
||||
clip.assign_asset("a1")
|
||||
assert clip.has_asset is True
|
||||
|
||||
|
||||
class TestEditPlanClipStatusFlow:
|
||||
"""状态流转测试."""
|
||||
|
||||
def _make_clip(self):
|
||||
return EditPlanClip.create(plan_id="p1", clip_type="video", order=0)
|
||||
|
||||
def test_initial_status_pending(self):
|
||||
"""初始状态 pending."""
|
||||
clip = self._make_clip()
|
||||
assert clip.status == EditPlanClipStatus.PENDING
|
||||
|
||||
def test_pending_to_ready(self):
|
||||
"""pending -> ready."""
|
||||
clip = self._make_clip()
|
||||
clip.mark_ready()
|
||||
assert clip.status == EditPlanClipStatus.READY
|
||||
|
||||
def test_ready_to_rendered(self):
|
||||
"""ready -> rendered."""
|
||||
clip = self._make_clip()
|
||||
clip.mark_ready()
|
||||
clip.mark_rendered()
|
||||
assert clip.status == EditPlanClipStatus.RENDERED
|
||||
|
||||
def test_ready_to_failed(self):
|
||||
"""ready -> failed."""
|
||||
clip = self._make_clip()
|
||||
clip.mark_ready()
|
||||
clip.mark_failed()
|
||||
assert clip.status == EditPlanClipStatus.FAILED
|
||||
|
||||
def test_cannot_ready_from_rendered(self):
|
||||
"""rendered 状态不能再 mark_ready."""
|
||||
clip = self._make_clip()
|
||||
clip.mark_ready()
|
||||
clip.mark_rendered()
|
||||
try:
|
||||
clip.mark_ready()
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "pending" in str(e).lower()
|
||||
|
||||
def test_cannot_ready_from_failed(self):
|
||||
"""failed 状态不能 mark_ready."""
|
||||
clip = self._make_clip()
|
||||
clip.mark_ready()
|
||||
clip.mark_failed()
|
||||
try:
|
||||
clip.mark_ready()
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "pending" in str(e).lower()
|
||||
|
||||
def test_cannot_render_from_pending(self):
|
||||
"""pending 不能直接 mark_rendered."""
|
||||
clip = self._make_clip()
|
||||
try:
|
||||
clip.mark_rendered()
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "ready" in str(e).lower()
|
||||
|
||||
def test_cannot_failed_from_pending(self):
|
||||
"""pending 不能直接 mark_failed."""
|
||||
clip = self._make_clip()
|
||||
try:
|
||||
clip.mark_failed()
|
||||
assert False
|
||||
except ValueError as e:
|
||||
assert "ready" in str(e).lower()
|
||||
|
||||
def test_status_change_updates_timestamp(self):
|
||||
"""状态变更更新 updated_at."""
|
||||
clip = self._make_clip()
|
||||
old_updated = clip.updated_at
|
||||
clip.mark_ready()
|
||||
assert clip.updated_at >= old_updated
|
||||
|
||||
|
||||
class TestEditPlanClipProperties:
|
||||
"""属性计算测试."""
|
||||
|
||||
def test_end_time(self):
|
||||
"""end_time = start_time + duration."""
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="p1",
|
||||
clip_type="v",
|
||||
order=0,
|
||||
start_time=10.0,
|
||||
duration=5.5,
|
||||
)
|
||||
assert clip.end_time == 15.5
|
||||
|
||||
def test_end_time_zero_duration(self):
|
||||
"""零时长 end_time = start_time."""
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="p1",
|
||||
clip_type="v",
|
||||
order=0,
|
||||
start_time=10.0,
|
||||
duration=0.0,
|
||||
)
|
||||
assert clip.end_time == 10.0
|
||||
|
||||
def test_end_time_zero_start(self):
|
||||
"""零起点 end_time = duration."""
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="p1",
|
||||
clip_type="v",
|
||||
order=0,
|
||||
start_time=0.0,
|
||||
duration=7.0,
|
||||
)
|
||||
assert clip.end_time == 7.0
|
||||
|
||||
def test_has_asset_empty_string(self):
|
||||
"""空字符串无素材."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, asset_id="")
|
||||
assert clip.has_asset is False
|
||||
|
||||
def test_has_asset_with_value(self):
|
||||
"""有值则有素材."""
|
||||
clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, asset_id="a1")
|
||||
assert clip.has_asset is True
|
||||
|
||||
def test_config_independent_between_clips(self):
|
||||
"""不同 clip 的 config 独立."""
|
||||
c1 = EditPlanClip.create(plan_id="p1", clip_type="v", order=0)
|
||||
c2 = EditPlanClip.create(plan_id="p1", clip_type="v", order=1)
|
||||
c1.config["key"] = "value"
|
||||
assert "key" not in c2.config
|
||||
@@ -11,8 +11,8 @@ import math
|
||||
|
||||
from packages.domain.quota import (
|
||||
QUOTA_TIERS,
|
||||
QuotaCheckResult,
|
||||
QuotaChecker,
|
||||
QuotaCheckResult,
|
||||
QuotaDimension,
|
||||
QuotaRegistry,
|
||||
QuotaTier,
|
||||
|
||||
Reference in New Issue
Block a user