Как реализовать end-to-end-тестирование telegram-бота

Каждый, кто когда-либо писал telegram-ботов, задавался вопросом: «А как их тестировать?» Сложно найти однозначный ответ. Например, при написании тестов для веб-приложений и API можно воспользоваться тестовым клиентом DRF или FastAPI: просто пишешь запрос и делаешь assert на полученный ответ. Мне захотелось получить подобный функционал и для тестирования telegram-бота.

Привет. Я Михаил Выборный, python-разработчик, backend-developer в облачном провайдере beeline cloud. В этой статье я хочу поделиться опытом написания автоматизированных end-to-end-тестов без эмуляции Telegram Bot API, но с использованием тестовых аккаунтов. Мы зайдем в изолированное тестовое пространство Telegram, создадим тестового бота, подготовим фикстуру для запуска нашего приложения и напишем авторизацию для тестовых клиентов.

По ходу статьи я буду использовать инструменты следующих библиотек:

• Python Telegram Bot — для написания бота (сокращенно PTB);

• Pytest — для организации тестов;

• Anyio — для асинхронных тестов и фикстур;

• Pyrogram — для отправки тестовых сообщений;

• asyncio.Event — для оповещения о получении ожидаемых сообщений и предотвращения излишних ожиданий;

• contexlib — для удобного синтаксиса написания контекстных менеджеров.

Статья написана на примере реализации несложного бота для хранения фотографий и отправки их в ответ при запросе. Вы можете посмотреть исходный код приложения и тестов на GitHub. Ниже я использую короткие выдержки из официальной документации Telegram на английском – к ним даю пояснения на русском своими словами.

Альтернативы

Перед тем как что-то писать самому, я пытался найти готовое решение и был уверен, что Google мне поможет. Но оказалось, что готового оптимального решения просто не существует. Здесь я кратко перечислю то, что мне удалось найти и почему это не подошло.

PTB Test

• больше не поддерживается

• интеграция с unittest (не pytest)

Aiogram Tests

• подходит для тестирования ботов, написанных только на aiogram

• реализует unit-тестирование через Mock-объекты (без интеграции Telegram Bot API)

Все тест-кейсы описываются как вызов конкретных handler-ов, из-за чего нет возможности протестировать их поведения целиком — от отправки пользователем сообщения до получения ответа. К примеру, непонятно, как тестировать фильтрацию входящих сообщений и их распределение по различным handler-ам и т. п.

Tg Integration

• также не поддерживается

• тянет устаревшие зависимости (pyrogram < 2.0.0 typing-extensions < 4.0.0)

• практически отсутствует документация

Перед тем как начать. Регистрация APP

Предположим, наш бот уже готов и мы переходим к тестированию. Для этого мы подготовим необходимые инструменты, чтобы написание самих тестов было кратким и понятным. Отправить сообщение, получить ответ, проверить его. Да, придется потратить время, чтобы реализовать вспомогательные фикстуры по регистрации клиентов и обработки входящих сообщений. Но, заморочившись однажды, мы сможем писать сами тесты всего в несколько строк. Их будет легко обновлять и поддерживать.

Отправлять тестовые сообщения мы будем с помощью Pyrogram. Для этого нужно зарегистрировать будущее клиентское приложение.

📍 Тут важно понимать, что для взаимодействия с Telegram API необходимо зарегистрировать свое приложение в системе Telegram — для этого потребуется действующий номер телефона. Но для запуска самих тестов будут использоваться тестовые телефонные номера. Поэтому переживать за сохранность личных данных или получение флуд-бана не стоит.

После регистрации мы получаем api_key и api_id, которые можем использовать для создания множества клиентских приложений.

The API key defines a token for a Telegram application you are going to build. This means that you are able to authorize multiple users or bots with a single API key. (© Telegram Documentation)

Тестовый бот

Регистрация бота

Для тестирования пользовательских приложений Telegram дает доступ к Dedicated test environment (выделенная тестовая среда). Это изолированная песочница, где можно регистрировать пользователей и создавать ботов. Если аккаунты мы будем создавать динамически, используя тестовые номера, то зарегистрировать тестового бота у @BotFather проще вручную, используя свой настоящий номер.

The test environment is completely separate from the main environment, so you will need to create a new user account and a new bot with @BotFather. (© Telegram Documentation)

Как это сделать, можно прочитать тут. Я использовал приложение Telegram для iOS, кликнув 10 раз на иконку Settings (Настройки).

После этого в приложении становится доступен второй аккаунт. Это очень удобное свойство Telegram, о котором знают немногие (лично я раньше не знал). Можно использовать для различного тестирования или даже завести секретный чат с другим человеком (который тоже зарегистрировался в тестовом пространстве). Далее, как обычно, регистрируем бота у @BotFather и получаем токен. Для обращения к любому из методов нужно добавить /test/ в путь запроса.

https://api.telegram.org/bot<token>/test/METHOD_NAME

Запуск бота во время тестов

Теперь при запуске тестов мы можем запустить наше приложение в тестовом пространстве. Для этого подготовим фикстуру.

@pytest.fixture(scope='session') # fixture runs Bot app only once for entire tess session
async def application():
	app = Application.builder().token('<token>' + '/test').build()

	await app.initialize()
	await app.post_init(app) # initialize does *not* call `post_init` - that is only done by run_polling/webhook
	await app.start()
	await app.updater.start_polling()

	yield app

	await app.updater.stop()
	await app.stop()
	await app.shutdown()

Обычно для запуска приложения мы используем app.run_polling(), но это заблокирует дальнейшее исполнение программы. Мы же будем запускать тесты в параллель с тем, как работает наш бот. Подробнее о том, как запускать PTB с другим асинхронным кодом, — Running PTB alongside other asyncio frameworks

📍 pytest сам по себе не асинхронный, но может работать с асинхронными тестами и фикстурами, если добавить pytestmark = pytest.mark.anyio. Подробнее: Testing with AnyIO

📍PTB на данный момент не поддерживает работу бота в тестовом пространстве, но можно передать токен вместе с приставкой /test. Класс telegram.Bot формирует базовый путь сложением base_url и token.

# content of /telegram/_bot.py
self._base_url: str = base_url + self._token
self._base_file_url: str = base_file_url + self._token

Перехватывание исключений

По умолчанию все необработанные исключение в PTB пишутся в лог. Но при запуске тестов мы хотим убедиться, что в приложении не возникало ошибок. Для этого добавим обработчик ошибок, где будем сохранять возникшие исключения в отдельное место (в моем случае это переменная класса). Позже мы сможем проверить на наличие непредвиденных исключений.

async def collect_app_exceptions_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
  self.collected_exception = context.error
  self.collection_event.set()     # special asyncio event we are wating for while collecting replyes
  raise ApplicationHandlerStop  	# prevent any other error handlers
  
app.add_error_handler(self.collect_app_exceptions_callback)

📍 collection_event объект класса asyncio.Event — событие, исполнение которого мы ожидаем, пока ждем ответа от бота. В случае перехватывания исключений ждать ответа дальше смысла нет, так как произошла ошибка. Но pytest об этой ошибке ничего не знает, поэтому сохраним ее здесь и сделаем re-rise позже. Вы увидите инициализацию и полное применение collection_event в следующем блоке.

Отлично, наш бот работает. Теперь можем отправлять ему тестовые сообщения. Для этого понадобится аккаунт.

Тестовый клиент-аккаунт

Тестовые номера

Для тестирования приложения мы будем использовать тестовые номера, которые предоставляет Telegram. Мы можем сформировать тестовые номера заранее и хранить их, к примеру, в .env файле или генерировать на лету для каждой тест-сессии отдельно. Тестовый номер — это:

99966XYYYY
X - номер номер DC от 1 до 3
Y - любая цифра от 0 до 9

Тестовый номер позволяет в автоматическом режиме проходить авторизацию, и мы можем создать сколь угодно много аккаунтов. К примеру, мой бот позволяет давать другим людям доступ к общей базе фотографий. Так я смогу протестировать, что один клиент загрузил фотографию, другой получил к ней доступ, а третий нет.

📍Telegram периодически сбрасывает тестовые номера, поэтому все дальнейшие действия по авторизации и настройке telegram-клиента лучше производить непосредственно перед каждым запуском тестов.

Do not store any important or private information in the messages of such test accounts; anyone can make use of the simplified authorization mechanism – and we periodically wipe all information stored there. (© Telegram Documentation)

Создание клиента

Для упрощенной регистрации клиентов и перехватывания всех сообщений от бота отдельно для каждого клиента я инкапсулировал всю логику в отдельный класс ClientIntegration. Но в примерах здесь я удалил все опциональные настройки, чтобы не отвлекаться от сути. Код вспомогательного класса целиком можно посмотреть тут.

1. Создадим клиента. В отличие от PTB, у Pyrogram есть нативная поддержка тестового пространства Telegram.

self.client = Client(
	'test-client',
	api_id='<api_id>',
	api_hash='<api_hash>',
	test_mode=True,
	in_memory=True,
	phone_number='99966' + '1' + '2023',
	phone_code='1' * 5,
)

• in_memory — после авторизации сессия не будет записываться в файл test-client.session, а останется в памяти. Это позволит изолировать тесты друг от друга и не подчищать *.session файлы вручную после тестов.

• phone_code — код подтверждения авторизации. Для тестовых номеров всегда равен номеру DC х 5 раз.

2. Мы уже указали телефонный номер и код подтверждения. Но может быть такое, что тестовый номер не зарегистрирован. В этом случае Pyrogram попросит ввести имя и фамилию. Чтобы пройти процесс регистрации автоматически, заменим stdout и stdin:

def mock_input_callback(prompt: str = ''):
	if 'Enter first name: ' == prompt:
		return '<first name>'
	if 'Enter last name (empty to skip): ' == prompt:
		return '<last name>'
	raise ValueError(prompt)

def mock_print_callback(self, *args, **kwargs):
	...

with pytest.MonkeyPatch.context() as monkeypatch:
	monkeypatch.setattr(builtins, 'input', mock_input_callback)
	monkeypatch.setattr(builtins, 'print', mock_print_callback)
	...

3. Теперь мы можем запустить приложение клиента Pyrogram. Для удобства я описал все действия в отдельном методе ClientIntegration и обернул его в asynccontextmanager. Я использую здесь контекст-менеджер, чтобы явно описать setup и teardown. Позже это будет удобно интегрировать в Pytest в качестве yield фикстуры.

@asynccontextmanager
async def session_context(self)
	...
	try:
		# Some phonenumbers are registered already, some other not.
		# To be sure, handle sign up action by patching stdin/stdout.
		with pytest.MonkeyPatch.context() as monkeypatch:
			monkeypatch.setattr(builtins, 'input', mock_input_callback)
			await self.client.start()

		# Update Telegram User properties, if they are needed for tests cases.
		await self.client.update_profile('<first name>', '<last name>', '<bio>')
		if self.client.me.username != self.credits.username:
			await self.client.set_username('<username>')

		# Clear messages to make each test isolated from others.
		await client.invoke(
			DeleteHistory(peer=await client.resolve_peer('<bot name>'), max_id=0, just_clear=False)
		)

		yield self

	finally:
		await self.client.set_username(None) # cleaning up
		await self.client.stop()

📍 Если для авторизации клиента использовать случайно сгенерированный тестовый номер, то стоит учесть, что другой разработчик мог установить двухфакторную авторизацию для этого номера, защитив его паролем. В таком случае остается только попробовать любой другой номер.

4. В итоге можем использовать ClientIntegration.session_context для описания фикстур.

@pytest.fixture(scope='session')
async def integration_1():
	async with ClientIntegration().session_context() as integration_1:
		yield integration_1

@pytest.fixture(scope='session')
async def integration_2():
	async with ClientIntegration().session_context() as integration_2:
		yield integration_2

...

📍 Я использую параметр scope='session', чтобы инициализировать клиента один раз для всей сессии. Это сократит время на подготовку фикстур, а поскольку сам клиент не хранит данные, это не нарушит изолированности тестов.

Тестирование

Сборщик ответов

Для тестов необходимо отправлять сообщения боту и получать от него ответы. Если с первым все понятно, то для второго нужно повесить обработчик сообщений, где мы будем складывать все полученные сообщения в список, чтобы потом сделать assert на содержание полученных сообщений.

Но как долго нам ждать ответа от бота? Можно задать максимальное значение timeout, а можно указать конкретное количество. К примеру, сейчас ожидаем получить только одно сообщение, и как только оно будет получено, можно двигаться дальше.

# messages collector:
async def collect_replyes_callback(self, client: Client, message: Message)
	self.collected_replyes.append(message)

	# If True, set event flag to True. No more waiting for messages:
	if len(self.collected_replyes) == self.collection_required_amount:
		self.collection_event.set()

# timeout checker:
async def collection_max_timeout_waiting(self, timeout: float):
	await sleep(timeout)
	self.collection_event.set()

# apply handler and create Event:
self.client.on_message(filters.chat('<bot name>'))(self.collect_replyes_callback)
self.collection_event = asyncio.Event()

# Wait until all messages are received or reaching timeout.
timout_task = asyncio.create_task(self.collection_max_timeout_waiting(timeout))
await self.collection_event.wait()
timout_task.cansel()

Теперь мы можем проверить полученные сообщения и то, что в приложении не было ошибок (мы использовали error_handler при инициализации приложения PTB в предыдущем блоке).

if self.collected_exceptions:
	raise self.collected_exception
assert len(self.collected_replyes) == amount, 'Received unexpected messages amount. '

Для удобства я описал все эти действия в отдельном методе ClientIntegration и также обернул его в @asynccontextmanager. При выходе из контекстного менеджера collect сработает блок finally, где мы будем ожидать ответы от бота.

@asynccontextmanager
async def collect(self, *, amount: int, timeout: float = 2.0):
	self.collection_amount = amount
	self.collected_replyes.clear()
	self.collection_event = asyncio.Event()

	try:
		yield self.collected_replyes
	finally:
		timout_event = asyncio.create_task(self.collection_max_timeout_waiting(timeout))
		await self.collection_event.wait()
		timout_event.cancel()

		if self.collected_exceptions:
			raise self.collected_exception
		assert len(self.collected_replyes) == amount, 'Received unexpected messages amount. '

Описание тестов

К примеру, мы хотим протестировать отправку сообщения /start и получение ответа Hello, {username}!. Благодаря фикстурам, описанным выше, этот тест-кейс может быть описан буквально тремя строчками.

pytestmark = [
	pytest.mark.anyio,
	pytest.mark.usefixtures('application')  # PTB Application
]

async def test_start_handler(integration: ClientIntegration):
	# test action:
	async with integration.collect(amount=1) as replyes:
		await integration.client.send_message('<bot name>', '/start')
	# at __aexit__ wait until all messages are received

	# test assertion:
	assert replyes[0].text == f'Hellow, {username}!'

📍 В качестве альтернативы Pyrogram может быть Telethon. Это схожая библиотека для работы с клиентом Telegram. Из плюсов — Telethon реализует метод conversation, что упрощает сборку ответов клиенту от бота. Подробнее можно посмотреть тут.

with client.conversation('<bot name>') as conv:
	await conv.send_message("/start")

	reply: Message = await conv.get_response()
	assert reply.raw_text == f'Hellow, {username}!'

Заключение

Потратив дополнительное время на подготовку фикстур, мы получаем все необходимые инструменты для написания тестов от начала и до конца. Мы можем создать несколько пользователей, отправлять им разные тестовые сообщения и проверять ответы. Это удобно не только для написания тестов в целом, но и во время разработки — как точка входа для запуска и отладки. Вместо того чтобы каждый раз доставать телефон и вручную отправлять сообщения боту, можно описать простой тест на отправку сообщения клиентом.

Но стоит понимать, что такой способ тестирования не самый дешевый с точки зрения реализации, поддержки и времени исполнения самих тестов. Каждый запрос на регистрацию клиента, отправку сообщения клиентом, на отправку ответа ботом происходит посредством реальных запрос к серверу Telegram. А это занимает время. Когда подобных тестов станет много, это может стать проблемой. Вот почему для написания тестов стоит придерживаться парадигмы Testing Pyramid и реализовывать подобным методом только базовые и важные функции вашего приложения.