373 lines
13 KiB
Python
Executable File
373 lines
13 KiB
Python
Executable File
"""分类领域模型单元测试 - 纯逻辑部分。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from domain.classification import (
|
|
AssetClassification,
|
|
AssetLibraryKind,
|
|
ClassificationJob,
|
|
ClassificationJobStatus,
|
|
ClassificationStatus,
|
|
IngestJobStatus,
|
|
)
|
|
|
|
|
|
class TestAssetLibraryKind:
|
|
"""素材库类型枚举。"""
|
|
|
|
def test_video_value(self):
|
|
assert AssetLibraryKind.VIDEO == "video"
|
|
|
|
def test_voice_value(self):
|
|
assert AssetLibraryKind.VOICE == "voice"
|
|
|
|
def test_image_value(self):
|
|
assert AssetLibraryKind.IMAGE == "image"
|
|
|
|
def test_is_str_enum(self):
|
|
assert isinstance(AssetLibraryKind.VIDEO, str)
|
|
assert AssetLibraryKind.VIDEO + "_test" == "video_test"
|
|
|
|
def test_members_count(self):
|
|
assert len(AssetLibraryKind) == 3
|
|
|
|
|
|
class TestIngestJobStatus:
|
|
"""导入任务状态枚举。"""
|
|
|
|
def test_pending_value(self):
|
|
assert IngestJobStatus.PENDING == "pending"
|
|
|
|
def test_processing_value(self):
|
|
assert IngestJobStatus.PROCESSING == "processing"
|
|
|
|
def test_completed_value(self):
|
|
assert IngestJobStatus.COMPLETED == "completed"
|
|
|
|
def test_failed_value(self):
|
|
assert IngestJobStatus.FAILED == "failed"
|
|
|
|
def test_members_count(self):
|
|
assert len(IngestJobStatus) == 4
|
|
|
|
|
|
class TestClassificationJobStatusMissing:
|
|
"""ClassificationJobStatus._missing_ 兼容性测试。"""
|
|
|
|
def test_standard_values(self):
|
|
"""标准值正常解析。"""
|
|
assert ClassificationJobStatus("pending") == ClassificationJobStatus.PENDING
|
|
assert ClassificationJobStatus("processing") == ClassificationJobStatus.PROCESSING
|
|
assert ClassificationJobStatus("completed") == ClassificationJobStatus.COMPLETED
|
|
assert ClassificationJobStatus("failed") == ClassificationJobStatus.FAILED
|
|
|
|
def test_done_maps_to_completed(self):
|
|
"""历史值 done 映射到 COMPLETED。"""
|
|
assert ClassificationJobStatus("done") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_success_maps_to_completed(self):
|
|
assert ClassificationJobStatus("success") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_finished_maps_to_completed(self):
|
|
assert ClassificationJobStatus("finished") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_complete_maps_to_completed(self):
|
|
assert ClassificationJobStatus("complete") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_fail_maps_to_failed(self):
|
|
assert ClassificationJobStatus("fail") == ClassificationJobStatus.FAILED
|
|
|
|
def test_error_maps_to_failed(self):
|
|
assert ClassificationJobStatus("error") == ClassificationJobStatus.FAILED
|
|
|
|
def test_err_maps_to_failed(self):
|
|
assert ClassificationJobStatus("err") == ClassificationJobStatus.FAILED
|
|
|
|
def test_process_maps_to_processing(self):
|
|
assert ClassificationJobStatus("process") == ClassificationJobStatus.PROCESSING
|
|
|
|
def test_running_maps_to_processing(self):
|
|
assert ClassificationJobStatus("running") == ClassificationJobStatus.PROCESSING
|
|
|
|
def test_run_maps_to_processing(self):
|
|
assert ClassificationJobStatus("run") == ClassificationJobStatus.PROCESSING
|
|
|
|
def test_unknown_value_defaults_to_pending(self):
|
|
"""未知值兜底为 PENDING。"""
|
|
assert ClassificationJobStatus("unknown") == ClassificationJobStatus.PENDING
|
|
assert ClassificationJobStatus("whatever") == ClassificationJobStatus.PENDING
|
|
assert ClassificationJobStatus("") == ClassificationJobStatus.PENDING
|
|
|
|
def test_case_insensitive(self):
|
|
"""大小写不敏感。"""
|
|
assert ClassificationJobStatus("DONE") == ClassificationJobStatus.COMPLETED
|
|
assert ClassificationJobStatus("Done") == ClassificationJobStatus.COMPLETED
|
|
assert ClassificationJobStatus("FAIL") == ClassificationJobStatus.FAILED
|
|
assert ClassificationJobStatus("Error") == ClassificationJobStatus.FAILED
|
|
|
|
def test_stripped(self):
|
|
"""前后空白字符被忽略。"""
|
|
assert ClassificationJobStatus(" done ") == ClassificationJobStatus.COMPLETED
|
|
assert ClassificationJobStatus("\tfail\n") == ClassificationJobStatus.FAILED
|
|
|
|
def test_none_returns_pending(self):
|
|
"""None 值也返回 PENDING（不报错）。"""
|
|
assert ClassificationJobStatus(None) == ClassificationJobStatus.PENDING
|
|
|
|
def test_integer_returns_pending(self):
|
|
"""非字符串值返回 PENDING。"""
|
|
assert ClassificationJobStatus(123) == ClassificationJobStatus.PENDING
|
|
|
|
|
|
class TestClassificationStatusAlias:
|
|
"""向后兼容别名。"""
|
|
|
|
def test_alias_same_class(self):
|
|
assert ClassificationStatus is ClassificationJobStatus
|
|
|
|
def test_alias_values_same(self):
|
|
assert ClassificationStatus.PENDING == ClassificationJobStatus.PENDING
|
|
assert ClassificationStatus.COMPLETED == ClassificationJobStatus.COMPLETED
|
|
|
|
|
|
class TestAssetClassification:
|
|
"""素材分类枚举。"""
|
|
|
|
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_members_count(self):
|
|
assert len(AssetClassification) == 9
|
|
|
|
|
|
class TestClassificationJobCreate:
|
|
"""ClassificationJob.create 工厂方法。"""
|
|
|
|
def test_create_basic(self):
|
|
job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
|
|
assert job.project_id == "proj-1"
|
|
assert job.asset_id == "asset-1"
|
|
assert job.status == ClassificationJobStatus.PENDING
|
|
assert job.classification == ""
|
|
assert job.confidence == 0.0
|
|
assert job.error_message == ""
|
|
assert job.id # 自动生成的 ID 非空
|
|
|
|
def test_create_strips_whitespace(self):
|
|
job = ClassificationJob.create(project_id=" proj-1 ", asset_id="\tasset-1\n")
|
|
assert job.project_id == "proj-1"
|
|
assert job.asset_id == "asset-1"
|
|
|
|
def test_create_empty_project_id_raises(self):
|
|
with pytest.raises(ValueError, match="project_id 不能为空"):
|
|
ClassificationJob.create(project_id="", asset_id="asset-1")
|
|
|
|
def test_create_whitespace_project_id_raises(self):
|
|
with pytest.raises(ValueError, match="project_id 不能为空"):
|
|
ClassificationJob.create(project_id=" ", asset_id="asset-1")
|
|
|
|
def test_create_empty_asset_id_raises(self):
|
|
with pytest.raises(ValueError, match="asset_id 不能为空"):
|
|
ClassificationJob.create(project_id="proj-1", asset_id="")
|
|
|
|
def test_create_whitespace_asset_id_raises(self):
|
|
with pytest.raises(ValueError, match="asset_id 不能为空"):
|
|
ClassificationJob.create(project_id="proj-1", asset_id=" \t ")
|
|
|
|
def test_create_generates_unique_ids(self):
|
|
job1 = ClassificationJob.create(project_id="p", asset_id="a")
|
|
job2 = ClassificationJob.create(project_id="p", asset_id="a")
|
|
assert job1.id != job2.id
|
|
|
|
def test_create_id_is_hex(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
assert job.created_at.tzinfo is not None
|
|
assert job.updated_at.tzinfo is not None
|
|
|
|
|
|
class TestClassificationJobState:
|
|
"""ClassificationJob 状态操作测试"""
|
|
|
|
def test_set_processing(self):
|
|
job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
|
|
job.status = ClassificationJobStatus.PROCESSING
|
|
assert job.status == ClassificationJobStatus.PROCESSING
|
|
|
|
def test_set_completed_with_result(self):
|
|
job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
|
|
job.status = ClassificationJobStatus.COMPLETED
|
|
job.classification = AssetClassification.SCENIC
|
|
job.confidence = 0.95
|
|
assert job.status == ClassificationJobStatus.COMPLETED
|
|
assert job.classification == "scenic"
|
|
assert job.confidence == pytest.approx(0.95)
|
|
|
|
def test_set_failed_with_error(self):
|
|
job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
|
|
job.status = ClassificationJobStatus.FAILED
|
|
job.error_message = "model timeout"
|
|
assert job.status == ClassificationJobStatus.FAILED
|
|
assert job.error_message == "model timeout"
|
|
|
|
def test_confidence_range_zero(self):
|
|
job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
|
|
job.confidence = 0.0
|
|
assert job.confidence == 0.0
|
|
|
|
def test_confidence_range_one(self):
|
|
job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
|
|
job.confidence = 1.0
|
|
assert job.confidence == 1.0
|
|
|
|
|
|
class TestClassificationJobStatusMissing:
|
|
"""ClassificationJobStatus._missing_ 兼容行为测试"""
|
|
|
|
def test_done_maps_to_completed(self):
|
|
assert ClassificationJobStatus("done") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_success_maps_to_completed(self):
|
|
assert ClassificationJobStatus("success") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_finished_maps_to_completed(self):
|
|
assert ClassificationJobStatus("finished") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_complete_maps_to_completed(self):
|
|
assert ClassificationJobStatus("complete") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_fail_maps_to_failed(self):
|
|
assert ClassificationJobStatus("fail") == ClassificationJobStatus.FAILED
|
|
|
|
def test_error_maps_to_failed(self):
|
|
assert ClassificationJobStatus("error") == ClassificationJobStatus.FAILED
|
|
|
|
def test_err_maps_to_failed(self):
|
|
assert ClassificationJobStatus("err") == ClassificationJobStatus.FAILED
|
|
|
|
def test_unknown_maps_to_pending(self):
|
|
assert ClassificationJobStatus("unknown_status") == ClassificationJobStatus.PENDING
|
|
|
|
def test_case_insensitive_mapping(self):
|
|
assert ClassificationJobStatus("DONE") == ClassificationJobStatus.COMPLETED
|
|
assert ClassificationJobStatus("Done") == ClassificationJobStatus.COMPLETED
|
|
|
|
def test_whitespace_stripped(self):
|
|
assert ClassificationJobStatus(" done ") == ClassificationJobStatus.COMPLETED
|
|
|
|
|
|
class TestClassificationJobExtended:
|
|
"""ClassificationJob 深度补充测试"""
|
|
|
|
def test_id_is_hex(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
int(job.id, 16)
|
|
|
|
def test_empty_classification(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
assert job.classification == ""
|
|
|
|
def test_zero_confidence(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
assert job.confidence == 0.0
|
|
|
|
def test_high_confidence(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
job.confidence = 0.99
|
|
assert job.confidence == pytest.approx(0.99)
|
|
|
|
def test_negative_confidence(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
job.confidence = -0.1
|
|
assert job.confidence == pytest.approx(-0.1)
|
|
|
|
def test_confidence_over_one(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
job.confidence = 1.5
|
|
assert job.confidence == pytest.approx(1.5)
|
|
|
|
def test_empty_error_message(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
assert job.error_message == ""
|
|
|
|
def test_long_error_message(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
long_msg = "error" * 100
|
|
job.error_message = long_msg
|
|
assert job.error_message == long_msg
|
|
assert len(job.error_message) == 500
|
|
|
|
def test_status_with_string_assignment(self):
|
|
job = ClassificationJob.create(project_id="p", asset_id="a")
|
|
job.status = "processing"
|
|
assert job.status == ClassificationJobStatus.PROCESSING
|
|
|
|
|
|
class TestAssetLibraryKindExtended:
|
|
"""AssetLibraryKind 深度补充测试"""
|
|
|
|
def test_image_value(self):
|
|
assert AssetLibraryKind.IMAGE == "image"
|
|
|
|
def test_all_three_kinds(self):
|
|
assert len(AssetLibraryKind) == 3
|
|
|
|
def test_is_string_enum(self):
|
|
assert isinstance(AssetLibraryKind.VIDEO, str)
|
|
|
|
def test_from_string(self):
|
|
assert AssetLibraryKind("video") == AssetLibraryKind.VIDEO
|
|
|
|
|
|
class TestIngestJobStatusExtended:
|
|
"""IngestJobStatus 深度补充测试"""
|
|
|
|
def test_is_string_enum(self):
|
|
assert isinstance(IngestJobStatus.PENDING, str)
|
|
|
|
def test_total_count(self):
|
|
assert len(IngestJobStatus) == 4
|
|
|
|
def test_from_string(self):
|
|
assert IngestJobStatus("pending") == IngestJobStatus.PENDING
|
|
|
|
|
|
class TestAssetClassificationExtended:
|
|
"""AssetClassification 深度补充测试"""
|
|
|
|
def test_total_count(self):
|
|
assert len(AssetClassification) == 9
|
|
|
|
def test_is_string_enum(self):
|
|
assert isinstance(AssetClassification.SCENIC, str)
|
|
|
|
def test_from_string(self):
|
|
assert AssetClassification("scenic") == AssetClassification.SCENIC
|
|
|
|
def test_other_category(self):
|
|
assert AssetClassification.OTHER == "other"
|