"""标题库 Use Cases 单元测试 — wave217""" from __future__ import annotations 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, PickTitleUseCase, UpdateTitleLibraryUseCase, ) from packages.domain.exceptions import NotFoundError, QuotaExceededError from packages.domain.title_library import TitleLibraryItem # ── helpers ────────────────────────────────────────────────────────────────── def _make_item( item_id="t1", user_id="u1", name="标题A", text="这是一个标题", category="default", usage_count=0, is_active=True, description="", tags=None, metadata_=None, ): return TitleLibraryItem( id=item_id, user_id=user_id, name=name, text=text, category=category, description=description, tags=tags or [], usage_count=usage_count, is_active=is_active, metadata_=metadata_ or {}, ) # ── ListTitleLibraryUseCase ───────────────────────────────────────────────── class TestListTitleLibraryUseCase: def test_list_default_params(self): items = [_make_item("t1"), _make_item("t2")] repo = MagicMock() repo.list_by_user.return_value = items uc = ListTitleLibraryUseCase(repo) result = uc.execute("u1") assert len(result) == 2 repo.list_by_user.assert_called_once_with("u1", category=None, skip=0, limit=50) def test_list_with_category(self): repo = MagicMock() repo.list_by_user.return_value = [] uc = ListTitleLibraryUseCase(repo) result = uc.execute("u1", category="marketing") repo.list_by_user.assert_called_once_with("u1", category="marketing", skip=0, limit=50) assert result == [] def test_list_pagination(self): repo = MagicMock() repo.list_by_user.return_value = [] uc = ListTitleLibraryUseCase(repo) uc.execute("u1", skip=10, limit=20) repo.list_by_user.assert_called_once_with("u1", category=None, skip=10, limit=20) # ── GetTitleLibraryUseCase ────────────────────────────────────────────────── class TestGetTitleLibraryUseCase: def test_get_found(self): item = _make_item() repo = MagicMock() repo.get.return_value = item uc = GetTitleLibraryUseCase(repo) result = uc.execute("t1", "u1") assert result.id == "t1" repo.get.assert_called_once_with("t1", "u1") def test_get_not_found(self): repo = MagicMock() repo.get.return_value = None uc = GetTitleLibraryUseCase(repo) result = uc.execute("t999", "u1") assert result is None # ── CreateTitleLibraryUseCase ─────────────────────────────────────────────── class TestCreateTitleLibraryUseCase: def test_create_success_free_plan_within_quota(self): repo = MagicMock() repo.count_by_user.return_value = 0 # 已用数量 repo.create.side_effect = lambda x: x uc = CreateTitleLibraryUseCase(repo) cmd = CreateTitleLibraryCommand(user_id="u1", name="好标题", text="这是一个好标题的内容") with patch("packages.application.title_library.use_cases.quota_checker") as mock_qc: mock_result = MagicMock() mock_result.allowed = True mock_result.limit = 10 mock_result.used = 0 mock_qc.check.return_value = mock_result result = uc.execute(cmd, plan_name="free") assert result.name == "好标题" assert result.user_id == "u1" repo.create.assert_called_once() def test_create_quota_exceeded_raises(self): repo = MagicMock() repo.count_by_user.return_value = 100 uc = CreateTitleLibraryUseCase(repo) cmd = CreateTitleLibraryCommand(user_id="u1", name="超了", text="配额超限了") with patch("packages.application.title_library.use_cases.quota_checker") as mock_qc: mock_result = MagicMock() mock_result.allowed = False mock_result.limit = 5 mock_result.used = 5 mock_qc.check.return_value = mock_result with pytest.raises(QuotaExceededError): uc.execute(cmd, plan_name="free") def test_create_with_tags_and_metadata(self): repo = MagicMock() repo.count_by_user.return_value = 0 repo.create.side_effect = lambda x: x uc = CreateTitleLibraryUseCase(repo) cmd = CreateTitleLibraryCommand( user_id="u1", name="标题", text="内容", category="marketing", description="描述", tags=["tag1", "tag2"], metadata_={"key": "value"}, ) with patch("packages.application.title_library.use_cases.quota_checker") as mock_qc: mock_result = MagicMock() mock_result.allowed = True mock_result.limit = 100 mock_result.used = 0 mock_qc.check.return_value = mock_result result = uc.execute(cmd) assert result.category == "marketing" assert result.description == "描述" assert result.tags == ["tag1", "tag2"] assert result.metadata_ == {"key": "value"} # ── UpdateTitleLibraryUseCase ─────────────────────────────────────────────── class TestUpdateTitleLibraryUseCase: def test_update_name(self): existing = _make_item(name="old") repo = MagicMock() repo.get.return_value = existing repo.update.return_value = existing uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t1", user_id="u1", name="new") result = uc.execute(cmd) assert result.name == "new" repo.update.assert_called_once() def test_update_text(self): existing = _make_item(text="old") repo = MagicMock() repo.get.return_value = existing repo.update.return_value = existing uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t1", user_id="u1", text="new text") result = uc.execute(cmd) assert result.text == "new text" def test_update_category(self): existing = _make_item(category="old") repo = MagicMock() repo.get.return_value = existing repo.update.return_value = existing uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t1", user_id="u1", category="new_cat") result = uc.execute(cmd) assert result.category == "new_cat" def test_update_tags(self): existing = _make_item(tags=["old"]) repo = MagicMock() repo.get.return_value = existing repo.update.return_value = existing uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t1", user_id="u1", tags=["a", "b"]) result = uc.execute(cmd) assert result.tags == ["a", "b"] def test_update_is_active(self): existing = _make_item(is_active=True) repo = MagicMock() repo.get.return_value = existing repo.update.return_value = existing uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t1", user_id="u1", is_active=False) result = uc.execute(cmd) assert result.is_active is False def test_update_not_found_raises(self): repo = MagicMock() repo.get.return_value = None uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t999", user_id="u1", name="x") with pytest.raises(NotFoundError): uc.execute(cmd) def test_update_none_fields_not_modified(self): existing = _make_item(name="keep", category="keep_cat", description="keep_desc") repo = MagicMock() repo.get.return_value = existing repo.update.return_value = existing uc = UpdateTitleLibraryUseCase(repo) cmd = UpdateTitleLibraryCommand(title_id="t1", user_id="u1") # 全None result = uc.execute(cmd) assert result.name == "keep" assert result.category == "keep_cat" assert result.description == "keep_desc" # ── DeleteTitleLibraryUseCase ─────────────────────────────────────────────── class TestDeleteTitleLibraryUseCase: def test_delete_success(self): repo = MagicMock() repo.delete.return_value = True uc = DeleteTitleLibraryUseCase(repo) result = uc.execute("t1", "u1") assert result is True repo.delete.assert_called_once_with("t1", "u1") def test_delete_not_found(self): repo = MagicMock() repo.delete.return_value = False uc = DeleteTitleLibraryUseCase(repo) result = uc.execute("t999", "u1") assert result is False # ── IncrementTitleUsageUseCase ────────────────────────────────────────────── class TestIncrementTitleUsageUseCase: def test_increment_default_1(self): repo = MagicMock() repo.increment_usage_count.return_value = True uc = IncrementTitleUsageUseCase(repo) cmd = IncrementTitleUsageCommand(title_id="t1", user_id="u1") result = uc.execute(cmd) assert result is True repo.increment_usage_count.assert_called_once_with("t1", "u1", increment=1) def test_increment_custom_amount(self): repo = MagicMock() repo.increment_usage_count.return_value = True uc = IncrementTitleUsageUseCase(repo) cmd = IncrementTitleUsageCommand(title_id="t1", user_id="u1", increment=5) result = uc.execute(cmd) repo.increment_usage_count.assert_called_once_with("t1", "u1", increment=5) def test_increment_zero_returns_false(self): repo = MagicMock() uc = IncrementTitleUsageUseCase(repo) cmd = IncrementTitleUsageCommand(title_id="t1", user_id="u1", increment=0) result = uc.execute(cmd) assert result is False repo.increment_usage_count.assert_not_called() def test_increment_negative_returns_false(self): repo = MagicMock() uc = IncrementTitleUsageUseCase(repo) cmd = IncrementTitleUsageCommand(title_id="t1", user_id="u1", increment=-1) result = uc.execute(cmd) assert result is False repo.increment_usage_count.assert_not_called() # ── PickTitleUseCase ──────────────────────────────────────────────────────── class TestPickTitleUseCase: def test_pick_from_multiple_returns_least_used_in_pool(self): items = [ _make_item("t1", usage_count=10), _make_item("t2", usage_count=1), # 最少 _make_item("t3", usage_count=5), _make_item("t4", usage_count=3), _make_item("t5", usage_count=8), _make_item("t6", usage_count=2), ] repo = MagicMock() repo.list_by_user.return_value = items uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1") # 由于有随机性，多次验证都在候选池（最少使用的5个）中 for _ in range(10): result = uc.execute(cmd) assert result is not None # 最少使用的5个是: t2(1), t6(2), t4(3), t3(5), t5(8) assert result.id in {"t1", "t2", "t3", "t4", "t5", "t6"} # 选中的一定是使用次数最少的5个之一 (usage_count <= 8) assert result.usage_count <= 8 # 验证查询参数 repo.list_by_user.assert_called() call_args = repo.list_by_user.call_args assert call_args[0][0] == "u1" assert call_args[1]["is_active"] is True def test_pick_empty_returns_none(self): repo = MagicMock() repo.list_by_user.return_value = [] uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1") result = uc.execute(cmd) assert result is None def test_pick_single_item(self): item = _make_item("t1") repo = MagicMock() repo.list_by_user.return_value = [item] uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1") result = uc.execute(cmd) assert result.id == "t1" def test_pick_with_category_filter(self): repo = MagicMock() repo.list_by_user.return_value = [_make_item("t1", category="marketing")] uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1", category="marketing") result = uc.execute(cmd) assert result is not None repo.list_by_user.assert_called_once() assert repo.list_by_user.call_args[1]["category"] == "marketing" def test_pick_exclude_ids(self): items = [ _make_item("t1", usage_count=1), _make_item("t2", usage_count=2), _make_item("t3", usage_count=3), ] repo = MagicMock() repo.list_by_user.return_value = items uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1", exclude_ids=["t1", "t2"]) # 排除 t1, t2 后只剩 t3 result = uc.execute(cmd) assert result.id == "t3" def test_pick_exclude_all_fallback_to_all(self): items = [ _make_item("t1", usage_count=1), _make_item("t2", usage_count=2), ] repo = MagicMock() repo.list_by_user.return_value = items uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1", exclude_ids=["t1", "t2"]) # 排除后没了，回退到从全部选 result = uc.execute(cmd) assert result is not None assert result.id in {"t1", "t2"} def test_pick_pool_size_is_5(self): # 10个标题，使用次数从 1~10 items = [_make_item(f"t{i}", usage_count=i) for i in range(1, 11)] repo = MagicMock() repo.list_by_user.return_value = items uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1") # 运行多次，确保选中的都在前5个使用最少的里（t1~t5, usage 1~5） for _ in range(20): result = uc.execute(cmd) assert int(result.id[1:]) <= 5 # 只从前5个里选 def test_pick_fewer_than_pool_size(self): # 只有3个标题，不足5个池大小 items = [ _make_item("t1", usage_count=3), _make_item("t2", usage_count=1), _make_item("t3", usage_count=2), ] repo = MagicMock() repo.list_by_user.return_value = items uc = PickTitleUseCase(repo) cmd = PickTitleCommand(user_id="u1") results = set() for _ in range(30): result = uc.execute(cmd) results.add(result.id) # 3个都有可能被选中（随机性+少量样本，大概率至少出现2个） assert len(results) >= 1 assert results.issubset({"t1", "t2", "t3"})