/tgintegration

Integration test and automation library for Telegram Messenger bots based on Pyrogram.

Primary LanguagePythonMIT LicenseMIT

TgIntegration

An integration test and automation library for Telegram Bots based on Pyrogram.
Test your bot in realtime scenarios!

Are you a user of TgIntegration? I'm actively looking for feedback and ways to improve the library, come and let me know in the official group!

PyPI - Python Version PyPI - Downloads PyPI GitHub top language GitHub Workflow Status (branch) GitHub Workflow Status

FeaturesRequirementsInstallationQuick Start GuideTest Frameworks

Features

▶️ See it in action! 🎬

  • 👤 Log into a Telegram user account and interact with bots or other users
  • ✅ Write realtime integration tests to ensure that your bot works as expected! ▶️ Pytest examples
  • ⚡️ Automate any interaction on Telegram! ▶️ Automatically play @IdleTownBot | More examples
  • 🛡 Fully typed for safety and autocompletion with your favorite IDE
  • 🐍 Built for modern Python (3.8+) with high test coverage

Prerequisites

Same as Pyrogram:

  • A Telegram API key.
  • A user session (seeing things happen in your own account is great for getting started)
  • But: Python 3.8 or higher!

A basic understanding of async/await and asynchronous context managers is assumed, as TgIntegration heavily relies on the latter to automate conversations.

Installation

All hail pip!

$ pip install tgintegration --upgrade


Feeling adventurous?

For bleeding edge, install the master branch:

$ pip install git+https://github.com/JosXa/tgintegration.git

Quick Start Guide

You can follow along by running the example (README)

Setup

Suppose we want to write integration tests for @BotListBot by sending it a couple of messages and checking that it responds the way it should.

After configuring a Pyrogram user client, let's start by creating a BotController:

from tgintegration import BotController

controller = BotController(
    peer="@BotListBot",      # The bot under test is https://t.me/BotListBot 🤖
    client=client,           # This assumes you already have a Pyrogram user client available
    max_wait=8,              # Maximum timeout for responses (optional)
    wait_consecutive=2,      # Minimum time to wait for more/consecutive messages (optional)
    raise_no_response=True,  # Raise `InvalidResponseError` when no response is received (defaults to True)
    global_action_delay=2.5  # Choosing a rather high delay so we can observe what's happening (optional)
)

await controller.clear_chat()  # Start with a blank screen (⚠️)

Now, let's send /start to the bot and wait until exactly three messages have been received by using the asynchronous collect context manager:

async with controller.collect(count=3) as response:
    await controller.send_command("start")

assert response.num_messages == 3  # Three messages received, bundled under a `Response` object
assert response.messages[0].sticker  # The first message is a sticker

The result should look like this:

image

Examining the buttons in the response...

# Get first (and only) inline keyboard from the replies
inline_keyboard = response.inline_keyboards[0]

# Three buttons in the first row
assert len(inline_keyboard.rows[0]) == 3

We can also press the inline keyboard buttons, for example based on a regular expression:

examples = await inline_keyboard.click(pattern=r".*Examples")

As the bot edits the message, .click() automatically listens for "message edited" updates and returns the new state as another Response.

image

assert "Examples for contributing to the BotList" in examples.full_text

Error handling

So what happens when we send an invalid query or the peer fails to respond?

The following instruction will raise an InvalidResponseError after controller.max_wait seconds. This is because we passed raise_no_response=True during controller initialization.

try:
    async with controller.collect():
        await controller.send_command("ayylmao")
except InvalidResponseError:
    pass  # OK

Let's explicitly set raise_ to False so that no exception occurs:

async with controller.collect(raise_=False) as response:
    await client.send_message(controller.peer_id, "Henlo Fren")

In this case, tgintegration will simply emit a warning, but you can still assert that no response has been received by using the is_empty property:

assert response.is_empty

Integrating with Test Frameworks

Pytest is the recommended test framework for use with tgintegration. You can browse through several examples and tgintegration also uses pytest for its own test suite.

unittest

I haven't tried out the builtin unittest library in combination with tgintegration yet, but theoretically I don't see any problems with it. If you do decide to try it, it would be awesome if you could tell me about your experience and whether anything could be improved 🙂 Let us know at 👉 https://t.me/TgIntegration or in an issue.