tests/unit/domain/test_entities.py

@@ -0,0 +1,520 @@
+"""Domain entities 单元测试 - wave161
+
+覆盖：AssetLibraryKind / IngestJobStatus / AssetStatus / ClassificationStatus 枚举
+      User / Project / AssetLibrary / Asset / IngestJob 领域模型
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+
+from packages.domain.entities import (
+    Asset,
+    AssetLibrary,
+    AssetLibraryKind,
+    AssetStatus,
+    ClassificationStatus,
+    IngestJob,
+    IngestJobStatus,
+    Project,
+    User,
+)
+
+# ============================================================
+# 枚举测试
+# ============================================================
+
+
+class TestAssetLibraryKind:
+    def test_values(self):
+        assert AssetLibraryKind.VIDEO == "video"
+        assert AssetLibraryKind.VOICE == "voice"
+        assert AssetLibraryKind.IMAGE == "image"
+
+    def test_is_str_enum(self):
+        assert isinstance(AssetLibraryKind.VIDEO, str)
+        assert AssetLibraryKind.VIDEO == "video"
+
+    def test_members_count(self):
+        assert len(AssetLibraryKind) == 3
+
+
+class TestIngestJobStatus:
+    def test_values(self):
+        assert IngestJobStatus.PENDING == "pending"
+        assert IngestJobStatus.PROCESSING == "processing"
+        assert IngestJobStatus.COMPLETED == "completed"
+        assert IngestJobStatus.FAILED == "failed"
+
+    def test_members_count(self):
+        assert len(IngestJobStatus) == 4
+
+
+class TestAssetStatus:
+    def test_values(self):
+        assert AssetStatus.UPLOADING == "uploading"
+        assert AssetStatus.READY == "ready"
+        assert AssetStatus.PROCESSING == "processing"
+        assert AssetStatus.ERROR == "error"
+        assert AssetStatus.DELETED == "deleted"
+
+    def test_members_count(self):
+        assert len(AssetStatus) == 5
+
+
+class TestClassificationStatus:
+    def test_values(self):
+        assert ClassificationStatus.PENDING == "pending"
+        assert ClassificationStatus.PROCESSING == "processing"
+        assert ClassificationStatus.COMPLETED == "completed"
+        assert ClassificationStatus.FAILED == "failed"
+
+    def test_members_count(self):
+        assert len(ClassificationStatus) == 4
+
+
+# ============================================================
+# User 测试
+# ============================================================
+
+
+class TestUser:
+    def test_create_minimal(self):
+        user = User(id="u1", email="test@example.com", display_name="Test")
+        assert user.id == "u1"
+        assert user.email == "test@example.com"
+        assert user.display_name == "Test"
+
+    def test_default_values(self):
+        user = User(id="u1", email="t@e.com", display_name="T")
+        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.wechat_unionid is None
+
+    def test_has_created_at(self):
+        before = datetime.now(timezone.utc)
+        user = User(id="u1", email="t@e.com", display_name="T")
+        after = datetime.now(timezone.utc)
+        assert before <= user.created_at <= after
+
+    def test_full_fields(self):
+        user = User(
+            id="u1",
+            email="admin@example.com",
+            display_name="Admin",
+            username="admin",
+            is_admin=True,
+            subscription_plan="enterprise",
+            max_projects=100,
+            max_storage_gb=1000,
+        )
+        assert user.username == "admin"
+        assert user.is_admin is True
+        assert user.subscription_plan == "enterprise"
+        assert user.max_projects == 100
+        assert user.max_storage_gb == 1000
+
+
+# ============================================================
+# Project 测试
+# ============================================================
+
+
+class TestProjectCreate:
+    def test_create_minimal(self):
+        p = Project.create(owner_user_id="u1", name="My Project")
+        assert p.id
+        assert len(p.id) == 32  # uuid4 hex
+        assert p.owner_user_id == "u1"
+        assert p.name == "My Project"
+        assert p.description == ""
+        assert p.shared_users == []
+
+    def test_create_with_description(self):
+        p = Project.create(owner_user_id="u1", name="P", description="  desc  ")
+        assert p.description == "desc"  # strip
+
+    def test_create_strips_name(self):
+        p = Project.create(owner_user_id="u1", name="  My Project  ")
+        assert p.name == "My Project"
+
+    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_create_unique_ids(self):
+        p1 = Project.create(owner_user_id="u1", name="P1")
+        p2 = Project.create(owner_user_id="u1", name="P2")
+        assert p1.id != p2.id
+
+    def test_create_has_timestamp(self):
+        before = datetime.now(timezone.utc)
+        p = Project.create(owner_user_id="u1", name="P")
+        after = datetime.now(timezone.utc)
+        assert before <= p.created_at <= after
+
+
+class TestProjectAccess:
+    def test_is_owner_true(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        assert p.is_owner("u1") is True
+
+    def test_is_owner_false(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        assert p.is_owner("u2") is False
+
+    def test_is_shared_with_true(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        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="P")
+        p.shared_users = ["u2"]
+        assert p.is_shared_with("u3") is False
+
+    def test_is_shared_with_empty(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        assert p.is_shared_with("u2") is False
+
+    def test_can_access_owner(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        assert p.can_access("u1") is True
+
+    def test_can_access_shared(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        p.shared_users = ["u2"]
+        assert p.can_access("u2") is True
+
+    def test_cannot_access_other(self):
+        p = Project.create(owner_user_id="u1", name="P")
+        assert p.can_access("u3") is False
+
+
+# ============================================================
+# AssetLibrary 测试
+# ============================================================
+
+
+class TestAssetLibraryCreate:
+    def test_create_minimal(self):
+        lib = AssetLibrary.create(project_id="p1", name="Videos", kind=AssetLibraryKind.VIDEO)
+        assert lib.id
+        assert len(lib.id) == 32
+        assert lib.project_id == "p1"
+        assert lib.name == "Videos"
+        assert lib.kind == AssetLibraryKind.VIDEO
+        assert lib.asset_count == 0
+        assert lib.total_size == 0
+
+    def test_create_strips_name(self):
+        lib = AssetLibrary.create(project_id="p1", name="  Voices  ", kind=AssetLibraryKind.VOICE)
+        assert lib.name == "Voices"
+
+    def test_create_empty_name_raises(self):
+        with pytest.raises(ValueError, match="素材库名称不能为空"):
+            AssetLibrary.create(project_id="p1", name="", kind=AssetLibraryKind.IMAGE)
+
+    def test_create_whitespace_name_raises(self):
+        with pytest.raises(ValueError, match="素材库名称不能为空"):
+            AssetLibrary.create(project_id="p1", name="   ", kind=AssetLibraryKind.VIDEO)
+
+    def test_create_unique_ids(self):
+        lib1 = AssetLibrary.create("p1", "L1", AssetLibraryKind.VIDEO)
+        lib2 = AssetLibrary.create("p1", "L2", AssetLibraryKind.IMAGE)
+        assert lib1.id != lib2.id
+
+    def test_create_has_timestamps(self):
+        before = datetime.now(timezone.utc)
+        lib = AssetLibrary.create("p1", "L", AssetLibraryKind.VIDEO)
+        after = datetime.now(timezone.utc)
+        assert before <= lib.created_at <= after
+        assert before <= lib.updated_at <= after
+
+
+# ============================================================
+# Asset 测试
+# ============================================================
+
+
+class TestAssetCreate:
+    def test_create_minimal(self):
+        asset = Asset.create(
+            project_id="p1",
+            library_id="lib1",
+            name="video.mp4",
+            storage_key="uploads/v1.mp4",
+            mime_type="video/mp4",
+        )
+        assert asset.id
+        assert len(asset.id) == 32
+        assert asset.project_id == "p1"
+        assert asset.library_id == "lib1"
+        assert asset.name == "video.mp4"
+        assert asset.storage_key == "uploads/v1.mp4"
+        assert asset.mime_type == "video/mp4"
+        assert asset.file_size == 0
+        assert asset.status == AssetStatus.UPLOADING
+        assert asset.classification_status == ClassificationStatus.PENDING
+        assert asset.tag_ids == []
+        assert asset.metadata == {}
+
+    def test_create_full(self):
+        asset = Asset.create(
+            project_id="p1",
+            library_id="lib1",
+            name="  clip.mov  ",
+            storage_key="  s3://bucket/clip.mov  ",
+            mime_type="  video/quicktime  ",
+            metadata={"resolution": "1080p"},
+            file_size=1024000,
+            thumbnail_url="https://img/thumb.jpg",
+            duration=30.5,
+            width=1920,
+            height=1080,
+            fps=29.97,
+            codec="h264",
+            status=AssetStatus.READY,
+            classification_status=ClassificationStatus.COMPLETED,
+            quality_score=0.95,
+            uploaded_by_user_id="  u1  ",
+            file_hash="  abc123  ",
+        )
+        assert asset.name == "clip.mov"  # stripped
+        assert asset.storage_key == "s3://bucket/clip.mov"
+        assert asset.mime_type == "video/quicktime"
+        assert asset.file_size == 1024000
+        assert asset.status == AssetStatus.READY
+        assert asset.classification_status == ClassificationStatus.COMPLETED
+        assert asset.quality_score == 0.95
+        assert asset.uploaded_by_user_id == "u1"
+        assert asset.file_hash == "abc123"
+        assert asset.metadata == {"resolution": "1080p"}
+
+    def test_create_empty_name_raises(self):
+        with pytest.raises(ValueError, match="素材名称不能为空"):
+            Asset.create(
+                project_id="p1",
+                library_id="lib1",
+                name="",
+                storage_key="k",
+                mime_type="video/mp4",
+            )
+
+    def test_create_whitespace_name_raises(self):
+        with pytest.raises(ValueError, match="素材名称不能为空"):
+            Asset.create(
+                project_id="p1",
+                library_id="lib1",
+                name="   ",
+                storage_key="k",
+                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="lib1",
+                name="v.mp4",
+                storage_key="",
+                mime_type="video/mp4",
+            )
+
+    def test_create_whitespace_storage_key_raises(self):
+        with pytest.raises(ValueError, match="storage_key 不能为空"):
+            Asset.create(
+                project_id="p1",
+                library_id="lib1",
+                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="lib1",
+                name="v.mp4",
+                storage_key="k",
+                mime_type="",
+            )
+
+    def test_create_metadata_none_defaults_empty_dict(self):
+        asset = Asset.create(
+            project_id="p1",
+            library_id="lib1",
+            name="v.mp4",
+            storage_key="k",
+            mime_type="video/mp4",
+            metadata=None,
+        )
+        assert asset.metadata == {}
+
+    def test_create_unique_ids(self):
+        a1 = Asset.create("p1", "lib1", "a.mp4", "k1", "video/mp4")
+        a2 = Asset.create("p1", "lib1", "b.mp4", "k2", "video/mp4")
+        assert a1.id != a2.id
+
+    def test_create_has_timestamps(self):
+        before = datetime.now(timezone.utc)
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        after = datetime.now(timezone.utc)
+        assert before <= a.created_at <= after
+        assert before <= a.updated_at <= after
+
+
+class TestAssetTags:
+    def test_add_tag(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        assert "tag1" in a.tag_ids
+        assert len(a.tag_ids) == 1
+
+    def test_add_tag_strips(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("  tag1  ")
+        assert a.tag_ids == ["tag1"]
+
+    def test_add_tag_deduplicates(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        a.add_tag("tag1")
+        assert a.tag_ids == ["tag1"]
+
+    def test_add_empty_tag_raises(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        with pytest.raises(ValueError, match="标签 ID 不能为空"):
+            a.add_tag("")
+
+    def test_add_whitespace_tag_raises(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        with pytest.raises(ValueError, match="标签 ID 不能为空"):
+            a.add_tag("   ")
+
+    def test_add_tag_updates_updated_at(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        old_updated = a.updated_at
+        # 确保时间差
+        import time
+
+        time.sleep(0.001)
+        a.add_tag("tag1")
+        assert a.updated_at >= old_updated
+
+    def test_add_duplicate_tag_no_updated_at_change(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        old_updated = a.updated_at
+        a.add_tag("tag1")  # 重复
+        assert a.updated_at == old_updated
+
+    def test_remove_tag(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        a.add_tag("tag2")
+        a.remove_tag("tag1")
+        assert a.tag_ids == ["tag2"]
+
+    def test_remove_tag_strips(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        a.remove_tag("  tag1  ")
+        assert a.tag_ids == []
+
+    def test_remove_nonexistent_tag_idempotent(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        old_updated = a.updated_at
+        a.remove_tag("nonexistent")  # 不报错
+        assert a.tag_ids == ["tag1"]
+        assert a.updated_at == old_updated  # 没修改就不更新时间
+
+    def test_remove_tag_updates_updated_at(self):
+        a = Asset.create("p1", "lib1", "v.mp4", "k", "video/mp4")
+        a.add_tag("tag1")
+        old_updated = a.updated_at
+        import time
+
+        time.sleep(0.001)
+        a.remove_tag("tag1")
+        assert a.updated_at >= old_updated
+
+
+# ============================================================
+# IngestJob 测试
+# ============================================================
+
+
+class TestIngestJobCreate:
+    def test_create_minimal(self):
+        job = IngestJob.create(project_id="p1", library_id="lib1", storage_key="uploads/v1.mp4")
+        assert job.id
+        assert len(job.id) == 32
+        assert job.project_id == "p1"
+        assert job.library_id == "lib1"
+        assert job.storage_key == "uploads/v1.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(
+            project_id="p1",
+            library_id="lib1",
+            storage_key="k",
+            file_hash="  abc123  ",
+        )
+        assert job.file_hash == "abc123"  # stripped
+
+    def test_create_strips_fields(self):
+        job = IngestJob.create(
+            project_id="  p1  ",
+            library_id="  lib1  ",
+            storage_key="  key1  ",
+        )
+        assert job.project_id == "p1"
+        assert job.library_id == "lib1"
+        assert job.storage_key == "key1"
+
+    def test_create_empty_project_id_raises(self):
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            IngestJob.create(project_id="", library_id="lib1", storage_key="k")
+
+    def test_create_whitespace_project_id_raises(self):
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            IngestJob.create(project_id="   ", library_id="lib1", storage_key="k")
+
+    def test_create_empty_library_id_raises(self):
+        with pytest.raises(ValueError, match="library_id 不能为空"):
+            IngestJob.create(project_id="p1", library_id="", storage_key="k")
+
+    def test_create_empty_storage_key_raises(self):
+        with pytest.raises(ValueError, match="storage_key 不能为空"):
+            IngestJob.create(project_id="p1", library_id="lib1", storage_key="")
+
+    def test_create_unique_ids(self):
+        j1 = IngestJob.create("p1", "lib1", "k1")
+        j2 = IngestJob.create("p1", "lib1", "k2")
+        assert j1.id != j2.id
+
+    def test_create_has_timestamps(self):
+        before = datetime.now(timezone.utc)
+        job = IngestJob.create("p1", "lib1", "k")
+        after = datetime.now(timezone.utc)
+        assert before <= job.created_at <= after
+        assert before <= job.updated_at <= after