Skip to main content
Version: 2.2.1

测试事件响应与会话操作

在 NoneBot 接收到事件时,事件响应器根据优先级依次通过权限、响应规则来判断当前事件是否应该触发。事件响应流程中,机器人可能会通过 send 发送消息或者调用平台接口来执行预期的操作。因此,我们需要对这两种操作进行单元测试。

在上一节中,我们对单个事件响应器进行了简单测试。但是在实际场景中,机器人可能定义了多个事件响应器,由于优先级和响应规则的存在,预期的事件响应器可能并不会被触发。NoneBug 支持同时测试多个事件响应器,以此来测试机器人的整体行为。

测试事件响应

NoneBug 提供了六种定义 RulePermission 预期行为的方法:

  • should_pass_rule
  • should_not_pass_rule
  • should_ignore_rule
  • should_pass_permission
  • should_not_pass_permission
  • should_ignore_permission
提示

事件响应器类型的检查属于 Permission 的一部分,因此可以通过 should_pass_permissionshould_not_pass_permission 方法来断言事件响应器类型的检查。

下面我们根据插件示例来测试事件响应行为,我们首先定义两个事件响应器作为测试的对象:

example.py
from nonebot import on_command

def never_pass():
return False

foo = on_command("foo")
bar = on_command("bar", permission=never_pass)

在这两个事件响应器中,foo 当收到 /foo 消息时会执行,而 bar 则不会执行。我们使用 NoneBug 来测试它们:

tests/test_example.py
from datetime import datetime

import pytest
from nonebug import App
from nonebot.adapters.console import User, Message, MessageEvent

def make_event(message: str = "") -> MessageEvent:
return MessageEvent(
time=datetime.now(),
self_id="test",
message=Message(message),
user=User(user_id=123456789),
)

@pytest.mark.asyncio
async def test_example(app: App):
from awesome_bot.plugins.example import foo, bar

async with app.test_matcher(foo) as ctx:
bot = ctx.create_bot()
event = make_event("/foo")
ctx.receive_event(bot, event)
ctx.should_pass_rule()
ctx.should_pass_permission()

async with app.test_matcher(bar) as ctx:
bot = ctx.create_bot()
event = make_event("/foo")
ctx.receive_event(bot, event)
ctx.should_not_pass_rule()
ctx.should_not_pass_permission()

在上面的代码中,我们分别对 foobar 事件响应器进行响应测试。我们使用 ctx.should_pass_rulectx.should_pass_permission 断言 foo 事件响应器应该被触发,使用 ctx.should_not_pass_rulectx.should_not_pass_permission 断言 bar 事件响应器应该被忽略。

当然,如果需要忽略某个事件响应器的响应规则和权限检查,强行进入响应流程,我们可以使用 should_ignore_ruleshould_ignore_permission 方法:

tests/test_example.py
from datetime import datetime

import pytest
from nonebug import App
from nonebot.adapters.console import User, Message, MessageEvent

def make_event(message: str = "") -> MessageEvent:
return MessageEvent(
time=datetime.now(),
self_id="test",
message=Message(message),
user=User(user_id=123456789),
)

@pytest.mark.asyncio
async def test_example(app: App):
from awesome_bot.plugins.example import foo, bar

async with app.test_matcher(bar) as ctx:
bot = ctx.create_bot()
event = make_event("/foo")
ctx.receive_event(bot, event)
ctx.should_ignore_rule(bar)
ctx.should_ignore_permission(bar)

在忽略了响应规则和权限检查之后,就会进入 bar 事件响应器的响应流程。

测试平台接口使用

上一节的示例插件测试中,我们已经尝试了测试插件对事件的消息回复。通常情况下,事件处理流程中对平台接口的使用会通过事件响应器操作或者调用平台 API 两种途径进行。针对这两种途径,NoneBug 分别提供了 ctx.should_call_sendctx.should_call_api 方法来测试平台接口的使用情况。

  1. should_call_send

    定义事件响应器预期发送的消息,即通过事件响应器操作 send进行的操作。should_call_send 有四个参数:

    • event:回复的目标事件。
    • message:预期的消息对象,可以是 strMessageMessageSegment
    • result:send 的返回值,将会返回给插件。
    • bot(可选):发送消息的 bot 对象。
    • **kwargs:send 方法的额外参数。
  2. should_call_api 定义事件响应器预期调用的平台 API 接口,即通过调用平台 API进行的操作。should_call_api 有四个参数:

    • api:API 名称。
    • data:预期的请求数据。
    • result:call_api 的返回值,将会返回给插件。
    • adapter(可选):调用 API 的平台适配器对象。
    • **kwargs:call_api 方法的额外参数。

下面是一个使用 should_call_sendshould_call_api 方法的示例:

我们先定义一个测试插件,在响应流程中向用户发送一条消息并调用 Console 适配器的 bell API。

example.py
from nonebot import on_command
from nonebot.adapters.console import Bot

foo = on_command("foo")

@foo.handle()
async def _(bot: Bot):
await foo.send("message")
await bot.bell()

然后我们对该插件进行测试:

tests/test_example.py
from datetime import datetime

import pytest
import nonebot
from nonebug import App
from nonebot.adapters.console import Bot, User, Adapter, Message, MessageEvent

def make_event(message: str = "") -> MessageEvent:
return MessageEvent(
time=datetime.now(),
self_id="test",
message=Message(message),
user=User(user_id=123456789),
)

@pytest.mark.asyncio
async def test_example(app: App):
from awesome_bot.plugins.example import foo

async with app.test_matcher(foo) as ctx:
adapter = nonebot.get_adapter(Adapter)
bot = ctx.create_bot(base=Bot, adapter=adapter)
event = make_event("/foo")
ctx.receive_event(bot, event)
ctx.should_call_send(event, "message", result=None, bot=bot)
ctx.should_call_api("bell", {}, result=None, adapter=adapter)

请注意,对于在依赖注入中使用了非基类对象的情况,我们需要在 create_bot 方法中指定 baseadapter 参数,确保不会因为重载功能而出现非预期情况。

测试会话控制

会话控制一节中,我们介绍了如何使用事件响应器操作来实现对用户的交互式会话。在上一节的示例插件测试中,我们其实已经使用了 ctx.should_finished 来断言会话结束。NoneBug 针对各种流程控制操作分别提供了相应的方法来定义预期的会话处理行为。它们分别是:

  • should_finished:断言会话结束,对应 matcher.finish 操作。
  • should_rejected:断言会话等待用户输入并重新执行当前事件处理函数,对应 matcher.reject 系列操作。
  • should_paused: 断言会话等待用户输入并执行下一个事件处理函数,对应 matcher.pause 操作。

我们仅需在测试用例中的正确位置调用这些方法,就可以断言会话的预期行为。例如:

example.py
from nonebot import on_command
from nonebot.typing import T_State

foo = on_command("foo")

@foo.got("key", prompt="请输入密码")
async def _(state: T_State, key: str = ArgPlainText()):
if key != "some password":
try_count = state.get("try_count", 1)
if try_count >= 3:
await foo.finish("密码错误次数过多")
else:
state["try_count"] = try_count + 1
await foo.reject("密码错误,请重新输入")
await foo.finish("密码正确")
tests/test_example.py
from datetime import datetime

import pytest
from nonebug import App
from nonebot.adapters.console import User, Message, MessageEvent

def make_event(message: str = "") -> MessageEvent:
return MessageEvent(
time=datetime.now(),
self_id="test",
message=Message(message),
user=User(user_id=123456789),
)

@pytest.mark.asyncio
async def test_example(app: App):
from awesome_bot.plugins.example import foo

async with app.test_matcher(foo) as ctx:
bot = ctx.create_bot()
event = make_event("/foo")
ctx.receive_event(bot, event)
ctx.should_call_send(event, "请输入密码", result=None)
ctx.should_rejected(foo)
event = make_event("wrong password")
ctx.receive_event(bot, event)
ctx.should_call_send(event, "密码错误,请重新输入", result=None)
ctx.should_rejected(foo)
event = make_event("wrong password")
ctx.receive_event(bot, event)
ctx.should_call_send(event, "密码错误,请重新输入", result=None)
ctx.should_rejected(foo)
event = make_event("wrong password")
ctx.receive_event(bot, event)
ctx.should_call_send(event, "密码错误次数过多", result=None)
ctx.should_finished(foo)