From 301d07e1100128e9009732450ab07dcafe7dc545 Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Sun, 26 Jul 2026 17:46:07 +0800
Subject: [PATCH] test(wave100): 40 unit tests for 3 application layer modules
 (milestone)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Milestone wave #100 — cleaning up zero-test application modules:
- generated_videos.py: 8 UseCases fully covered (25 tests)
  - ListGeneratedVideosUseCase: 4 tests
  - ListGeneratedVideosPaginatedUseCase: 7 tests (page clamp / page_size clamp / filters)
  - GetGeneratedVideoUseCase: 2 tests
  - ListGeneratedVideosByTaskUseCase: 3 tests
  - GetGeneratedVideoDownloadUrlUseCase: 2 tests
  - UpdateVideoReviewStatusUseCase: 7 tests (3 valid statuses + validation)
  - GetVideosByIdsUseCase: 2 tests
- assets.py: 6 tests (CreateAssetCommand + ListAssets + CreateAsset)
- asset_libraries.py: 5 tests (Command + List + Create)
Total: 40 tests across 3 modules
---
 tests/unit/test_application_modules.py | 499 +++++++++++++++++++++++++
 1 file changed, 499 insertions(+)
 create mode 100755 tests/unit/test_application_modules.py

diff --git a/tests/unit/test_application_modules.py b/tests/unit/test_application_modules.py
new file mode 100755
index 000000000..09711b97a
--- /dev/null
+++ b/tests/unit/test_application_modules.py
@@ -0,0 +1,499 @@
+"""Application 层零测试模块合集 — 第100波里程碑。
+
+覆盖:
+- packages/application/generated_videos.py (8个UseCase)
+- packages/application/assets.py (ListAssets + CreateAsset)
+- packages/application/asset_libraries.py (ListLibraries + CreateLibrary)
+
+策略: Mock repository，测参数校验 + 委托行为
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from packages.application.asset_libraries import (
+    CreateAssetLibraryCommand,
+    CreateAssetLibraryUseCase,
+    ListAssetLibrariesUseCase,
+)
+from packages.application.assets import (
+    CreateAssetCommand,
+    CreateAssetUseCase,
+    ListAssetsUseCase,
+)
+from packages.application.generated_videos import (
+    GetGeneratedVideoDownloadUrlUseCase,
+    GetGeneratedVideoUseCase,
+    GetVideosByIdsUseCase,
+    ListGeneratedVideosByTaskUseCase,
+    ListGeneratedVideosPaginatedUseCase,
+    ListGeneratedVideosUseCase,
+    UpdateVideoReviewStatusUseCase,
+)
+from packages.domain import AssetLibraryKind, AssetStatus, ClassificationStatus, GeneratedVideo
+
+# ── generated_videos.py ────────────────────────────────────────────────────────
+
+
+class TestListGeneratedVideosUseCase:
+    def test_success(self):
+        mock_repo = MagicMock()
+        mock_repo.list_by_project.return_value = [MagicMock(spec=GeneratedVideo)]
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        result = use_case.execute("proj1")
+
+        assert len(result) == 1
+        mock_repo.list_by_project.assert_called_once_with("proj1")
+
+    def test_strips_project_id(self):
+        mock_repo = MagicMock()
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        use_case.execute("  proj1  ")
+
+        mock_repo.list_by_project.assert_called_once_with("proj1")
+
+    def test_empty_project_id_raises(self):
+        mock_repo = MagicMock()
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            use_case.execute("")
+
+    def test_whitespace_project_id_raises(self):
+        mock_repo = MagicMock()
+        use_case = ListGeneratedVideosUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            use_case.execute("  \t  ")
+
+
+class TestListGeneratedVideosPaginatedUseCase:
+    def test_default_params(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        result, total = use_case.execute()
+
+        assert total == 0
+        assert result == []
+        mock_repo.list_paginated.assert_called_once_with(
+            user_id=None,
+            project_id=None,
+            status=None,
+            review_status=None,
+            page=1,
+            page_size=20,
+        )
+
+    def test_page_below_1_clamps_to_1(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page=0)
+
+        mock_repo.list_paginated.assert_called_once()
+        call_kwargs = mock_repo.list_paginated.call_args.kwargs
+        assert call_kwargs["page"] == 1
+
+    def test_negative_page_clamps(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page=-5)
+
+        assert mock_repo.list_paginated.call_args.kwargs["page"] == 1
+
+    def test_page_size_zero_clamps(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page_size=0)
+
+        assert mock_repo.list_paginated.call_args.kwargs["page_size"] == 20
+
+    def test_page_size_over_100_clamps(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page_size=200)
+
+        assert mock_repo.list_paginated.call_args.kwargs["page_size"] == 20
+
+    def test_page_size_50_ok(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(page_size=50)
+
+        assert mock_repo.list_paginated.call_args.kwargs["page_size"] == 50
+
+    def test_with_all_filters(self):
+        mock_repo = MagicMock()
+        mock_repo.list_paginated.return_value = ([], 0)
+        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)
+
+        use_case.execute(
+            user_id="u1",
+            project_id="p1",
+            status="completed",
+            review_status="approved",
+            page=2,
+            page_size=10,
+        )
+
+        mock_repo.list_paginated.assert_called_once_with(
+            user_id="u1",
+            project_id="p1",
+            status="completed",
+            review_status="approved",
+            page=2,
+            page_size=10,
+        )
+
+
+class TestGetGeneratedVideoUseCase:
+    def test_found(self):
+        mock_repo = MagicMock()
+        expected = MagicMock(spec=GeneratedVideo)
+        mock_repo.get.return_value = expected
+        use_case = GetGeneratedVideoUseCase(mock_repo)
+
+        result = use_case.execute("vid1")
+
+        assert result == expected
+        mock_repo.get.assert_called_once_with("vid1")
+
+    def test_not_found(self):
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+        use_case = GetGeneratedVideoUseCase(mock_repo)
+
+        result = use_case.execute("vid1")
+
+        assert result is None
+
+
+class TestListGeneratedVideosByTaskUseCase:
+    def test_success(self):
+        mock_repo = MagicMock()
+        mock_repo.list_by_generation_task.return_value = [MagicMock()]
+        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)
+
+        result = use_case.execute("task1")
+
+        assert len(result) == 1
+        mock_repo.list_by_generation_task.assert_called_once_with("task1")
+
+    def test_strips_task_id(self):
+        mock_repo = MagicMock()
+        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)
+
+        use_case.execute("  task1  ")
+
+        mock_repo.list_by_generation_task.assert_called_once_with("task1")
+
+    def test_empty_task_id_raises(self):
+        mock_repo = MagicMock()
+        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="generation_task_id 不能为空"):
+            use_case.execute("")
+
+
+class TestGetGeneratedVideoDownloadUrlUseCase:
+    def test_found(self):
+        mock_repo = MagicMock()
+        mock_item = MagicMock()
+        mock_item.file_url = "https://cdn/v.mp4"
+        mock_repo.get.return_value = mock_item
+        use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo)
+
+        result = use_case.execute("vid1")
+
+        assert result == "https://cdn/v.mp4"
+
+    def test_not_found_returns_none(self):
+        mock_repo = MagicMock()
+        mock_repo.get.return_value = None
+        use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo)
+
+        result = use_case.execute("vid1")
+
+        assert result is None
+
+
+class TestUpdateVideoReviewStatusUseCase:
+    def test_pending_review(self):
+        mock_repo = MagicMock()
+        mock_repo.update_review_status.return_value = MagicMock()
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("vid1", "pending_review")
+
+        mock_repo.update_review_status.assert_called_once_with("vid1", "pending_review")
+
+    def test_approved(self):
+        mock_repo = MagicMock()
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("vid1", "approved")
+
+        mock_repo.update_review_status.assert_called_once_with("vid1", "approved")
+
+    def test_rejected(self):
+        mock_repo = MagicMock()
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("vid1", "rejected")
+
+        mock_repo.update_review_status.assert_called_once_with("vid1", "rejected")
+
+    def test_strips_video_id(self):
+        mock_repo = MagicMock()
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        use_case.execute("  vid1  ", "approved")
+
+        mock_repo.update_review_status.assert_called_once_with("vid1", "approved")
+
+    def test_empty_video_id_raises(self):
+        mock_repo = MagicMock()
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="video_id 不能为空"):
+            use_case.execute("", "approved")
+
+    def test_invalid_status_raises(self):
+        mock_repo = MagicMock()
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="无效的 review_status"):
+            use_case.execute("vid1", "invalid_status")
+
+    def test_not_found_returns_none(self):
+        mock_repo = MagicMock()
+        mock_repo.update_review_status.return_value = None
+        use_case = UpdateVideoReviewStatusUseCase(mock_repo)
+
+        result = use_case.execute("vid1", "approved")
+
+        assert result is None
+
+
+class TestGetVideosByIdsUseCase:
+    def test_success(self):
+        mock_repo = MagicMock()
+        mock_repo.get_by_ids.return_value = [MagicMock(), MagicMock()]
+        use_case = GetVideosByIdsUseCase(mock_repo)
+
+        result = use_case.execute(["id1", "id2", "id3"])
+
+        assert len(result) == 2
+        mock_repo.get_by_ids.assert_called_once_with(["id1", "id2", "id3"])
+
+    def test_empty_list(self):
+        mock_repo = MagicMock()
+        mock_repo.get_by_ids.return_value = []
+        use_case = GetVideosByIdsUseCase(mock_repo)
+
+        result = use_case.execute([])
+
+        assert result == []
+        mock_repo.get_by_ids.assert_called_once_with([])
+
+
+# ── assets.py ──────────────────────────────────────────────────────────────────
+
+
+class TestCreateAssetCommand:
+    def test_minimal(self):
+        cmd = CreateAssetCommand(
+            project_id="p1",
+            library_id="l1",
+            name="test.mp4",
+            storage_key="k",
+            mime_type="video/mp4",
+        )
+        assert cmd.project_id == "p1"
+        assert cmd.library_id == "l1"
+        assert cmd.name == "test.mp4"
+        assert cmd.storage_key == "k"
+        assert cmd.mime_type == "video/mp4"
+        assert cmd.file_size == 0
+        assert cmd.status == AssetStatus.UPLOADING
+        assert cmd.classification_status == ClassificationStatus.PENDING
+
+    def test_full(self):
+        cmd = CreateAssetCommand(
+            project_id="p1",
+            library_id="l1",
+            name="test.mp4",
+            storage_key="k",
+            mime_type="video/mp4",
+            metadata={"k": "v"},
+            file_size=1024,
+            duration=10.0,
+            width=1920,
+            height=1080,
+            fps=30.0,
+            codec="h264",
+            status=AssetStatus.READY,
+            quality_score=0.9,
+            uploaded_by_user_id="u1",
+        )
+        assert cmd.file_size == 1024
+        assert cmd.duration == 10.0
+        assert cmd.status == AssetStatus.READY
+        assert cmd.quality_score == 0.9
+
+
+class TestListAssetsUseCase:
+    def test_success(self):
+        mock_repo = MagicMock()
+        mock_repo.find_by_library.return_value = []
+        use_case = ListAssetsUseCase(mock_repo)
+
+        result = use_case.execute("lib1")
+
+        assert result == []
+        mock_repo.find_by_library.assert_called_once_with("lib1")
+
+    def test_strips_library_id(self):
+        mock_repo = MagicMock()
+        use_case = ListAssetsUseCase(mock_repo)
+
+        use_case.execute("  lib1  ")
+
+        mock_repo.find_by_library.assert_called_once_with("lib1")
+
+    def test_empty_library_id_raises(self):
+        mock_repo = MagicMock()
+        use_case = ListAssetsUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="library_id 不能为空"):
+            use_case.execute("")
+
+
+class TestCreateAssetUseCase:
+    def test_creates_asset_via_repo(self):
+        mock_repo = MagicMock()
+        mock_repo.create.return_value = MagicMock()
+        use_case = CreateAssetUseCase(mock_repo)
+
+        cmd = CreateAssetCommand(
+            project_id="p1",
+            library_id="l1",
+            name="test.mp4",
+            storage_key="videos/t.mp4",
+            mime_type="video/mp4",
+            file_size=1024,
+        )
+        result = use_case.execute(cmd)
+
+        assert result is not None
+        mock_repo.create.assert_called_once()
+        created_asset = mock_repo.create.call_args[0][0]
+        assert created_asset.project_id == "p1"
+        assert created_asset.name == "test.mp4"
+        assert created_asset.file_size == 1024
+        assert created_asset.status == AssetStatus.UPLOADING
+
+    def test_asset_create_validation_propagates(self):
+        mock_repo = MagicMock()
+        use_case = CreateAssetUseCase(mock_repo)
+
+        cmd = CreateAssetCommand(
+            project_id="p1",
+            library_id="l1",
+            name="",
+            storage_key="k",
+            mime_type="video/mp4",
+        )
+
+        with pytest.raises(ValueError, match="素材名称不能为空"):
+            use_case.execute(cmd)
+
+
+# ── asset_libraries.py ────────────────────────────────────────────────────────
+
+
+class TestCreateAssetLibraryCommand:
+    def test_creation(self):
+        cmd = CreateAssetLibraryCommand(
+            project_id="p1",
+            name="我的库",
+            kind=AssetLibraryKind.VIDEO,
+        )
+        assert cmd.project_id == "p1"
+        assert cmd.name == "我的库"
+        assert cmd.kind == AssetLibraryKind.VIDEO
+
+
+class TestListAssetLibrariesUseCase:
+    def test_success(self):
+        mock_repo = MagicMock()
+        mock_repo.find_by_project.return_value = []
+        use_case = ListAssetLibrariesUseCase(mock_repo)
+
+        result = use_case.execute("p1")
+
+        assert result == []
+        mock_repo.find_by_project.assert_called_once_with("p1")
+
+    def test_strips_project_id(self):
+        mock_repo = MagicMock()
+        use_case = ListAssetLibrariesUseCase(mock_repo)
+
+        use_case.execute("  p1  ")
+
+        mock_repo.find_by_project.assert_called_once_with("p1")
+
+    def test_empty_project_id_raises(self):
+        mock_repo = MagicMock()
+        use_case = ListAssetLibrariesUseCase(mock_repo)
+
+        with pytest.raises(ValueError, match="project_id 不能为空"):
+            use_case.execute("")
+
+
+class TestCreateAssetLibraryUseCase:
+    def test_creates_library_via_repo(self):
+        mock_repo = MagicMock()
+        mock_repo.create.return_value = MagicMock()
+        use_case = CreateAssetLibraryUseCase(mock_repo)
+
+        cmd = CreateAssetLibraryCommand(
+            project_id="p1",
+            name="视频库",
+            kind=AssetLibraryKind.VIDEO,
+        )
+        result = use_case.execute(cmd)
+
+        assert result is not None
+        mock_repo.create.assert_called_once()
+        created = mock_repo.create.call_args[0][0]
+        assert created.project_id == "p1"
+        assert created.name == "视频库"
+        assert created.kind == AssetLibraryKind.VIDEO
+
+    def test_validation_propagates(self):
+        mock_repo = MagicMock()
+        use_case = CreateAssetLibraryUseCase(mock_repo)
+
+        cmd = CreateAssetLibraryCommand(
+            project_id="p1",
+            name="",
+            kind=AssetLibraryKind.VIDEO,
+        )
+
+        with pytest.raises(ValueError, match="素材库名称不能为空"):
+            use_case.execute(cmd)