test(p3-1): wave31 素材库+素材UseCase单测20个 #772

Merged
xiaoxia merged 2 commits from test/p3-1-wave31-asset-library-use-cases into develop 2026-07-23 20:46:04 +08:00
2 changed files with 794 additions and 0 deletions
+342
View File
@@ -0,0 +1,342 @@
"""素材库 + 素材 UseCase 单元测试(P3-1 第31波)。"""
from __future__ import annotations
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.domain import AssetLibraryKind, AssetStatus, ClassificationStatus
# ── ListAssetLibrariesUseCase ────────────────────────────────────────────────
class TestListAssetLibrariesUseCase:
"""ListAssetLibrariesUseCase 单元测试。"""
def test_list_libraries_success(self):
"""正常列出项目素材库。"""
mock_repo = MagicMock()
mock_repo.find_by_project.return_value = ["lib1", "lib2"]
use_case = ListAssetLibrariesUseCase(mock_repo)
result = use_case.execute("proj_001")
assert result == ["lib1", "lib2"]
mock_repo.find_by_project.assert_called_once_with("proj_001")
def test_list_libraries_empty(self):
"""项目无素材库返回空列表。"""
mock_repo = MagicMock()
mock_repo.find_by_project.return_value = []
use_case = ListAssetLibrariesUseCase(mock_repo)
result = use_case.execute("proj_001")
assert result == []
def test_list_libraries_empty_project_id_raises(self):
"""空 project_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = ListAssetLibrariesUseCase(mock_repo)
with pytest.raises(ValueError, match="project_id 不能为空"):
use_case.execute("")
def test_list_libraries_whitespace_project_id_raises(self):
"""全空格 project_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = ListAssetLibrariesUseCase(mock_repo)
with pytest.raises(ValueError, match="project_id 不能为空"):
use_case.execute(" ")
def test_list_libraries_project_id_stripped(self):
"""project_id 首尾空格被清理。"""
mock_repo = MagicMock()
mock_repo.find_by_project.return_value = []
use_case = ListAssetLibrariesUseCase(mock_repo)
use_case.execute(" proj_001 ")
mock_repo.find_by_project.assert_called_once_with("proj_001")
# ── CreateAssetLibraryUseCase ────────────────────────────────────────────────
class TestCreateAssetLibraryUseCase:
"""CreateAssetLibraryUseCase 单元测试。"""
def test_create_video_library_success(self):
"""正常创建视频素材库。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetLibraryUseCase(mock_repo)
command = CreateAssetLibraryCommand(
project_id="proj_001",
name="我的视频",
kind=AssetLibraryKind.VIDEO,
)
result = use_case.execute(command)
assert result.project_id == "proj_001"
assert result.name == "我的视频"
assert result.kind == AssetLibraryKind.VIDEO
mock_repo.create.assert_called_once()
def test_create_audio_library_success(self):
"""创建音频素材库。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetLibraryUseCase(mock_repo)
command = CreateAssetLibraryCommand(
project_id="proj_001",
name="配乐库",
kind=AssetLibraryKind.VOICE,
)
result = use_case.execute(command)
assert result.kind == AssetLibraryKind.VOICE
assert result.asset_count == 0
assert result.total_size == 0
def test_create_image_library_success(self):
"""创建图片素材库。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetLibraryUseCase(mock_repo)
command = CreateAssetLibraryCommand(
project_id="proj_001",
name="图片素材",
kind=AssetLibraryKind.IMAGE,
)
result = use_case.execute(command)
assert result.kind == AssetLibraryKind.IMAGE
def test_create_library_name_stripped(self):
"""素材库名称首尾空格被清理。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetLibraryUseCase(mock_repo)
command = CreateAssetLibraryCommand(
project_id="proj_001",
name=" 我的素材库 ",
kind=AssetLibraryKind.VIDEO,
)
result = use_case.execute(command)
assert result.name == "我的素材库"
# ── ListAssetsUseCase ────────────────────────────────────────────────────────
class TestListAssetsUseCase:
"""ListAssetsUseCase 单元测试。"""
def test_list_assets_success(self):
"""正常列出素材库素材。"""
mock_repo = MagicMock()
mock_repo.find_by_library.return_value = ["asset1", "asset2"]
use_case = ListAssetsUseCase(mock_repo)
result = use_case.execute("lib_001")
assert result == ["asset1", "asset2"]
mock_repo.find_by_library.assert_called_once_with("lib_001")
def test_list_assets_empty(self):
"""素材库为空返回空列表。"""
mock_repo = MagicMock()
mock_repo.find_by_library.return_value = []
use_case = ListAssetsUseCase(mock_repo)
result = use_case.execute("lib_001")
assert result == []
def test_list_assets_empty_library_id_raises(self):
"""空 library_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = ListAssetsUseCase(mock_repo)
with pytest.raises(ValueError, match="library_id 不能为空"):
use_case.execute("")
def test_list_assets_whitespace_library_id_raises(self):
"""全空格 library_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = ListAssetsUseCase(mock_repo)
with pytest.raises(ValueError, match="library_id 不能为空"):
use_case.execute(" ")
def test_list_assets_library_id_stripped(self):
"""library_id 首尾空格被清理。"""
mock_repo = MagicMock()
mock_repo.find_by_library.return_value = []
use_case = ListAssetsUseCase(mock_repo)
use_case.execute(" lib_001 ")
mock_repo.find_by_library.assert_called_once_with("lib_001")
# ── CreateAssetUseCase ──────────────────────────────────────────────────────
class TestCreateAssetUseCase:
"""CreateAssetUseCase 单元测试。"""
def test_create_video_asset_success(self):
"""正常创建视频素材。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetUseCase(mock_repo)
command = CreateAssetCommand(
project_id="proj_001",
library_id="lib_001",
name="video.mp4",
storage_key="assets/video.mp4",
mime_type="video/mp4",
file_size=1024000,
duration=30.5,
width=1920,
height=1080,
fps=25.0,
codec="h264",
)
result = use_case.execute(command)
assert result.name == "video.mp4"
assert result.project_id == "proj_001"
assert result.library_id == "lib_001"
assert result.storage_key == "assets/video.mp4"
assert result.mime_type == "video/mp4"
assert result.file_size == 1024000
assert result.duration == 30.5
assert result.width == 1920
assert result.height == 1080
assert result.fps == 25.0
assert result.codec == "h264"
assert result.status == AssetStatus.UPLOADING
assert result.classification_status == ClassificationStatus.PENDING
mock_repo.create.assert_called_once()
def test_create_audio_asset_success(self):
"""创建音频素材。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetUseCase(mock_repo)
command = CreateAssetCommand(
project_id="proj_001",
library_id="lib_002",
name="bgm.mp3",
storage_key="audio/bgm.mp3",
mime_type="audio/mpeg",
file_size=512000,
duration=180.0,
)
result = use_case.execute(command)
assert result.mime_type == "audio/mpeg"
assert result.duration == 180.0
assert result.width is None
assert result.height is None
def test_create_asset_default_status(self):
"""默认状态为 UPLOADING + PENDING。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetUseCase(mock_repo)
command = CreateAssetCommand(
project_id="proj_001",
library_id="lib_001",
name="test.mp4",
storage_key="test.mp4",
mime_type="video/mp4",
)
result = use_case.execute(command)
assert result.status == AssetStatus.UPLOADING
assert result.classification_status == ClassificationStatus.PENDING
def test_create_asset_custom_status(self):
"""可以指定自定义状态。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetUseCase(mock_repo)
command = CreateAssetCommand(
project_id="proj_001",
library_id="lib_001",
name="test.mp4",
storage_key="test.mp4",
mime_type="video/mp4",
status=AssetStatus.READY,
classification_status=ClassificationStatus.COMPLETED,
quality_score=85.5,
uploaded_by_user_id="user_001",
)
result = use_case.execute(command)
assert result.status == AssetStatus.READY
assert result.classification_status == ClassificationStatus.COMPLETED
assert result.quality_score == 85.5
assert result.uploaded_by_user_id == "user_001"
def test_create_asset_with_metadata(self):
"""带 metadata 创建素材。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetUseCase(mock_repo)
metadata = {"location": "beijing", "camera": "sony"}
command = CreateAssetCommand(
project_id="proj_001",
library_id="lib_001",
name="test.mp4",
storage_key="test.mp4",
mime_type="video/mp4",
metadata=metadata,
)
result = use_case.execute(command)
assert result.metadata == metadata
def test_create_asset_with_thumbnail(self):
"""带缩略图创建素材。"""
mock_repo = MagicMock()
mock_repo.create.side_effect = lambda x: x
use_case = CreateAssetUseCase(mock_repo)
command = CreateAssetCommand(
project_id="proj_001",
library_id="lib_001",
name="test.mp4",
storage_key="test.mp4",
mime_type="video/mp4",
thumbnail_url="http://cdn.com/thumb.jpg",
)
result = use_case.execute(command)
assert result.thumbnail_url == "http://cdn.com/thumb.jpg"
+452
View File
@@ -0,0 +1,452 @@
"""Project 领域模型 + UseCase 单元测试(P3-1 第30波)。"""
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.entities import Project
# ── Project 领域模型 ────────────────────────────────────────────────────────
class TestProjectDomain:
"""Project 领域模型单元测试。"""
def test_create_project_basic(self):
"""创建项目基本字段正确。"""
project = Project.create(owner_user_id="user_001", name="我的项目")
assert project.id
assert project.owner_user_id == "user_001"
assert project.name == "我的项目"
assert project.description == ""
assert project.shared_users == []
assert project.created_at
def test_create_project_with_description(self):
"""创建项目带描述。"""
project = Project.create(
owner_user_id="user_001",
name="测试项目",
description="这是一个测试项目",
)
assert project.name == "测试项目"
assert project.description == "这是一个测试项目"
def test_create_project_name_stripped(self):
"""项目名称首尾空格被清理。"""
project = Project.create(owner_user_id="user_001", name=" 我的项目 ")
assert project.name == "我的项目"
def test_create_project_description_stripped(self):
"""项目描述首尾空格被清理。"""
project = Project.create(
owner_user_id="user_001",
name="测试",
description=" 描述内容 ",
)
assert project.description == "描述内容"
def test_create_project_empty_name_raises(self):
"""空项目名抛 ValueError。"""
with pytest.raises(ValueError, match="项目名称不能为空"):
Project.create(owner_user_id="user_001", name="")
def test_create_project_whitespace_name_raises(self):
"""全空格项目名抛 ValueError。"""
with pytest.raises(ValueError, match="项目名称不能为空"):
Project.create(owner_user_id="user_001", name=" ")
def test_is_owner_true(self):
"""is_owner 所有者返回 True。"""
project = Project.create(owner_user_id="user_001", name="项目")
assert project.is_owner("user_001") is True
def test_is_owner_false(self):
"""is_owner 非所有者返回 False。"""
project = Project.create(owner_user_id="user_001", name="项目")
assert project.is_owner("user_002") is False
def test_is_shared_with_true(self):
"""is_shared_with 已共享用户返回 True。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002", "user_003"]
assert project.is_shared_with("user_002") is True
assert project.is_shared_with("user_003") is True
def test_is_shared_with_false(self):
"""is_shared_with 未共享用户返回 False。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002"]
assert project.is_shared_with("user_004") is False
def test_is_shared_with_empty_list(self):
"""共享列表为空时返回 False。"""
project = Project.create(owner_user_id="user_001", name="项目")
assert project.is_shared_with("user_002") is False
def test_can_access_owner(self):
"""所有者可以访问。"""
project = Project.create(owner_user_id="user_001", name="项目")
assert project.can_access("user_001") is True
def test_can_access_shared_user(self):
"""共享用户可以访问。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002"]
assert project.can_access("user_002") is True
def test_can_access_other_user(self):
"""其他用户不能访问。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002"]
assert project.can_access("user_003") is False
# ── ListProjectsUseCase ──────────────────────────────────────────────────────
class TestListProjectsUseCase:
"""ListProjectsUseCase 单元测试。"""
def test_list_projects_success(self):
"""正常列出用户项目。"""
mock_repo = MagicMock()
mock_repo.find_accessible_projects.return_value = [
Project.create(owner_user_id="user_001", name="项目1"),
Project.create(owner_user_id="user_001", name="项目2"),
]
use_case = ListProjectsUseCase(mock_repo)
result = use_case.execute("user_001")
assert len(result) == 2
mock_repo.find_accessible_projects.assert_called_once_with("user_001")
def test_list_projects_empty(self):
"""用户没有项目返回空列表。"""
mock_repo = MagicMock()
mock_repo.find_accessible_projects.return_value = []
use_case = ListProjectsUseCase(mock_repo)
result = use_case.execute("user_001")
assert result == []
def test_list_projects_empty_user_id_raises(self):
"""空 user_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = ListProjectsUseCase(mock_repo)
with pytest.raises(ValueError, match="user_id 不能为空"):
use_case.execute("")
def test_list_projects_whitespace_user_id_raises(self):
"""全空格 user_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = ListProjectsUseCase(mock_repo)
with pytest.raises(ValueError, match="user_id 不能为空"):
use_case.execute(" ")
def test_list_projects_user_id_stripped(self):
"""user_id 首尾空格被清理。"""
mock_repo = MagicMock()
mock_repo.find_accessible_projects.return_value = []
use_case = ListProjectsUseCase(mock_repo)
use_case.execute(" user_001 ")
mock_repo.find_accessible_projects.assert_called_once_with("user_001")
# ── GetProjectUseCase ───────────────────────────────────────────────────────
class TestGetProjectUseCase:
"""GetProjectUseCase 单元测试。"""
def test_get_project_success(self):
"""正常获取项目。"""
project = Project.create(owner_user_id="user_001", name="测试项目")
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
use_case = GetProjectUseCase(mock_repo)
result = use_case.execute("proj_001")
assert result == project
mock_repo.find_by_id.assert_called_once_with("proj_001")
def test_get_project_not_found(self):
"""项目不存在返回 None。"""
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = None
use_case = GetProjectUseCase(mock_repo)
result = use_case.execute("proj_001")
assert result is None
def test_get_project_empty_id_raises(self):
"""空 project_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = GetProjectUseCase(mock_repo)
with pytest.raises(ValueError, match="project_id 不能为空"):
use_case.execute("")
def test_get_project_whitespace_id_raises(self):
"""全空格 project_id 抛 ValueError。"""
mock_repo = MagicMock()
use_case = GetProjectUseCase(mock_repo)
with pytest.raises(ValueError, match="project_id 不能为空"):
use_case.execute(" ")
def test_get_project_id_stripped(self):
"""project_id 首尾空格被清理。"""
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = None
use_case = GetProjectUseCase(mock_repo)
use_case.execute(" proj_001 ")
mock_repo.find_by_id.assert_called_once_with("proj_001")
# ── CreateProjectUseCase ────────────────────────────────────────────────────
class TestCreateProjectUseCase:
"""CreateProjectUseCase 单元测试。"""
def test_create_project_success(self):
"""正常创建项目。"""
mock_repo = MagicMock()
mock_repo.save.side_effect = lambda p: p
use_case = CreateProjectUseCase(mock_repo)
command = CreateProjectCommand(name="新项目", description="项目描述")
result = use_case.execute(command, "user_001")
assert result.name == "新项目"
assert result.description == "项目描述"
assert result.owner_user_id == "user_001"
mock_repo.save.assert_called_once()
def test_create_project_default_description(self):
"""不传描述默认为空。"""
mock_repo = MagicMock()
mock_repo.save.side_effect = lambda p: p
use_case = CreateProjectUseCase(mock_repo)
command = CreateProjectCommand(name="新项目")
result = use_case.execute(command, "user_001")
assert result.description == ""
def test_create_project_empty_name_raises(self):
"""空项目名抛 ValueError(领域层校验)。"""
mock_repo = MagicMock()
use_case = CreateProjectUseCase(mock_repo)
command = CreateProjectCommand(name="")
with pytest.raises(ValueError, match="项目名称不能为空"):
use_case.execute(command, "user_001")
mock_repo.save.assert_not_called()
# ── ShareProjectUseCase ─────────────────────────────────────────────────────
class TestShareProjectUseCase:
"""ShareProjectUseCase 单元测试。"""
def test_share_project_success(self):
"""正常共享项目给用户。"""
project = Project.create(owner_user_id="user_001", name="项目")
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
mock_repo.save.side_effect = lambda p: p
use_case = ShareProjectUseCase(mock_repo)
result = use_case.execute(project.id, "user_001", "user_002")
assert "user_002" in result.shared_users
mock_repo.save.assert_called_once()
def test_share_project_already_shared_no_duplicate(self):
"""已共享用户再次共享不重复添加。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002"]
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
mock_repo.save.side_effect = lambda p: p
use_case = ShareProjectUseCase(mock_repo)
result = use_case.execute(project.id, "user_001", "user_002")
assert result.shared_users == ["user_002"]
mock_repo.save.assert_not_called()
def test_share_project_not_found_raises(self):
"""项目不存在抛 ValueError。"""
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = None
use_case = ShareProjectUseCase(mock_repo)
with pytest.raises(ValueError, match="项目不存在"):
use_case.execute("proj_xxx", "user_001", "user_002")
mock_repo.save.assert_not_called()
def test_share_project_not_owner_raises(self):
"""非所有者不能共享项目。"""
project = Project.create(owner_user_id="user_001", name="项目")
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
use_case = ShareProjectUseCase(mock_repo)
with pytest.raises(ValueError, match="只有项目所有者可以共享项目"):
use_case.execute(project.id, "user_002", "user_003")
mock_repo.save.assert_not_called()
# ── UnshareProjectUseCase ───────────────────────────────────────────────────
class TestUnshareProjectUseCase:
"""UnshareProjectUseCase 单元测试。"""
def test_unshare_project_success(self):
"""正常取消共享。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002", "user_003"]
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
mock_repo.save.side_effect = lambda p: p
use_case = UnshareProjectUseCase(mock_repo)
result = use_case.execute(project.id, "user_001", "user_002")
assert result.shared_users == ["user_003"]
mock_repo.save.assert_called_once()
def test_unshare_project_not_shared_noop(self):
"""未共享的用户取消共享不操作。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_003"]
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
mock_repo.save.side_effect = lambda p: p
use_case = UnshareProjectUseCase(mock_repo)
result = use_case.execute(project.id, "user_001", "user_002")
assert result.shared_users == ["user_003"]
mock_repo.save.assert_not_called()
def test_unshare_project_not_found_raises(self):
"""项目不存在抛 ValueError。"""
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = None
use_case = UnshareProjectUseCase(mock_repo)
with pytest.raises(ValueError, match="项目不存在"):
use_case.execute("proj_xxx", "user_001", "user_002")
mock_repo.save.assert_not_called()
def test_unshare_project_not_owner_raises(self):
"""非所有者不能取消共享。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002"]
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
use_case = UnshareProjectUseCase(mock_repo)
with pytest.raises(ValueError, match="只有项目所有者可以取消共享"):
use_case.execute(project.id, "user_002", "user_003")
mock_repo.save.assert_not_called()
# ── DeleteProjectUseCase ────────────────────────────────────────────────────
class TestDeleteProjectUseCase:
"""DeleteProjectUseCase 单元测试。"""
def test_delete_project_success(self):
"""所有者正常删除项目。"""
project = Project.create(owner_user_id="user_001", name="项目")
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
mock_repo.delete.return_value = True
use_case = DeleteProjectUseCase(mock_repo)
result = use_case.execute(project.id, "user_001")
assert result is True
mock_repo.delete.assert_called_once_with(project.id)
def test_delete_project_not_found(self):
"""项目不存在返回 False。"""
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = None
use_case = DeleteProjectUseCase(mock_repo)
result = use_case.execute("proj_xxx", "user_001")
assert result is False
mock_repo.delete.assert_not_called()
def test_delete_project_not_owner_raises(self):
"""非所有者删除抛 PermissionError。"""
project = Project.create(owner_user_id="user_001", name="项目")
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
use_case = DeleteProjectUseCase(mock_repo)
with pytest.raises(PermissionError, match="只有项目所有者可以删除项目"):
use_case.execute(project.id, "user_002")
mock_repo.delete.assert_not_called()
def test_delete_project_shared_user_raises(self):
"""共享用户不能删除项目。"""
project = Project.create(owner_user_id="user_001", name="项目")
project.shared_users = ["user_002"]
mock_repo = MagicMock()
mock_repo.find_by_id.return_value = project
use_case = DeleteProjectUseCase(mock_repo)
with pytest.raises(PermissionError, match="只有项目所有者可以删除项目"):
use_case.execute(project.id, "user_002")
mock_repo.delete.assert_not_called()