diff --git a/tests/unit/test_recipe_use_cases.py b/tests/unit/test_recipe_use_cases.py old mode 100644 new mode 100755 index 43b2de396..d664c918b --- a/tests/unit/test_recipe_use_cases.py +++ b/tests/unit/test_recipe_use_cases.py @@ -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) diff --git a/tests/unit/test_title_library_use_cases.py b/tests/unit/test_title_library_use_cases.py old mode 100644 new mode 100755 index 213b9127e..a44341f4d --- a/tests/unit/test_title_library_use_cases.py +++ b/tests/unit/test_title_library_use_cases.py @@ -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