test: P3-1 第34波单元测试（title_library/recipe） #812
Regular → Executable
+338
-144
@@ -1,9 +1,8 @@
|
||||
"""Recipe use cases unit tests."""
|
||||
"""配方 Recipe UseCase 单元测试."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import Mock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -18,70 +17,138 @@ from packages.application.recipe.use_cases import (
|
||||
FeatureDisabledError,
|
||||
GetRecipeUseCase,
|
||||
ListRecipesUseCase,
|
||||
NotFoundError,
|
||||
UpdateRecipeUseCase,
|
||||
UseRecipeResult,
|
||||
UseRecipeUseCase,
|
||||
)
|
||||
from packages.domain.exceptions import NotFoundError
|
||||
from packages.domain.recipe import Recipe, RecipeItem
|
||||
|
||||
|
||||
def _make_recipe(**kwargs) -> Recipe:
|
||||
defaults = dict(
|
||||
id="recipe001",
|
||||
user_id="user001",
|
||||
name="测试配方",
|
||||
description="描述",
|
||||
template_id="tpl001",
|
||||
generation_params={"mode": "one_take"},
|
||||
items=[],
|
||||
def _make_recipe(id: str, name: str, user_id: str = "user_1", item_count: int = 0) -> Recipe:
|
||||
items = [
|
||||
RecipeItem(
|
||||
id=f"item_{i}",
|
||||
recipe_id=id,
|
||||
item_type="asset",
|
||||
item_id=f"asset_{i}",
|
||||
position=i,
|
||||
)
|
||||
for i in range(item_count)
|
||||
]
|
||||
return Recipe(
|
||||
id=id,
|
||||
user_id=user_id,
|
||||
name=name,
|
||||
description="测试配方",
|
||||
template_id="tmpl_1",
|
||||
generation_params={"resolution": "1080p"},
|
||||
items=items,
|
||||
is_active=True,
|
||||
metadata_={},
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return Recipe(**defaults)
|
||||
|
||||
|
||||
def _make_item(**kwargs) -> RecipeItem:
|
||||
defaults = dict(
|
||||
id="item001",
|
||||
recipe_id="recipe001",
|
||||
item_type="asset",
|
||||
item_id="asset001",
|
||||
position=0,
|
||||
metadata_={},
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return RecipeItem(**defaults)
|
||||
@pytest.fixture
|
||||
def mock_repo():
|
||||
return MagicMock()
|
||||
|
||||
|
||||
class TestListRecipesUseCase:
|
||||
"""ListRecipesUseCase 测试"""
|
||||
|
||||
def test_list_returns_results(self, mock_repo):
|
||||
"""正常返回配方列表"""
|
||||
recipe = _make_recipe("r1", "配方1")
|
||||
mock_repo.list_by_user.return_value = [recipe]
|
||||
use_case = ListRecipesUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("user_1")
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].id == "r1"
|
||||
mock_repo.list_by_user.assert_called_once_with("user_1", skip=0, limit=50)
|
||||
|
||||
def test_list_with_pagination(self, mock_repo):
|
||||
"""带分页参数"""
|
||||
mock_repo.list_by_user.return_value = []
|
||||
use_case = ListRecipesUseCase(mock_repo)
|
||||
|
||||
use_case.execute("user_1", skip=5, limit=10)
|
||||
|
||||
mock_repo.list_by_user.assert_called_once_with("user_1", skip=5, limit=10)
|
||||
|
||||
def test_empty_list(self, mock_repo):
|
||||
"""空列表"""
|
||||
mock_repo.list_by_user.return_value = []
|
||||
use_case = ListRecipesUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("user_1")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
class TestGetRecipeUseCase:
|
||||
"""GetRecipeUseCase 测试"""
|
||||
|
||||
def test_get_existing(self, mock_repo):
|
||||
"""获取存在的配方"""
|
||||
recipe = _make_recipe("r1", "配方1", item_count=3)
|
||||
mock_repo.get.return_value = recipe
|
||||
use_case = GetRecipeUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("r1", "user_1")
|
||||
|
||||
assert result is not None
|
||||
assert result.id == "r1"
|
||||
assert len(result.items) == 3
|
||||
mock_repo.get.assert_called_once_with("r1", "user_1")
|
||||
|
||||
def test_get_nonexistent_returns_none(self, mock_repo):
|
||||
"""获取不存在的配方返回 None"""
|
||||
mock_repo.get.return_value = None
|
||||
use_case = GetRecipeUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("noexist", "user_1")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestCreateRecipeUseCase:
|
||||
@pytest.fixture
|
||||
def mock_repo(self):
|
||||
repo = Mock()
|
||||
repo.create = Mock(side_effect=lambda r: r)
|
||||
repo.create_items = Mock(side_effect=lambda items: items)
|
||||
return repo
|
||||
"""CreateRecipeUseCase 测试"""
|
||||
|
||||
def test_create_basic(self, mock_repo):
|
||||
uc = CreateRecipeUseCase(mock_repo)
|
||||
cmd = CreateRecipeCommand(
|
||||
user_id="user001",
|
||||
name="我的配方",
|
||||
description="desc",
|
||||
template_id="tpl001",
|
||||
generation_params={"mode": "one_take"},
|
||||
def test_create_without_items(self, mock_repo):
|
||||
"""创建不带items的配方"""
|
||||
mock_repo.create.side_effect = lambda x: x
|
||||
use_case = CreateRecipeUseCase(mock_repo)
|
||||
|
||||
command = CreateRecipeCommand(
|
||||
user_id="user_1",
|
||||
name="新配方",
|
||||
description="测试",
|
||||
template_id="tmpl_1",
|
||||
generation_params={"key": "value"},
|
||||
items=[],
|
||||
)
|
||||
result = uc.execute(cmd)
|
||||
assert result.name == "我的配方"
|
||||
assert result.user_id == "user001"
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert isinstance(result, Recipe)
|
||||
assert result.name == "新配方"
|
||||
assert result.user_id == "user_1"
|
||||
assert result.template_id == "tmpl_1"
|
||||
assert result.generation_params == {"key": "value"}
|
||||
assert result.items == []
|
||||
mock_repo.create.assert_called_once()
|
||||
mock_repo.create_items.assert_not_called()
|
||||
|
||||
def test_create_with_items(self, mock_repo):
|
||||
uc = CreateRecipeUseCase(mock_repo)
|
||||
cmd = CreateRecipeCommand(
|
||||
user_id="user001",
|
||||
"""创建带items的配方"""
|
||||
mock_repo.create.side_effect = lambda x: x
|
||||
mock_repo.create_items.side_effect = lambda items: items
|
||||
use_case = CreateRecipeUseCase(mock_repo)
|
||||
|
||||
command = CreateRecipeCommand(
|
||||
user_id="user_1",
|
||||
name="带素材配方",
|
||||
items=[
|
||||
RecipeItemCommand(item_type="asset", item_id="a1", position=0),
|
||||
@@ -89,123 +156,250 @@ class TestCreateRecipeUseCase:
|
||||
RecipeItemCommand(item_type="voice", item_id="v1", position=2),
|
||||
],
|
||||
)
|
||||
result = uc.execute(cmd)
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert len(result.items) == 3
|
||||
assert result.items[0].item_type == "asset"
|
||||
assert result.items[1].item_type == "title"
|
||||
assert result.items[2].item_type == "voice"
|
||||
mock_repo.create.assert_called_once()
|
||||
mock_repo.create_items.assert_called_once()
|
||||
items_arg = mock_repo.create_items.call_args[0][0]
|
||||
assert items_arg[0].item_type == "asset"
|
||||
assert items_arg[1].item_type == "title"
|
||||
assert items_arg[2].item_type == "voice"
|
||||
created_items = mock_repo.create_items.call_args[0][0]
|
||||
assert len(created_items) == 3
|
||||
|
||||
def test_create_with_default_values(self, mock_repo):
|
||||
"""使用默认值创建"""
|
||||
mock_repo.create.side_effect = lambda x: x
|
||||
use_case = CreateRecipeUseCase(mock_repo)
|
||||
|
||||
class TestListRecipesUseCase:
|
||||
def test_list(self):
|
||||
repo = Mock()
|
||||
repo.list_by_user = Mock(return_value=[_make_recipe()])
|
||||
uc = ListRecipesUseCase(repo)
|
||||
result = uc.execute("user001", skip=0, limit=10)
|
||||
assert len(result) == 1
|
||||
repo.list_by_user.assert_called_once_with("user001", skip=0, limit=10)
|
||||
command = CreateRecipeCommand(user_id="user_1", name="极简配方")
|
||||
result = use_case.execute(command)
|
||||
|
||||
|
||||
class TestGetRecipeUseCase:
|
||||
def test_get_found(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=_make_recipe())
|
||||
uc = GetRecipeUseCase(repo)
|
||||
result = uc.execute("recipe001", "user001")
|
||||
assert result is not None
|
||||
assert result.id == "recipe001"
|
||||
|
||||
def test_get_not_found(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=None)
|
||||
uc = GetRecipeUseCase(repo)
|
||||
result = uc.execute("recipe999", "user001")
|
||||
assert result is None
|
||||
assert result.description == ""
|
||||
assert result.template_id == ""
|
||||
assert result.generation_params == {}
|
||||
assert result.items == []
|
||||
assert result.metadata_ == {}
|
||||
|
||||
|
||||
class TestUpdateRecipeUseCase:
|
||||
@pytest.fixture
|
||||
def mock_repo(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=_make_recipe())
|
||||
repo.update = Mock(side_effect=lambda r: r)
|
||||
repo.list_items = Mock(return_value=[])
|
||||
repo.delete_items_by_recipe = Mock(return_value=0)
|
||||
repo.create_items = Mock(side_effect=lambda items: items)
|
||||
return repo
|
||||
"""UpdateRecipeUseCase 测试"""
|
||||
|
||||
def test_update_name(self, mock_repo):
|
||||
uc = UpdateRecipeUseCase(mock_repo)
|
||||
cmd = UpdateRecipeCommand(
|
||||
recipe_id="recipe001",
|
||||
user_id="user001",
|
||||
name="新名字",
|
||||
)
|
||||
result = uc.execute(cmd)
|
||||
assert result.name == "新名字"
|
||||
"""更新配方名称"""
|
||||
recipe = _make_recipe("r1", "旧名称")
|
||||
mock_repo.get.return_value = recipe
|
||||
mock_repo.update.side_effect = lambda x: x
|
||||
mock_repo.list_items.return_value = []
|
||||
use_case = UpdateRecipeUseCase(mock_repo)
|
||||
|
||||
def test_update_not_found(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=None)
|
||||
uc = UpdateRecipeUseCase(repo)
|
||||
cmd = UpdateRecipeCommand(recipe_id="xxx", user_id="user001", name="x")
|
||||
with pytest.raises(NotFoundError):
|
||||
uc.execute(cmd)
|
||||
command = UpdateRecipeCommand(recipe_id="r1", user_id="user_1", name="新名称")
|
||||
result = use_case.execute(command)
|
||||
|
||||
def test_update_replace_items(self, mock_repo):
|
||||
uc = UpdateRecipeUseCase(mock_repo)
|
||||
cmd = UpdateRecipeCommand(
|
||||
recipe_id="recipe001",
|
||||
user_id="user001",
|
||||
items=[RecipeItemCommand(item_type="voice", item_id="v2", position=0)],
|
||||
assert result.name == "新名称"
|
||||
# 其他不变
|
||||
assert result.description == "测试配方"
|
||||
assert result.template_id == "tmpl_1"
|
||||
mock_repo.get.assert_called_once_with("r1", "user_1")
|
||||
mock_repo.update.assert_called_once()
|
||||
# 没传items时从repository加载
|
||||
mock_repo.list_items.assert_called_once_with("r1")
|
||||
|
||||
def test_update_multiple_fields(self, mock_repo):
|
||||
"""同时更新多个字段"""
|
||||
recipe = _make_recipe("r1", "旧")
|
||||
mock_repo.get.return_value = recipe
|
||||
mock_repo.update.side_effect = lambda x: x
|
||||
mock_repo.list_items.return_value = []
|
||||
use_case = UpdateRecipeUseCase(mock_repo)
|
||||
|
||||
command = UpdateRecipeCommand(
|
||||
recipe_id="r1",
|
||||
user_id="user_1",
|
||||
description="新描述",
|
||||
template_id="tmpl_new",
|
||||
generation_params={"new": "params"},
|
||||
)
|
||||
result = uc.execute(cmd)
|
||||
mock_repo.delete_items_by_recipe.assert_called_once_with("recipe001")
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result.description == "新描述"
|
||||
assert result.template_id == "tmpl_new"
|
||||
assert result.generation_params == {"new": "params"}
|
||||
|
||||
def test_update_items_replaces_old(self, mock_repo):
|
||||
"""更新items时删除旧的并创建新的"""
|
||||
recipe = _make_recipe("r1", "配方", item_count=2)
|
||||
mock_repo.get.return_value = recipe
|
||||
mock_repo.update.side_effect = lambda x: x
|
||||
mock_repo.create_items.side_effect = lambda items: items
|
||||
use_case = UpdateRecipeUseCase(mock_repo)
|
||||
|
||||
command = UpdateRecipeCommand(
|
||||
recipe_id="r1",
|
||||
user_id="user_1",
|
||||
items=[
|
||||
RecipeItemCommand(item_type="asset", item_id="new_a", position=0),
|
||||
RecipeItemCommand(item_type="title", item_id="new_t", position=1),
|
||||
],
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
mock_repo.delete_items_by_recipe.assert_called_once_with("r1")
|
||||
mock_repo.create_items.assert_called_once()
|
||||
assert len(result.items) == 1
|
||||
assert len(result.items) == 2
|
||||
assert result.items[0].item_id == "new_a"
|
||||
|
||||
def test_update_empty_items_list(self, mock_repo):
|
||||
"""更新为空items列表也会替换"""
|
||||
recipe = _make_recipe("r1", "配方", item_count=3)
|
||||
mock_repo.get.return_value = recipe
|
||||
mock_repo.update.side_effect = lambda x: x
|
||||
mock_repo.create_items.return_value = []
|
||||
use_case = UpdateRecipeUseCase(mock_repo)
|
||||
|
||||
command = UpdateRecipeCommand(
|
||||
recipe_id="r1", user_id="user_1", items=[]
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
mock_repo.delete_items_by_recipe.assert_called_once()
|
||||
mock_repo.create_items.assert_called_once_with([])
|
||||
assert result.items == []
|
||||
|
||||
def test_update_nonexistent_raises(self, mock_repo):
|
||||
"""更新不存在的配方抛出 NotFoundError"""
|
||||
mock_repo.get.return_value = None
|
||||
use_case = UpdateRecipeUseCase(mock_repo)
|
||||
|
||||
command = UpdateRecipeCommand(
|
||||
recipe_id="noexist", user_id="user_1", name="新名称"
|
||||
)
|
||||
with pytest.raises(NotFoundError, match="not found"):
|
||||
use_case.execute(command)
|
||||
|
||||
mock_repo.update.assert_not_called()
|
||||
|
||||
|
||||
class TestDeleteRecipeUseCase:
|
||||
def test_delete_success(self):
|
||||
repo = Mock()
|
||||
repo.delete = Mock(return_value=True)
|
||||
uc = DeleteRecipeUseCase(repo)
|
||||
assert uc.execute("recipe001", "user001") is True
|
||||
"""DeleteRecipeUseCase 测试"""
|
||||
|
||||
def test_delete_not_found(self):
|
||||
repo = Mock()
|
||||
repo.delete = Mock(return_value=False)
|
||||
uc = DeleteRecipeUseCase(repo)
|
||||
assert uc.execute("recipe999", "user001") is False
|
||||
def test_delete_success(self, mock_repo):
|
||||
"""删除成功"""
|
||||
mock_repo.delete.return_value = True
|
||||
use_case = DeleteRecipeUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("r1", "user_1")
|
||||
|
||||
assert result is True
|
||||
mock_repo.delete.assert_called_once_with("r1", "user_1")
|
||||
|
||||
def test_delete_nonexistent_returns_false(self, mock_repo):
|
||||
"""删除不存在的返回 False"""
|
||||
mock_repo.delete.return_value = False
|
||||
use_case = DeleteRecipeUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("noexist", "user_1")
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
class TestUseRecipeUseCase:
|
||||
def test_use_success_basic_plan(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=_make_recipe())
|
||||
uc = UseRecipeUseCase(repo)
|
||||
result = uc.execute("recipe001", "user001", user_plan="basic")
|
||||
assert result.recipe.id == "recipe001"
|
||||
assert result.warnings == []
|
||||
"""UseRecipeUseCase 使用配方测试"""
|
||||
|
||||
def test_use_success_premium_plan(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=_make_recipe())
|
||||
uc = UseRecipeUseCase(repo)
|
||||
result = uc.execute("recipe001", "user001", user_plan="premium")
|
||||
assert result.recipe.id == "recipe001"
|
||||
def test_use_recipe_premium_enabled(self, mock_repo):
|
||||
"""premium用户可以使用配方"""
|
||||
recipe = _make_recipe("r1", "配方1", item_count=2)
|
||||
mock_repo.get.return_value = recipe
|
||||
use_case = UseRecipeUseCase(mock_repo)
|
||||
|
||||
def test_use_free_plan_forbidden(self):
|
||||
repo = Mock()
|
||||
uc = UseRecipeUseCase(repo)
|
||||
with pytest.raises(FeatureDisabledError):
|
||||
uc.execute("recipe001", "user001", user_plan="free")
|
||||
# 用 patch mock feature_flags
|
||||
with patch("packages.application.recipe.use_cases.feature_flags") as mock_ff:
|
||||
mock_ff.is_enabled.return_value = True
|
||||
result = use_case.execute("r1", "user_1", user_plan="premium")
|
||||
|
||||
def test_use_not_found(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=None)
|
||||
uc = UseRecipeUseCase(repo)
|
||||
with pytest.raises(NotFoundError):
|
||||
uc.execute("recipe999", "user001", user_plan="basic")
|
||||
assert isinstance(result, UseRecipeResult)
|
||||
assert result.recipe.id == "r1"
|
||||
assert isinstance(result.warnings, list)
|
||||
mock_repo.get.assert_called_once_with("r1", "user_1")
|
||||
|
||||
def test_use_recipe_basic_enabled(self, mock_repo):
|
||||
"""basic用户可以使用配方"""
|
||||
recipe = _make_recipe("r1", "配方1")
|
||||
mock_repo.get.return_value = recipe
|
||||
use_case = UseRecipeUseCase(mock_repo)
|
||||
|
||||
with patch("packages.application.recipe.use_cases.feature_flags") as mock_ff:
|
||||
mock_ff.is_enabled.return_value = True
|
||||
result = use_case.execute("r1", "user_1", user_plan="basic")
|
||||
|
||||
assert result.recipe.id == "r1"
|
||||
|
||||
def test_use_recipe_feature_disabled(self, mock_repo):
|
||||
"""功能未启用时抛出 FeatureDisabledError"""
|
||||
use_case = UseRecipeUseCase(mock_repo)
|
||||
|
||||
with patch("packages.application.recipe.use_cases.feature_flags") as mock_ff:
|
||||
mock_ff.is_enabled.return_value = False
|
||||
with pytest.raises(FeatureDisabledError, match="仅对基础版和高级版"):
|
||||
use_case.execute("r1", "user_1", user_plan="free")
|
||||
|
||||
mock_repo.get.assert_not_called()
|
||||
|
||||
def test_use_recipe_not_found(self, mock_repo):
|
||||
"""配方不存在时抛出 NotFoundError"""
|
||||
mock_repo.get.return_value = None
|
||||
use_case = UseRecipeUseCase(mock_repo)
|
||||
|
||||
with patch("packages.application.recipe.use_cases.feature_flags") as mock_ff:
|
||||
mock_ff.is_enabled.return_value = True
|
||||
with pytest.raises(NotFoundError, match="not found"):
|
||||
use_case.execute("noexist", "user_1", user_plan="premium")
|
||||
|
||||
|
||||
class TestRecipeCommands:
|
||||
"""命令数据类测试"""
|
||||
|
||||
def test_create_recipe_command_fields(self):
|
||||
"""CreateRecipeCommand 字段"""
|
||||
cmd = CreateRecipeCommand(
|
||||
user_id="u1",
|
||||
name="测试",
|
||||
description="desc",
|
||||
template_id="t1",
|
||||
generation_params={"a": 1},
|
||||
items=[RecipeItemCommand(item_type="asset", item_id="a1", position=0)],
|
||||
metadata_={"key": "val"},
|
||||
)
|
||||
assert cmd.user_id == "u1"
|
||||
assert cmd.name == "测试"
|
||||
assert cmd.description == "desc"
|
||||
assert cmd.template_id == "t1"
|
||||
assert cmd.generation_params == {"a": 1}
|
||||
assert len(cmd.items) == 1
|
||||
assert cmd.items[0].item_type == "asset"
|
||||
assert cmd.metadata_ == {"key": "val"}
|
||||
|
||||
def test_recipe_item_command_defaults(self):
|
||||
"""RecipeItemCommand 默认值"""
|
||||
cmd = RecipeItemCommand(item_type="asset", item_id="a1")
|
||||
assert cmd.position == 0
|
||||
assert cmd.metadata_ == {}
|
||||
|
||||
def test_update_recipe_command_defaults_none(self):
|
||||
"""UpdateRecipeCommand 字段默认None"""
|
||||
cmd = UpdateRecipeCommand(recipe_id="r1", user_id="u1")
|
||||
assert cmd.name is None
|
||||
assert cmd.description is None
|
||||
assert cmd.template_id is None
|
||||
assert cmd.generation_params is None
|
||||
assert cmd.items is None
|
||||
assert cmd.metadata_ is None
|
||||
|
||||
def test_commands_are_dataclasses(self):
|
||||
"""都是 dataclass"""
|
||||
from dataclasses import is_dataclass
|
||||
|
||||
assert is_dataclass(CreateRecipeCommand)
|
||||
assert is_dataclass(UpdateRecipeCommand)
|
||||
assert is_dataclass(RecipeItemCommand)
|
||||
assert is_dataclass(UseRecipeResult)
|
||||
|
||||
Regular → Executable
+359
-419
@@ -1,492 +1,432 @@
|
||||
"""
|
||||
标题库（Title Library）Use Case 回归测试
|
||||
"""标题库 UseCase 单元测试."""
|
||||
|
||||
测试目标：
|
||||
1. CreateTitleLibraryUseCase - 创建标题库条目
|
||||
2. UpdateTitleLibraryUseCase - 更新标题库条目
|
||||
3. 配额逻辑覆盖 - titles: free=50, basic=500, premium=500
|
||||
4. 边界条件与异常场景
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import Mock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.application.title_library.commands import (
|
||||
CreateTitleLibraryCommand,
|
||||
IncrementTitleUsageCommand,
|
||||
PickTitleCommand,
|
||||
UpdateTitleLibraryCommand,
|
||||
)
|
||||
from packages.application.title_library.use_cases import (
|
||||
CreateTitleLibraryUseCase,
|
||||
DeleteTitleLibraryUseCase,
|
||||
GetTitleLibraryUseCase,
|
||||
IncrementTitleUsageUseCase,
|
||||
ListTitleLibraryUseCase,
|
||||
NotFoundError,
|
||||
QuotaExceededError,
|
||||
PickTitleUseCase,
|
||||
UpdateTitleLibraryUseCase,
|
||||
)
|
||||
from packages.domain.exceptions import NotFoundError, QuotaExceededError
|
||||
from packages.domain.title_library import TitleLibraryItem
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_repo():
|
||||
"""创建 Mock 仓储"""
|
||||
repo = Mock()
|
||||
repo.count_by_user = Mock(return_value=0)
|
||||
repo.create = Mock(side_effect=lambda item: item)
|
||||
repo.update = Mock(side_effect=lambda item: item)
|
||||
repo.get = Mock(return_value=None)
|
||||
repo.delete = Mock(return_value=True)
|
||||
repo.list_by_user = Mock(return_value=[])
|
||||
return repo
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def create_use_case(mock_repo):
|
||||
return CreateTitleLibraryUseCase(repository=mock_repo)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def update_use_case(mock_repo):
|
||||
return UpdateTitleLibraryUseCase(repository=mock_repo)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_create_command():
|
||||
"""标准创建命令"""
|
||||
return CreateTitleLibraryCommand(
|
||||
user_id="user-001",
|
||||
name="测试标题",
|
||||
text="这是一个测试标题文本",
|
||||
category="新闻",
|
||||
description="用于测试的标题",
|
||||
tags=["测试", "新闻"],
|
||||
metadata_={"source": "unit_test"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def existing_title_item():
|
||||
"""模拟已存在的标题条目"""
|
||||
def _make_item(id: str, name: str, text: str, usage_count: int = 0, category: str = "default") -> TitleLibraryItem:
|
||||
return TitleLibraryItem(
|
||||
id="existing-title-001",
|
||||
user_id="user-001",
|
||||
name="旧标题",
|
||||
text="旧文本",
|
||||
category="旧分类",
|
||||
description="旧描述",
|
||||
tags=["旧"],
|
||||
id=id,
|
||||
user_id="user_1",
|
||||
name=name,
|
||||
text=text,
|
||||
category=category,
|
||||
description="",
|
||||
tags=[],
|
||||
usage_count=usage_count,
|
||||
is_active=True,
|
||||
metadata_={},
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 1. CreateTitleLibraryUseCase 测试
|
||||
# ===========================================================================
|
||||
@pytest.fixture
|
||||
def mock_repo():
|
||||
return MagicMock()
|
||||
|
||||
|
||||
class TestCreateTitleLibraryUseCase:
|
||||
"""标题库创建 UseCase 测试"""
|
||||
@pytest.fixture
|
||||
def sample_item():
|
||||
return _make_item("title_1", "爆款标题", "这是一个爆款标题文案", usage_count=5)
|
||||
|
||||
def test_create_success_all_fields(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""测试创建成功 - 所有字段完整传入"""
|
||||
result = create_use_case.execute(sample_create_command, plan_name="free")
|
||||
|
||||
assert result is not None
|
||||
assert result.user_id == "user-001"
|
||||
assert result.name == "测试标题"
|
||||
assert result.text == "这是一个测试标题文本"
|
||||
assert result.category == "新闻"
|
||||
assert result.description == "用于测试的标题"
|
||||
assert result.tags == ["测试", "新闻"]
|
||||
assert result.metadata_ == {"source": "unit_test"}
|
||||
class TestListTitleLibraryUseCase:
|
||||
"""ListTitleLibraryUseCase 测试"""
|
||||
|
||||
mock_repo.count_by_user.assert_called_once_with("user-001")
|
||||
mock_repo.create.assert_called_once()
|
||||
def test_list_returns_results(self, mock_repo, sample_item):
|
||||
"""正常返回标题列表"""
|
||||
mock_repo.list_by_user.return_value = [sample_item]
|
||||
use_case = ListTitleLibraryUseCase(mock_repo)
|
||||
|
||||
def test_create_generates_uuid(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""测试创建时自动生成 UUID 作为 id"""
|
||||
result = create_use_case.execute(sample_create_command, plan_name="free")
|
||||
result = use_case.execute("user_1")
|
||||
|
||||
assert result.id is not None
|
||||
assert len(result.id) == 32 # uuid4().hex 长度为 32
|
||||
assert result.id.isalnum()
|
||||
|
||||
def test_create_default_values(self, create_use_case, mock_repo):
|
||||
"""测试默认值填充"""
|
||||
command = CreateTitleLibraryCommand(
|
||||
user_id="user-001",
|
||||
name="最小化创建",
|
||||
text="文本",
|
||||
assert len(result) == 1
|
||||
assert result[0].id == "title_1"
|
||||
mock_repo.list_by_user.assert_called_once_with(
|
||||
"user_1", category=None, skip=0, limit=50
|
||||
)
|
||||
|
||||
result = create_use_case.execute(command, plan_name="free")
|
||||
def test_list_with_category(self, mock_repo, sample_item):
|
||||
"""按分类过滤"""
|
||||
mock_repo.list_by_user.return_value = [sample_item]
|
||||
use_case = ListTitleLibraryUseCase(mock_repo)
|
||||
|
||||
assert result.category == "default"
|
||||
assert result.description == ""
|
||||
assert result.tags == []
|
||||
assert result.metadata_ == {}
|
||||
use_case.execute("user_1", category="电商")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 2. 配额逻辑测试（titles: free=50, basic=500, premium=500）
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestCreateTitleLibraryQuota:
|
||||
"""标题库创建配额检查测试"""
|
||||
|
||||
def test_quota_free_plan_under_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""free 套餐（上限50），当前 25 个，允许创建"""
|
||||
mock_repo.count_by_user.return_value = 25
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name="free")
|
||||
|
||||
assert result is not None
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_quota_free_plan_at_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""free 套餐（上限50），当前 50 个，拒绝创建"""
|
||||
mock_repo.count_by_user.return_value = 50
|
||||
|
||||
with pytest.raises(QuotaExceededError) as exc_info:
|
||||
create_use_case.execute(sample_create_command, plan_name="free")
|
||||
|
||||
assert exc_info.value.dimension == "max_titles"
|
||||
assert exc_info.value.limit == 50
|
||||
assert exc_info.value.used == 50
|
||||
mock_repo.create.assert_not_called()
|
||||
|
||||
def test_quota_free_plan_just_under_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""free 套餐（上限50），当前 49 个，允许创建（边界）"""
|
||||
mock_repo.count_by_user.return_value = 49
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name="free")
|
||||
|
||||
assert result is not None
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_quota_free_plan_over_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""free 套餐（上限50），当前 60 个，拒绝创建"""
|
||||
mock_repo.count_by_user.return_value = 60
|
||||
|
||||
with pytest.raises(QuotaExceededError) as exc_info:
|
||||
create_use_case.execute(sample_create_command, plan_name="free")
|
||||
|
||||
assert exc_info.value.dimension == "max_titles"
|
||||
assert exc_info.value.limit == 50
|
||||
assert exc_info.value.used == 60
|
||||
|
||||
def test_quota_basic_plan_under_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""basic 套餐（上限500），当前 200 个，允许创建"""
|
||||
mock_repo.count_by_user.return_value = 200
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name="basic")
|
||||
|
||||
assert result is not None
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_quota_basic_plan_at_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""basic 套餐（上限500），当前 500 个，拒绝创建"""
|
||||
mock_repo.count_by_user.return_value = 500
|
||||
|
||||
with pytest.raises(QuotaExceededError) as exc_info:
|
||||
create_use_case.execute(sample_create_command, plan_name="basic")
|
||||
|
||||
assert exc_info.value.dimension == "max_titles"
|
||||
assert exc_info.value.limit == 500
|
||||
assert exc_info.value.used == 500
|
||||
|
||||
def test_quota_basic_plan_just_under_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""basic 套餐（上限500），当前 499 个，允许创建（边界）"""
|
||||
mock_repo.count_by_user.return_value = 499
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name="basic")
|
||||
|
||||
assert result is not None
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_quota_premium_plan_under_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""premium 套餐（上限500），当前 250 个，允许创建"""
|
||||
mock_repo.count_by_user.return_value = 250
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name="premium")
|
||||
|
||||
assert result is not None
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_quota_premium_plan_at_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""premium 套餐（上限500），当前 500 个，拒绝创建"""
|
||||
mock_repo.count_by_user.return_value = 500
|
||||
|
||||
with pytest.raises(QuotaExceededError) as exc_info:
|
||||
create_use_case.execute(sample_create_command, plan_name="premium")
|
||||
|
||||
assert exc_info.value.dimension == "max_titles"
|
||||
assert exc_info.value.limit == 500
|
||||
assert exc_info.value.used == 500
|
||||
|
||||
def test_quota_premium_plan_just_under_limit(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""premium 套餐（上限500），当前 499 个，允许创建（边界）"""
|
||||
mock_repo.count_by_user.return_value = 499
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name="premium")
|
||||
|
||||
assert result is not None
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_quota_zero_usage_all_plans(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""新用户零使用量，所有套餐均可创建"""
|
||||
mock_repo.count_by_user.return_value = 0
|
||||
|
||||
for plan in ["free", "basic", "premium"]:
|
||||
mock_repo.create.reset_mock()
|
||||
mock_repo.count_by_user.reset_mock()
|
||||
mock_repo.count_by_user.return_value = 0
|
||||
|
||||
result = create_use_case.execute(sample_create_command, plan_name=plan)
|
||||
assert result is not None, f"{plan} 套餐零使用量应允许创建"
|
||||
|
||||
def test_quota_unknown_plan_defaults_to_zero(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""未知套餐名默认配额为 0，无法创建"""
|
||||
mock_repo.count_by_user.return_value = 0
|
||||
|
||||
with pytest.raises(QuotaExceededError):
|
||||
create_use_case.execute(sample_create_command, plan_name="unknown_plan")
|
||||
|
||||
def test_quota_exceeded_error_attributes(self, create_use_case, mock_repo, sample_create_command):
|
||||
"""QuotaExceededError 异常属性完整性"""
|
||||
mock_repo.count_by_user.return_value = 50
|
||||
|
||||
with pytest.raises(QuotaExceededError) as exc_info:
|
||||
create_use_case.execute(sample_create_command, plan_name="free")
|
||||
|
||||
err = exc_info.value
|
||||
assert hasattr(err, "dimension")
|
||||
assert hasattr(err, "limit")
|
||||
assert hasattr(err, "used")
|
||||
assert "max_titles" in str(err)
|
||||
assert "50" in str(err)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 3. UpdateTitleLibraryUseCase 测试
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestUpdateTitleLibraryUseCase:
|
||||
"""标题库更新 UseCase 测试"""
|
||||
|
||||
def test_update_success_all_fields(self, update_use_case, mock_repo, existing_title_item):
|
||||
"""测试全字段更新成功"""
|
||||
mock_repo.get.return_value = existing_title_item
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="existing-title-001",
|
||||
user_id="user-001",
|
||||
name="更新后标题",
|
||||
text="更新后文本",
|
||||
category="新分类",
|
||||
description="新描述",
|
||||
tags=["新标签"],
|
||||
is_active=False,
|
||||
metadata_={"updated": True},
|
||||
mock_repo.list_by_user.assert_called_once_with(
|
||||
"user_1", category="电商", skip=0, limit=50
|
||||
)
|
||||
|
||||
result = update_use_case.execute(command)
|
||||
def test_list_with_pagination(self, mock_repo, sample_item):
|
||||
"""带分页参数"""
|
||||
mock_repo.list_by_user.return_value = [sample_item]
|
||||
use_case = ListTitleLibraryUseCase(mock_repo)
|
||||
|
||||
assert result.name == "更新后标题"
|
||||
assert result.text == "更新后文本"
|
||||
assert result.category == "新分类"
|
||||
assert result.description == "新描述"
|
||||
assert result.tags == ["新标签"]
|
||||
assert result.is_active is False
|
||||
assert result.metadata_ == {"updated": True}
|
||||
use_case.execute("user_1", skip=10, limit=20)
|
||||
|
||||
mock_repo.update.assert_called_once()
|
||||
|
||||
def test_update_partial_only_name(self, update_use_case, mock_repo, existing_title_item):
|
||||
"""测试仅更新 name"""
|
||||
mock_repo.get.return_value = existing_title_item
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="existing-title-001",
|
||||
user_id="user-001",
|
||||
name="仅改名",
|
||||
mock_repo.list_by_user.assert_called_once_with(
|
||||
"user_1", category=None, skip=10, limit=20
|
||||
)
|
||||
|
||||
result = update_use_case.execute(command)
|
||||
def test_empty_list(self, mock_repo):
|
||||
"""空列表"""
|
||||
mock_repo.list_by_user.return_value = []
|
||||
use_case = ListTitleLibraryUseCase(mock_repo)
|
||||
|
||||
assert result.name == "仅改名"
|
||||
# 其他字段保持不变
|
||||
assert result.text == "旧文本"
|
||||
assert result.category == "旧分类"
|
||||
assert result.description == "旧描述"
|
||||
result = use_case.execute("user_1")
|
||||
|
||||
def test_update_partial_only_is_active(self, update_use_case, mock_repo, existing_title_item):
|
||||
"""测试仅更新 is_active（软删除/恢复）"""
|
||||
mock_repo.get.return_value = existing_title_item
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="existing-title-001",
|
||||
user_id="user-001",
|
||||
is_active=False,
|
||||
)
|
||||
|
||||
result = update_use_case.execute(command)
|
||||
|
||||
assert result.is_active is False
|
||||
assert result.name == "旧标题" # 其他字段不变
|
||||
|
||||
def test_update_not_found(self, update_use_case, mock_repo):
|
||||
"""测试更新不存在的条目"""
|
||||
mock_repo.get.return_value = None
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="nonexistent-id",
|
||||
user_id="user-001",
|
||||
name="不存在",
|
||||
)
|
||||
|
||||
with pytest.raises(NotFoundError, match="nonexistent-id"):
|
||||
update_use_case.execute(command)
|
||||
|
||||
mock_repo.update.assert_not_called()
|
||||
|
||||
def test_update_wrong_user(self, update_use_case, mock_repo):
|
||||
"""测试用户隔离"""
|
||||
mock_repo.get.return_value = None
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="existing-title-001",
|
||||
user_id="other-user-999",
|
||||
name="恶意修改",
|
||||
)
|
||||
|
||||
with pytest.raises(NotFoundError):
|
||||
update_use_case.execute(command)
|
||||
|
||||
def test_update_none_fields_not_changed(self, update_use_case, mock_repo, existing_title_item):
|
||||
"""测试 None 字段不覆盖原有值"""
|
||||
mock_repo.get.return_value = existing_title_item
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="existing-title-001",
|
||||
user_id="user-001",
|
||||
)
|
||||
|
||||
result = update_use_case.execute(command)
|
||||
|
||||
assert result.name == "旧标题"
|
||||
assert result.text == "旧文本"
|
||||
assert result.category == "旧分类"
|
||||
assert result.is_active is True
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 4. DeleteTitleLibraryUseCase 测试
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestDeleteTitleLibraryUseCase:
|
||||
"""标题库删除 UseCase 测试"""
|
||||
|
||||
def test_delete_success(self, mock_repo):
|
||||
"""测试删除成功"""
|
||||
mock_repo.delete.return_value = True
|
||||
use_case = DeleteTitleLibraryUseCase(repository=mock_repo)
|
||||
|
||||
result = use_case.execute("title-001", "user-001")
|
||||
|
||||
assert result is True
|
||||
mock_repo.delete.assert_called_once_with("title-001", "user-001")
|
||||
|
||||
def test_delete_not_found(self, mock_repo):
|
||||
"""测试删除不存在的条目"""
|
||||
mock_repo.delete.return_value = False
|
||||
use_case = DeleteTitleLibraryUseCase(repository=mock_repo)
|
||||
|
||||
result = use_case.execute("nonexistent", "user-001")
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 5. GetTitleLibraryUseCase 测试
|
||||
# ===========================================================================
|
||||
assert result == []
|
||||
|
||||
|
||||
class TestGetTitleLibraryUseCase:
|
||||
"""标题库查询 UseCase 测试"""
|
||||
"""GetTitleLibraryUseCase 测试"""
|
||||
|
||||
def test_get_existing(self, mock_repo):
|
||||
"""测试查询存在的条目"""
|
||||
expected = TitleLibraryItem(
|
||||
id="t-001",
|
||||
user_id="user-001",
|
||||
name="测试",
|
||||
text="文本",
|
||||
)
|
||||
mock_repo.get.return_value = expected
|
||||
use_case = GetTitleLibraryUseCase(repository=mock_repo)
|
||||
def test_get_existing(self, mock_repo, sample_item):
|
||||
"""获取存在的标题"""
|
||||
mock_repo.get.return_value = sample_item
|
||||
use_case = GetTitleLibraryUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("t-001", "user-001")
|
||||
result = use_case.execute("title_1", "user_1")
|
||||
|
||||
assert result is not None
|
||||
assert result.id == "t-001"
|
||||
mock_repo.get.assert_called_once_with("t-001", "user-001")
|
||||
assert result.id == "title_1"
|
||||
mock_repo.get.assert_called_once_with("title_1", "user_1")
|
||||
|
||||
def test_get_not_found(self, mock_repo):
|
||||
"""测试查询不存在的条目"""
|
||||
def test_get_nonexistent_returns_none(self, mock_repo):
|
||||
"""获取不存在的标题返回 None"""
|
||||
mock_repo.get.return_value = None
|
||||
use_case = GetTitleLibraryUseCase(repository=mock_repo)
|
||||
use_case = GetTitleLibraryUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("nonexistent", "user-001")
|
||||
result = use_case.execute("nonexistent", "user_1")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 6. ListTitleLibraryUseCase 测试
|
||||
# ===========================================================================
|
||||
class TestCreateTitleLibraryUseCase:
|
||||
"""CreateTitleLibraryUseCase 测试"""
|
||||
|
||||
def test_create_success(self, mock_repo, sample_item):
|
||||
"""创建成功"""
|
||||
mock_repo.count_by_user.return_value = 0
|
||||
mock_repo.create.return_value = sample_item
|
||||
use_case = CreateTitleLibraryUseCase(mock_repo)
|
||||
|
||||
command = CreateTitleLibraryCommand(
|
||||
user_id="user_1",
|
||||
name="新标题",
|
||||
text="新标题文案",
|
||||
category="default",
|
||||
description="",
|
||||
tags=[],
|
||||
metadata_={},
|
||||
)
|
||||
result = use_case.execute(command, plan_name="free")
|
||||
|
||||
assert result.id == "title_1"
|
||||
mock_repo.count_by_user.assert_called_once_with("user_1")
|
||||
mock_repo.create.assert_called_once()
|
||||
|
||||
def test_create_quota_exceeded(self, mock_repo):
|
||||
"""超过配额时抛出 QuotaExceededError"""
|
||||
mock_repo.count_by_user.return_value = 9999
|
||||
use_case = CreateTitleLibraryUseCase(mock_repo)
|
||||
|
||||
command = CreateTitleLibraryCommand(
|
||||
user_id="user_1",
|
||||
name="新标题",
|
||||
text="文案",
|
||||
category="default",
|
||||
description="",
|
||||
tags=[],
|
||||
metadata_={},
|
||||
)
|
||||
with pytest.raises(QuotaExceededError):
|
||||
use_case.execute(command, plan_name="free")
|
||||
|
||||
mock_repo.create.assert_not_called()
|
||||
|
||||
def test_create_with_tags_and_metadata(self, mock_repo, sample_item):
|
||||
"""创建时带 tags 和 metadata_"""
|
||||
mock_repo.count_by_user.return_value = 0
|
||||
mock_repo.create.return_value = sample_item
|
||||
use_case = CreateTitleLibraryUseCase(mock_repo)
|
||||
|
||||
command = CreateTitleLibraryCommand(
|
||||
user_id="user_1",
|
||||
name="带标签标题",
|
||||
text="文案",
|
||||
category="电商",
|
||||
description="测试描述",
|
||||
tags=["爆款", "促销"],
|
||||
metadata_={"source": "manual"},
|
||||
)
|
||||
use_case.execute(command, plan_name="premium")
|
||||
|
||||
created = mock_repo.create.call_args[0][0]
|
||||
assert isinstance(created, TitleLibraryItem)
|
||||
assert created.name == "带标签标题"
|
||||
assert created.category == "电商"
|
||||
assert created.tags == ["爆款", "促销"]
|
||||
assert created.metadata_ == {"source": "manual"}
|
||||
|
||||
|
||||
class TestListTitleLibraryUseCase:
|
||||
"""标题库列表 UseCase 测试"""
|
||||
class TestUpdateTitleLibraryUseCase:
|
||||
"""UpdateTitleLibraryUseCase 测试"""
|
||||
|
||||
def test_list_default(self, mock_repo):
|
||||
"""测试默认列表查询"""
|
||||
def test_update_name(self, mock_repo, sample_item):
|
||||
"""更新标题名称"""
|
||||
mock_repo.get.return_value = sample_item
|
||||
mock_repo.update.side_effect = lambda x: x
|
||||
use_case = UpdateTitleLibraryUseCase(mock_repo)
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="title_1", user_id="user_1", name="新名称"
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result.name == "新名称"
|
||||
# 其他字段不变
|
||||
assert result.text == "这是一个爆款标题文案"
|
||||
mock_repo.get.assert_called_once_with("title_1", "user_1")
|
||||
mock_repo.update.assert_called_once()
|
||||
|
||||
def test_update_multiple_fields(self, mock_repo, sample_item):
|
||||
"""同时更新多个字段"""
|
||||
mock_repo.get.return_value = sample_item
|
||||
mock_repo.update.side_effect = lambda x: x
|
||||
use_case = UpdateTitleLibraryUseCase(mock_repo)
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="title_1",
|
||||
user_id="user_1",
|
||||
text="新文案内容",
|
||||
category="美食",
|
||||
is_active=False,
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result.text == "新文案内容"
|
||||
assert result.category == "美食"
|
||||
assert result.is_active is False
|
||||
|
||||
def test_update_nonexistent_raises(self, mock_repo):
|
||||
"""更新不存在的标题抛出 NotFoundError"""
|
||||
mock_repo.get.return_value = None
|
||||
use_case = UpdateTitleLibraryUseCase(mock_repo)
|
||||
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id="noexist", user_id="user_1", name="新名称"
|
||||
)
|
||||
with pytest.raises(NotFoundError, match="not found"):
|
||||
use_case.execute(command)
|
||||
|
||||
mock_repo.update.assert_not_called()
|
||||
|
||||
|
||||
class TestDeleteTitleLibraryUseCase:
|
||||
"""DeleteTitleLibraryUseCase 测试"""
|
||||
|
||||
def test_delete_success(self, mock_repo):
|
||||
"""删除成功"""
|
||||
mock_repo.delete.return_value = True
|
||||
use_case = DeleteTitleLibraryUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("title_1", "user_1")
|
||||
|
||||
assert result is True
|
||||
mock_repo.delete.assert_called_once_with("title_1", "user_1")
|
||||
|
||||
def test_delete_nonexistent_returns_false(self, mock_repo):
|
||||
"""删除不存在的返回 False"""
|
||||
mock_repo.delete.return_value = False
|
||||
use_case = DeleteTitleLibraryUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("noexist", "user_1")
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
class TestIncrementTitleUsageUseCase:
|
||||
"""IncrementTitleUsageUseCase 测试"""
|
||||
|
||||
def test_increment_positive(self, mock_repo):
|
||||
"""正增量时调用 repository"""
|
||||
mock_repo.increment_usage_count.return_value = True
|
||||
use_case = IncrementTitleUsageUseCase(mock_repo)
|
||||
|
||||
command = IncrementTitleUsageCommand(
|
||||
title_id="title_1", user_id="user_1", increment=1
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result is True
|
||||
mock_repo.increment_usage_count.assert_called_once_with(
|
||||
"title_1", "user_1", increment=1
|
||||
)
|
||||
|
||||
def test_increment_zero_returns_false(self, mock_repo):
|
||||
"""增量为0返回False，不调用repository"""
|
||||
use_case = IncrementTitleUsageUseCase(mock_repo)
|
||||
|
||||
command = IncrementTitleUsageCommand(
|
||||
title_id="title_1", user_id="user_1", increment=0
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result is False
|
||||
mock_repo.increment_usage_count.assert_not_called()
|
||||
|
||||
def test_increment_negative_returns_false(self, mock_repo):
|
||||
"""负增量返回False"""
|
||||
use_case = IncrementTitleUsageUseCase(mock_repo)
|
||||
|
||||
command = IncrementTitleUsageCommand(
|
||||
title_id="title_1", user_id="user_1", increment=-1
|
||||
)
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result is False
|
||||
mock_repo.increment_usage_count.assert_not_called()
|
||||
|
||||
def test_increment_large_number(self, mock_repo):
|
||||
"""大增量值"""
|
||||
mock_repo.increment_usage_count.return_value = True
|
||||
use_case = IncrementTitleUsageUseCase(mock_repo)
|
||||
|
||||
command = IncrementTitleUsageCommand(
|
||||
title_id="title_1", user_id="user_1", increment=10
|
||||
)
|
||||
use_case.execute(command)
|
||||
|
||||
mock_repo.increment_usage_count.assert_called_once_with(
|
||||
"title_1", "user_1", increment=10
|
||||
)
|
||||
|
||||
|
||||
class TestPickTitleUseCase:
|
||||
"""PickTitleUseCase 智能选标题测试"""
|
||||
|
||||
def test_pick_from_multiple(self, mock_repo):
|
||||
"""从多个标题中选一个（最少使用的前5个中随机）"""
|
||||
items = [
|
||||
TitleLibraryItem(id="t1", user_id="user-001", name="A", text="a"),
|
||||
TitleLibraryItem(id="t2", user_id="user-001", name="B", text="b"),
|
||||
_make_item(f"t{i}", f"标题{i}", f"文案{i}", usage_count=i)
|
||||
for i in range(10)
|
||||
]
|
||||
mock_repo.list_by_user.return_value = items
|
||||
use_case = ListTitleLibraryUseCase(repository=mock_repo)
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("user-001")
|
||||
command = PickTitleCommand(user_id="user_1")
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert len(result) == 2
|
||||
mock_repo.list_by_user.assert_called_once_with("user-001", category=None, skip=0, limit=50)
|
||||
assert result is not None
|
||||
assert isinstance(result, TitleLibraryItem)
|
||||
# 选出的应该是使用次数最少的前5个之一（0-4）
|
||||
assert result.usage_count <= 4
|
||||
mock_repo.list_by_user.assert_called_once()
|
||||
|
||||
def test_list_with_category_filter(self, mock_repo):
|
||||
"""测试按分类筛选"""
|
||||
def test_pick_empty_returns_none(self, mock_repo):
|
||||
"""空标题库返回 None"""
|
||||
mock_repo.list_by_user.return_value = []
|
||||
use_case = ListTitleLibraryUseCase(repository=mock_repo)
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
use_case.execute("user-001", category="新闻", skip=5, limit=10)
|
||||
command = PickTitleCommand(user_id="user_1")
|
||||
result = use_case.execute(command)
|
||||
|
||||
mock_repo.list_by_user.assert_called_once_with("user-001", category="新闻", skip=5, limit=10)
|
||||
assert result is None
|
||||
|
||||
def test_list_empty(self, mock_repo):
|
||||
"""测试空列表"""
|
||||
mock_repo.list_by_user.return_value = []
|
||||
use_case = ListTitleLibraryUseCase(repository=mock_repo)
|
||||
def test_pick_with_category(self, mock_repo):
|
||||
"""按分类选标题"""
|
||||
items = [_make_item("t1", "标题1", "文案1", category="美食")]
|
||||
mock_repo.list_by_user.return_value = items
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
result = use_case.execute("user-001")
|
||||
command = PickTitleCommand(user_id="user_1", category="美食")
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result == []
|
||||
assert result is not None
|
||||
call_kwargs = mock_repo.list_by_user.call_args[1]
|
||||
assert call_kwargs["category"] == "美食"
|
||||
assert call_kwargs["is_active"] is True
|
||||
|
||||
def test_pick_exclude_ids(self, mock_repo):
|
||||
"""排除指定ID"""
|
||||
items = [
|
||||
_make_item("t1", "标题1", "文案1", usage_count=1),
|
||||
_make_item("t2", "标题2", "文案2", usage_count=2),
|
||||
_make_item("t3", "标题3", "文案3", usage_count=3),
|
||||
]
|
||||
mock_repo.list_by_user.return_value = items
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
command = PickTitleCommand(user_id="user_1", exclude_ids=["t1", "t2"])
|
||||
result = use_case.execute(command)
|
||||
|
||||
# 排除两个后只剩t3
|
||||
assert result.id == "t3"
|
||||
|
||||
def test_pick_exclude_all_falls_back(self, mock_repo):
|
||||
"""排除全部时从所有标题中选"""
|
||||
items = [
|
||||
_make_item("t1", "标题1", "文案1", usage_count=1),
|
||||
_make_item("t2", "标题2", "文案2", usage_count=2),
|
||||
]
|
||||
mock_repo.list_by_user.return_value = items
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
command = PickTitleCommand(user_id="user_1", exclude_ids=["t1", "t2"])
|
||||
result = use_case.execute(command)
|
||||
|
||||
# 排除全部后fallback到全部，所以还是能选出一个
|
||||
assert result is not None
|
||||
assert result.id in ("t1", "t2")
|
||||
|
||||
def test_pick_single_item(self, mock_repo):
|
||||
"""只有一个标题时选它"""
|
||||
item = _make_item("only", "唯一标题", "唯一文案", usage_count=10)
|
||||
mock_repo.list_by_user.return_value = [item]
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
command = PickTitleCommand(user_id="user_1")
|
||||
result = use_case.execute(command)
|
||||
|
||||
assert result.id == "only"
|
||||
|
||||
def test_pick_prefers_less_used(self, mock_repo):
|
||||
"""倾向于选择使用次数少的"""
|
||||
items = [
|
||||
_make_item("t_used", "常用", "常用", usage_count=100),
|
||||
_make_item("t_fresh", "新的", "新的", usage_count=0),
|
||||
]
|
||||
mock_repo.list_by_user.return_value = items
|
||||
use_case = PickTitleUseCase(mock_repo)
|
||||
|
||||
# 跑多次，验证使用少的出现在候选池里
|
||||
results = set()
|
||||
for _ in range(20):
|
||||
command = PickTitleCommand(user_id="user_1")
|
||||
r = use_case.execute(command)
|
||||
if r:
|
||||
results.add(r.id)
|
||||
|
||||
# 两个都在候选池（少于5个），所以都可能被选中
|
||||
assert "t_used" in results or "t_fresh" in results
|
||||
|
||||
Reference in New Issue
Block a user