diff --git a/changes/unreleased/5229.87PBN4GFkuAaDhhgFwCYkY.toml b/changes/unreleased/5229.87PBN4GFkuAaDhhgFwCYkY.toml index 7bb836c47c6..04754f99219 100644 --- a/changes/unreleased/5229.87PBN4GFkuAaDhhgFwCYkY.toml +++ b/changes/unreleased/5229.87PBN4GFkuAaDhhgFwCYkY.toml @@ -3,4 +3,6 @@ highlights = "Full Support for Bot API 10.0" pull_requests = [ { uid = "5229", author_uid = "aelkheir", closes_threads = ["5228"] }, { uid = "5230", author_uid = "harshil21" }, + { uid = "5235", author_uid = "harshil21" }, + { uid = "5238", author_uid = "harshil21" }, ] diff --git a/docs/source/inclusions/bot_methods.rst b/docs/source/inclusions/bot_methods.rst index 8ee2734a81f..3b9fdfd626c 100644 --- a/docs/source/inclusions/bot_methods.rst +++ b/docs/source/inclusions/bot_methods.rst @@ -171,6 +171,8 @@ - Used for unpinning a message * - :meth:`~telegram.Bot.unpin_all_chat_messages` - Used for unpinning all pinned chat messages + * - :meth:`~telegram.Bot.get_user_personal_chat_messages` + - Used for obtaining the personal chat messages of a user * - :meth:`~telegram.Bot.get_user_profile_audios` - Used for obtaining user's profile audios * - :meth:`~telegram.Bot.get_user_profile_photos` @@ -241,6 +243,10 @@ - Used for obtaining the menu button of a private chat or the default menu button * - :meth:`~telegram.Bot.set_chat_menu_button` - Used for setting the menu button of a private chat or the default menu button + * - :meth:`~telegram.Bot.set_managed_bot_access_settings` + - Used for changing the access settings of a managed bot + * - :meth:`~telegram.Bot.get_managed_bot_access_settings` + - Used for obtaining the access settings of a managed bot * - :meth:`~telegram.Bot.set_my_description` - Used for setting the description of the bot * - :meth:`~telegram.Bot.get_my_description` diff --git a/docs/source/telegram.at-tree.rst b/docs/source/telegram.at-tree.rst index 25d9c269838..716f7ab4361 100644 --- a/docs/source/telegram.at-tree.rst +++ b/docs/source/telegram.at-tree.rst @@ -8,6 +8,7 @@ Available Types telegram.animation telegram.audio telegram.birthdate + telegram.botaccesssettings telegram.botcommand telegram.botcommandscope telegram.botcommandscopeallchatadministrators diff --git a/docs/source/telegram.botaccesssettings.rst b/docs/source/telegram.botaccesssettings.rst new file mode 100644 index 00000000000..788689b9266 --- /dev/null +++ b/docs/source/telegram.botaccesssettings.rst @@ -0,0 +1,6 @@ +BotAccessSettings +================= + +.. autoclass:: telegram.BotAccessSettings + :members: + :show-inheritance: \ No newline at end of file diff --git a/src/telegram/__init__.py b/src/telegram/__init__.py index 362f3253198..db9aa78a7fb 100644 --- a/src/telegram/__init__.py +++ b/src/telegram/__init__.py @@ -35,6 +35,7 @@ "BackgroundTypeWallpaper", "Birthdate", "Bot", + "BotAccessSettings", "BotCommand", "BotCommandScope", "BotCommandScopeAllChatAdministrators", @@ -348,6 +349,7 @@ from . import _version, constants, error, helpers, request, warnings from ._birthdate import Birthdate from ._bot import Bot +from ._botaccesssettings import BotAccessSettings from ._botcommand import BotCommand from ._botcommandscope import ( BotCommandScope, diff --git a/src/telegram/_bot.py b/src/telegram/_bot.py index 3491a75cffa..5493cb21709 100644 --- a/src/telegram/_bot.py +++ b/src/telegram/_bot.py @@ -47,7 +47,8 @@ serialization = None # type: ignore[assignment] CRYPTO_INSTALLED = False -from telegram._botcommand import BotCommand # pylint: disable=ungrouped-imports +from telegram._botaccesssettings import BotAccessSettings # pylint: disable=ungrouped-imports +from telegram._botcommand import BotCommand from telegram._botcommandscope import BotCommandScope from telegram._botdescription import BotDescription, BotShortDescription from telegram._botname import BotName @@ -1209,7 +1210,7 @@ async def send_message_draft( self, chat_id: int, draft_id: int, - text: str, + text: str | None = None, message_thread_id: int | None = None, parse_mode: ODVInput[str] = DEFAULT_NONE, entities: Sequence["MessageEntity"] | None = None, @@ -1221,7 +1222,9 @@ async def send_message_draft( api_kwargs: JSONDict | None = None, ) -> bool: """Use this method to stream a partial message to a user while the message is being - generated. + generated. Note that the streamed draft is ephemeral and acts as a temporary 30-second + preview - once the output is finalized, you must call :meth:`~Bot.send_message` with + the complete message to persist it in the user's chat. .. versionadded:: 22.6 @@ -1233,19 +1236,21 @@ async def send_message_draft( chat_id (:obj:`int`): Unique identifier for the target private chat. draft_id (:obj:`int`): Unique identifier of the message draft; must be non-zero. Changes of drafts with the same identifier are animated. - text (:obj:`str`): Text of the message to be sent, - :tg-const:`telegram.constants.MessageLimit.MIN_TEXT_LENGTH`- - :tg-const:`telegram.constants.MessageLimit.MAX_TEXT_LENGTH` characters after - entities parsing. + text (:obj:`str`, optional): Text of the message to be sent, + 0-:tg-const:`telegram.constants.MessageLimit.MAX_TEXT_LENGTH` characters after + entities parsing. Pass an empty text to show a "Thinking..." placeholder. + + .. versionchanged:: NEXT.VERSION + Bot API 10.0 now makes this an optional parameter. + + message_thread_id (:obj:`int`, optional): Unique identifier for the target + message thread. parse_mode (:obj:`str`): |parse_mode| entities (Sequence[:class:`telegram.MessageEntity`], optional): Sequence of special entities that appear in message text, which can be specified instead of :paramref:`parse_mode`. |sequenceargs| - message_thread_id (:obj:`int`, optional): Unique identifier for the target - message thread. - Returns: :obj:`bool`: On success, :obj:`True` is returned. @@ -11127,6 +11132,145 @@ async def edit_user_star_subscription( api_kwargs=api_kwargs, ) + async def get_managed_bot_access_settings( + self, + user_id: int, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + ) -> BotAccessSettings: + """ + Use this method to get the access settings of a managed bot. + + .. versionadded:: NEXT.VERSION + + Args: + user_id (:obj:`int`): User identifier of the managed bot whose access settings will be + returned. + + Returns: + :class:`telegram.BotAccessSettings`: The access settings of the managed bot. + + Raises: + :class:`telegram.error.TelegramError` + """ + + data: JSONDict = { + "user_id": user_id, + } + + return BotAccessSettings.de_json( + await self._post( + "getManagedBotAccessSettings", + data, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=api_kwargs, + ), + self, + ) + + async def set_managed_bot_access_settings( + self, + user_id: int, + is_access_restricted: bool, + added_user_ids: Sequence[int] | None = None, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + ) -> bool: + """ + Use this method to change the access settings of a managed bot. + + .. versionadded:: NEXT.VERSION + + Args: + user_id (:obj:`int`): User identifier of the managed bot whose access settings will be + changed. + is_access_restricted (:obj:`bool`): Pass :obj:`True`, if only selected users can access + the bot. The bot's owner can always access it. + added_user_ids (Sequence[:obj:`int`], optional): A list of up to + :tg-const:`telegram.constants.ManagedBotAccessLimit.MAX_ALLOWED_USERS` + identifiers of users who will have access to the bot in addition to its owner. + Ignored if :paramref:`is_access_restricted` is :obj:`False`. + + Returns: + :obj:`bool`: On success, :obj:`True` is returned. + + Raises: + :class:`telegram.error.TelegramError` + """ + + data: JSONDict = { + "user_id": user_id, + "is_access_restricted": is_access_restricted, + "added_user_ids": added_user_ids, + } + + return await self._post( + "setManagedBotAccessSettings", + data, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=api_kwargs, + ) + + async def get_user_personal_chat_messages( + self, + user_id: int, + limit: int, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + ) -> tuple[Message, ...]: + """ + Use this method to get the last messages from the personal chat (i.e., the chat currently + added to their profile) of a given user. + + .. versionadded:: NEXT.VERSION + + Args: + user_id (:obj:`int`): Unique identifier of the target user. + limit (:obj:`int`): The maximum number of messages to return; + :tg-const:`telegram.constants.PersonalChatMessagesLimit.MIN_LIMIT`- + :tg-const:`telegram.constants.PersonalChatMessagesLimit.MAX_LIMIT`. + + Returns: + tuple[:class:`telegram.Message`, ...]: On success, a tuple of + :class:`telegram.Message` objects is returned. + + Raises: + :class:`telegram.error.TelegramError` + """ + + data: JSONDict = {"user_id": user_id, "limit": limit} + + return Message.de_list( + await self._post( + "getUserPersonalChatMessages", + data, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=api_kwargs, + ), + self, + ) + async def send_paid_media( self, chat_id: str | int, @@ -12733,6 +12877,12 @@ def to_dict(self, recursive: bool = True) -> JSONDict: # noqa: ARG002 """Alias for :meth:`replace_managed_bot_token`""" savePreparedKeyboardButton = save_prepared_keyboard_button """Alias for :meth:`save_prepared_keyboard_button`""" + getManagedBotAccessSettings = get_managed_bot_access_settings + """Alias for :meth:`get_managed_bot_access_settings`""" + setManagedBotAccessSettings = set_managed_bot_access_settings + """Alias for :meth:`set_managed_bot_access_settings`""" + getUserPersonalChatMessages = get_user_personal_chat_messages + """Alias for :meth:`get_user_personal_chat_messages`""" deleteMessageReaction = delete_message_reaction """Alias for :meth:`delete_message_reaction`""" deleteAllMessageReactions = delete_all_message_reactions diff --git a/src/telegram/_botaccesssettings.py b/src/telegram/_botaccesssettings.py new file mode 100644 index 00000000000..745d63ae67b --- /dev/null +++ b/src/telegram/_botaccesssettings.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2026 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +"""This module contains an object that represents a Telegram Bot Access Settings.""" + +from collections.abc import Sequence +from typing import TYPE_CHECKING + +from telegram._telegramobject import TelegramObject +from telegram._user import User +from telegram._utils.argumentparsing import de_list_optional, parse_sequence_arg +from telegram._utils.types import JSONDict + +if TYPE_CHECKING: + from telegram import Bot + + +class BotAccessSettings(TelegramObject): + """ + This object describes the access settings of a bot. + + Objects of this class are comparable in terms of equality. Two objects of this class are + considered equal, if their :attr:`is_access_restricted` and :attr:`added_users` are equal. + + .. versionadded:: NEXT.VERSION + + Args: + is_access_restricted (:obj:`bool`): :obj:`True`, if only selected users can access the bot. + The bot's owner can always access it. + added_users (Sequence[:class:`telegram.User`], optional): The list of other users who + have access to the bot if the access is restricted. + + Attributes: + is_access_restricted (:obj:`bool`): :obj:`True`, if only selected users can access the bot. + The bot's owner can always access it. + added_users (Sequence[:class:`telegram.User`]): Optional. The list of other users who + have access to the bot if the access is restricted. + """ + + __slots__ = ("added_users", "is_access_restricted") + + def __init__( + self, + is_access_restricted: bool, + added_users: Sequence[User] | None = None, + *, + api_kwargs: JSONDict | None = None, + ): + super().__init__(api_kwargs=api_kwargs) + self.is_access_restricted: bool = is_access_restricted + self.added_users: tuple[User, ...] = parse_sequence_arg(added_users) + + self._id_attrs = (self.is_access_restricted, self.added_users) + self._freeze() + + @classmethod + def de_json(cls, data: JSONDict, bot: "Bot | None" = None) -> "BotAccessSettings": + """See :meth:`telegram.TelegramObject.de_json`.""" + data = cls._parse_data(data) + data["added_users"] = de_list_optional(data.get("added_users"), User, bot) + + return super().de_json(data=data, bot=bot) diff --git a/src/telegram/_chat.py b/src/telegram/_chat.py index c0164f36270..f3842aaf4ec 100644 --- a/src/telegram/_chat.py +++ b/src/telegram/_chat.py @@ -1090,7 +1090,7 @@ async def send_message( async def send_message_draft( self, draft_id: int, - text: str, + text: str | None = None, message_thread_id: int | None = None, parse_mode: ODVInput[str] = DEFAULT_NONE, entities: Sequence["MessageEntity"] | None = None, @@ -1107,6 +1107,9 @@ async def send_message_draft( For the documentation of the arguments, please see :meth:`telegram.Bot.send_message_draft`. + .. versionchanged:: NEXT.VERSION + Bot API 10.0 makes the ``text`` argument optional. + Returns: :obj:`bool`: On success, :obj:`True` is returned. diff --git a/src/telegram/_message.py b/src/telegram/_message.py index b87b8bdae4b..9ac9ee3eeb2 100644 --- a/src/telegram/_message.py +++ b/src/telegram/_message.py @@ -2186,7 +2186,7 @@ async def reply_text( async def reply_text_draft( self, draft_id: int, - text: str, + text: str | None = None, parse_mode: ODVInput[str] = DEFAULT_NONE, entities: Sequence["MessageEntity"] | None = None, message_thread_id: ODVInput[int] = DEFAULT_NONE, @@ -2213,6 +2213,9 @@ async def reply_text_draft( .. versionadded:: 22.6 + .. versionchanged:: NEXT.VERSION + Bot API 10.0 makes the ``text`` argument optional. + Returns: :obj:`bool`: On success, :obj:`True` is returned. diff --git a/src/telegram/_user.py b/src/telegram/_user.py index a2b53d6fede..2bf6c8af2e9 100644 --- a/src/telegram/_user.py +++ b/src/telegram/_user.py @@ -42,6 +42,7 @@ from telegram import ( Animation, Audio, + BotAccessSettings, Contact, Document, Gift, @@ -528,7 +529,7 @@ async def send_message( async def send_message_draft( self, draft_id: int, - text: str, + text: str | None = None, message_thread_id: int | None = None, parse_mode: ODVInput[str] = DEFAULT_NONE, entities: Sequence["MessageEntity"] | None = None, @@ -550,6 +551,9 @@ async def send_message_draft( .. versionadded:: 22.6 + .. versionchanged:: NEXT.VERSION + Bot API 10.0 makes the ``text`` argument optional. + Returns: :obj:`bool`: On success, :obj:`True` is returned. @@ -2772,6 +2776,81 @@ async def replace_token( api_kwargs=api_kwargs, ) + async def get_managed_bot_access_settings( + self, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + ) -> "BotAccessSettings": + """ + Shortcut for:: + + await bot.get_managed_bot_access_settings( + user_id=update.effective_user.id, + *args, **kwargs + ) + + For the documentation of the arguments, please see + :meth:`telegram.Bot.get_managed_bot_access_settings`. + + .. versionadded:: NEXT.VERSION + + Returns: + :class:`telegram.BotAccessSettings`: On success, returns the access settings of the bot + managed by the user. + """ + + return await self.get_bot().get_managed_bot_access_settings( + user_id=self.id, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=api_kwargs, + ) + + async def set_managed_bot_access_settings( + self, + is_access_restricted: bool, + added_user_ids: Sequence[int] | None = None, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + ) -> bool: + """ + Shortcut for:: + + await bot.set_managed_bot_access_settings( + user_id=update.effective_user.id, + *args, **kwargs + ) + + For the documentation of the arguments, please see + :meth:`telegram.Bot.set_managed_bot_access_settings`. + + .. versionadded:: NEXT.VERSION + + Returns: + :obj:`bool`: On success, :obj:`True` is returned. + """ + + return await self.get_bot().set_managed_bot_access_settings( + user_id=self.id, + is_access_restricted=is_access_restricted, + added_user_ids=added_user_ids, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=api_kwargs, + ) + async def delete_reaction( self, chat_id: int | str, @@ -2813,6 +2892,44 @@ async def delete_reaction( api_kwargs=api_kwargs, ) + async def get_personal_chat_messages( + self, + limit: int, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + ) -> tuple["Message", ...]: + """ + Shortcut for:: + + await bot.get_user_personal_chat_messages( + user_id=update.effective_user.id, + *args, **kwargs + ) + + For the documentation of the arguments, please see + :meth:`telegram.Bot.get_user_personal_chat_messages`. + + .. versionadded:: NEXT.VERSION + + Returns: + tuple[:class:`telegram.Message`]: On success, a tuple of messages from the personal + channel chat is returned. + """ + + return await self.get_bot().get_user_personal_chat_messages( + user_id=self.id, + limit=limit, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=api_kwargs, + ) + async def delete_all_reactions( self, chat_id: int | str, diff --git a/src/telegram/constants.py b/src/telegram/constants.py index fb0be0b48d2..a71ade00d3d 100644 --- a/src/telegram/constants.py +++ b/src/telegram/constants.py @@ -87,6 +87,7 @@ "KeyboardButtonRequestUsersLimit", "KeyboardButtonStyle", "LocationLimit", + "ManagedBotAccessLimit", "MaskPosition", "MediaGroupLimit", "MenuButtonType", @@ -101,6 +102,7 @@ "OwnedGiftType", "PaidMediaType", "ParseMode", + "PersonalChatMessagesLimit", "PollLimit", "PollType", "PollingLimit", @@ -2568,6 +2570,28 @@ class PaidMediaType(StringEnum): """:obj:`str`: The type of :class:`telegram.PaidMediaPhoto`.""" +class PersonalChatMessagesLimit(IntEnum): + """This enum contains limitations for + :paramref:`telegram.Bot.get_user_personal_chat_messages.limit`. + The enum members of this enumeration are instances of :class:`int` and can be treated as such. + + .. versionadded:: NEXT.VERSION + """ + + __slots__ = () + + MIN_LIMIT = 1 + """:obj:`int`: Minimum value allowed for the + :paramref:`~telegram.Bot.get_user_personal_chat_messages.limit` + parameter of :meth:`telegram.Bot.get_user_personal_chat_messages`. + """ + MAX_LIMIT = 20 + """:obj:`int`: Maximum value allowed for the + :paramref:`~telegram.Bot.get_user_personal_chat_messages.limit` + parameter of :meth:`telegram.Bot.get_user_personal_chat_messages`. + """ + + class PollingLimit(IntEnum): """This enum contains limitations for :paramref:`telegram.Bot.get_updates.limit`. The enum members of this enumeration are instances of :class:`int` and can be treated as such. @@ -4039,6 +4063,22 @@ class ReactionEmoji(StringEnum): """:obj:`str`: Pouting face""" +class ManagedBotAccessLimit(IntEnum): + """This enum contains limitations for :meth:`~telegram.Bot.set_managed_bot_access_settings`. + The enum members of this enumeration are instances of :class:`int` and can be treated as such. + + .. versionadded:: NEXT.VERSION + """ + + __slots__ = () + + MAX_ALLOWED_USERS = 10 + """:obj:`int`: Maximum number of users that can be allowed to access a managed bot in the + :paramref:`~telegram.Bot.set_managed_bot_access_settings.added_user_ids` parameter of + :meth:`~telegram.Bot.set_managed_bot_access_settings`. + """ + + class VerifyLimit(IntEnum): """This enum contains limitations for :meth:`~telegram.Bot.verify_chat` and :meth:`~telegram.Bot.verify_user`. diff --git a/src/telegram/ext/_extbot.py b/src/telegram/ext/_extbot.py index b6595fe0e98..7b468dade61 100644 --- a/src/telegram/ext/_extbot.py +++ b/src/telegram/ext/_extbot.py @@ -38,6 +38,7 @@ Animation, Audio, Bot, + BotAccessSettings, BotCommand, BotCommandScope, BotDescription, @@ -3150,7 +3151,7 @@ async def send_message_draft( self, chat_id: int, draft_id: int, - text: str, + text: str | None = None, message_thread_id: int | None = None, parse_mode: ODVInput[str] = DEFAULT_NONE, entities: Sequence["MessageEntity"] | None = None, @@ -5034,6 +5035,74 @@ async def edit_user_star_subscription( api_kwargs=self._merge_api_rl_kwargs(api_kwargs, rate_limit_args), ) + async def get_managed_bot_access_settings( + self, + user_id: int, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + rate_limit_args: RLARGS | None = None, + ) -> BotAccessSettings: + + return await super().get_managed_bot_access_settings( + user_id=user_id, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=self._merge_api_rl_kwargs(api_kwargs, rate_limit_args), + ) + + async def set_managed_bot_access_settings( + self, + user_id: int, + is_access_restricted: bool, + added_user_ids: Sequence[int] | None = None, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + rate_limit_args: RLARGS | None = None, + ) -> bool: + + return await super().set_managed_bot_access_settings( + user_id=user_id, + is_access_restricted=is_access_restricted, + added_user_ids=added_user_ids, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=self._merge_api_rl_kwargs(api_kwargs, rate_limit_args), + ) + + async def get_user_personal_chat_messages( + self, + user_id: int, + limit: int, + *, + read_timeout: ODVInput[float] = DEFAULT_NONE, + write_timeout: ODVInput[float] = DEFAULT_NONE, + connect_timeout: ODVInput[float] = DEFAULT_NONE, + pool_timeout: ODVInput[float] = DEFAULT_NONE, + api_kwargs: JSONDict | None = None, + rate_limit_args: RLARGS | None = None, + ) -> tuple[Message, ...]: + return await super().get_user_personal_chat_messages( + user_id=user_id, + limit=limit, + read_timeout=read_timeout, + write_timeout=write_timeout, + connect_timeout=connect_timeout, + pool_timeout=pool_timeout, + api_kwargs=self._merge_api_rl_kwargs(api_kwargs, rate_limit_args), + ) + async def send_paid_media( self, chat_id: str | int, @@ -5814,5 +5883,8 @@ async def delete_all_message_reactions( getManagedBotToken = get_managed_bot_token replaceManagedBotToken = replace_managed_bot_token savePreparedKeyboardButton = save_prepared_keyboard_button + getManagedBotAccessSettings = get_managed_bot_access_settings + setManagedBotAccessSettings = set_managed_bot_access_settings + getUserPersonalChatMessages = get_user_personal_chat_messages deleteMessageReaction = delete_message_reaction deleteAllMessageReactions = delete_all_message_reactions diff --git a/tests/auxil/dummy_objects.py b/tests/auxil/dummy_objects.py index c0ff0ce0cf5..4a0b0748fb6 100644 --- a/tests/auxil/dummy_objects.py +++ b/tests/auxil/dummy_objects.py @@ -4,6 +4,7 @@ from telegram import ( AcceptedGiftTypes, + BotAccessSettings, BotCommand, BotDescription, BotName, @@ -63,6 +64,7 @@ _PREPARED_DUMMY_OBJECTS: dict[str, object] = { "bool": True, + "BotAccessSettings": BotAccessSettings(is_access_restricted=True, added_users=[_DUMMY_USER]), "BotCommand": BotCommand(command="dummy_command", description="dummy_description"), "BotDescription": BotDescription(description="dummy_description"), "BotName": BotName(name="dummy_name"), diff --git a/tests/test_bot.py b/tests/test_bot.py index 15673347538..8d5296d01cd 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -33,6 +33,7 @@ from telegram import ( Bot, + BotAccessSettings, BotCommand, BotCommandScopeChat, BotDescription, @@ -2920,6 +2921,42 @@ async def make_assertion(url, request_data: RequestData, *args, **kwargs): ) assert isinstance(inst, PreparedKeyboardButton) + async def test_get_managed_bot_access_settings(self, offline_bot, monkeypatch): + async def make_assertion(url, request_data: RequestData, *args, **kwargs): + assert request_data.parameters.get("user_id") == 1234 + return BotAccessSettings( + is_access_restricted=True, + added_users=[User(1, "first", False)], + ).to_dict() + + monkeypatch.setattr(offline_bot.request, "post", make_assertion) + settings = await offline_bot.get_managed_bot_access_settings(1234) + assert isinstance(settings, BotAccessSettings) + + async def test_set_managed_bot_access_settings(self, offline_bot, monkeypatch): + async def make_assertion(url, request_data: RequestData, *args, **kwargs): + assert request_data.parameters.get("user_id") == 1234 + assert request_data.parameters.get("is_access_restricted") is True + assert request_data.parameters.get("added_user_ids") == [1, 2, 3] + + monkeypatch.setattr(offline_bot.request, "post", make_assertion) + await offline_bot.set_managed_bot_access_settings( + 1234, + is_access_restricted=True, + added_user_ids=[1, 2, 3], + ) + + async def test_get_user_personal_chat_messages(self, offline_bot, monkeypatch): + async def make_assertion(url, request_data: RequestData, *args, **kwargs): + assert request_data.parameters.get("user_id") == 1234 + assert request_data.parameters.get("limit") == 1 + return [make_message("dummy reply").to_dict()] + + monkeypatch.setattr(offline_bot.request, "post", make_assertion) + msgs = await offline_bot.get_user_personal_chat_messages(1234, limit=1) + assert isinstance(msgs, tuple) + assert all(isinstance(msg, Message) for msg in msgs) + # Bots cannot delete their own reaction from my testing, so we aren't making a real request async def test_delete_message_reaction(self, offline_bot, monkeypatch): async def make_assertion(url, request_data: RequestData, *args, **kwargs): @@ -4948,6 +4985,12 @@ async def test_my_profile_photo(self, bot): bot_profile_photos = await bot.get_user_profile_photos(bot.id) assert bot_profile_photos.total_count == 1 + async def test_get_user_personal_chat_messages(self, bot): + # id is of the Test User + messages = await bot.get_user_personal_chat_messages(user_id=675666224, limit=2) + assert isinstance(messages, tuple) + assert len(messages) == 2 + async def test_initialize_tracks_requests_and_bot_separately(self, offline_bot, monkeypatch): """Test that requests and bot user are initialized separately and only once.""" request_init_count = 0 diff --git a/tests/test_botaccesssettings.py b/tests/test_botaccesssettings.py new file mode 100644 index 00000000000..56cb8d05f50 --- /dev/null +++ b/tests/test_botaccesssettings.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2026 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +"""This module contains an object that represents a Telegram Bot Access Settings.""" + +import pytest + +from telegram import BotAccessSettings, Dice +from telegram._user import User +from tests.auxil.slots import mro_slots + + +@pytest.fixture(scope="module") +def bot_access_settings(): + return BotAccessSettings( + is_access_restricted=BotAccessSettingsTestBase.is_access_restricted, + added_users=BotAccessSettingsTestBase.added_users, + ) + + +class BotAccessSettingsTestBase: + is_access_restricted = True + added_users = [User(id=123, first_name="John", is_bot=False)] + + +class TestBotAccessSettingsWithoutRequest(BotAccessSettingsTestBase): + def test_slot_behaviour(self, bot_access_settings): + inst = bot_access_settings + for attr in inst.__slots__: + assert getattr(inst, attr, "err") != "err", f"got extra slot '{attr}'" + assert len(mro_slots(inst)) == len(set(mro_slots(inst))), "duplicate slot" + + def test_de_json(self, offline_bot): + json_dict = { + "is_access_restricted": self.is_access_restricted, + "added_users": [user.to_dict() for user in self.added_users], + } + bot_access_settings = BotAccessSettings.de_json(json_dict, offline_bot) + assert bot_access_settings.api_kwargs == {} + + assert bot_access_settings.is_access_restricted == self.is_access_restricted + assert bot_access_settings.added_users == tuple(self.added_users) + + def test_to_dict(self, bot_access_settings): + bot_access_settings_dict = bot_access_settings.to_dict() + + assert isinstance(bot_access_settings_dict, dict) + assert ( + bot_access_settings_dict["is_access_restricted"] + == bot_access_settings.is_access_restricted + ) + assert isinstance(bot_access_settings_dict["added_users"], list) + assert bot_access_settings_dict["added_users"][0] == self.added_users[0].to_dict() + + def test_equality(self): + a = BotAccessSettings(is_access_restricted=True, added_users=self.added_users) + b = BotAccessSettings(is_access_restricted=True, added_users=self.added_users) + c = BotAccessSettings(is_access_restricted=False, added_users=self.added_users) + d = BotAccessSettings(is_access_restricted=True, added_users=None) + e = Dice(emoji="🎲", value=4) + + assert a == b + assert hash(a) == hash(b) + + assert a != c + assert hash(a) != hash(c) + + assert a != d + assert hash(a) != hash(d) + + assert a != e + assert hash(a) != hash(e) diff --git a/tests/test_user.py b/tests/test_user.py index 4d6d35755e5..6ebebcf00ce 100644 --- a/tests/test_user.py +++ b/tests/test_user.py @@ -927,6 +927,73 @@ async def make_assertion(*_, **kwargs): monkeypatch.setattr(user.get_bot(), "replace_managed_bot_token", make_assertion) assert await user.replace_token() + async def test_instance_method_get_managed_bot_access_settings(self, monkeypatch, user): + async def make_assertion(*_, **kwargs): + return kwargs["user_id"] == user.id + + assert check_shortcut_signature( + user.get_managed_bot_access_settings, + Bot.get_managed_bot_access_settings, + ["user_id"], + [], + ) + assert await check_shortcut_call( + user.get_managed_bot_access_settings, + user.get_bot(), + "get_managed_bot_access_settings", + ) + assert await check_defaults_handling(user.get_managed_bot_access_settings, user.get_bot()) + + monkeypatch.setattr(user.get_bot(), "get_managed_bot_access_settings", make_assertion) + assert await user.get_managed_bot_access_settings() + + async def test_instance_method_set_managed_bot_access_settings(self, monkeypatch, user): + async def make_assertion(*_, **kwargs): + return ( + kwargs["user_id"] == user.id + and kwargs["is_access_restricted"] is True + and kwargs["added_user_ids"] == [123] + ) + + assert check_shortcut_signature( + user.set_managed_bot_access_settings, + Bot.set_managed_bot_access_settings, + ["user_id"], + [], + ) + assert await check_shortcut_call( + user.set_managed_bot_access_settings, + user.get_bot(), + "set_managed_bot_access_settings", + ) + assert await check_defaults_handling(user.set_managed_bot_access_settings, user.get_bot()) + + monkeypatch.setattr(user.get_bot(), "set_managed_bot_access_settings", make_assertion) + assert await user.set_managed_bot_access_settings( + is_access_restricted=True, + added_user_ids=[123], + ) + + async def test_instance_method_get_personal_chat_messages(self, monkeypatch, user): + async def make_assertion(*_, **kwargs): + return kwargs["user_id"] == user.id and kwargs["limit"] == 2 + + assert check_shortcut_signature( + user.get_personal_chat_messages, + Bot.get_user_personal_chat_messages, + ["user_id"], + [], + ) + assert await check_shortcut_call( + user.get_personal_chat_messages, + user.get_bot(), + "get_user_personal_chat_messages", + ) + assert await check_defaults_handling(user.get_personal_chat_messages, user.get_bot()) + + monkeypatch.setattr(user.get_bot(), "get_user_personal_chat_messages", make_assertion) + assert await user.get_personal_chat_messages(limit=2) + async def test_instance_method_delete_reaction(self, monkeypatch, user): async def make_assertion(*_, **kwargs): return (