500 lines
16 KiB
Python
Executable File
500 lines
16 KiB
Python
Executable File
"""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)
|