tests/unit/test_duplication_use_cases.py

@@ -1,12 +1,4 @@
-"""查重应用层用例单元测试。
-
-覆盖：
-- UploadForDuplicationUseCase — 创建查重记录
-- ListDuplicationRecordsUseCase — 列表查询（含分页）
-- GetDuplicationDetailUseCase — 详情查询
-- DeleteDuplicationRecordUseCase — 删除记录
-- RetryDuplicationUseCase — 重试查重（含状态校验）
-"""
+"""查重 UseCase 单元测试."""
 
 from __future__ import annotations
 
@@ -25,181 +17,211 @@ from packages.application.duplication import (
 from packages.domain.duplication import DuplicationRecord
 
 
-def _make_record(status="pending", **kwargs):
-    """创建测试用 DuplicationRecord。"""
-    record = DuplicationRecord.create(
-        user_id=kwargs.get("user_id", "user-1"),
-        filename=kwargs.get("filename", "test.mp4"),
-        file_size=kwargs.get("file_size", 1024),
-        storage_key=kwargs.get("storage_key", "oss/key"),
-        duration_seconds=kwargs.get("duration", 30.0),
+@pytest.fixture
+def mock_repo():
+    return MagicMock()
+
+
+@pytest.fixture
+def sample_record():
+    r = DuplicationRecord.create(
+        user_id="user_1",
+        filename="test_video.mp4",
+        file_size=1048576,
+        storage_key="uploads/test_video.mp4",
+        duration_seconds=30.5,
     )
-    if status != "pending":
-        record.mark_processing()
-    if status == "completed":
-        record.mark_completed(duplicate_rate=15.0, duplicate_count=1, segments=[])
-    elif status == "failed":
-        record.mark_failed("处理失败")
-    return record
+    r.id = "dup_123"
+    return r
+
+
+@pytest.fixture
+def failed_record():
+    r = DuplicationRecord.create(
+        user_id="user_1",
+        filename="failed.mp4",
+        file_size=512000,
+        storage_key="uploads/failed.mp4",
+    )
+    r.id = "dup_456"
+    r.mark_failed("网络超时")
+    return r
 
 
 class TestUploadForDuplicationUseCase:
-    """上传查重用例测试。"""
-
-    def test_execute_creates_and_persists_record(self):
-        mock_repo = MagicMock()
-        mock_repo.create.side_effect = lambda r: r
+    """UploadForDuplicationUseCase 测试"""
 
+    def test_upload_success(self, mock_repo, sample_record):
+        """上传查重成功"""
+        mock_repo.create.return_value = sample_record
         use_case = UploadForDuplicationUseCase(mock_repo)
+
         command = UploadForDuplicationCommand(
-            user_id="user-1",
-            filename="video.mp4",
-            file_size=2048,
-            storage_key="oss/video.mp4",
-            duration_seconds=60.0,
+            user_id="user_1",
+            filename="test_video.mp4",
+            file_size=1048576,
+            storage_key="uploads/test_video.mp4",
+            duration_seconds=30.5,
         )
         result = use_case.execute(command)
 
-        assert result.user_id == "user-1"
-        assert result.filename == "video.mp4"
-        assert result.file_size == 2048
-        assert result.status == "pending"
+        assert result.id == "dup_123"
+        assert result.user_id == "user_1"
         mock_repo.create.assert_called_once()
+        created = mock_repo.create.call_args[0][0]
+        assert isinstance(created, DuplicationRecord)
+        assert created.status == "pending"
 
-    def test_execute_with_default_duration(self):
-        mock_repo = MagicMock()
-        mock_repo.create.side_effect = lambda r: r
-
+    def test_upload_default_duration(self, mock_repo):
+        """不传 duration 默认 0.0"""
+        mock_repo.create.side_effect = lambda x: x
         use_case = UploadForDuplicationUseCase(mock_repo)
+
         command = UploadForDuplicationCommand(
-            user_id="user-1",
-            filename="video.mp4",
-            file_size=1024,
-            storage_key="oss/key",
+            user_id="user_1",
+            filename="test.mp4",
+            file_size=1000,
+            storage_key="key",
         )
         result = use_case.execute(command)
+
         assert result.duration_seconds == 0.0
 
-    def test_execute_invalid_user_id_raises(self):
-        mock_repo = MagicMock()
+    def test_upload_empty_user_id_raises(self, mock_repo):
+        """空 user_id 在 domain 层抛出"""
         use_case = UploadForDuplicationUseCase(mock_repo)
+
         command = UploadForDuplicationCommand(
             user_id="",
-            filename="video.mp4",
-            file_size=1024,
-            storage_key="oss/key",
+            filename="test.mp4",
+            file_size=1000,
+            storage_key="key",
         )
-        with pytest.raises(ValueError, match="user_id"):
+        with pytest.raises(ValueError, match="user_id cannot be empty"):
             use_case.execute(command)
+
         mock_repo.create.assert_not_called()
 
+    def test_upload_zero_file_size_raises(self, mock_repo):
+        """文件大小为0抛出"""
+        use_case = UploadForDuplicationUseCase(mock_repo)
+
+        command = UploadForDuplicationCommand(
+            user_id="user_1",
+            filename="test.mp4",
+            file_size=0,
+            storage_key="key",
+        )
+        with pytest.raises(ValueError, match="file_size must be positive"):
+            use_case.execute(command)
+
 
 class TestListDuplicationRecordsUseCase:
-    """列表查询用例测试。"""
+    """ListDuplicationRecordsUseCase 测试"""
 
-    def test_execute_returns_records(self):
-        records = [_make_record(), _make_record(filename="b.mp4")]
-        mock_repo = MagicMock()
-        mock_repo.list_by_user.return_value = records
-
-        use_case = ListDuplicationRecordsUseCase(mock_repo)
-        result = use_case.execute("user-1")
-
-        assert len(result) == 2
-        mock_repo.list_by_user.assert_called_once_with("user-1", offset=0, limit=50)
-
-    def test_execute_with_pagination(self):
-        mock_repo = MagicMock()
-        mock_repo.list_by_user.return_value = []
-
-        use_case = ListDuplicationRecordsUseCase(mock_repo)
-        use_case.execute("user-1", offset=10, limit=20)
-
-        mock_repo.list_by_user.assert_called_once_with("user-1", offset=10, limit=20)
-
-    def test_execute_empty_user_id_raises(self):
-        mock_repo = MagicMock()
+    def test_list_returns_results(self, mock_repo, sample_record):
+        """正常返回用户查重记录列表"""
+        mock_repo.list_by_user.return_value = [sample_record]
         use_case = ListDuplicationRecordsUseCase(mock_repo)
 
-        with pytest.raises(ValueError):
+        result = use_case.execute("user_1")
+
+        assert len(result) == 1
+        assert result[0].id == "dup_123"
+        mock_repo.list_by_user.assert_called_once_with("user_1", offset=0, limit=50)
+
+    def test_list_with_offset_limit(self, mock_repo, sample_record):
+        """带 offset 和 limit 参数"""
+        mock_repo.list_by_user.return_value = [sample_record]
+        use_case = ListDuplicationRecordsUseCase(mock_repo)
+
+        use_case.execute("user_1", offset=10, limit=20)
+
+        mock_repo.list_by_user.assert_called_once_with("user_1", offset=10, limit=20)
+
+    def test_empty_user_id_raises(self, mock_repo):
+        """空 user_id 抛出"""
+        use_case = ListDuplicationRecordsUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="user_id 不能为空"):
             use_case.execute("")
+
         mock_repo.list_by_user.assert_not_called()
 
-    def test_execute_whitespace_user_id_raises(self):
-        mock_repo = MagicMock()
+    def test_whitespace_user_id_raises(self, mock_repo):
+        """纯空格 user_id 抛出"""
         use_case = ListDuplicationRecordsUseCase(mock_repo)
 
-        with pytest.raises(ValueError):
+        with pytest.raises(ValueError, match="user_id 不能为空"):
             use_case.execute("   ")
 
-    def test_execute_strips_user_id(self):
-        mock_repo = MagicMock()
-        mock_repo.list_by_user.return_value = []
-
+    def test_user_id_stripped(self, mock_repo, sample_record):
+        """user_id 被 strip"""
+        mock_repo.list_by_user.return_value = [sample_record]
         use_case = ListDuplicationRecordsUseCase(mock_repo)
-        use_case.execute("  user-1  ")
 
-        mock_repo.list_by_user.assert_called_once_with("user-1", offset=0, limit=50)
+        use_case.execute("  user_1  ")
+
+        mock_repo.list_by_user.assert_called_once_with("user_1", offset=0, limit=50)
 
 
 class TestGetDuplicationDetailUseCase:
-    """详情查询用例测试。"""
-
-    def test_execute_returns_record(self):
-        record = _make_record()
-        mock_repo = MagicMock()
-        mock_repo.get.return_value = record
+    """GetDuplicationDetailUseCase 测试"""
 
+    def test_get_existing(self, mock_repo, sample_record):
+        """获取存在的记录"""
+        mock_repo.get.return_value = sample_record
         use_case = GetDuplicationDetailUseCase(mock_repo)
-        result = use_case.execute(record.id)
 
-        assert result is record
-        mock_repo.get.assert_called_once_with(record.id)
+        result = use_case.execute("dup_123")
 
-    def test_execute_returns_none_for_missing(self):
-        mock_repo = MagicMock()
+        assert result is not None
+        assert result.id == "dup_123"
+        mock_repo.get.assert_called_once_with("dup_123")
+
+    def test_get_nonexistent_returns_none(self, mock_repo):
+        """获取不存在的记录返回 None"""
         mock_repo.get.return_value = None
-
         use_case = GetDuplicationDetailUseCase(mock_repo)
+
         result = use_case.execute("nonexistent")
 
         assert result is None
 
 
 class TestDeleteDuplicationRecordUseCase:
-    """删除用例测试。"""
+    """DeleteDuplicationRecordUseCase 测试"""
 
-    def test_execute_deletes_record(self):
-        mock_repo = MagicMock()
+    def test_delete_success(self, mock_repo):
+        """删除成功"""
         mock_repo.delete.return_value = True
-
         use_case = DeleteDuplicationRecordUseCase(mock_repo)
-        result = use_case.execute("record-1")
+
+        result = use_case.execute("dup_123")
 
         assert result is True
-        mock_repo.delete.assert_called_once_with("record-1")
+        mock_repo.delete.assert_called_once_with("dup_123")
 
-    def test_execute_returns_false_for_missing(self):
-        mock_repo = MagicMock()
+    def test_delete_nonexistent_returns_false(self, mock_repo):
+        """删除不存在的记录返回 False"""
         mock_repo.delete.return_value = False
-
         use_case = DeleteDuplicationRecordUseCase(mock_repo)
+
         result = use_case.execute("nonexistent")
 
         assert result is False
 
 
 class TestRetryDuplicationUseCase:
-    """重试用例测试。"""
-
-    def test_execute_resets_failed_record(self):
-        record = _make_record(status="failed")
-        mock_repo = MagicMock()
-        mock_repo.get.return_value = record
-        mock_repo.update.side_effect = lambda r: r
+    """RetryDuplicationUseCase 测试"""
 
+    def test_retry_failed_record(self, mock_repo, failed_record):
+        """失败记录可以重试，状态重置为 pending"""
+        mock_repo.get.return_value = failed_record
+        mock_repo.update.side_effect = lambda x: x
         use_case = RetryDuplicationUseCase(mock_repo)
-        result = use_case.execute(record.id)
+
+        result = use_case.execute("dup_456")
 
         assert result is not None
         assert result.status == "pending"
@@ -209,24 +231,69 @@ class TestRetryDuplicationUseCase:
         assert result.segments == []
         mock_repo.update.assert_called_once()
 
-    def test_execute_returns_none_for_missing(self):
-        mock_repo = MagicMock()
+    def test_retry_nonexistent_returns_none(self, mock_repo):
+        """重试不存在的记录返回 None"""
         mock_repo.get.return_value = None
-
         use_case = RetryDuplicationUseCase(mock_repo)
+
         result = use_case.execute("nonexistent")
 
         assert result is None
         mock_repo.update.assert_not_called()
 
-    def test_execute_calls_repo_get_and_update(self):
-        record = _make_record(status="failed")
-        mock_repo = MagicMock()
-        mock_repo.get.return_value = record
-        mock_repo.update.side_effect = lambda r: r
-
+    def test_retry_pending_raises(self, mock_repo, sample_record):
+        """pending 状态的记录不能重试"""
+        assert sample_record.status == "pending"
+        mock_repo.get.return_value = sample_record
         use_case = RetryDuplicationUseCase(mock_repo)
-        use_case.execute(record.id)
 
-        mock_repo.get.assert_called_once_with(record.id)
-        mock_repo.update.assert_called_once()
+        with pytest.raises(ValueError, match="只有 failed 状态的记录可以重试"):
+            use_case.execute("dup_123")
+
+        mock_repo.update.assert_not_called()
+
+    def test_retry_completed_raises(self, mock_repo, sample_record):
+        """completed 状态的记录不能重试"""
+        sample_record.mark_completed(duplicate_rate=25.5, duplicate_count=3, segments=[])
+        mock_repo.get.return_value = sample_record
+        use_case = RetryDuplicationUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="只有 failed 状态的记录可以重试"):
+            use_case.execute("dup_123")
+
+        mock_repo.update.assert_not_called()
+
+
+class TestUploadForDuplicationCommand:
+    """UploadForDuplicationCommand 数据类测试"""
+
+    def test_command_fields(self):
+        """命令对象字段正确"""
+        cmd = UploadForDuplicationCommand(
+            user_id="user_1",
+            filename="test.mp4",
+            file_size=1024,
+            storage_key="uploads/test.mp4",
+            duration_seconds=15.0,
+        )
+        assert cmd.user_id == "user_1"
+        assert cmd.filename == "test.mp4"
+        assert cmd.file_size == 1024
+        assert cmd.storage_key == "uploads/test.mp4"
+        assert cmd.duration_seconds == 15.0
+
+    def test_default_duration(self):
+        """duration_seconds 默认 0.0"""
+        cmd = UploadForDuplicationCommand(
+            user_id="user_1",
+            filename="test.mp4",
+            file_size=1024,
+            storage_key="key",
+        )
+        assert cmd.duration_seconds == 0.0
+
+    def test_command_is_dataclass(self):
+        """是 dataclass"""
+        from dataclasses import is_dataclass
+
+        assert is_dataclass(UploadForDuplicationCommand)

tests/unit/test_generated_videos_use_cases.py

@@ -0,0 +1,319 @@
+"""生成视频 UseCase 单元测试."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from packages.application.generated_videos import (
+    GetGeneratedVideoDownloadUrlUseCase,
+    GetGeneratedVideoUseCase,
+    GetVideosByIdsUseCase,
+    ListGeneratedVideosByTaskUseCase,
+    ListGeneratedVideosPaginatedUseCase,
+    ListGeneratedVideosUseCase,
+    UpdateVideoReviewStatusUseCase,
+)
+from packages.domain.generated_video import GeneratedVideo
+
+
+@pytest.fixture
+def mock_repo():
+    return MagicMock()
+
+
+@pytest.fixture
+def sample_video():
+    v = GeneratedVideo.create(
+        project_id="proj_1",
+        generation_task_id="task_1",
+        name="测试视频",
+        file_url="https://oss.example.com/videos/test.mp4",
+        user_id="user_1",
+        file_size=1024000,
+        duration=30.5,
+        width=1920,
+        height=1080,
+        fps=30.0,
+    )
+    v.id = "video_123"
+    return v
+
+
+class TestListGeneratedVideosUseCase:
+    """ListGeneratedVideosUseCase 测试"""
+
+    def test_list_returns_results(self, mock_repo, sample_video):
+        """正常返回项目生成视频列表"""
+        mock_repo.list_by_project.return_value = [sample_video]
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        result = use_case.execute("proj_1")
+
+        assert len(result) == 1
+        assert result[0].id == "video_123"
+        mock_repo.list_by_project.assert_called_once_with("proj_1")
+
+    def test_empty_project_id_raises(self, mock_repo):
+        """空 project_id 抛出"""
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            use_case.execute("")
+
+        mock_repo.list_by_project.assert_not_called()
+
+    def test_whitespace_project_id_raises(self, mock_repo):
+        """纯空格 project_id 抛出"""
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            use_case.execute("   ")
+
+    def test_project_id_stripped(self, mock_repo, sample_video):
+        """project_id 会被 strip"""
+        mock_repo.list_by_project.return_value = [sample_video]
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        use_case.execute("  proj_1  ")
+
+        mock_repo.list_by_project.assert_called_once_with("proj_1")
+
+
+class TestListGeneratedVideosPaginatedUseCase:
+    """ListGeneratedVideosPaginatedUseCase 测试"""
+
+    def test_paginated_default_params(self, mock_repo, sample_video):
+        """默认分页参数正确传递"""
+        mock_repo.list_paginated.return_value = ([sample_video], 1)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        results, total = use_case.execute(user_id="user_1")
+
+        assert len(results) == 1
+        assert total == 1
+        mock_repo.list_paginated.assert_called_once_with(
+            user_id="user_1",
+            project_id=None,
+            status=None,
+            review_status=None,
+            page=1,
+            page_size=20,
+        )
+
+    def test_page_less_than_1_clamped(self, mock_repo, sample_video):
+        """page < 1 被修正为 1"""
+        mock_repo.list_paginated.return_value = ([sample_video], 1)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page=0)
+
+        call_kwargs = mock_repo.list_paginated.call_args[1]
+        assert call_kwargs["page"] == 1
+
+    def test_page_size_less_than_1_clamped(self, mock_repo, sample_video):
+        """page_size < 1 被修正为 20"""
+        mock_repo.list_paginated.return_value = ([sample_video], 1)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page_size=0)
+
+        call_kwargs = mock_repo.list_paginated.call_args[1]
+        assert call_kwargs["page_size"] == 20
+
+    def test_page_size_greater_than_100_clamped(self, mock_repo, sample_video):
+        """page_size > 100 被修正为 20"""
+        mock_repo.list_paginated.return_value = ([sample_video], 1)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page_size=200)
+
+        call_kwargs = mock_repo.list_paginated.call_args[1]
+        assert call_kwargs["page_size"] == 20
+
+    def test_full_filter_params(self, mock_repo, sample_video):
+        """所有过滤参数正确传递"""
+        mock_repo.list_paginated.return_value = ([sample_video], 1)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(
+            user_id="user_1",
+            project_id="proj_1",
+            status="completed",
+            review_status="approved",
+            page=2,
+            page_size=50,
+        )
+
+        mock_repo.list_paginated.assert_called_once_with(
+            user_id="user_1",
+            project_id="proj_1",
+            status="completed",
+            review_status="approved",
+            page=2,
+            page_size=50,
+        )
+
+
+class TestGetGeneratedVideoUseCase:
+    """GetGeneratedVideoUseCase 测试"""
+
+    def test_get_existing(self, mock_repo, sample_video):
+        """获取存在的视频"""
+        mock_repo.get.return_value = sample_video
+        use_case = GetGeneratedVideoUseCase(mock_repo)
+
+        result = use_case.execute("video_123")
+
+        assert result is not None
+        assert result.id == "video_123"
+
+    def test_get_nonexistent_returns_none(self, mock_repo):
+        """获取不存在的视频返回 None"""
+        mock_repo.get.return_value = None
+        use_case = GetGeneratedVideoUseCase(mock_repo)
+
+        result = use_case.execute("nonexistent")
+
+        assert result is None
+
+
+class TestListGeneratedVideosByTaskUseCase:
+    """ListGeneratedVideosByTaskUseCase 测试"""
+
+    def test_list_by_task(self, mock_repo, sample_video):
+        """按任务ID查询视频"""
+        mock_repo.list_by_generation_task.return_value = [sample_video]
+        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)
+
+        result = use_case.execute("task_1")
+
+        assert len(result) == 1
+        mock_repo.list_by_generation_task.assert_called_once_with("task_1")
+
+    def test_empty_task_id_raises(self, mock_repo):
+        """空任务ID抛出"""
+        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="generation_task_id 不能为空"):
+            use_case.execute("")
+
+    def test_task_id_stripped(self, mock_repo, sample_video):
+        """任务ID被 strip"""
+        mock_repo.list_by_generation_task.return_value = [sample_video]
+        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)
+
+        use_case.execute("  task_1  ")
+
+        mock_repo.list_by_generation_task.assert_called_once_with("task_1")
+
+
+class TestGetGeneratedVideoDownloadUrlUseCase:
+    """GetGeneratedVideoDownloadUrlUseCase 测试"""
+
+    def test_get_url_success(self, mock_repo, sample_video):
+        """成功获取下载URL"""
+        mock_repo.get.return_value = sample_video
+        use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo)
+
+        result = use_case.execute("video_123")
+
+        assert result == sample_video.file_url
+        assert "test.mp4" in result
+
+    def test_get_url_nonexistent_returns_none(self, mock_repo):
+        """视频不存在返回 None"""
+        mock_repo.get.return_value = None
+        use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo)
+
+        result = use_case.execute("nonexistent")
+
+        assert result is None
+
+
+class TestUpdateVideoReviewStatusUseCase:
+    """UpdateVideoReviewStatusUseCase 测试"""
+
+    def test_update_status_pending_review(self, mock_repo, sample_video):
+        """更新为待审核"""
+        mock_repo.update_review_status.return_value = sample_video
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        result = use_case.execute("video_123", "pending_review")
+
+        assert result is not None
+        mock_repo.update_review_status.assert_called_once_with("video_123", "pending_review")
+
+    def test_update_status_approved(self, mock_repo, sample_video):
+        """更新为审核通过"""
+        mock_repo.update_review_status.return_value = sample_video
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("video_123", "approved")
+
+        mock_repo.update_review_status.assert_called_once_with("video_123", "approved")
+
+    def test_update_status_rejected(self, mock_repo, sample_video):
+        """更新为审核拒绝"""
+        mock_repo.update_review_status.return_value = sample_video
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("video_123", "rejected")
+
+        mock_repo.update_review_status.assert_called_once_with("video_123", "rejected")
+
+    def test_invalid_status_raises(self, mock_repo):
+        """无效状态抛出 ValueError"""
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="无效的 review_status"):
+            use_case.execute("video_123", "invalid_status")
+
+        mock_repo.update_review_status.assert_not_called()
+
+    def test_empty_video_id_raises(self, mock_repo):
+        """空视频ID抛出"""
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="video_id 不能为空"):
+            use_case.execute("", "approved")
+
+    def test_video_id_stripped(self, mock_repo, sample_video):
+        """视频ID被 strip"""
+        mock_repo.update_review_status.return_value = sample_video
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("  video_123  ", "approved")
+
+        mock_repo.update_review_status.assert_called_once_with("video_123", "approved")
+
+
+class TestGetVideosByIdsUseCase:
+    """GetVideosByIdsUseCase 测试"""
+
+    def test_get_by_ids(self, mock_repo, sample_video):
+        """按ID批量获取"""
+        video2 = GeneratedVideo.create(
+            project_id="proj_1",
+            generation_task_id="task_2",
+            name="视频2",
+            file_url="https://oss.example.com/videos/v2.mp4",
+        )
+        video2.id = "video_456"
+        mock_repo.get_by_ids.return_value = [sample_video, video2]
+        use_case = GetVideosByIdsUseCase(mock_repo)
+
+        result = use_case.execute(["video_123", "video_456"])
+
+        assert len(result) == 2
+        mock_repo.get_by_ids.assert_called_once_with(["video_123", "video_456"])
+
+    def test_empty_list(self, mock_repo):
+        """空ID列表返回空"""
+        mock_repo.get_by_ids.return_value = []
+        use_case = GetVideosByIdsUseCase(mock_repo)
+
+        result = use_case.execute([])
+
+        assert result == []

tests/unit/test_projects_use_cases.py

@@ -0,0 +1,298 @@
+"""项目 UseCase 单元测试."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from packages.application.projects import (
+    CreateProjectCommand,
+    CreateProjectUseCase,
+    DeleteProjectUseCase,
+    GetProjectUseCase,
+    ListProjectsUseCase,
+    ShareProjectUseCase,
+    UnshareProjectUseCase,
+)
+from packages.domain import Project
+
+
+@pytest.fixture
+def mock_repo():
+    return MagicMock()
+
+
+@pytest.fixture
+def sample_project():
+    p = Project.create(owner_user_id="user_1", name="测试项目", description="测试描述")
+    p.id = "proj_123"
+    return p
+
+
+class TestListProjectsUseCase:
+    """ListProjectsUseCase 测试"""
+
+    def test_list_returns_repo_results(self, mock_repo, sample_project):
+        """正常返回 repository 查询结果"""
+        mock_repo.find_accessible_projects.return_value = [sample_project]
+        use_case = ListProjectsUseCase(mock_repo)
+
+        result = use_case.execute("user_1")
+
+        assert len(result) == 1
+        assert result[0].id == "proj_123"
+        mock_repo.find_accessible_projects.assert_called_once_with("user_1")
+
+    def test_empty_user_id_raises(self, mock_repo):
+        """空 user_id 抛出 ValueError"""
+        use_case = ListProjectsUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="user_id 不能为空"):
+            use_case.execute("")
+
+        mock_repo.find_accessible_projects.assert_not_called()
+
+    def test_whitespace_user_id_raises(self, mock_repo):
+        """纯空格 user_id 也抛出"""
+        use_case = ListProjectsUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="user_id 不能为空"):
+            use_case.execute("   ")
+
+    def test_user_id_stripped(self, mock_repo, sample_project):
+        """user_id 会被 strip 后查询"""
+        mock_repo.find_accessible_projects.return_value = [sample_project]
+        use_case = ListProjectsUseCase(mock_repo)
+
+        use_case.execute("  user_1  ")
+
+        mock_repo.find_accessible_projects.assert_called_once_with("user_1")
+
+
+class TestGetProjectUseCase:
+    """GetProjectUseCase 测试"""
+
+    def test_get_existing_project(self, mock_repo, sample_project):
+        """获取存在的项目"""
+        mock_repo.find_by_id.return_value = sample_project
+        use_case = GetProjectUseCase(mock_repo)
+
+        result = use_case.execute("proj_123")
+
+        assert result is not None
+        assert result.id == "proj_123"
+        mock_repo.find_by_id.assert_called_once_with("proj_123")
+
+    def test_get_nonexistent_returns_none(self, mock_repo):
+        """获取不存在的项目返回 None"""
+        mock_repo.find_by_id.return_value = None
+        use_case = GetProjectUseCase(mock_repo)
+
+        result = use_case.execute("nonexistent")
+
+        assert result is None
+
+    def test_empty_project_id_raises(self, mock_repo):
+        """空 project_id 抛出"""
+        use_case = GetProjectUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            use_case.execute("")
+
+    def test_project_id_stripped(self, mock_repo, sample_project):
+        """project_id 会被 strip"""
+        mock_repo.find_by_id.return_value = sample_project
+        use_case = GetProjectUseCase(mock_repo)
+
+        use_case.execute("  proj_123  ")
+
+        mock_repo.find_by_id.assert_called_once_with("proj_123")
+
+
+class TestCreateProjectUseCase:
+    """CreateProjectUseCase 测试"""
+
+    def test_create_success(self, mock_repo, sample_project):
+        """创建成功返回 Project"""
+        mock_repo.save.return_value = sample_project
+        use_case = CreateProjectUseCase(mock_repo)
+
+        command = CreateProjectCommand(name="新项目", description="新描述")
+        result = use_case.execute(command, "user_1")
+
+        assert result.id == "proj_123"
+        mock_repo.save.assert_called_once()
+        saved = mock_repo.save.call_args[0][0]
+        assert isinstance(saved, Project)
+        assert saved.owner_user_id == "user_1"
+        assert saved.name == "新项目"
+        assert saved.description == "新描述"
+
+    def test_create_without_description(self, mock_repo):
+        """不传 description 使用默认值"""
+        mock_repo.save.side_effect = lambda x: x
+        use_case = CreateProjectUseCase(mock_repo)
+
+        command = CreateProjectCommand(name="极简项目")
+        result = use_case.execute(command, "user_1")
+
+        assert result.name == "极简项目"
+        assert result.description == ""
+
+    def test_create_empty_name_raises(self, mock_repo):
+        """空项目名在 domain 层抛出"""
+        use_case = CreateProjectUseCase(mock_repo)
+
+        command = CreateProjectCommand(name="")
+        with pytest.raises(ValueError, match="项目名称不能为空"):
+            use_case.execute(command, "user_1")
+
+        mock_repo.save.assert_not_called()
+
+
+class TestShareProjectUseCase:
+    """ShareProjectUseCase 测试"""
+
+    def test_share_success(self, mock_repo, sample_project):
+        """所有者成功共享项目"""
+        mock_repo.find_by_id.return_value = sample_project
+        mock_repo.save.side_effect = lambda x: x
+        use_case = ShareProjectUseCase(mock_repo)
+
+        result = use_case.execute("proj_123", "user_1", "user_2")
+
+        assert "user_2" in result.shared_users
+        mock_repo.save.assert_called_once()
+
+    def test_share_nonexistent_project_raises(self, mock_repo):
+        """项目不存在时抛出"""
+        mock_repo.find_by_id.return_value = None
+        use_case = ShareProjectUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="项目不存在"):
+            use_case.execute("noexist", "user_1", "user_2")
+
+    def test_share_not_owner_raises(self, mock_repo, sample_project):
+        """非所有者不能共享"""
+        mock_repo.find_by_id.return_value = sample_project
+        use_case = ShareProjectUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="只有项目所有者可以共享"):
+            use_case.execute("proj_123", "user_other", "user_2")
+
+        mock_repo.save.assert_not_called()
+
+    def test_share_already_shared_no_duplicate(self, mock_repo, sample_project):
+        """已共享的用户不会重复添加"""
+        sample_project.shared_users = ["user_2"]
+        mock_repo.find_by_id.return_value = sample_project
+        mock_repo.save.side_effect = lambda x: x
+        use_case = ShareProjectUseCase(mock_repo)
+
+        result = use_case.execute("proj_123", "user_1", "user_2")
+
+        assert result.shared_users.count("user_2") == 1
+        mock_repo.save.assert_not_called()
+
+
+class TestUnshareProjectUseCase:
+    """UnshareProjectUseCase 测试"""
+
+    def test_unshare_success(self, mock_repo, sample_project):
+        """所有者成功取消共享"""
+        sample_project.shared_users = ["user_2", "user_3"]
+        mock_repo.find_by_id.return_value = sample_project
+        mock_repo.save.side_effect = lambda x: x
+        use_case = UnshareProjectUseCase(mock_repo)
+
+        result = use_case.execute("proj_123", "user_1", "user_2")
+
+        assert "user_2" not in result.shared_users
+        assert "user_3" in result.shared_users
+        mock_repo.save.assert_called_once()
+
+    def test_unshare_nonexistent_project_raises(self, mock_repo):
+        """项目不存在时抛出"""
+        mock_repo.find_by_id.return_value = None
+        use_case = UnshareProjectUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="项目不存在"):
+            use_case.execute("noexist", "user_1", "user_2")
+
+    def test_unshare_not_owner_raises(self, mock_repo, sample_project):
+        """非所有者不能取消共享"""
+        sample_project.shared_users = ["user_2"]
+        mock_repo.find_by_id.return_value = sample_project
+        use_case = UnshareProjectUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="只有项目所有者可以取消共享"):
+            use_case.execute("proj_123", "user_other", "user_2")
+
+        mock_repo.save.assert_not_called()
+
+    def test_unshare_not_shared_no_save(self, mock_repo, sample_project):
+        """用户未被共享时不触发 save"""
+        mock_repo.find_by_id.return_value = sample_project
+        use_case = UnshareProjectUseCase(mock_repo)
+
+        result = use_case.execute("proj_123", "user_1", "user_not_shared")
+
+        assert result is sample_project
+        mock_repo.save.assert_not_called()
+
+
+class TestDeleteProjectUseCase:
+    """DeleteProjectUseCase 测试"""
+
+    def test_delete_owner_success(self, mock_repo, sample_project):
+        """所有者删除成功"""
+        mock_repo.find_by_id.return_value = sample_project
+        mock_repo.delete.return_value = True
+        use_case = DeleteProjectUseCase(mock_repo)
+
+        result = use_case.execute("proj_123", "user_1")
+
+        assert result is True
+        mock_repo.delete.assert_called_once_with("proj_123")
+
+    def test_delete_nonexistent_returns_false(self, mock_repo):
+        """删除不存在的项目返回 False"""
+        mock_repo.find_by_id.return_value = None
+        use_case = DeleteProjectUseCase(mock_repo)
+
+        result = use_case.execute("noexist", "user_1")
+
+        assert result is False
+        mock_repo.delete.assert_not_called()
+
+    def test_delete_not_owner_raises(self, mock_repo, sample_project):
+        """非所有者删除抛出 PermissionError"""
+        mock_repo.find_by_id.return_value = sample_project
+        use_case = DeleteProjectUseCase(mock_repo)
+
+        with pytest.raises(PermissionError, match="只有项目所有者可以删除"):
+            use_case.execute("proj_123", "user_other")
+
+        mock_repo.delete.assert_not_called()
+
+
+class TestCreateProjectCommand:
+    """CreateProjectCommand 数据类测试"""
+
+    def test_command_fields(self):
+        """命令对象字段正确"""
+        cmd = CreateProjectCommand(name="test", description="desc")
+        assert cmd.name == "test"
+        assert cmd.description == "desc"
+
+    def test_command_default_description(self):
+        """description 默认空字符串"""
+        cmd = CreateProjectCommand(name="test")
+        assert cmd.description == ""
+
+    def test_command_is_dataclass(self):
+        """是 dataclass"""
+        from dataclasses import is_dataclass
+
+        assert is_dataclass(CreateProjectCommand)