test(wave125): entities领域实体单测 +76

覆盖Asset/Project/AssetLibrary/IngestJob/AssetStatus兼容

tests/unit/test_entities.py

@@ -0,0 +1,510 @@
+"""entities 领域实体单测 — Asset/Project/AssetLibrary/IngestJob/AssetStatus."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.classification import (
+    AssetLibraryKind,
+    ClassificationStatus,
+    IngestJobStatus,
+)
+from packages.domain.entities import (
+    Asset,
+    AssetLibrary,
+    AssetStatus,
+    IngestJob,
+    Project,
+)
+
+
+# ── AssetStatus 枚举兼容 ────────────────────────────────────────────────────
+
+
+class TestAssetStatus:
+    def test_basic_statuses(self):
+        assert AssetStatus.UPLOADING == "uploading"
+        assert AssetStatus.READY == "ready"
+        assert AssetStatus.PROCESSING == "processing"
+        assert AssetStatus.ERROR == "error"
+        assert AssetStatus.DELETED == "deleted"
+
+    def test_missing_uploaded_maps_to_ready(self):
+        # 兼容历史：uploaded → READY
+        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_unknown_value_defaults_to_ready(self):
+        # 兜底：未知值 → READY，不阻塞业务
+        assert AssetStatus("some_weird_value") == AssetStatus.READY
+
+    def test_case_insensitive(self):
+        assert AssetStatus("READY") == AssetStatus.READY
+        assert AssetStatus("Ready") == AssetStatus.READY
+
+    def test_whitespace_stripped(self):
+        assert AssetStatus("  ready  ") == AssetStatus.READY
+
+
+# ── Project ────────────────────────────────────────────────────────────────
+
+
+class TestProject:
+    def test_create_basic(self):
+        p = Project.create(owner_user_id="user_1", name="我的项目")
+        assert p.id  # 自动生成
+        assert p.owner_user_id == "user_1"
+        assert p.name == "我的项目"
+        assert p.description == ""
+        assert p.shared_users == []
+
+    def test_create_with_description(self):
+        p = Project.create(owner_user_id="u1", name="P1", description="测试描述")
+        assert p.description == "测试描述"
+
+    def test_create_name_stripped(self):
+        p = Project.create(owner_user_id="u1", name="  我的项目  ")
+        assert p.name == "我的项目"
+
+    def test_create_empty_name_raises(self):
+        with pytest.raises(ValueError, match="项目名称不能为空"):
+            Project.create(owner_user_id="u1", name="")
+
+    def test_create_whitespace_name_raises(self):
+        with pytest.raises(ValueError, match="项目名称不能为空"):
+            Project.create(owner_user_id="u1", name="   ")
+
+    def test_is_owner_true(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        assert p.is_owner("u1") is True
+
+    def test_is_owner_false(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        assert p.is_owner("u2") is False
+
+    def test_is_shared_with_true(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        p.shared_users = ["u2", "u3"]
+        assert p.is_shared_with("u2") is True
+
+    def test_is_shared_with_false(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        assert p.is_shared_with("u2") is False
+
+    def test_can_access_owner(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        assert p.can_access("u1") is True
+
+    def test_can_access_shared_user(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        p.shared_users = ["u2"]
+        assert p.can_access("u2") is True
+
+    def test_can_access_stranger(self):
+        p = Project.create(owner_user_id="u1", name="P1")
+        assert p.can_access("u3") is False
+
+    def test_description_stripped(self):
+        p = Project.create(owner_user_id="u1", name="P1", description="  desc  ")
+        assert p.description == "desc"
+
+
+# ── AssetLibrary ───────────────────────────────────────────────────────────
+
+
+class TestAssetLibrary:
+    def test_create_basic(self):
+        lib = AssetLibrary.create(
+            project_id="proj_1",
+            name="视频素材",
+            kind=AssetLibraryKind.VIDEO,
+        )
+        assert lib.id
+        assert lib.project_id == "proj_1"
+        assert lib.name == "视频素材"
+        assert lib.kind == AssetLibraryKind.VIDEO
+        assert lib.asset_count == 0
+        assert lib.total_size == 0
+
+    def test_create_name_stripped(self):
+        lib = AssetLibrary.create(
+            project_id="p1", name="  我的素材库  ", kind=AssetLibraryKind.VOICE
+        )
+        assert lib.name == "我的素材库"
+
+    def test_create_empty_name_raises(self):
+        with pytest.raises(ValueError, match="素材库名称不能为空"):
+            AssetLibrary.create(project_id="p1", name="", kind=AssetLibraryKind.VIDEO)
+
+    def test_create_whitespace_name_raises(self):
+        with pytest.raises(ValueError, match="素材库名称不能为空"):
+            AssetLibrary.create(project_id="p1", name="  ", kind=AssetLibraryKind.VIDEO)
+
+    def test_create_with_string_kind(self):
+        lib = AssetLibrary.create(project_id="p1", name="L1", kind="video")
+        assert lib.kind == AssetLibraryKind.VIDEO
+
+
+# ── Asset ──────────────────────────────────────────────────────────────────
+
+
+class TestAssetCreate:
+    def test_create_basic(self):
+        asset = Asset.create(
+            project_id="p1",
+            library_id="lib1",
+            name="测试视频.mp4",
+            storage_key="videos/test.mp4",
+            mime_type="video/mp4",
+        )
+        assert asset.id
+        assert asset.project_id == "p1"
+        assert asset.library_id == "lib1"
+        assert asset.name == "测试视频.mp4"
+        assert asset.storage_key == "videos/test.mp4"
+        assert asset.mime_type == "video/mp4"
+        assert asset.status == AssetStatus.UPLOADING
+        assert asset.classification_status == ClassificationStatus.PENDING
+        assert asset.file_size == 0
+        assert asset.tag_ids == []
+        assert asset.metadata == {}
+
+    def test_create_name_stripped(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1",
+            name="  video.mp4  ",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        assert asset.name == "video.mp4"
+
+    def test_create_storage_key_stripped(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="  key1  ", mime_type="video/mp4",
+        )
+        assert asset.storage_key == "key1"
+
+    def test_create_mime_type_stripped(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="  video/mp4  ",
+        )
+        assert asset.mime_type == "video/mp4"
+
+    def test_create_empty_name_raises(self):
+        with pytest.raises(ValueError, match="素材名称不能为空"):
+            Asset.create(
+                project_id="p1", library_id="l1", name="",
+                storage_key="k1", mime_type="video/mp4",
+            )
+
+    def test_create_empty_storage_key_raises(self):
+        with pytest.raises(ValueError, match="storage_key 不能为空"):
+            Asset.create(
+                project_id="p1", library_id="l1", name="v.mp4",
+                storage_key="  ", mime_type="video/mp4",
+            )
+
+    def test_create_empty_mime_type_raises(self):
+        with pytest.raises(ValueError, match="mime_type 不能为空"):
+            Asset.create(
+                project_id="p1", library_id="l1", name="v.mp4",
+                storage_key="k1", mime_type="",
+            )
+
+    def test_create_with_file_size(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4", file_size=1024000,
+        )
+        assert asset.file_size == 1024000
+
+    def test_create_with_duration(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4", duration=30.5,
+        )
+        assert asset.duration == 30.5
+
+    def test_create_with_dimensions(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+            width=1080, height=1920, fps=30.0, codec="h264",
+        )
+        assert asset.width == 1080
+        assert asset.height == 1920
+        assert asset.fps == 30.0
+        assert asset.codec == "h264"
+
+    def test_create_with_metadata(self):
+        meta = {"bitrate": 5000, "codec": "h264"}
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4", metadata=meta,
+        )
+        assert asset.metadata == meta
+
+    def test_create_metadata_none_defaults_empty(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4", metadata=None,
+        )
+        assert asset.metadata == {}
+
+    def test_create_thumbnail_url_none(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4", thumbnail_url=None,
+        )
+        assert asset.thumbnail_url is None
+
+    def test_create_uploaded_by_stripped(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+            uploaded_by_user_id="  user1  ",
+        )
+        assert asset.uploaded_by_user_id == "user1"
+
+    def test_create_file_hash_stripped(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+            file_hash="  abc123  ",
+        )
+        assert asset.file_hash == "abc123"
+
+
+class TestAssetFileType:
+    def test_video_mime(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        assert asset.file_type == "video"
+
+    def test_audio_mime(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="a.mp3",
+            storage_key="k1", mime_type="audio/mpeg",
+        )
+        assert asset.file_type == "audio"
+
+    def test_image_mime(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="i.jpg",
+            storage_key="k1", mime_type="image/jpeg",
+        )
+        assert asset.file_type == "image"
+
+    def test_invalid_mime_returns_full(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="f.xxx",
+            storage_key="k1", mime_type="application/octet-stream",
+        )
+        assert asset.file_type == "application"
+
+
+class TestAssetTags:
+    def test_add_tag(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("tag_1")
+        assert "tag_1" in asset.tag_ids
+        assert len(asset.tag_ids) == 1
+
+    def test_add_tag_dedup(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("tag_1")
+        asset.add_tag("tag_1")
+        assert len(asset.tag_ids) == 1
+
+    def test_add_tag_empty_raises(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        with pytest.raises(ValueError, match="标签 ID 不能为空"):
+            asset.add_tag("")
+
+    def test_add_tag_whitespace_raises(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        with pytest.raises(ValueError, match="标签 ID 不能为空"):
+            asset.add_tag("   ")
+
+    def test_add_tag_strips_whitespace(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("  tag_1  ")
+        assert asset.tag_ids == ["tag_1"]
+
+    def test_remove_tag(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("tag_1")
+        asset.add_tag("tag_2")
+        asset.remove_tag("tag_1")
+        assert asset.tag_ids == ["tag_2"]
+
+    def test_remove_nonexistent_tag_no_error(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("tag_1")
+        asset.remove_tag("nonexistent")  # 幂等，不报错
+        assert asset.tag_ids == ["tag_1"]
+
+    def test_remove_tag_strips_whitespace(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("tag_1")
+        asset.remove_tag("  tag_1  ")
+        assert len(asset.tag_ids) == 0
+
+    def test_add_tag_updates_updated_at(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        old_updated = asset.updated_at
+        import time; time.sleep(0.001)
+        asset.add_tag("tag_1")
+        assert asset.updated_at > old_updated
+
+    def test_remove_tag_updates_updated_at(self):
+        asset = Asset.create(
+            project_id="p1", library_id="l1", name="v.mp4",
+            storage_key="k1", mime_type="video/mp4",
+        )
+        asset.add_tag("tag_1")
+        old_updated = asset.updated_at
+        import time; time.sleep(0.001)
+        asset.remove_tag("tag_1")
+        assert asset.updated_at > old_updated
+
+
+# ── IngestJob ──────────────────────────────────────────────────────────────
+
+
+class TestIngestJob:
+    def test_create_basic(self):
+        job = IngestJob.create(
+            project_id="p1",
+            library_id="lib1",
+            storage_key="videos/test.mp4",
+        )
+        assert job.id
+        assert job.project_id == "p1"
+        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_file_hash(self):
+        job = IngestJob.create(
+            project_id="p1", library_id="l1",
+            storage_key="k1", file_hash="abc123",
+        )
+        assert job.file_hash == "abc123"
+
+    def test_create_project_id_stripped(self):
+        job = IngestJob.create(
+            project_id="  p1  ", library_id="l1", storage_key="k1",
+        )
+        assert job.project_id == "p1"
+
+    def test_create_library_id_stripped(self):
+        job = IngestJob.create(
+            project_id="p1", library_id="  l1  ", storage_key="k1",
+        )
+        assert job.library_id == "l1"
+
+    def test_create_storage_key_stripped(self):
+        job = IngestJob.create(
+            project_id="p1", library_id="l1", storage_key="  k1  ",
+        )
+        assert job.storage_key == "k1"
+
+    def test_create_empty_project_id_raises(self):
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            IngestJob.create(project_id="", library_id="l1", storage_key="k1")
+
+    def test_create_empty_library_id_raises(self):
+        with pytest.raises(ValueError, match="library_id 不能为空"):
+            IngestJob.create(project_id="p1", library_id="", storage_key="k1")
+
+    def test_create_empty_storage_key_raises(self):
+        with pytest.raises(ValueError, match="storage_key 不能为空"):
+            IngestJob.create(project_id="p1", library_id="l1", storage_key="")
+
+    def test_create_whitespace_project_id_raises(self):
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            IngestJob.create(project_id="   ", library_id="l1", storage_key="k1")
+
+    def test_create_file_hash_stripped(self):
+        job = IngestJob.create(
+            project_id="p1", library_id="l1",
+            storage_key="k1", file_hash="  hash123  ",
+        )
+        assert job.file_hash == "hash123"
+
+    def test_created_at_auto_set(self):
+        job = IngestJob.create(project_id="p1", library_id="l1", storage_key="k1")
+        assert job.created_at is not None
+        assert job.updated_at is not None