tgintegration

https://josxa.github.io/tgintegration/
Admin
tgintegration is a Python library that helps you test your bots and automate routine tasks on...
Tokens:3,352
Snippets:38
Trust Score:9.7
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install TGIntegration via pip

Source: https://josxa.github.io/tgintegration/setup

Standard installation command for the latest stable version of the library.

```bash
pip install tgintegration --upgrade
```

--------------------------------

### Install TGIntegration from master branch

Source: https://josxa.github.io/tgintegration/setup

Installation command for the bleeding edge version directly from the GitHub repository.

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

--------------------------------

### Initialize BotController

Source: https://josxa.github.io/tgintegration/getting-started

Configures the controller with the target bot and Pyrogram client. Ensure a Pyrogram user client is already available before initialization.

```python
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 (⚠️)
```

--------------------------------

### ReplyKeyboard.click

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.containers.reply_keyboard

Clicks a button by pattern and waits for the bot's response.

```APIDOC
## click(pattern, filters=None, quote=False)

### Description
Locates a button by pattern, sends the button's text as a message to the chat, and waits for the bot's response.

### Parameters
#### Arguments
- **pattern** (Pattern) - Required - The regex pattern to match the button caption.
- **filters** (Filter) - Optional - Additional filters for collecting the response.
- **quote** (bool) - Optional - Whether to reply to the message containing the buttons.

### Response
- **Response** - The bot's response object.
```

--------------------------------

### ReplyKeyboard Class Implementation

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.containers.reply_keyboard

The ReplyKeyboard class handles button retrieval and interaction within a Telegram chat context.

```python
"""

"""
import re
from typing import List
from typing import Pattern
from typing import TYPE_CHECKING
from typing import Union

from pyrogram import filters as f
from pyrogram.filters import Filter
from pyrogram.types import KeyboardButton
from pyrogram.types import Message

from tgintegration.containers import NoButtonFound

if TYPE_CHECKING:
    from tgintegration.botcontroller import BotController
    from tgintegration.containers.responses import Response


class ReplyKeyboard:DOCS
    """
    Represents a regular keyboard in the Telegram UI and allows to click buttons in the menu.

    See Also:
        [InlineKeyboard](tgintegration.InlineKeyboard)
    """

    def __init__(
        self,
        controller: "BotController",
        chat_id: Union[int, str],
        message_id: int,
        button_rows: List[List[KeyboardButton]],
    ):
        self._controller: BotController = controller
        self._message_id = message_id
        self._peer_id = chat_id
        self.rows = button_rows

    def find_button(self, pattern: Pattern) -> KeyboardButton:DOCS
        """
        Attempts to retrieve a clickable button anywhere in the underlying `rows` by matching the button captions with
        the given `pattern`. If no button could be found, **this method raises** `NoButtonFound`.

        Args:
            pattern: The button caption to look for (by `re.match`).

        Returns:
            The `KeyboardButton` if found.
        """
        compiled = re.compile(pattern)
        for row in self.rows:
            for button in row:
                # TODO: Investigate why sometimes it's a button and other times a string
                if compiled.match(button.text if hasattr(button, "text") else button):
                    return button
        raise NoButtonFound(f"No clickable entity found for pattern r'{pattern}'")

    async def _click_nowait(self, pattern, quote=False) -> Message:
        button = self.find_button(pattern)

        return await self._controller.client.send_message(
            self._peer_id,
            button.text,
            reply_to_message_id=self._message_id if quote else None,
        )

    @propertyDOCS
    def num_buttons(self) -> int:
        """
        Returns the total number of buttons in all underlying rows.
        """
        return sum(len(row) for row in self.rows)

    async def click(DOCS
        self, pattern: Pattern, filters: Filter = None, quote: bool = False
    ) -> "Response":
        """
        Uses `find_button` with the given `pattern`, clicks the button if found, and waits for the bot to react. For
        a `ReplyKeyboard`, this means that a message with the button's caption will be sent to the same chat.

        If not button could be found, `NoButtonFound` will be raised.

        Args:
            pattern: The button caption to look for (by `re.match`).
            filters: Additional filters to be given to `collect`. Will be merged with a "same chat" filter and
                `filters.text | filters.edited`.
            quote: Whether to reply to the message containing the buttons.

        Returns:
            The bot's `Response`.
        """
        button = self.find_button(pattern)

        filters = (
            filters & f.chat(self._peer_id) if filters else f.chat(self._peer_id)
        ) & (f.text | f.edited)

        async with self._controller.collect(filters=filters) as res:  # type: Response
            await self._controller.client.send_message(
                self._controller.peer,
                button.text if hasattr(button, "text") else button,
                reply_to_message_id=self._message_id if quote else None,
            )

        return res
```

--------------------------------

### Define Expected Peer Reaction with Expectation

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.expectation

Use the Expectation class to define the minimum and maximum number of messages expected from a peer. Initialize with NotSet for unbounded limits.

```python
from dataclasses import dataclass
from typing import List
from typing import Union

from pyrogram.types import Message

from tgintegration.containers.responses import InvalidResponseError
from tgintegration.timeout_settings import TimeoutSettings
from tgintegration.utils.sentinel import NotSet

logger = logging.getLogger(__name__)


@dataclass
class Expectation:
    min_messages: Union[int, NotSet] = NotSet
    max_messages: Union[int, NotSet] = NotSet
```

--------------------------------

### Inspect Inline Keyboards

Source: https://josxa.github.io/tgintegration/getting-started

Accesses inline keyboards from the response object to verify button layout.

```python
# 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
```

--------------------------------

### Expectation Class

Source: https://josxa.github.io/tgintegration/api/tgintegration.expectation

Defines the expected reaction of a peer using min_messages and max_messages constraints.

```APIDOC
## Class: tgintegration.expectation.Expectation

### Description
Defines the expected reaction of a peer.

### Parameters
- **min_messages** (NotSet) - Optional - The minimum number of messages expected.
- **max_messages** (NotSet) - Optional - The maximum number of messages expected.
```

--------------------------------

### Handle Response Errors

Source: https://josxa.github.io/tgintegration/getting-started

Demonstrates how to handle InvalidResponseError when a bot fails to respond within the configured timeout.

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

--------------------------------

### ReplyKeyboard.find_button

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.containers.reply_keyboard

Searches for a button within the keyboard rows using a regex pattern.

```APIDOC
## find_button(pattern)

### Description
Attempts to retrieve a clickable button from the keyboard rows by matching the button caption against a regex pattern.

### Parameters
#### Arguments
- **pattern** (Pattern) - Required - The regex pattern to match against the button caption.

### Response
- **KeyboardButton** - The found button object.

### Errors
- **NoButtonFound** - Raised if no button matches the provided pattern.
```

--------------------------------

### Interact with Inline Buttons

Source: https://josxa.github.io/tgintegration/getting-started

Clicks buttons on an inline keyboard using a regex pattern and verifies the resulting message text.

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

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

--------------------------------

### Collect Bot Responses

Source: https://josxa.github.io/tgintegration/getting-started

Uses the collect context manager to send a command and wait for a specific number of incoming messages.

```python
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
```

--------------------------------

### Check if Minimum Messages Received

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.expectation

The is_sufficient method checks if at least the minimum number of messages have been received. If min_messages is NotSet, it defaults to requiring at least one message.

```python
def is_sufficient(self, messages: List[Message]) -> bool:
    n = len(messages)
    if self.min_messages is NotSet:
        return n >= 1
    return n >= self.min_messages
```

--------------------------------

### Verify Message Count Against Expectation

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.expectation

The verify method checks if the received messages meet the min_messages and max_messages criteria. It raises an InvalidResponseError or logs a debug message if the criteria are not met, based on timeout settings.

```python
def _is_match(self, messages: List[Message]) -> bool:
    n = len(messages)
    return (self.min_messages is NotSet or n >= self.min_messages) and (
        self.max_messages is NotSet or n <= self.max_messages
    )

def verify(self, messages: List[Message], timeouts: TimeoutSettings) -> None:
    if self._is_match(messages):
        return

    n = len(messages)

    if n < self.min_messages:
        _raise_or_log(
            timeouts,
            "Expected {} messages but only received {} after waiting {} seconds.",
            self.min_messages,
            n,
            timeouts.max_wait,
        )
        return

    if n > self.max_messages:
        _raise_or_log(
            timeouts,
            "Expected only {} messages but received {}.",
            self.max_messages,
            n,
        )
        return
```

--------------------------------

### InvalidResponseError Exception

Source: https://josxa.github.io/tgintegration/api/tgintegration.containers.responses

Details about the InvalidResponseError exception, raised when a peer's response does not match expectations.

```APIDOC
## InvalidResponseError

### Description
Raised when peer's response did not match the expectation.

### Type
Exception
```

--------------------------------

### Raise or Log Response Errors

Source: https://josxa.github.io/tgintegration/api/source/tgintegration.expectation

The _raise_or_log helper function handles error reporting. It raises an InvalidResponseError if raise_on_timeout is True, otherwise it logs a debug message.

```python
def _raise_or_log(timeouts: TimeoutSettings, msg: str, *fmt) -> None:
    if timeouts.raise_on_timeout:
        if fmt:
            raise InvalidResponseError(msg.format(*fmt))
        else:
            raise InvalidResponseError(msg)
    logger.debug(msg, *fmt)
```

--------------------------------

### Disable Exception Raising

Source: https://josxa.github.io/tgintegration/getting-started

Configures the collect context manager to suppress exceptions when no response is received, allowing for manual verification.

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

```python
assert response.is_empty
```
Tokens:
Raw