From 8c43ef53dcc57cc8da94e09b4a51d41630780cd3 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 26 Jul 2026 17:32:43 +0800 Subject: [PATCH] test(wave95): 82 unit tests for domain entities (Project/Asset/AssetStatus/IngestJob/User/AssetLibrary) - Project: create validation + access control (is_owner/is_shared_with/can_access) = 13 tests - AssetLibrary: create validation + kind variants = 6 tests - AssetStatus: enum values + _missing_ backward compatibility (20+ aliases) = 26 tests - Asset: create validation + file_type property + tag operations (add/remove/dedup) = 23 tests - IngestJob: create validation + field stripping = 11 tests - User: dataclass default values + admin + pro subscription = 3 tests Total: 82 unit tests for entities.py 292 lines --- tests/unit/test_domain_entities.py | 488 +++++++++++++++++++++++++++++ 1 file changed, 488 insertions(+) create mode 100755 tests/unit/test_domain_entities.py diff --git a/tests/unit/test_domain_entities.py b/tests/unit/test_domain_entities.py new file mode 100755 index 000000000..95184c0fe --- /dev/null +++ b/tests/unit/test_domain_entities.py @@ -0,0 +1,488 @@ +"""Domain entities 单元测试。""" + +from datetime import datetime, timezone + +import pytest + +from packages.domain.classification import ( + AssetLibraryKind, + ClassificationStatus, + IngestJobStatus, +) +from packages.domain.entities import ( + Asset, + AssetLibrary, + AssetStatus, + IngestJob, + Project, + User, +) + + +class TestProjectCreate: + def test_create_success(self): + project = Project.create(owner_user_id="user1", name="我的项目") + assert project.id is not None + assert len(project.id) == 32 + assert project.owner_user_id == "user1" + assert project.name == "我的项目" + assert project.description == "" + assert project.shared_users == [] + assert isinstance(project.created_at, datetime) + + def test_create_with_description(self): + project = Project.create("u1", "Test Project", "A test description") + assert project.description == "A test description" + + def test_create_strips_name(self): + project = Project.create("u1", " 带空格的项目 ") + assert project.name == "带空格的项目" + + def test_create_strips_description(self): + project = Project.create("u1", "P1", " desc ") + assert project.description == "desc" + + def test_create_empty_name(self): + with pytest.raises(ValueError, match="项目名称不能为空"): + Project.create("u1", "") + + def test_create_whitespace_name(self): + with pytest.raises(ValueError, match="项目名称不能为空"): + Project.create("u1", " \t ") + + def test_create_unique_ids(self): + p1 = Project.create("u1", "P1") + p2 = Project.create("u1", "P2") + assert p1.id != p2.id + + +class TestProjectAccess: + def test_is_owner_true(self): + project = Project.create("owner1", "P1") + assert project.is_owner("owner1") is True + + def test_is_owner_false(self): + project = Project.create("owner1", "P1") + assert project.is_owner("other") is False + + def test_is_shared_with_true(self): + project = Project.create("owner1", "P1") + project.shared_users = ["user_a", "user_b"] + assert project.is_shared_with("user_a") is True + assert project.is_shared_with("user_b") is True + + def test_is_shared_with_false(self): + project = Project.create("owner1", "P1") + project.shared_users = ["user_a"] + assert project.is_shared_with("user_c") is False + + def test_can_access_owner(self): + project = Project.create("owner1", "P1") + assert project.can_access("owner1") is True + + def test_can_access_shared_user(self): + project = Project.create("owner1", "P1") + project.shared_users = ["shared_user"] + assert project.can_access("shared_user") is True + + def test_cannot_access_other(self): + project = Project.create("owner1", "P1") + assert project.can_access("stranger") is False + + def test_empty_shared_users(self): + project = Project.create("owner1", "P1") + assert project.shared_users == [] + assert project.is_shared_with("anyone") is False + + +class TestAssetLibraryCreate: + def test_create_video_library(self): + lib = AssetLibrary.create("proj1", "视频素材库", AssetLibraryKind.VIDEO) + assert lib.id is not None + assert len(lib.id) == 32 + assert lib.project_id == "proj1" + assert lib.name == "视频素材库" + assert lib.kind == AssetLibraryKind.VIDEO + assert lib.asset_count == 0 + assert lib.total_size == 0 + + def test_create_voice_library(self): + lib = AssetLibrary.create("proj1", "音乐库", AssetLibraryKind.VOICE) + assert lib.kind == AssetLibraryKind.VOICE + + def test_create_image_library(self): + lib = AssetLibrary.create("proj1", "图片库", AssetLibraryKind.IMAGE) + assert lib.kind == AssetLibraryKind.IMAGE + + def test_create_strips_name(self): + lib = AssetLibrary.create("p1", " 我的库 ", AssetLibraryKind.VIDEO) + assert lib.name == "我的库" + + def test_create_empty_name(self): + with pytest.raises(ValueError, match="素材库名称不能为空"): + AssetLibrary.create("p1", "", AssetLibraryKind.VIDEO) + + def test_create_whitespace_name(self): + with pytest.raises(ValueError, match="素材库名称不能为空"): + AssetLibrary.create("p1", " \t ", AssetLibraryKind.VIDEO) + + +class TestAssetStatusEnum: + def test_basic_values(self): + assert AssetStatus.UPLOADING.value == "uploading" + assert AssetStatus.READY.value == "ready" + assert AssetStatus.PROCESSING.value == "processing" + assert AssetStatus.ERROR.value == "error" + assert AssetStatus.DELETED.value == "deleted" + + def test_missing_uploaded_maps_to_ready(self): + assert AssetStatus("uploaded") == AssetStatus.READY + + def test_missing_success_maps_to_ready(self): + assert AssetStatus("success") == AssetStatus.READY + + def test_missing_ok_maps_to_ready(self): + assert AssetStatus("ok") == AssetStatus.READY + + def test_missing_done_maps_to_ready(self): + assert AssetStatus("done") == AssetStatus.READY + + def test_missing_complete_maps_to_ready(self): + assert AssetStatus("complete") == AssetStatus.READY + + def test_missing_upload_maps_to_uploading(self): + assert AssetStatus("upload") == AssetStatus.UPLOADING + + def test_missing_uploading_start_maps_to_uploading(self): + assert AssetStatus("uploading_start") == AssetStatus.UPLOADING + + def test_missing_upload_start_maps_to_uploading(self): + assert AssetStatus("upload_start") == AssetStatus.UPLOADING + + def test_missing_failed_maps_to_error(self): + assert AssetStatus("failed") == AssetStatus.ERROR + + def test_missing_fail_maps_to_error(self): + assert AssetStatus("fail") == AssetStatus.ERROR + + def test_missing_err_maps_to_error(self): + assert AssetStatus("err") == AssetStatus.ERROR + + def test_missing_process_maps_to_processing(self): + assert AssetStatus("process") == AssetStatus.PROCESSING + + def test_missing_running_maps_to_processing(self): + assert AssetStatus("running") == AssetStatus.PROCESSING + + def test_missing_run_maps_to_processing(self): + assert AssetStatus("run") == AssetStatus.PROCESSING + + def test_missing_unknown_value_falls_back_to_ready(self): + assert AssetStatus("completely_unknown_status") == AssetStatus.READY + + def test_missing_empty_string_falls_back_to_ready(self): + assert AssetStatus("") == AssetStatus.READY + + def test_missing_case_insensitive(self): + assert AssetStatus("UPLOADED") == AssetStatus.READY + assert AssetStatus("Success") == AssetStatus.READY + assert AssetStatus("FAILED") == AssetStatus.ERROR + + def test_missing_with_whitespace(self): + assert AssetStatus(" uploaded ") == AssetStatus.READY + assert AssetStatus("\tfailed\n") == AssetStatus.ERROR + + def test_missing_non_string_value(self): + assert AssetStatus(None) == AssetStatus.READY + assert AssetStatus(123) == AssetStatus.READY + + def test_known_values_still_work(self): + assert AssetStatus("uploading") == AssetStatus.UPLOADING + assert AssetStatus("ready") == AssetStatus.READY + assert AssetStatus("processing") == AssetStatus.PROCESSING + assert AssetStatus("error") == AssetStatus.ERROR + assert AssetStatus("deleted") == AssetStatus.DELETED + + +class TestAssetCreate: + def test_create_minimal(self): + asset = Asset.create( + project_id="proj1", + library_id="lib1", + name="test.mp4", + storage_key="videos/test.mp4", + mime_type="video/mp4", + ) + assert asset.id is not None + assert len(asset.id) == 32 + assert asset.project_id == "proj1" + assert asset.library_id == "lib1" + assert asset.name == "test.mp4" + assert asset.storage_key == "videos/test.mp4" + assert asset.mime_type == "video/mp4" + assert asset.file_size == 0 + assert asset.thumbnail_url is None + assert asset.duration is None + assert asset.width is None + assert asset.height is None + assert asset.status == AssetStatus.UPLOADING + assert asset.classification_status == ClassificationStatus.PENDING + assert asset.quality_score is None + assert asset.tag_ids == [] + assert isinstance(asset.created_at, datetime) + assert isinstance(asset.updated_at, datetime) + + def test_create_with_all_fields(self): + asset = Asset.create( + project_id="proj1", + library_id="lib1", + name="movie.mp4", + storage_key="v/m.mp4", + mime_type="video/mp4", + metadata={"key": "val"}, + file_size=1024000, + thumbnail_url="http://cdn/thumb.jpg", + duration=120.5, + width=1920, + height=1080, + fps=30.0, + codec="h264", + status=AssetStatus.READY, + classification_status=ClassificationStatus.COMPLETED, + quality_score=0.85, + uploaded_by_user_id="user1", + file_hash="abc123", + ) + assert asset.file_size == 1024000 + assert asset.thumbnail_url == "http://cdn/thumb.jpg" + assert asset.duration == 120.5 + assert asset.width == 1920 + assert asset.height == 1080 + assert asset.fps == 30.0 + assert asset.codec == "h264" + assert asset.status == AssetStatus.READY + assert asset.classification_status == ClassificationStatus.COMPLETED + assert asset.quality_score == 0.85 + assert asset.uploaded_by_user_id == "user1" + assert asset.file_hash == "abc123" + assert asset.metadata == {"key": "val"} + + def test_create_strips_name(self): + asset = Asset.create("p1", "l1", " test.mp4 ", "k", "video/mp4") + assert asset.name == "test.mp4" + + def test_create_strips_storage_key(self): + asset = Asset.create("p1", "l1", "n", " key.mp4 ", "video/mp4") + assert asset.storage_key == "key.mp4" + + def test_create_strips_mime_type(self): + asset = Asset.create("p1", "l1", "n", "k", " video/mp4 ") + assert asset.mime_type == "video/mp4" + + def test_create_empty_name(self): + with pytest.raises(ValueError, match="素材名称不能为空"): + Asset.create("p1", "l1", "", "k", "video/mp4") + + def test_create_empty_storage_key(self): + with pytest.raises(ValueError, match="storage_key 不能为空"): + Asset.create("p1", "l1", "n", "", "video/mp4") + + def test_create_empty_mime_type(self): + with pytest.raises(ValueError, match="mime_type 不能为空"): + Asset.create("p1", "l1", "n", "k", "") + + def test_create_whitespace_storage_key(self): + with pytest.raises(ValueError, match="storage_key 不能为空"): + Asset.create("p1", "l1", "n", " \t ", "video/mp4") + + def test_create_none_metadata_defaults_to_empty_dict(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4", metadata=None) + assert asset.metadata == {} + + def test_create_unique_ids(self): + a1 = Asset.create("p1", "l1", "n1", "k1", "video/mp4") + a2 = Asset.create("p1", "l1", "n2", "k2", "video/mp4") + assert a1.id != a2.id + + +class TestAssetFileType: + def test_video_mime(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + assert asset.file_type == "video" + + def test_audio_mime(self): + asset = Asset.create("p1", "l1", "n", "k", "audio/mpeg") + assert asset.file_type == "audio" + + def test_image_mime(self): + asset = Asset.create("p1", "l1", "n", "k", "image/jpeg") + assert asset.file_type == "image" + + def test_simple_mime_no_slash(self): + asset = Asset.create("p1", "l1", "n", "k", "application") + assert asset.file_type == "application" + + +class TestAssetTags: + def test_add_tag(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("tag1") + assert "tag1" in asset.tag_ids + assert len(asset.tag_ids) == 1 + + def test_add_tag_strips(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag(" tag_trim ") + assert "tag_trim" in asset.tag_ids + + def test_add_tag_duplicate_prevented(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("tag1") + asset.add_tag("tag1") + assert asset.tag_ids.count("tag1") == 1 + assert len(asset.tag_ids) == 1 + + def test_add_tag_empty(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + with pytest.raises(ValueError, match="标签 ID 不能为空"): + asset.add_tag("") + + def test_add_tag_whitespace(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + with pytest.raises(ValueError, match="标签 ID 不能为空"): + asset.add_tag(" \t ") + + def test_add_multiple_tags(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + asset.add_tag("t2") + asset.add_tag("t3") + assert asset.tag_ids == ["t1", "t2", "t3"] + + def test_remove_tag(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + asset.add_tag("t2") + asset.remove_tag("t1") + assert asset.tag_ids == ["t2"] + + def test_remove_nonexistent_tag_idempotent(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + # 删除不存在的标签不报错 + asset.remove_tag("nonexistent") + assert asset.tag_ids == ["t1"] + + def test_remove_tag_strips(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + asset.remove_tag(" t1 ") + assert asset.tag_ids == [] + + def test_add_tag_updates_updated_at(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + old_time = asset.updated_at + asset.add_tag("t1") + assert asset.updated_at >= old_time + + def test_remove_tag_updates_updated_at(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + old_time = asset.updated_at + asset.remove_tag("t1") + assert asset.updated_at >= old_time + + +class TestIngestJobCreate: + def test_create_success(self): + job = IngestJob.create( + project_id="proj1", + library_id="lib1", + storage_key="videos/test.mp4", + ) + assert job.id is not None + assert len(job.id) == 32 + assert job.project_id == "proj1" + assert job.library_id == "lib1" + assert job.storage_key == "videos/test.mp4" + assert job.status == IngestJobStatus.PENDING + assert job.error_message == "" + assert job.result_asset_id == "" + assert job.file_hash == "" + + def test_create_with_hash(self): + job = IngestJob.create("p1", "l1", "k", file_hash="abcdef123456") + assert job.file_hash == "abcdef123456" + + def test_create_strips_project_id(self): + job = IngestJob.create(" p1 ", "l1", "k") + assert job.project_id == "p1" + + def test_create_strips_library_id(self): + job = IngestJob.create("p1", " l1 ", "k") + assert job.library_id == "l1" + + def test_create_strips_storage_key(self): + job = IngestJob.create("p1", "l1", " k ") + assert job.storage_key == "k" + + def test_create_strips_file_hash(self): + job = IngestJob.create("p1", "l1", "k", file_hash=" hash ") + assert job.file_hash == "hash" + + def test_create_empty_project_id(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + IngestJob.create("", "l1", "k") + + def test_create_empty_library_id(self): + with pytest.raises(ValueError, match="library_id 不能为空"): + IngestJob.create("p1", "", "k") + + def test_create_empty_storage_key(self): + with pytest.raises(ValueError, match="storage_key 不能为空"): + IngestJob.create("p1", "l1", "") + + def test_create_whitespace_project_id(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + IngestJob.create(" \t ", "l1", "k") + + def test_create_unique_ids(self): + j1 = IngestJob.create("p1", "l1", "k1") + j2 = IngestJob.create("p1", "l1", "k2") + assert j1.id != j2.id + + +class TestUserDataclass: + def test_default_values(self): + user = User(id="u1", email="test@example.com", display_name="Test User") + assert user.id == "u1" + assert user.email == "test@example.com" + assert user.display_name == "Test User" + assert user.username == "" + assert user.password_hash == "" + assert user.email_verified is False + assert user.subscription_plan == "free" + assert user.subscription_status == "active" + assert user.max_projects == 3 + assert user.max_storage_gb == 10 + assert user.used_storage_gb == 0.0 + assert user.is_admin is False + assert user.wechat_openid is None + assert user.phone is None + assert user.phone_verified is False + assert isinstance(user.created_at, datetime) + + def test_admin_user(self): + user = User(id="admin", email="admin@test.com", display_name="Admin", is_admin=True) + assert user.is_admin is True + + def test_pro_subscription(self): + user = User( + id="u1", + email="u@t.com", + display_name="U", + subscription_plan="pro", + max_storage_gb=100, + ) + assert user.subscription_plan == "pro" + assert user.max_storage_gb == 100