Модульное тестирование (ErisPulse-Testing)
ErisPulse-Testing — это официальный инструмент тестирования (RFC EPRFC-2026-001, направление 3):
предоставляет TestBot, фабрику тестовых событий, захват исходящих сообщений и интерфейс утверждений, делая модульное тестирование таким же простым, как написание обычных pytest.
pip install ErisPulse-Testing
Инструмент разработки с односторонней зависимостью от фреймворка, нулевое вмешательство во время выполнения. Для проверки адаптера на реальной платформе используйте
tests/devs/test_adapter.pyиз репозитория фреймворка.
Быстрый старт
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("Сегодня уже зарегистрирован") # Второе попадание в кэш
Рекомендуется использовать TestBot с async with: при запуске регистрируется MockAdapter (захватывает все исходящие), отключается дедупликация событий, применяются перезаписи конфигурации; при выходе автоматически очищаются глобальные состояния фреймворка, тесты не влияют друг на друга.
Сопутствующие fixtures для pytest (автоматически доступны после установки):
testbot: стандартный TestBot уровня function (platform=test, префикс/)make_testbot(**kwargs): фабрика с пользовательскими параметрами (prefix/config/platform/bot_id...)
Рекомендуется в конфигурации проекта установить asyncio_mode = "auto" ([tool.pytest.ini_options]),
или добавить @pytest.mark.asyncio к тестовым функциям.
Фабрика событий
| Функция | Описание |
|---|---|
create_message_event(text, user_id=..., group_id=None, ...) |
Событие сообщения; если group_id пустой, то личное сообщение |
create_command_event("roll 3", prefix="/") |
Командное событие (автоматически добавляет префикс, если уже есть — не дублирует) |
create_notice_event(type, ...) |
Событие уведомления (например friend_add) |
create_request_event(type, ...) |
Событие запроса (например, запрос на добавление в друзья) |
create_meta_event("connect", ...) |
Мета-событие (connect позволяет Bot войти в сеть) |
Все события имеют уникальный id на основе uuid, что изначально избегает дедупликации в фреймворке.
Важно: сгенерированные события не содержат оригинальный платформенный запрос (event.get_raw() возвращает пустой словарь). Для определения сценариев группового чата / личных сообщений используйте доступные методы event.is_group_message() / event.get_detail_type() / event.get_group_id() и т.д., не читайте raw.
API TestBot
Распределение
trace = await bot.dispatch(event) # Распределяет и ждет, пока обработчики примут решение, возвращает цепочку решений
await bot.dispatch(event, drain=False) # Первая интерактивная отправка: не ждать (обработчики wait_reply остаются активными)
await bot.send_message("Привет") # Упрощённый способ отправки сообщения
await bot.reply_as("18", user_id="u1") # Симуляция ответа пользователя (автоматически ждёт готовности waiter)
dispatch() после emit собирает все задачи обработчиков, возвращает сразу после завершения — в тестах не нужно использовать sleep.
Утверждения исходящих сообщений
bot.replies # Все исходящие сообщения (список SentMessage)
bot.last_reply.text # Текст последнего сообщения
bot.replies_to("123") # Фильтрация по цели
bot.clear_replies() # Изоляция утверждений между этапами
bot.assert_replied() # Есть хотя бы одно исходящее сообщение
bot.assert_replied(contains="Регистрация", to="123")
bot.assert_not_replied() # Нет ни одного исходящего сообщения
bot.assert_reply_contains("Регистрация прошла успешно") # Есть сообщение, содержащее указанный текст
await bot.wait_for_reply(timeout=2) # Ждёт появления асинхронного ответа
Поля SentMessage: text (первый текстовый сегмент), segments (полный набор сообщений),
target_type / target_id / bot_id (контекст отправки), has_modifier("at") и т.д.
Загрузка модулей
await bot.load_module("MyModule") # Имя уже зарегистрированного модуля (требует завершения discovery entry-point в SDK)
await bot.load_module(MyModule) # Или подкласс BaseModule (автоматически register + load, рекомендуется)
await bot.unload_module("MyModule")
Обработчики команд / событий, зарегистрированные в on_load, принадлежат модулю, при выгрузке автоматически удаляются, можно напрямую утверждать "после выгрузки команда не работает".
Важно: строковая форма не выполняет scan entry-point (TestBot не инициализирует процесс discovery фреймворка); для тестирования модулей с мягкими зависимостями передавайте напрямую класс или зарегистрируйте модуль и передайте имя.
Замена зависимостей (требуется 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
Заменяется функция, объявленная в таблице команд Depends(get_session), автоматически восстанавливается при выходе из with.
Перезапись конфигурации
bot = TestBot(prefix="//", config={
"ErisPulse.event.command.case_sensitive": False,
"MyModule.api_key": "test-key", # Конфигурация модуля (self.cfg доступна для чтения)
})
Изменения применяются на уровне памяти, префикс команд и т.д. изменяются мгновенно. Два важных замечания:
- Сохранение на диск: перезапись будет сохранена по стратегии отложенного сохранения фреймворка (по умолчанию около 5 секунд) в
config/config.tomlв рабочей директории — если проект находится в git, добавьтеconfig/в.gitignore; - Конфликт с перезаписью конфигурации в модуле во время выполнения (известный лимит): если модуль перезаписывает конфигурацию целиком (например,
self.cfg = ..., как в списке подписок) и здесь происходит точечная перезапись, возникает проблема согласованности чтения/записи ConfigManager — модуль может не видеть перезаписанные значения при чтении целиком, а перезапись может быть перезаписана при сохранении на диск (в ErisPulse 2.9.0-dev.1 исправлено, в 2.8.x всё ещё актуально). Для тестов, использующих "перезапись конфигурации во время выполнения", в 2.8.x рекомендуется в fixture перезаписывать соответствующие разделы конфигурации целиком.
Цепочка распределения (для отладки "почему команда не сработала"; требует EP>=2.9.0-dev)
dispatch() возвращает DispatchTrace — цепочку причинно-следственных связей каждого этапа распределения:
trace = await bot.dispatch(create_command_event("dailyx", user_id="123"))
trace.verdict # executed / rejected / dropped / failed / no_match / passed
trace.explain() # Пошаговое объяснение причинно-следственных связей (на текущем языке)
trace.command # Название команды, которая сработала (если не сработала — None)
trace.steps("cooldown") # Фильтрация по этапам
trace.assert_executed("daily") # Утверждение выполнения (в случае сбоя — полная цепочка причинно-следственных связей)
trace.assert_rejected() # Утверждение, что команда была отклонена по правилам доступа
trace.assert_dropped() # Утверждение, что команда была тихо отброшена (например, из-за кэша)
trace.assert_no_match() # Утверждение, что команда не сработала
Покрытые этапы: проверка текста команды, сопоставление команды (если не сработала — предложение похожих команд), область действия, ACL пользователя, проверка владельца, функции прав, тихая отбраковка из-за кэша, парсинг параметров, результат выполнения, промежуточные отказы middleware.
В продакшене также можно использовать встроенные средства фреймворка ErisPulse.Core.Event.trace (start_dispatch_trace() / format_dispatch_trace()) для сбора и отображения цепочки распределения.