### 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 ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.