Module Testing (ErisPulse-Testing)

ErisPulse-Testing is the official testing toolkit (RFC EPRFC-2026-001 Direction 3):
It provides TestBot, test event factories, outbound message capturing and assertion utilities, making module testing as simple as writing regular pytest.

pip install ErisPulse-Testing

A development-time tool that depends on the framework unidirectionally, with zero runtime interference. For smoke tests that truly connect to adapter platforms, use tests/devs/test_adapter.py in the framework repository.

Quick Start

import pytest
from ErisPulse.Core.Event.command import command
from ErisPulse_Testing import TestBot, create_command_event

async def test_daily(make_testbot):
    async with make_testbot(prefix="/") as bot:
        @command("daily", cooldown="1d", cooldown_reply="签到已过期")
        async def daily(event):
            await event.reply("签到成功！")

        await bot.dispatch(create_command_event("daily", user_id="123"))
        assert bot.last_reply.text == "签到成功！"

        await bot.dispatch(create_command_event("daily", user_id="123"))
        bot.assert_reply_contains("签到已过期")   # Second call hits cooldown

TestBot is recommended to be used with async with: it registers a MockAdapter at startup (captures all outbound messages), disables event deduplication, applies configuration overrides; on exit, it automatically cleans up the framework's global state, ensuring test cases do not pollute each other.

Accompanying pytest fixtures (automatically available after installation):

• testbot: Standard TestBot at function level (platform=test, prefix /)
• make_testbot(**kwargs): Custom parameter factory (prefix / config / platform / bot_id ...)

It is recommended to configure asyncio_mode = "auto" in your test project ([tool.pytest.ini_options]), or add @pytest.mark.asyncio to test cases.

Event Factories

All events use a unique id generated by uuid, naturally avoiding the framework's event deduplication.

Note: Synthetic events do not contain raw platform messages (event.get_raw() returns an empty dict). To determine scenarios like group chat / private chat, use accessors such as event.is_group_message() / event.get_detail_type() / event.get_group_id(), and do not read raw data.

TestBot API

Dispatching

trace = await bot.dispatch(event)          # Dispatch and wait for handler to land, returns decision chain
await bot.dispatch(event, drain=False)     # Interactive first message: do not wait (wait_reply handlers remain active)
await bot.send_message("你好")             # Shortcut for message dispatch
await bot.reply_as("18", user_id="u1")     # Simulate wait_reply user reply (automatically wait for waiter to be ready)

dispatch() gathers all in-flight handler Tasks after emit, and returns immediately after processing is complete — no need to sleep in tests.

Outbound Assertions

bot.replies                # All outbound messages (list of SentMessage)
bot.last_reply.text        # Text of the most recent reply
bot.replies_to("123")      # Filter by target
bot.clear_replies()        # Isolate assertions between stages
bot.assert_replied()                       # Assert there is at least one outbound message
bot.assert_replied(contains="签到", to="123")
bot.assert_not_replied()                   # Assert there are no outbound messages
bot.assert_reply_contains("签到成功")       # Assert there is an outbound message containing the specified text
await bot.wait_for_reply(timeout=2)        # Wait for asynchronous reply to appear

SentMessage fields: text (first text segment), segments (full message segments), target_type / target_id / bot_id (sending context), has_modifier("at"), etc.

Module Loading

await bot.load_module("MyModule")   # Already registered module name (requires framework sdk.init() to complete entry-point discovery)
await bot.load_module(MyModule)     # Or BaseModule subclass (auto register + load, recommended)
await bot.unload_module("MyModule")

Event handlers registered in on_load are tied to the module and are automatically cleaned up on unload, allowing direct assertions like "command is disabled after unload".
Note: Using string forms does not perform entry-point scanning (TestBot does not initialize the framework discovery process); for testing soft-dependency modules, pass the class directly (or register manually before passing the name).

Dependency Replacement (requires EP>=2.9.0-dev)

with bot.patch_dependency(get_session, fake_session) as mock:
    await bot.dispatch(create_command_event("query"))
    assert mock.called

This replaces functions declared with Depends(get_session) in the command registry; the original is restored automatically when exiting the with block.

Configuration Overwrite

bot = TestBot(prefix="//", config={
    "ErisPulse.event.command.case_sensitive": False,
    "MyModule.api_key": "test-key",     # Module configuration (readable via self.cfg)
})

Overwrites are injected at the memory layer, and changes like command prefixes take effect immediately via hot updates. Two points to note:

1. Persistence: Overwrites are written to disk via the framework's delayed write strategy (default ~5 seconds) into config/config.toml in the cwd — for projects being tested, add config/ to .gitignore;
2. Conflict with module runtime configuration writes (known limitation): When a module writes back configuration as a whole section (e.g., self.cfg = ..., such as subscription lists) and this conflicts with point-based overwrites here, there is a consistency issue in the ConfigManager's read/write — modules reading the whole section may not see the overwrite values, and overwrites may be overwritten during disk writes (fixed in ErisPulse 2.9.0-dev.1, still affected in 2.8.x). For use cases involving "runtime configuration writes", in 2.8.x it is recommended to reset the relevant configuration section via full section overwrite in the fixture.

Dispatch Decision Chain (for troubleshooting "why command didn't trigger"; requires EP>=2.9.0-dev)

dispatch() returns a DispatchTrace — a causal chain of every decision point in this dispatch:

trace = await bot.dispatch(create_command_event("dailyx", user_id="123"))

trace.verdict        # executed / rejected / dropped / failed / no_match / passed
trace.explain()      # Causal explanation line by line (in current language)
trace.command        # Command name matched (None if no match)
trace.steps("cooldown")  # Filter decision records by stage

trace.assert_executed("daily")  # Assert execution (includes full causal chain on failure)
trace.assert_rejected()         # Assert rejection by permission-related decisions
trace.assert_dropped()          # Assert silent dropping (e.g. cooldown)
trace.assert_no_match()         # Assert no command matched

Decision coverage: command text matching, command hit (with spelling suggestions if no match), scope, user ACL, owner check, permission functions, cooldown silent dropping, parameter parsing, execution result, middleware rejection.

The framework's built-in ErisPulse.Core.Event.trace (start_dispatch_trace() / format_dispatch_trace()) can also be used in production environments to collect and render decision chains.