test(p3-1): wave31 素材库+素材UseCase单测20个 #772
Executable
+342
@@ -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"
|
||||
Executable
+452
@@ -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()
|
||||
Reference in New Issue
Block a user