You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

已安装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()

另外可以做两个优化:

  1. 移除未使用的MockBot类,因为你已经用AsyncMock模拟了bot对象
  2. 使用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.12 17:01:02