from __future__ import annotations import asyncio import sys import types from datetime import UTC, datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock import pytest @pytest.fixture def giveaway_module(app_modules, monkeypatch): permissions = types.ModuleType("wbb.core.decorators.permissions") permissions.adminsOnly = lambda *_args, **_kwargs: lambda function: function functions = types.ModuleType("wbb.utils.functions") async def extract_user_and_reason(_message): return None, "" functions.extract_user_and_reason = extract_user_and_reason monkeypatch.setitem( sys.modules, "wbb.core.decorators.permissions", permissions ) monkeypatch.setitem(sys.modules, "wbb.utils.functions", functions) def no_running_loop(): raise RuntimeError with monkeypatch.context() as scoped: scoped.setattr(asyncio, "get_running_loop", no_running_loop) return app_modules.load("wbb.modules.giveaway") async def _giveaway_with_participants( app_modules, count: int = 12, winner_count: int = 1 ): giveaway_db = app_modules.load("wbb.utils.dbgiveaway") giveaway = await giveaway_db.create_giveaway( chat_id=-100, creator_id=1, creator_name="Admin", title="每周抽奖", description="", prizes=[{"name": "Winner", "count": winner_count}], ends_at=datetime.now(UTC) + timedelta(hours=1), ) for user_id in range(1, count + 1): assert ( await giveaway_db.add_participant( giveaway_id=giveaway["giveaway_id"], user_id=user_id, username=f"user{user_id}", first_name=f"User {user_id}", ) == "ok" ) return giveaway async def test_giveaway_keyword_lists_draw_and_participant_button( app_modules, giveaway_module ): giveaway = await _giveaway_with_participants(app_modules) message = SimpleNamespace( chat=SimpleNamespace(id=-100), reply_text=AsyncMock(return_value=SimpleNamespace(id=1)), ) await giveaway_module.giveaway_keyword_command(None, message) text = message.reply_text.await_args.args[0] markup = message.reply_text.await_args.kwargs["reply_markup"] assert "每周抽奖" in text assert "12 人参与" in text callback_data = [button.callback_data for row in markup.inline_keyboard for button in row] assert f"giveaway_join:{giveaway['giveaway_id']}" in callback_data assert f"giveaway_participants:{giveaway['giveaway_id']}" in callback_data async def test_participant_callback_paginates_ten_users_per_page( app_modules, giveaway_module ): giveaway = await _giveaway_with_participants(app_modules) callback_message = SimpleNamespace( chat=SimpleNamespace(id=-100), reply_text=AsyncMock(return_value=SimpleNamespace(id=2)), edit_text=AsyncMock(), ) open_query = SimpleNamespace( data=f"giveaway_participants:{giveaway['giveaway_id']}", message=callback_message, answer=AsyncMock(), ) await giveaway_module.giveaway_participants_callback(None, open_query) first_page = callback_message.reply_text.await_args.args[0] first_markup = callback_message.reply_text.await_args.kwargs["reply_markup"] assert "第 1/2 页 · 共 12 人" in first_page assert "User 1" in first_page assert "User 10" in first_page assert "User 11" not in first_page assert any( button.callback_data == f"giveaway_participants_page:{giveaway['giveaway_id']}:2" for row in first_markup.inline_keyboard for button in row ) page_query = SimpleNamespace( data=f"giveaway_participants_page:{giveaway['giveaway_id']}:2", message=callback_message, answer=AsyncMock(), ) await giveaway_module.giveaway_participants_page_callback(None, page_query) second_page = callback_message.edit_text.await_args.args[0] assert "第 2/2 页 · 共 12 人" in second_page assert "11. " in second_page assert "User 11" in second_page assert "User 12" in second_page async def test_draw_keyword_shows_latest_winners_with_pagination( app_modules, giveaway_module ): giveaway = await _giveaway_with_participants( app_modules, count=12, winner_count=12 ) service = app_modules.load("wbb.services.giveaways") ok, _, _ = await service.finish_and_publish_giveaway( giveaway["giveaway_id"], publish=False ) assert ok is True message = SimpleNamespace( chat=SimpleNamespace(id=-100), reply_text=AsyncMock(return_value=SimpleNamespace(id=3)), ) await giveaway_module.giveaway_winners_keyword_command(None, message) first_page = message.reply_text.await_args.args[0] first_markup = message.reply_text.await_args.kwargs["reply_markup"] assert "抽奖结果" in first_page assert "第 1/2 页 · 共 12 位中奖者" in first_page assert any( button.callback_data == f"giveaway_winners_page:{giveaway['giveaway_id']}:2" for row in first_markup.inline_keyboard for button in row ) callback_message = SimpleNamespace( chat=SimpleNamespace(id=-100), edit_text=AsyncMock(), ) page_query = SimpleNamespace( data=f"giveaway_winners_page:{giveaway['giveaway_id']}:2", message=callback_message, answer=AsyncMock(), ) await giveaway_module.giveaway_winners_page_callback(None, page_query) second_page = callback_message.edit_text.await_args.args[0] assert "第 2/2 页 · 共 12 位中奖者" in second_page assert "11. Winner" in second_page assert "12. Winner" in second_page