已安装pytest-asyncio,测试Telegram Bot多异步函数仍报错求助
问题描述
我在用pytest编写测试用例,测试Telegram Bot的start、name等异步函数。单独测试单个函数时可正常运行,但同时测试多个函数时,出现如下错误提示:
PytestUnhandledCoroutineWarning: async def functions are not natively supported and have been skipped. You need to install a suitable plugin for your async framework.
尽管我已经安装了提示中提到的pytest-asyncio插件,但问题依然存在。
项目参考代码
import logging from telegram import Update from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler, MessageHandler, filters task_list = {} formatted_task_list = [] done_list = {} formatted_done_list = [] user_name = {} async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id = update.effective_user.id if user_id not in task_list: task_list[user_id] = [] done_list[user_id] = [] user_name[user_id] = "Task" await context.bot.send_message(chat_id=update.effective_chat.id, text="👋Hello! Welcome to Task List") async def name(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id = update.effective_user.id if user_id in task_list: user_id = update.effective_user.id user_name[user_id] = " ".join(context.args) user_name[user_id] = f"<b>{user_name[user_id]}</b>" await context.bot.send_message(chat_id=update.effective_chat.id, text=f"Your list will be named {user_name[user_id]}. You're set!", parse_mode="HTML") else: await context.bot.send_message(chat_id=update.effective_chat.id, text="Type /start to start using this bot.") def main(): application = ApplicationBuilder().token('My tokens').build() start_handler = CommandHandler(['start', 'hello'], start) name_handler = CommandHandler('name', name) application.add_handler(start_handler) application.add_handler(name_handler) application.run_polling() if __name__ == "__main__": main()
我的pytest测试代码
import pytest from unittest.mock import AsyncMock, patch from project import start, view, add, name # Mocks class MockUser: def __init__(self, id): self.id = id class MockChat: def __init__(self, id): self.id = id class MockBot: async def send_message(self, chat_id, text): pass # Test case @pytest.mark.asyncio @patch('project.task_list', {932872542: ['Example Task']}) @patch('project.done_list', {932872542: []}) @patch('project.user_name', {932872542: 'Task'}) async def test_start_function(): update_mock = AsyncMock() context_mock = AsyncMock() context_mock.bot = AsyncMock() update_mock.effective_user = MockUser(932872542) update_mock.effective_chat = MockChat(123) update_mock.message.text = "/start" await start(update_mock, context_mock) context_mock.bot.send_message.assert_called_once_with( chat_id=123, text="👋Hello! Welcome to Task List") async def test_name_function(): update_mock = AsyncMock() context_mock = AsyncMock() context_mock.bot = AsyncMock() update_mock.effective_user = MockUser(932872542) update_mock.effective_chat = MockChat(123) update_mock.message.text = "/name New List" context_mock.args = ["New", "List"] await name(update_mock, context_mock) context_mock.bot.send_message.assert_called_once_with( chat_id=123, text="Your list will be named <b>New List</b>. You're set!", parse_mode="HTML", ) if __name__ == "__main__": pytest.main()
解决方案
问题根源在于你仅给第一个测试函数test_start_function添加了@pytest.mark.asyncio装饰器,而第二个测试函数test_name_function没有加。pytest-asyncio要求每个异步测试函数都必须标记该装饰器,否则pytest会将未标记的异步函数视为普通同步函数,导致跳过执行并抛出警告。
修改后的测试代码如下(重点添加第二个函数的装饰器):
import pytest from unittest.mock import AsyncMock, patch from project import start, view, add, name # Mocks class MockUser: def __init__(self, id): self.id = id class MockChat: def __init__(self, id): self.id = id # Test case @pytest.mark.asyncio @patch('project.task_list', {932872542: ['Example Task']}) @patch('project.done_list', {932872542: []}) @patch('project.user_name', {932872542: 'Task'}) async def test_start_function(): update_mock = AsyncMock() context_mock = AsyncMock() context_mock.bot = AsyncMock() update_mock.effective_user = MockUser(932872542) update_mock.effective_chat = MockChat(123) update_mock.message.text = "/start" await start(update_mock, context_mock) context_mock.bot.send_message.assert_called_once_with( chat_id=123, text="👋Hello! Welcome to Task List") # 添加@pytest.mark.asyncio装饰器 @pytest.mark.asyncio @patch('project.task_list', {932872542: ['Example Task']}) @patch('project.user_name', {932872542: 'Task'}) async def test_name_function(): update_mock = AsyncMock() context_mock = AsyncMock() context_mock.bot = AsyncMock() update_mock.effective_user = MockUser(932872542) update_mock.effective_chat = MockChat(123) update_mock.message.text = "/name New List" context_mock.args = ["New", "List"] await name(update_mock, context_mock) context_mock.bot.send_message.assert_called_once_with( chat_id=123, text="Your list will be named <b>New List</b>. You're set!", parse_mode="HTML", ) if __name__ == "__main__": pytest.main()
另外可以做两个优化:
- 移除未使用的
MockBot类,因为你已经用AsyncMock模拟了bot对象 - 使用pytest fixture来管理全局变量的patch,减少重复代码,示例如下:
import pytest from unittest.mock import AsyncMock, patch from project import start, name # Mocks class MockUser: def __init__(self, id): self.id = id class MockChat: def __init__(self, id): self.id = id @pytest.fixture(autouse=True) def mock_global_vars(): with patch('project.task_list', {932872542: ['Example Task']}), \ patch('project.done_list', {932872542: []}), \ patch('project.user_name', {932872542: 'Task'}): yield @pytest.mark.asyncio async def test_start_function(): # 测试代码不变 pass @pytest.mark.asyncio async def test_name_function(): # 测试代码不变 pass
内容的提问来源于stack exchange,提问作者Lili
相关产品推荐
相关产品推荐

