mirror of
https://github.com/nonebot/nonebot2.git
synced 2024-12-24 20:35:54 +08:00
289 lines
11 KiB
Plaintext
289 lines
11 KiB
Plaintext
|
---
|
|||
|
sidebar_position: 2
|
|||
|
description: 测试事件响应、平台接口调用和会话控制
|
|||
|
---
|
|||
|
|
|||
|
# 测试事件响应与会话操作
|
|||
|
|
|||
|
import Tabs from "@theme/Tabs";
|
|||
|
import TabItem from "@theme/TabItem";
|
|||
|
|
|||
|
在 NoneBot 接收到事件时,事件响应器根据优先级依次通过权限、响应规则来判断当前事件是否应该触发。事件响应流程中,机器人可能会通过 `send` 发送消息或者调用平台接口来执行预期的操作。因此,我们需要对这两种操作进行单元测试。
|
|||
|
|
|||
|
在上一节中,我们对单个事件响应器进行了简单测试。但是在实际场景中,机器人可能定义了多个事件响应器,由于优先级和响应规则的存在,预期的事件响应器可能并不会被触发。NoneBug 支持同时测试多个事件响应器,以此来测试机器人的整体行为。
|
|||
|
|
|||
|
## 测试事件响应
|
|||
|
|
|||
|
NoneBug 提供了六种定义 `Rule` 和 `Permission` 预期行为的方法:
|
|||
|
|
|||
|
- `should_pass_rule`
|
|||
|
- `should_not_pass_rule`
|
|||
|
- `should_ignore_rule`
|
|||
|
- `should_pass_permission`
|
|||
|
- `should_not_pass_permission`
|
|||
|
- `should_ignore_permission`
|
|||
|
|
|||
|
:::tip 提示
|
|||
|
事件响应器类型的检查属于 `Permission` 的一部分,因此可以通过 `should_pass_permission` 和 `should_not_pass_permission` 方法来断言事件响应器类型的检查。
|
|||
|
:::
|
|||
|
|
|||
|
下面我们根据插件示例来测试事件响应行为,我们首先定义两个事件响应器作为测试的对象:
|
|||
|
|
|||
|
```python title=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 来测试它们:
|
|||
|
|
|||
|
<Tabs groupId="testScope">
|
|||
|
<TabItem value="separate" label="独立测试" default>
|
|||
|
|
|||
|
```python {21,22,28,29} title=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()
|
|||
|
```
|
|||
|
|
|||
|
在上面的代码中,我们分别对 `foo` 和 `bar` 事件响应器进行响应测试。我们使用 `ctx.should_pass_rule` 和 `ctx.should_pass_permission` 断言 `foo` 事件响应器应该被触发,使用 `ctx.should_not_pass_rule` 和 `ctx.should_not_pass_permission` 断言 `bar` 事件响应器应该被忽略。
|
|||
|
|
|||
|
</TabItem>
|
|||
|
<TabItem value="global" label="集成测试">
|
|||
|
|
|||
|
```python title=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() as ctx:
|
|||
|
bot = ctx.create_bot()
|
|||
|
event = make_event("/foo")
|
|||
|
ctx.receive_event(bot, event)
|
|||
|
ctx.should_pass_rule(foo)
|
|||
|
ctx.should_pass_permission(foo)
|
|||
|
ctx.should_not_pass_rule(bar)
|
|||
|
ctx.should_not_pass_permission(bar)
|
|||
|
```
|
|||
|
|
|||
|
在上面的代码中,我们对 `foo` 和 `bar` 事件响应器一起进行响应测试。我们使用 `ctx.should_pass_rule` 和 `ctx.should_pass_permission` 断言 `foo` 事件响应器应该被触发,使用 `ctx.should_not_pass_rule` 和 `ctx.should_not_pass_permission` 断言 `bar` 事件响应器应该被忽略。通过参数,我们可以指定断言的事件响应器。
|
|||
|
|
|||
|
</TabItem>
|
|||
|
</Tabs>
|
|||
|
|
|||
|
当然,如果需要忽略某个事件响应器的响应规则和权限检查,强行进入响应流程,我们可以使用 `should_ignore_rule` 和 `should_ignore_permission` 方法:
|
|||
|
|
|||
|
```python {21,22} title=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_send` 和 `ctx.should_call_api` 方法来测试平台接口的使用情况。
|
|||
|
|
|||
|
1. `should_call_send`
|
|||
|
|
|||
|
定义事件响应器预期发送的消息,即通过[事件响应器操作 send](../../appendices/session-control.mdx#send)进行的操作。`should_call_send` 有四个参数:
|
|||
|
|
|||
|
- `event`:回复的目标事件。
|
|||
|
- `message`:预期的消息对象,可以是 `str`、`Message` 或 `MessageSegment`。
|
|||
|
- `result`:send 的返回值,将会返回给插件。
|
|||
|
- `bot`(可选):发送消息的 bot 对象。
|
|||
|
- `**kwargs`:send 方法的额外参数。
|
|||
|
|
|||
|
2. `should_call_api`
|
|||
|
定义事件响应器预期调用的平台 API 接口,即通过[调用平台 API](../../appendices/api-calling.mdx#调用平台-API)进行的操作。`should_call_api` 有四个参数:
|
|||
|
|
|||
|
- `api`:API 名称。
|
|||
|
- `data`:预期的请求数据。
|
|||
|
- `result`:call_api 的返回值,将会返回给插件。
|
|||
|
- `adapter`(可选):调用 API 的平台适配器对象。
|
|||
|
- `**kwargs`:call_api 方法的额外参数。
|
|||
|
|
|||
|
下面是一个使用 `should_call_send` 和 `should_call_api` 方法的示例:
|
|||
|
|
|||
|
我们先定义一个测试插件,在响应流程中向用户发送一条消息并调用 `Console` 适配器的 `bell` API。
|
|||
|
|
|||
|
```python {8,9} title=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()
|
|||
|
```
|
|||
|
|
|||
|
然后我们对该插件进行测试:
|
|||
|
|
|||
|
```python {19,20,23,24} title=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` 方法中指定 `base` 和 `adapter` 参数,确保不会因为重载功能而出现非预期情况。
|
|||
|
|
|||
|
## 测试会话控制
|
|||
|
|
|||
|
在[会话控制](../../appendices/session-control.mdx)一节中,我们介绍了如何使用事件响应器操作来实现对用户的交互式会话。在上一节的示例插件测试中,我们其实已经使用了 `ctx.should_finished` 来断言会话结束。NoneBug 针对各种流程控制操作分别提供了相应的方法来定义预期的会话处理行为。它们分别是:
|
|||
|
|
|||
|
- `should_finished`:断言会话结束,对应 `matcher.finish` 操作。
|
|||
|
- `should_rejected`:断言会话等待用户输入并重新执行当前事件处理函数,对应 `matcher.reject` 系列操作。
|
|||
|
- `should_paused`: 断言会话等待用户输入并执行下一个事件处理函数,对应 `matcher.pause` 操作。
|
|||
|
|
|||
|
我们仅需在测试用例中的正确位置调用这些方法,就可以断言会话的预期行为。例如:
|
|||
|
|
|||
|
```python title=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("密码正确")
|
|||
|
```
|
|||
|
|
|||
|
```python title=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)
|
|||
|
```
|