OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2024-12-28 23:38:31 +01:00
Code Issues Projects Releases Packages Wiki Activity

API 6.1 (#3112)

Browse source 
Co-authored-by: Harshil <37377066+harshil21@users.noreply.github.com>
Co-authored-by: Bibo-Joshi <22366557+Bibo-Joshi@users.noreply.github.com>
This commit is contained in:
Poolitzer 2022-06-27 18:54:11 +02:00 committed by GitHub
parent 01d643913e
commit 08e223ba90
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
23 changed files with 652 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
README.rst
View file

@ -14,7 +14,7 @@
:target: https://pypi.org/project/python-telegram-bot/
:alt: Supported Python versions
.. image:: https://img.shields.io/badge/Bot%20API-6.0-blue?logo=telegram
.. image:: https://img.shields.io/badge/Bot%20API-6.1-blue?logo=telegram
:target: https://core.telegram.org/bots/api-changelog
:alt: Supported Bot API versions
@ -93,7 +93,7 @@ Installing both ``python-telegram-bot`` and ``python-telegram-bot-raw`` in conju
Telegram API support
====================
All types and methods of the Telegram Bot API **6.0** are supported.
All types and methods of the Telegram Bot API **6.1** are supported.
Installing
==========

4
README_RAW.rst
View file

@ -14,7 +14,7 @@
:target: https://pypi.org/project/python-telegram-bot-raw/
:alt: Supported Python versions
.. image:: https://img.shields.io/badge/Bot%20API-6.0-blue?logo=telegram
.. image:: https://img.shields.io/badge/Bot%20API-6.1-blue?logo=telegram
:target: https://core.telegram.org/bots/api-changelog
:alt: Supported Bot API versions
@ -89,7 +89,7 @@ Installing both ``python-telegram-bot`` and ``python-telegram-bot-raw`` in conju
Telegram API support
====================
All types and methods of the Telegram Bot API **6.0** are supported.
All types and methods of the Telegram Bot API **6.1** are supported.
Installing
==========

2
docs/source/bot_methods.rst
View file

@ -263,6 +263,8 @@
:align: left
:widths: 1 4
* - :meth:`~telegram.Bot.create_invoice_link`
- Used to generate an HTTP link for an invoice
* - :meth:`~telegram.Bot.close`
- Used for closing server instance when switching to another local server
* - :meth:`~telegram.Bot.log_out`

192
telegram/_bot.py
View file

@ -3795,6 +3795,7 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
allowed_updates: List[str] = None,
ip_address: str = None,
drop_pending_updates: bool = None,
secret_token: str = None,
*,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3808,9 +3809,9 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
specified url, containing An Update. In case of an unsuccessful request,
Telegram will give up after a reasonable amount of attempts.
If you'd like to make sure that the Webhook request comes from Telegram, Telegram
recommends using a secret path in the URL, e.g. https://www.example.com/<token>. Since
nobody else knows your bot's token, you can be pretty sure it's them.
If you'd like to make sure that the Webhook was set by you, you can specify secret data in
the parameter :paramref:`secret_token`. If specified, the request will contain a header
``X-Telegram-Bot-Api-Secret-Token`` with the secret token as content.
Note:
The certificate argument should be a file from disk ``open(filename, 'rb')``.
@ -3839,6 +3840,14 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
a short period of time.
drop_pending_updates (:obj:`bool`, optional): Pass :obj:`True` to drop all pending
updates.
secret_token (:obj:`str`, optional): A secret token to be sent in a header
``X-Telegram-Bot-Api-Secret-Token`` in every webhook request,
:tg-const:`telegram.constants.WebhookLimit.MIN_SECRET_TOKEN_LENGTH`-
:tg-const:`telegram.constants.WebhookLimit.MAX_SECRET_TOKEN_LENGTH` characters.
Only characters ``A-Z``, ``a-z``, ``0-9``, ``_`` and ``-`` are allowed.
The header is useful to ensure that the request comes from a webhook set by you.
.. versionadded:: 20.0
Keyword Args:
read_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
@ -3889,6 +3898,8 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
data["ip_address"] = ip_address
if drop_pending_updates:
data["drop_pending_updates"] = drop_pending_updates
if secret_token is not None:
data["secret_token"] = secret_token
result = await self._post(
"setWebhook",
@ -4593,26 +4604,32 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
Args:
chat_id (:obj:`int` | :obj:`str`): Unique identifier for the target chat or username
of the target channel (in the format ``@channelusername``).
title (:obj:`str`): Product name, 1-32 characters.
description (:obj:`str`): Product description, 1-255 characters.
payload (:obj:`str`): Bot-defined invoice payload, 1-128 bytes. This will not be
title (:obj:`str`): Product name. :tg-const:`telegram.Invoice.MIN_TITLE_LENGTH`-
:tg-const:`telegram.Invoice.MAX_TITLE_LENGTH` characters.
description (:obj:`str`): Product description.
:tg-const:`telegram.Invoice.MIN_DESCRIPTION_LENGTH`-
:tg-const:`telegram.Invoice.MAX_DESCRIPTION_LENGTH` characters.
payload (:obj:`str`): Bot-defined invoice payload.
:tg-const:`telegram.Invoice.MIN_PAYLOAD_LENGTH`-
:tg-const:`telegram.Invoice.MAX_PAYLOAD_LENGTH` bytes. This will not be
displayed to the user, use for your internal processes.
provider_token (:obj:`str`): Payments provider token, obtained via
`@BotFather <https://t.me/BotFather>`_.
currency (:obj:`str`): Three-letter ISO 4217 currency code.
currency (:obj:`str`): Three-letter ISO 4217 currency code, see `more on currencies
<https://core.telegram.org/bots/payments#supported-currencies>`_.
prices (List[:class:`telegram.LabeledPrice`)]: Price breakdown, a list
of components (e.g. product price, tax, discount, delivery cost, delivery tax,
bonus, etc.).
max_tip_amount (:obj:`int`, optional): The maximum accepted amount for tips in the
smallest units of the currency (integer, not float/double). For example, for a
maximum tip of US$ 1.45 pass ``max_tip_amount = 145``. See the exp parameter in
*smallest* units of the currency (integer, **not** float/double). For example, for
a maximum tip of US$ 1.45 pass ``max_tip_amount = 145``. See the exp parameter in
`currencies.json <https://core.telegram.org/bots/payments/currencies.json>`_, it
shows the number of digits past the decimal point for each currency (2 for the
majority of currencies). Defaults to ``0``.
.. versionadded:: 13.5
suggested_tip_amounts (List[:obj:`int`], optional): An array of
suggested amounts of tips in the smallest units of the currency (integer, not
suggested amounts of tips in the *smallest* units of the currency (integer, **not**
float/double). At most 4 suggested tip amounts can be specified. The suggested tip
amounts must be positive, passed in a strictly increased order and must not exceed
``max_tip_amount``.
@ -7693,6 +7710,159 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
)
return MenuButton.de_json(result, bot=self) # type: ignore[return-value, arg-type]
@_log
async def create_invoice_link(
self,
title: str,
description: str,
payload: str,
provider_token: str,
currency: str,
prices: List["LabeledPrice"],
max_tip_amount: int = None,
suggested_tip_amounts: List[int] = None,
provider_data: Union[str, object] = None,
photo_url: str = None,
photo_size: int = None,
photo_width: int = None,
photo_height: int = None,
need_name: bool = None,
need_phone_number: bool = None,
need_email: bool = None,
need_shipping_address: bool = None,
send_phone_number_to_provider: bool = None,
send_email_to_provider: bool = None,
is_flexible: bool = 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,
) -> str:
"""Use this method to create a link for an invoice.
.. versionadded:: 20.0
Args:
title (:obj:`str`): Product name. :tg-const:`telegram.Invoice.MIN_TITLE_LENGTH`-
:tg-const:`telegram.Invoice.MAX_TITLE_LENGTH` characters.
description (:obj:`str`): Product description.
:tg-const:`telegram.Invoice.MIN_DESCRIPTION_LENGTH`-
:tg-const:`telegram.Invoice.MAX_DESCRIPTION_LENGTH` characters.
payload (:obj:`str`): Bot-defined invoice payload.
:tg-const:`telegram.Invoice.MIN_PAYLOAD_LENGTH`-
:tg-const:`telegram.Invoice.MAX_PAYLOAD_LENGTH` bytes. This will not be
displayed to the user, use for your internal processes.
provider_token (:obj:`str`): Payments provider token, obtained via
`@BotFather <https://t.me/BotFather>`_.
currency (:obj:`str`): Three-letter ISO 4217 currency code, see `more on currencies
<https://core.telegram.org/bots/payments#supported-currencies>`_.
prices (List[:class:`telegram.LabeledPrice`)]: Price breakdown, a list
of components (e.g. product price, tax, discount, delivery cost, delivery tax,
bonus, etc.).
max_tip_amount (:obj:`int`, optional): The maximum accepted amount for tips in the
*smallest* units of the currency (integer, **not** float/double). For example, for
a maximum tip of US$ 1.45 pass ``max_tip_amount = 145``. See the exp parameter in
`currencies.json <https://core.telegram.org/bots/payments/currencies.json>`_, it
shows the number of digits past the decimal point for each currency (2 for the
majority of currencies). Defaults to ``0``.
suggested_tip_amounts (List[:obj:`int`], optional): An array of
suggested amounts of tips in the *smallest* units of the currency (integer, **not**
float/double). At most 4 suggested tip amounts can be specified. The suggested tip
amounts must be positive, passed in a strictly increased order and must not exceed
:paramref:`max_tip_amount`.
provider_data (:obj:`str` | :obj:`object`, optional): Data about the
invoice, which will be shared with the payment provider. A detailed description of
required fields should be provided by the payment provider. When an object is
passed, it will be encoded as JSON.
photo_url (:obj:`str`, optional): URL of the product photo for the invoice. Can be a
photo of the goods or a marketing image for a service.
photo_size (:obj:`int`, optional): Photo size in bytes.
photo_width (:obj:`int`, optional): Photo width.
photo_height (:obj:`int`, optional): Photo height.
need_name (:obj:`bool`, optional): Pass :obj:`True`, if you require the user's full
name to complete the order.
need_phone_number (:obj:`bool`, optional): Pass :obj:`True`, if you require the user's
phone number to complete the order.
need_email (:obj:`bool`, optional): Pass :obj:`True`, if you require the user's email
address to complete the order.
need_shipping_address (:obj:`bool`, optional): Pass :obj:`True`, if you require the
user's shipping address to complete the order.
send_phone_number_to_provider (:obj:`bool`, optional): Pass :obj:`True`, if user's
phone number should be sent to provider.
send_email_to_provider (:obj:`bool`, optional): Pass :obj:`True`, if user's email
address should be sent to provider.
is_flexible (:obj:`bool`, optional): Pass :obj:`True`, if the final price depends on
the shipping method.
Keyword Args:
read_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.read_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
write_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.write_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
connect_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.connect_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
pool_timeout (:obj:`float` | :obj:`None`, optional): Value to pass to
:paramref:`telegram.request.BaseRequest.post.pool_timeout`. Defaults to
:attr:`~telegram.request.BaseRequest.DEFAULT_NONE`.
api_kwargs (:obj:`dict`, optional): Arbitrary keyword arguments to be passed to the
Telegram API.
Returns:
:class:`str`: On success, the created invoice link is returned.
"""
data: JSONDict = {
"title": title,
"description": description,
"payload": payload,
"provider_token": provider_token,
"currency": currency,
"prices": prices,
}
if max_tip_amount is not None:
data["max_tip_amount"] = max_tip_amount
if suggested_tip_amounts is not None:
data["suggested_tip_amounts"] = suggested_tip_amounts
if provider_data is not None:
data["provider_data"] = provider_data
if photo_url is not None:
data["photo_url"] = photo_url
if photo_size is not None:
data["photo_size"] = photo_size
if photo_width is not None:
data["photo_width"] = photo_width
if photo_height is not None:
data["photo_height"] = photo_height
if need_name is not None:
data["need_name"] = need_name
if need_phone_number is not None:
data["need_phone_number"] = need_phone_number
if need_email is not None:
data["need_email"] = need_email
if need_shipping_address is not None:
data["need_shipping_address"] = need_shipping_address
if is_flexible is not None:
data["is_flexible"] = is_flexible
if send_phone_number_to_provider is not None:
data["send_phone_number_to_provider"] = send_phone_number_to_provider
if send_email_to_provider is not None:
data["send_email_to_provider"] = send_email_to_provider
return await self._post( # type: ignore[return-value]
"createInvoiceLink",
data,
read_timeout=read_timeout,
write_timeout=write_timeout,
connect_timeout=connect_timeout,
pool_timeout=pool_timeout,
api_kwargs=api_kwargs,
)
def to_dict(self) -> JSONDict:
"""See :meth:`telegram.TelegramObject.to_dict`."""
data: JSONDict = {"id": self.id, "username": self.username, "first_name": self.first_name}
@ -7881,3 +8051,5 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
"""Alias for :meth:`get_my_default_administrator_rights`"""
setMyDefaultAdministratorRights = set_my_default_administrator_rights
"""Alias for :meth:`set_my_default_administrator_rights`"""
createInvoiceLink = create_invoice_link
"""Alias for :meth:`create_invoice_link`"""

26
telegram/_chat.py
View file

@ -124,6 +124,16 @@ class Chat(TelegramObject):
chats. Returned only in :meth:`telegram.Bot.get_chat`.
location (:class:`telegram.ChatLocation`, optional): For supergroups, the location to which
the supergroup is connected. Returned only in :meth:`telegram.Bot.get_chat`.
join_to_send_messages (:obj:`bool`, optional): :obj:`True`, if users need to join the
supergroup before they can send messages. Returned only in
:meth:`telegram.Bot.get_chat`.
.. versionadded:: 20.0
join_by_request (:obj:`bool`, optional): :obj:`True`, if all users directly joining the
supergroup need to be approved by supergroup administrators. Returned only in
:meth:`telegram.Bot.get_chat`.
.. versionadded:: 20.0
**kwargs (:obj:`dict`): Arbitrary keyword arguments.
Attributes:
@ -168,6 +178,16 @@ class Chat(TelegramObject):
chats. Returned only in :meth:`telegram.Bot.get_chat`.
location (:class:`telegram.ChatLocation`): Optional. For supergroups, the location to which
the supergroup is connected. Returned only in :meth:`telegram.Bot.get_chat`.
join_to_send_messages (:obj:`bool`): Optional. :obj:`True`, if users need to join
the supergroup before they can send messages. Returned only in
:meth:`telegram.Bot.get_chat`.
.. versionadded:: 20.0
join_by_request (:obj:`bool`): Optional. :obj:`True`, if all users directly
joining the supergroup need to be approved by supergroup administrators. Returned only
in :meth:`telegram.Bot.get_chat`.
.. versionadded:: 20.0
"""
@ -193,6 +213,8 @@ class Chat(TelegramObject):
"message_auto_delete_time",
"has_protected_content",
"has_private_forwards",
"join_to_send_messages",
"join_by_request",
)
SENDER: ClassVar[str] = constants.ChatType.SENDER
@ -232,6 +254,8 @@ class Chat(TelegramObject):
message_auto_delete_time: int = None,
has_private_forwards: bool = None,
has_protected_content: bool = None,
join_to_send_messages: bool = None,
join_by_request: bool = None,
**_kwargs: Any,
):
# Required
@ -260,6 +284,8 @@ class Chat(TelegramObject):
self.can_set_sticker_set = can_set_sticker_set
self.linked_chat_id = linked_chat_id
self.location = location
self.join_to_send_messages = join_to_send_messages
self.join_by_request = join_by_request
self.set_bot(bot)
self._id_attrs = (self.id,)

13
telegram/_files/sticker.py
View file

@ -22,6 +22,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, List, Optional
from telegram import constants
from telegram._files._basethumbedmedium import _BaseThumbedMedium
from telegram._files.file import File
from telegram._files.photosize import PhotoSize
from telegram._telegramobject import TelegramObject
from telegram._utils.types import JSONDict
@ -62,6 +63,10 @@ class Sticker(_BaseThumbedMedium):
position where the mask should be placed.
file_size (:obj:`int`, optional): File size in bytes.
bot (:class:`telegram.Bot`, optional): The Bot to use for instance methods.
premium_animation (:class:`telegram.File`, optional): Premium animation for the sticker,
if the sticker is premium.
.. versionadded:: 20.0
_kwargs (:obj:`dict`): Arbitrary keyword arguments.
Attributes:
@ -83,6 +88,10 @@ class Sticker(_BaseThumbedMedium):
where the mask should be placed.
file_size (:obj:`int`): Optional. File size in bytes.
bot (:class:`telegram.Bot`): Optional. The Bot to use for instance methods.
premium_animation (:class:`telegram.File`): Optional. Premium animation for the
sticker, if the sticker is premium.
.. versionadded:: 20.0
"""
@ -94,6 +103,7 @@ class Sticker(_BaseThumbedMedium):
"mask_position",
"set_name",
"width",
"premium_animation",
)
def __init__(
@ -110,6 +120,7 @@ class Sticker(_BaseThumbedMedium):
set_name: str = None,
mask_position: "MaskPosition" = None,
bot: "Bot" = None,
premium_animation: "File" = None,
**_kwargs: Any,
):
super().__init__(
@ -128,6 +139,7 @@ class Sticker(_BaseThumbedMedium):
self.emoji = emoji
self.set_name = set_name
self.mask_position = mask_position
self.premium_animation = premium_animation
@classmethod
def de_json(cls, data: Optional[JSONDict], bot: "Bot") -> Optional["Sticker"]:
@ -139,6 +151,7 @@ class Sticker(_BaseThumbedMedium):
data["thumb"] = PhotoSize.de_json(data.get("thumb"), bot)
data["mask_position"] = MaskPosition.de_json(data.get("mask_position"), bot)
data["premium_animation"] = File.de_json(data.get("premium_animation"), bot)
return cls(bot=bot, **data)

32
telegram/_inline/inputinvoicemessagecontent.py
View file

@ -39,9 +39,14 @@ class InputInvoiceMessageContent(InputMessageContent):
.. versionadded:: 13.5
Args:
title (:obj:`str`): Product name, 1-32 characters
description (:obj:`str`): Product description, 1-255 characters
payload (:obj:`str`):Bot-defined invoice payload, 1-128 bytes. This will not be displayed
title (:obj:`str`): Product name. :tg-const:`telegram.Invoice.MIN_TITLE_LENGTH`-
:tg-const:`telegram.Invoice.MAX_TITLE_LENGTH` characters.
description (:obj:`str`): Product description.
:tg-const:`telegram.Invoice.MIN_DESCRIPTION_LENGTH`-
:tg-const:`telegram.Invoice.MAX_DESCRIPTION_LENGTH` characters.
payload (:obj:`str`): Bot-defined invoice payload.
:tg-const:`telegram.Invoice.MIN_PAYLOAD_LENGTH`-
:tg-const:`telegram.Invoice.MAX_PAYLOAD_LENGTH` bytes. This will not be displayed
to the user, use for your internal processes.
provider_token (:obj:`str`): Payment provider token, obtained via
`@Botfather <https://t.me/Botfather>`_.
@ -50,15 +55,15 @@ class InputInvoiceMessageContent(InputMessageContent):
prices (List[:class:`telegram.LabeledPrice`]): Price breakdown, a list of
components (e.g. product price, tax, discount, delivery cost, delivery tax, bonus,
etc.)
max_tip_amount (:obj:`int`, optional): The maximum accepted amount for tips in the smallest
units of the currency (integer, not float/double). For example, for a maximum tip of
US$ 1.45 pass ``max_tip_amount = 145``. See the ``exp`` parameter in
max_tip_amount (:obj:`int`, optional): The maximum accepted amount for tips in the
*smallest* units of the currency (integer, **not** float/double). For example, for a
maximum tip of US$ 1.45 pass ``max_tip_amount = 145``. See the ``exp`` parameter in
`currencies.json <https://core.telegram.org/bots/payments/currencies.json>`_, it
shows the number of digits past the decimal point for each currency (2 for the majority
of currencies). Defaults to ``0``.
suggested_tip_amounts (List[:obj:`int`], optional): An array of suggested
amounts of tip in the smallest units of the currency (integer, not float/double). At
most 4 suggested tip amounts can be specified. The suggested tip amounts must be
amounts of tip in the *smallest* units of the currency (integer, **not** float/double).
At most 4 suggested tip amounts can be specified. The suggested tip amounts must be
positive, passed in a strictly increased order and must not exceed
:attr:`max_tip_amount`.
provider_data (:obj:`str`, optional): An object for data about the invoice,
@ -87,9 +92,14 @@ class InputInvoiceMessageContent(InputMessageContent):
**kwargs (:obj:`dict`): Arbitrary keyword arguments.
Attributes:
title (:obj:`str`): Product name, 1-32 characters
description (:obj:`str`): Product description, 1-255 characters
payload (:obj:`str`):Bot-defined invoice payload, 1-128 bytes. This will not be displayed
title (:obj:`str`): Product name. :tg-const:`telegram.Invoice.MIN_TITLE_LENGTH`-
:tg-const:`telegram.Invoice.MAX_TITLE_LENGTH` characters.
description (:obj:`str`): Product description.
:tg-const:`telegram.Invoice.MIN_DESCRIPTION_LENGTH`-
:tg-const:`telegram.Invoice.MAX_DESCRIPTION_LENGTH` characters.
payload (:obj:`str`): Bot-defined invoice payload.
:tg-const:`telegram.Invoice.MIN_PAYLOAD_LENGTH`-
:tg-const:`telegram.Invoice.MAX_PAYLOAD_LENGTH` bytes. This will not be displayed
to the user, use for your internal processes.
provider_token (:obj:`str`): Payment provider token, obtained via
`@Botfather <https://t.me/Botfather>`_.

4
telegram/_loginurl.py
View file

@ -39,7 +39,7 @@ class LoginUrl(TelegramObject):
`Checking authorization <https://core.telegram.org/widgets/login#checking-authorization>`_
Args:
url (:obj:`str`): An HTTP URL to be opened with user authorization data added to the query
url (:obj:`str`): An HTTPS URL to be opened with user authorization data added to the query
string when the button is pressed. If the user refuses to provide authorization data,
the original URL without information about the user will be opened. The data added is
the same as described in
@ -59,7 +59,7 @@ class LoginUrl(TelegramObject):
for your bot to send messages to the user.
Attributes:
url (:obj:`str`): An HTTP URL to be opened with user authorization data.
url (:obj:`str`): An HTTPS URL to be opened with user authorization data.
forward_text (:obj:`str`): Optional. New text of the button in forwarded messages.
bot_username (:obj:`str`): Optional. Username of a bot, which will be used for user
authorization.

34
telegram/_payment/invoice.py
View file

@ -18,8 +18,9 @@
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains an object that represents a Telegram Invoice."""
from typing import Any
from typing import Any, ClassVar
from telegram import constants
from telegram._telegramobject import TelegramObject
@ -83,3 +84,34 @@ class Invoice(TelegramObject):
self.currency,
self.total_amount,
)
MIN_TITLE_LENGTH: ClassVar[int] = constants.InvoiceLimit.MIN_TITLE_LENGTH
""":const:`telegram.constants.InvoiceLimit.MIN_TITLE_LENGTH`
.. versionadded:: 20.0
"""
MAX_TITLE_LENGTH: ClassVar[int] = constants.InvoiceLimit.MAX_TITLE_LENGTH
""":const:`telegram.constants.InvoiceLimit.MAX_TITLE_LENGTH`
.. versionadded:: 20.0
"""
MIN_DESCRIPTION_LENGTH: ClassVar[int] = constants.InvoiceLimit.MIN_DESCRIPTION_LENGTH
""":const:`telegram.constants.InvoiceLimit.MIN_DESCRIPTION_LENGTH`
.. versionadded:: 20.0
"""
MAX_DESCRIPTION_LENGTH: ClassVar[int] = constants.InvoiceLimit.MAX_DESCRIPTION_LENGTH
""":const:`telegram.constants.InvoiceLimit.MAX_DESCRIPTION_LENGTH`
.. versionadded:: 20.0
"""
MIN_PAYLOAD_LENGTH: ClassVar[int] = constants.InvoiceLimit.MIN_PAYLOAD_LENGTH
""":const:`telegram.constants.InvoiceLimit.MIN_PAYLOAD_LENGTH`
.. versionadded:: 20.0
"""
MAX_PAYLOAD_LENGTH: ClassVar[int] = constants.InvoiceLimit.MAX_PAYLOAD_LENGTH
""":const:`telegram.constants.InvoiceLimit.MAX_PAYLOAD_LENGTH`
.. versionadded:: 20.0
"""

20
telegram/_user.py
View file

@ -83,6 +83,13 @@ class User(TelegramObject):
supports_inline_queries (:obj:`str`, optional): :obj:`True`, if the bot supports inline
queries. Returned only in :attr:`telegram.Bot.get_me` requests.
bot (:class:`telegram.Bot`, optional): The Bot to use for instance methods.
is_premium (:obj:`bool`, optional): :obj:`True`, if this user is a Telegram Premium user.
.. versionadded:: 20.0
added_to_attachment_menu (:obj:`bool`, optional): :obj:`True`, if this user added
the bot to the attachment menu.
.. versionadded:: 20.0
Attributes:
id (:obj:`int`): Unique identifier for this user or bot.
@ -98,7 +105,14 @@ class User(TelegramObject):
supports_inline_queries (:obj:`str`): Optional. :obj:`True`, if the bot supports inline
queries. Returned only in :attr:`telegram.Bot.get_me` requests.
bot (:class:`telegram.Bot`): Optional. The Bot to use for instance methods.
is_premium (:obj:`bool`): Optional. :obj:`True`, if this user is a Telegram
Premium user.
.. versionadded:: 20.0
added_to_attachment_menu (:obj:`bool`): Optional. :obj:`True`, if this user added
the bot to the attachment menu.
.. versionadded:: 20.0
"""
__slots__ = (
@ -111,6 +125,8 @@ class User(TelegramObject):
"supports_inline_queries",
"id",
"language_code",
"is_premium",
"added_to_attachment_menu",
)
def __init__(
@ -125,6 +141,8 @@ class User(TelegramObject):
can_read_all_group_messages: bool = None,
supports_inline_queries: bool = None,
bot: "Bot" = None,
is_premium: bool = None,
added_to_attachment_menu: bool = None,
**_kwargs: Any,
):
# Required
@ -138,6 +156,8 @@ class User(TelegramObject):
self.can_join_groups = can_join_groups
self.can_read_all_group_messages = can_read_all_group_messages
self.supports_inline_queries = supports_inline_queries
self.is_premium = is_premium
self.added_to_attachment_menu = added_to_attachment_menu
self.set_bot(bot)
self._id_attrs = (self.id,)

42
telegram/constants.py
View file

@ -45,6 +45,7 @@ __all__ = [
"InlineQueryLimit",
"InlineQueryResultType",
"InputMediaType",
"InvoiceLimit",
"LocationLimit",
"MaskPosition",
"MenuButtonType",
@ -56,6 +57,7 @@ __all__ = [
"PollLimit",
"PollType",
"SUPPORTED_WEBHOOK_PORTS",
"WebhookLimit",
"UpdateType",
]
@ -90,7 +92,7 @@ class _BotAPIVersion(NamedTuple):
#: :data:`telegram.__bot_api_version_info__`.
#:
#: .. versionadded:: 20.0
BOT_API_VERSION_INFO = _BotAPIVersion(major=6, minor=0)
BOT_API_VERSION_INFO = _BotAPIVersion(major=6, minor=1)
#: :obj:`str`: Telegram Bot API
#: version supported by this version of `python-telegram-bot`. Also available as
#: :data:`telegram.__bot_api_version__`.
@ -809,3 +811,41 @@ class UpdateType(StringEnum):
""":obj:`str`: Updates with :attr:`telegram.Update.chat_member`."""
CHAT_JOIN_REQUEST = "chat_join_request"
""":obj:`str`: Updates with :attr:`telegram.Update.chat_join_request`."""
class InvoiceLimit(IntEnum):
"""This enum contains limitations for :meth:`telegram.Bot.create_invoice_link`. The enum
members of this enumeration are instances of :class:`int` and can be treated as such.
.. versionadded:: 20.0
"""
__slots__ = ()
MIN_TITLE_LENGTH = 1
""":obj:`int`: Minimum number of characters of the invoice title."""
MAX_TITLE_LENGTH = 32
""":obj:`int`: Maximum number of characters of the invoice title."""
MIN_DESCRIPTION_LENGTH = 1
""":obj:`int`: Minimum number of characters of the invoice description."""
MAX_DESCRIPTION_LENGTH = 255
""":obj:`int`: Maximum number of characters of the invoice description."""
MIN_PAYLOAD_LENGTH = 1
""":obj:`int`: Minimum amount of bytes for the internal payload."""
MAX_PAYLOAD_LENGTH = 128
""":obj:`int`: Maximum amount of bytes for the internal payload."""
class WebhookLimit(IntEnum):
"""This enum contains limitations for :paramref:`telegram.Bot.set_webhook.secret_token`. The
enum members of this enumeration are instances of :class:`int` and can be treated as such.
.. versionadded:: 20.0
"""
__slots__ = ()
MIN_SECRET_TOKEN_LENGTH = 1
""":obj:`int`: Minimum length of the secret token."""
MAX_SECRET_TOKEN_LENGTH = 256
""":obj:`int`: Maximum length of the secret token."""

12
telegram/ext/_application.py
View file

@ -662,6 +662,7 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AbstractAsyncContextManager)
max_connections: int = 40,
close_loop: bool = True,
stop_signals: ODVInput[Sequence[int]] = DEFAULT_NONE,
secret_token: str = None,
) -> None:
"""Convenience method that takes care of initializing and starting the app,
polling updates from Telegram using :meth:`telegram.ext.Updater.start_webhook` and
@ -724,6 +725,16 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AbstractAsyncContextManager)
:meth:`asyncio.loop.add_signal_handler`. Most notably, the standard event loop
on Windows, :class:`asyncio.ProactorEventLoop`, does not implement this method.
If this method is not available, stop signals can not be set.
secret_token (:obj:`str`, optional): Secret token to ensure webhook requests originate
from Telegram. See :paramref:`telegram.Bot.set_webhook.secret_token` for more
details.
When added, the web server started by this call will expect the token to be set in
the ``X-Telegram-Bot-Api-Secret-Token`` header of an incoming request and will
raise a :class:`http.HTTPStatus.FORBIDDEN <http.HTTPStatus>` error if either the
header isn't set or it is set to a wrong token.
.. versionadded:: 20.0
"""
if not self.updater:
raise RuntimeError(
@ -743,6 +754,7 @@ class Application(Generic[BT, CCT, UD, CD, BD, JQ], AbstractAsyncContextManager)
allowed_updates=allowed_updates,
ip_address=ip_address,
max_connections=max_connections,
secret_token=secret_token,
),
close_loop=close_loop,
stop_signals=stop_signals,

20
telegram/ext/_updater.py
View file

@ -369,6 +369,7 @@ class Updater(AbstractAsyncContextManager):
drop_pending_updates: bool = None,
ip_address: str = None,
max_connections: int = 40,
secret_token: str = None,
) -> asyncio.Queue:
"""
Starts a small http server to listen for updates via webhook. If :paramref:`cert`
@ -395,6 +396,7 @@ class Updater(AbstractAsyncContextManager):
key (:class:`pathlib.Path` | :obj:`str`, optional): Path to the SSL key file.
drop_pending_updates (:obj:`bool`, optional): Whether to clean any pending updates on
Telegram servers before actually starting to poll. Default is :obj:`False`.
.. versionadded :: 13.4
bootstrap_retries (:obj:`int`, optional): Whether the bootstrapping phase of the
:class:`telegram.ext.Updater` will retry on failures on the Telegram server.
@ -407,12 +409,23 @@ class Updater(AbstractAsyncContextManager):
:paramref:`port`, :paramref:`url_path`, :paramref:`cert`, and :paramref:`key`.
ip_address (:obj:`str`, optional): Passed to :meth:`telegram.Bot.set_webhook`.
Defaults to :obj:`None`.
.. versionadded :: 13.4
allowed_updates (List[:obj:`str`], optional): Passed to
:meth:`telegram.Bot.set_webhook`. Defaults to :obj:`None`.
max_connections (:obj:`int`, optional): Passed to
:meth:`telegram.Bot.set_webhook`. Defaults to ``40``.
.. versionadded:: 13.6
secret_token (:obj:`str`, optional): Passed to :meth:`telegram.Bot.set_webhook`.
Defaults to :obj:`None`.
When added, the web server started by this call will expect the token to be set in
the ``X-Telegram-Bot-Api-Secret-Token`` header of an incoming request and will
raise a :class:`http.HTTPStatus.FORBIDDEN <http.HTTPStatus>` error if either the
header isn't set or it is set to a wrong token.
.. versionadded:: 20.0
Returns:
:class:`queue.Queue`: The update queue that can be filled from the main thread.
@ -444,6 +457,7 @@ class Updater(AbstractAsyncContextManager):
ready=webhook_ready,
ip_address=ip_address,
max_connections=max_connections,
secret_token=secret_token,
)
self._logger.debug("Waiting for webhook server to start")
@ -470,6 +484,7 @@ class Updater(AbstractAsyncContextManager):
ready: asyncio.Event = None,
ip_address: str = None,
max_connections: int = 40,
secret_token: str = None,
) -> None:
self._logger.debug("Updater thread started (webhook)")
@ -477,7 +492,7 @@ class Updater(AbstractAsyncContextManager):
url_path = f"/{url_path}"
# Create Tornado app instance
app = WebhookAppClass(url_path, self.bot, self.update_queue)
app = WebhookAppClass(url_path, self.bot, self.update_queue, secret_token)
# Form SSL Context
# An SSLError is raised if the private key does not match with the certificate
@ -517,6 +532,7 @@ class Updater(AbstractAsyncContextManager):
allowed_updates=allowed_updates,
ip_address=ip_address,
max_connections=max_connections,
secret_token=secret_token,
)
await self._httpd.serve_forever(ready=ready)
@ -591,6 +607,7 @@ class Updater(AbstractAsyncContextManager):
bootstrap_interval: float = 1,
ip_address: str = None,
max_connections: int = 40,
secret_token: str = None,
) -> None:
"""Prepares the setup for fetching updates: delete or set the webhook and drop pending
updates if appropriate. If there are unsuccessful attempts, this will retry as specified by
@ -616,6 +633,7 @@ class Updater(AbstractAsyncContextManager):
ip_address=ip_address,
drop_pending_updates=drop_pending_updates,
max_connections=max_connections,
secret_token=secret_token,
)
return False

32
telegram/ext/_utils/webhookhandler.py
View file

@ -83,8 +83,14 @@ class WebhookServer:
class WebhookAppClass(tornado.web.Application):
"""Application used in the Webserver"""
def __init__(self, webhook_path: str, bot: "Bot", update_queue: asyncio.Queue):
self.shared_objects = {"bot": bot, "update_queue": update_queue}
def __init__(
self, webhook_path: str, bot: "Bot", update_queue: asyncio.Queue, secret_token: str = None
):
self.shared_objects = {
"bot": bot,
"update_queue": update_queue,
"secret_token": secret_token,
}
handlers = [(rf"{webhook_path}/?", TelegramHandler, self.shared_objects)] # noqa
tornado.web.Application.__init__(self, handlers) # type: ignore
@ -96,16 +102,21 @@ class WebhookAppClass(tornado.web.Application):
class TelegramHandler(tornado.web.RequestHandler):
"""BaseHandler that processes incoming requests from Telegram"""
__slots__ = ("bot", "update_queue", "_logger")
__slots__ = ("bot", "update_queue", "_logger", "secret_token")
SUPPORTED_METHODS = ("POST",) # type: ignore[assignment]
def initialize(self, bot: "Bot", update_queue: asyncio.Queue) -> None:
def initialize(self, bot: "Bot", update_queue: asyncio.Queue, secret_token: str) -> None:
"""Initialize for each request - that's the interface provided by tornado"""
# pylint: disable=attribute-defined-outside-init
self.bot = bot
self.update_queue = update_queue
self._logger = logging.getLogger(__name__)
self.secret_token = secret_token
if secret_token:
self._logger.debug(
"The webhook server has a secret token, " "expecting it in incoming requests now"
)
def set_default_headers(self) -> None:
"""Sets default headers"""
@ -144,6 +155,19 @@ class TelegramHandler(tornado.web.RequestHandler):
ct_header = self.request.headers.get("Content-Type", None)
if ct_header != "application/json":
raise tornado.web.HTTPError(HTTPStatus.FORBIDDEN)
# verifying that the secret token is the one the user set when the user set one
if self.secret_token is not None:
token = self.request.headers.get("X-Telegram-Bot-Api-Secret-Token")
if not token:
self._logger.debug("Request did not include the secret token")
raise tornado.web.HTTPError(
HTTPStatus.FORBIDDEN, reason="Request did not include the secret token"
)
if token != self.secret_token:
self._logger.debug("Request had the wrong secret token: %s", token)
raise tornado.web.HTTPError(
HTTPStatus.FORBIDDEN, reason="Request had the wrong secret token"
)
def log_exception(
self,

52
telegram/ext/filters.py
View file

@ -76,6 +76,8 @@ __all__ = (
"TEXT",
"Text",
"USER",
"USER_ATTACHMENT",
"PREMIUM_USER",
"UpdateFilter",
"UpdateType",
"User",
@ -1949,6 +1951,21 @@ class Sticker:
.. versionadded:: 20.0
"""
class _Premium(MessageFilter):
__slots__ = ()
def filter(self, message: Message) -> bool:
return bool(message.sticker) and bool(
message.sticker.premium_animation # type: ignore
)
PREMIUM = _Premium(name="filters.Sticker.PREMIUM")
"""Messages that contain :attr:`telegram.Message.sticker` and have a
:attr:`premium animation <telegram.Sticker.premium_animation>`.
.. versionadded:: 20.0
"""
class _SuccessfulPayment(MessageFilter):
__slots__ = ()
@ -2185,6 +2202,41 @@ USER = _User(name="filters.USER")
"""This filter filters *any* message that has a :attr:`telegram.Message.from_user`."""
class _UserAttachment(UpdateFilter):
__slots__ = ()
def filter(self, update: Update) -> bool:
return bool(update.effective_user) and bool(
update.effective_user.added_to_attachment_menu # type: ignore
)
USER_ATTACHMENT = _UserAttachment(name="filters.USER_ATTACHMENT")
"""This filter filters *any* message that have a user who added the bot to their
:attr:`attachment menu <telegram.User.added_to_attachment_menu>` as
:attr:`telegram.Update.effective_user`.
.. versionadded:: 20.0
"""
class _UserPremium(UpdateFilter):
__slots__ = ()
def filter(self, update: Update) -> bool:
return bool(update.effective_user) and bool(
update.effective_user.is_premium # type: ignore
)
PREMIUM_USER = _UserPremium(name="filters.PREMIUM_USER")
"""This filter filters *any* message from a
:attr:`Telegram Premium user <telegram.User.is_premium>` as :attr:`telegram.Update.effective_user`.
.. versionadded:: 20.0
"""
class _Venue(MessageFilter):
__slots__ = ()

3
tests/conftest.py
View file

@ -830,10 +830,13 @@ async def send_webhook_message(
content_len: int = -1,
content_type: str = "application/json",
get_method: str = None,
secret_token: str = None,
) -> Response:
headers = {
"content-type": content_type,
}
if secret_token:
headers["X-Telegram-Bot-Api-Secret-Token"] = secret_token
if not payload_str:
content_len = None

31
tests/test_bot.py
View file

@ -1634,6 +1634,35 @@ class TestBot:
assert await bot.set_webhook("", drop_pending_updates=drop_pending_updates)
assert await bot.delete_webhook(drop_pending_updates=drop_pending_updates)
async def test_set_webhook_params(self, bot, monkeypatch):
# actually making calls to TG is done in
# test_set_webhook_get_webhook_info_and_delete_webhook. Sadly secret_token can't be tested
# there so we have this function \o/
async def make_assertion(*args, **_):
kwargs = args[1]
return (
kwargs["url"] == "example.com"
and kwargs["certificate"].input_file_content
== data_file("sslcert.pem").read_bytes()
and kwargs["max_connections"] == 7
and kwargs["allowed_updates"] == ["messages"]
and kwargs["ip_address"] == "127.0.0.1"
and kwargs["drop_pending_updates"]
and kwargs["secret_token"] == "SoSecretToken"
)
monkeypatch.setattr(bot, "_post", make_assertion)
assert await bot.set_webhook(
"example.com",
data_file("sslcert.pem").read_bytes(),
7,
["messages"],
"127.0.0.1",
True,
"SoSecretToken",
)
@flaky(3, 1)
async def test_leave_chat(self, bot):
with pytest.raises(BadRequest, match="Chat not found"):
@ -1833,7 +1862,7 @@ class TestBot:
# We assume that the other game score tests ran within 20 sec
assert high_scores[0].score == BASE_GAME_SCORE - 10
# send_invoice is tested in test_invoice
# send_invoice and create_invoice_link is tested in test_invoice
# TODO: Needs improvement. Need incoming shipping queries to test
async def test_answer_shipping_query_ok(self, monkeypatch, bot):

10
tests/test_chat.py
View file

@ -42,6 +42,8 @@ def chat(bot):
location=TestChat.location,
has_private_forwards=True,
has_protected_content=True,
join_to_send_messages=True,
join_by_request=True,
)
@ -64,6 +66,8 @@ class TestChat:
location = ChatLocation(Location(123, 456), "Barbie World")
has_protected_content = True
has_private_forwards = True
join_to_send_messages = True
join_by_request = True
def test_slot_behaviour(self, chat, mro_slots):
for attr in chat.__slots__:
@ -86,6 +90,8 @@ class TestChat:
"has_private_forwards": self.has_private_forwards,
"linked_chat_id": self.linked_chat_id,
"location": self.location.to_dict(),
"join_to_send_messages": self.join_to_send_messages,
"join_by_request": self.join_by_request,
}
chat = Chat.de_json(json_dict, bot)
@ -104,6 +110,8 @@ class TestChat:
assert chat.linked_chat_id == self.linked_chat_id
assert chat.location.location == self.location.location
assert chat.location.address == self.location.address
assert chat.join_to_send_messages == self.join_to_send_messages
assert chat.join_by_request == self.join_by_request
def test_to_dict(self, chat):
chat_dict = chat.to_dict()
@ -121,6 +129,8 @@ class TestChat:
assert chat_dict["has_protected_content"] == chat.has_protected_content
assert chat_dict["linked_chat_id"] == chat.linked_chat_id
assert chat_dict["location"] == chat.location.to_dict()
assert chat_dict["join_to_send_messages"] == chat.join_to_send_messages
assert chat_dict["join_by_request"] == chat.join_by_request
def test_enum_init(self):
chat = Chat(id=1, type="foo")

25
tests/test_filters.py
View file

@ -27,6 +27,7 @@ from telegram import (
Chat,
Dice,
Document,
File,
Message,
MessageEntity,
Sticker,
@ -830,15 +831,26 @@ class TestFilters:
update.message.sticker = Sticker("1", "uniq", 1, 2, False, False)
assert filters.Sticker.ALL.check_update(update)
assert filters.Sticker.STATIC.check_update(update)
assert not filters.Sticker.VIDEO.check_update(update)
assert not filters.Sticker.PREMIUM.check_update(update)
update.message.sticker.is_animated = True
assert filters.Sticker.ANIMATED.check_update(update)
assert not filters.Sticker.VIDEO.check_update(update)
assert not filters.Sticker.STATIC.check_update(update)
assert not filters.Sticker.PREMIUM.check_update(update)
update.message.sticker.is_animated = False
update.message.sticker.is_video = True
assert not filters.Sticker.ANIMATED.check_update(update)
assert not filters.Sticker.STATIC.check_update(update)
assert filters.Sticker.VIDEO.check_update(update)
assert not filters.Sticker.PREMIUM.check_update(update)
update.message.sticker.premium_animation = File("string", "uniqueString")
assert not filters.Sticker.ANIMATED.check_update(update)
# premium stickers can be animated, video, or probably also static,
# it doesn't really matter for the test
assert not filters.Sticker.STATIC.check_update(update)
assert filters.Sticker.VIDEO.check_update(update)
assert filters.Sticker.PREMIUM.check_update(update)
def test_filters_video(self, update):
assert not filters.VIDEO.check_update(update)
@ -1168,6 +1180,19 @@ class TestFilters:
with pytest.raises(RuntimeError, match="Cannot set name"):
f.name = "foo"
def test_filters_user_attributes(self, update):
assert not filters.USER_ATTACHMENT.check_update(update)
assert not filters.PREMIUM_USER.check_update(update)
update.message.from_user.added_to_attachment_menu = True
assert filters.USER_ATTACHMENT.check_update(update)
assert not filters.PREMIUM_USER.check_update(update)
update.message.from_user.is_premium = True
assert filters.USER_ATTACHMENT.check_update(update)
assert filters.PREMIUM_USER.check_update(update)
update.message.from_user.added_to_attachment_menu = False
assert not filters.USER_ATTACHMENT.check_update(update)
assert filters.PREMIUM_USER.check_update(update)
def test_filters_chat_init(self):
with pytest.raises(RuntimeError, match="in conjunction with"):
filters.Chat(chat_id=1, username="chat")

66
tests/test_invoice.py
View file

@ -98,8 +98,18 @@ class TestInvoice:
assert message.invoice.title == self.title
assert message.invoice.total_amount == self.total_amount
@flaky(3, 1)
async def test_send_all_args(self, bot, chat_id, provider_token, monkeypatch):
link = await bot.create_invoice_link(
title=self.title,
description=self.description,
payload=self.payload,
provider_token=provider_token,
currency=self.currency,
prices=self.prices,
)
assert isinstance(link, str)
assert link != ""
async def test_send_all_args_send_invoice(self, bot, chat_id, provider_token, monkeypatch):
message = await bot.send_invoice(
chat_id,
self.title,
@ -193,6 +203,58 @@ class TestInvoice:
protect_content=True,
)
async def test_send_all_args_create_invoice_link(
self, bot, chat_id, provider_token, monkeypatch
):
async def make_assertion(*args, **_):
kwargs = args[1]
return (
kwargs["title"] == "title"
and kwargs["description"] == "description"
and kwargs["payload"] == "payload"
and kwargs["provider_token"] == "provider_token"
and kwargs["currency"] == "currency"
and kwargs["prices"] == self.prices
and kwargs["max_tip_amount"] == "max_tip_amount"
and kwargs["suggested_tip_amounts"] == "suggested_tip_amounts"
and kwargs["provider_data"] == "provider_data"
and kwargs["photo_url"] == "photo_url"
and kwargs["photo_size"] == "photo_size"
and kwargs["photo_width"] == "photo_width"
and kwargs["photo_height"] == "photo_height"
and kwargs["need_name"] == "need_name"
and kwargs["need_phone_number"] == "need_phone_number"
and kwargs["need_email"] == "need_email"
and kwargs["need_shipping_address"] == "need_shipping_address"
and kwargs["send_phone_number_to_provider"] == "send_phone_number_to_provider"
and kwargs["send_email_to_provider"] == "send_email_to_provider"
and kwargs["is_flexible"] == "is_flexible"
)
monkeypatch.setattr(bot, "_post", make_assertion)
assert await bot.create_invoice_link(
title="title",
description="description",
payload="payload",
provider_token="provider_token",
currency="currency",
prices=self.prices,
max_tip_amount="max_tip_amount",
suggested_tip_amounts="suggested_tip_amounts",
provider_data="provider_data",
photo_url="photo_url",
photo_size="photo_size",
photo_width="photo_width",
photo_height="photo_height",
need_name="need_name",
need_phone_number="need_phone_number",
need_email="need_email",
need_shipping_address="need_shipping_address",
send_phone_number_to_provider="send_phone_number_to_provider",
send_email_to_provider="send_email_to_provider",
is_flexible="is_flexible",
)
async def test_send_object_as_provider_data(self, monkeypatch, bot, chat_id, provider_token):
async def make_assertion(url, request_data: RequestData, *args, **kwargs):
return request_data.json_parameters["provider_data"] == '{"test_data": 123456789}'

28
tests/test_sticker.py
View file

@ -23,7 +23,7 @@ from pathlib import Path
import pytest
from flaky import flaky
from telegram import Audio, Bot, MaskPosition, PhotoSize, Sticker, StickerSet
from telegram import Audio, Bot, File, MaskPosition, PhotoSize, Sticker, StickerSet
from telegram.error import BadRequest, TelegramError
from telegram.request import RequestData
from tests.conftest import (
@ -91,6 +91,8 @@ class TestSticker:
sticker_file_id = "5a3128a4d2a04750b5b58397f3b5e812"
sticker_file_unique_id = "adc3145fd2e84d95b64d68eaa22aa33e"
premium_animation = File("this_is_an_id", "this_is_an_unique_id")
def test_slot_behaviour(self, sticker, mro_slots, recwarn):
for attr in sticker.__slots__:
assert getattr(sticker, attr, "err") != "err", f"got extra slot '{attr}'"
@ -118,6 +120,8 @@ class TestSticker:
assert sticker.thumb.width == self.thumb_width
assert sticker.thumb.height == self.thumb_height
assert sticker.thumb.file_size == self.thumb_file_size
# we need to be a premium TG user to send a premium sticker, so the below is not tested
# assert sticker.premium_animation == self.premium_animation
@flaky(3, 1)
async def test_send_all_args(self, bot, chat_id, sticker_file, sticker):
@ -135,6 +139,8 @@ class TestSticker:
assert message.sticker.is_animated == sticker.is_animated
assert message.sticker.is_video == sticker.is_video
assert message.sticker.file_size == sticker.file_size
# we need to be a premium TG user to send a premium sticker, so the below is not tested
# assert message.sticker.premium_animation == sticker.premium_animation
assert isinstance(message.sticker.thumb, PhotoSize)
assert isinstance(message.sticker.thumb.file_id, str)
@ -212,6 +218,7 @@ class TestSticker:
"thumb": sticker.thumb.to_dict(),
"emoji": self.emoji,
"file_size": self.file_size,
"premium_animation": self.premium_animation.to_dict(),
}
json_sticker = Sticker.de_json(json_dict, bot)
@ -224,6 +231,7 @@ class TestSticker:
assert json_sticker.emoji == self.emoji
assert json_sticker.file_size == self.file_size
assert json_sticker.thumb == sticker.thumb
assert json_sticker.premium_animation == self.premium_animation
async def test_send_with_sticker(self, monkeypatch, bot, chat_id, sticker):
async def make_assertion(url, request_data: RequestData, *args, **kwargs):
@ -317,6 +325,24 @@ class TestSticker:
with pytest.raises(TypeError):
await bot.send_sticker(chat_id)
@flaky(3, 1)
async def test_premium_animation(self, bot):
# testing animation sucks a bit since we can't create a premium sticker. What we can do is
# get a sticker set which includes a premium sticker and check that specific one.
premium_sticker_set = await bot.get_sticker_set("Flame")
# the first one to appear here is a sticker with unique file id of AQADOBwAAifPOElr
# this could change in the future ofc.
premium_sticker = premium_sticker_set.stickers[20]
assert premium_sticker.premium_animation.file_unique_id == "AQADOBwAAifPOElr"
assert isinstance(premium_sticker.premium_animation.file_id, str)
assert premium_sticker.premium_animation.file_id != ""
premium_sticker_dict = {
"file_unique_id": "AQADOBwAAifPOElr",
"file_id": premium_sticker.premium_animation.file_id,
"file_size": premium_sticker.premium_animation.file_size,
}
assert premium_sticker.premium_animation.to_dict() == premium_sticker_dict
def test_equality(self, sticker):
a = Sticker(
sticker.file_id,

28
tests/test_updater.py
View file

@ -504,7 +504,10 @@ class TestUpdater:
@pytest.mark.parametrize("ext_bot", [True, False])
@pytest.mark.parametrize("drop_pending_updates", (True, False))
async def test_webhook_basic(self, monkeypatch, updater, drop_pending_updates, ext_bot):
@pytest.mark.parametrize("secret_token", ["SecretToken", None])
async def test_webhook_basic(
self, monkeypatch, updater, drop_pending_updates, ext_bot, secret_token
):
# Testing with both ExtBot and Bot to make sure any logic in WebhookHandler
# that depends on this distinction works
if ext_bot and not isinstance(updater.bot, ExtBot):
@ -533,13 +536,16 @@ class TestUpdater:
ip_address=ip,
port=port,
url_path="TOKEN",
secret_token=secret_token,
)
assert return_value is updater.update_queue
assert updater.running
# Now, we send an update to the server
update = make_message_update("Webhook")
await send_webhook_message(ip, port, update.to_json(), "TOKEN")
await send_webhook_message(
ip, port, update.to_json(), "TOKEN", secret_token=secret_token
)
assert (await updater.update_queue.get()).to_dict() == update.to_dict()
# Returns Not Found if path is incorrect
@ -550,6 +556,22 @@ class TestUpdater:
response = await send_webhook_message(ip, port, None, "TOKEN", get_method="HEAD")
assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED
if secret_token:
# Returns Forbidden if no secret token is set
response_text = "<html><title>403: {0}</title><body>403: {0}</body></html>"
response = await send_webhook_message(ip, port, update.to_json(), "TOKEN")
assert response.status_code == HTTPStatus.FORBIDDEN
assert response.text == response_text.format(
"Request did not include the secret token"
)
# Returns Forbidden if the secret token is wrong
response = await send_webhook_message(
ip, port, update.to_json(), "TOKEN", secret_token="NotTheSecretToken"
)
assert response.status_code == HTTPStatus.FORBIDDEN
assert response.text == response_text.format("Request had the wrong secret token")
await updater.stop()
assert not updater.running
@ -600,6 +622,7 @@ class TestUpdater:
max_connections=40,
allowed_updates=None,
ip_address=None,
secret_token=None,
**expected_delete_webhook,
)
@ -641,6 +664,7 @@ class TestUpdater:
max_connections=47,
allowed_updates=["message"],
ip_address="123.456.789",
secret_token=None,
**expected_delete_webhook,
)

12
tests/test_user.py
View file

@ -35,6 +35,8 @@ def json_dict():
"can_join_groups": TestUser.can_join_groups,
"can_read_all_group_messages": TestUser.can_read_all_group_messages,
"supports_inline_queries": TestUser.supports_inline_queries,
"is_premium": TestUser.is_premium,
"added_to_attachment_menu": TestUser.added_to_attachment_menu,
}
@ -51,6 +53,8 @@ def user(bot):
can_read_all_group_messages=TestUser.can_read_all_group_messages,
supports_inline_queries=TestUser.supports_inline_queries,
bot=bot,
is_premium=TestUser.is_premium,
added_to_attachment_menu=TestUser.added_to_attachment_menu,
)
@ -64,6 +68,8 @@ class TestUser:
can_join_groups = True
can_read_all_group_messages = True
supports_inline_queries = False
is_premium = True
added_to_attachment_menu = False
def test_slot_behaviour(self, user, mro_slots):
for attr in user.__slots__:
@ -82,6 +88,8 @@ class TestUser:
assert user.can_join_groups == self.can_join_groups
assert user.can_read_all_group_messages == self.can_read_all_group_messages
assert user.supports_inline_queries == self.supports_inline_queries
assert user.is_premium == self.is_premium
assert user.added_to_attachment_menu == self.added_to_attachment_menu
def test_de_json_without_username(self, json_dict, bot):
del json_dict["username"]
@ -97,6 +105,8 @@ class TestUser:
assert user.can_join_groups == self.can_join_groups
assert user.can_read_all_group_messages == self.can_read_all_group_messages
assert user.supports_inline_queries == self.supports_inline_queries
assert user.is_premium == self.is_premium
assert user.added_to_attachment_menu == self.added_to_attachment_menu
def test_de_json_without_username_and_last_name(self, json_dict, bot):
del json_dict["username"]
@ -113,6 +123,8 @@ class TestUser:
assert user.can_join_groups == self.can_join_groups
assert user.can_read_all_group_messages == self.can_read_all_group_messages
assert user.supports_inline_queries == self.supports_inline_queries
assert user.is_premium == self.is_premium
assert user.added_to_attachment_menu == self.added_to_attachment_menu
def test_name(self, user):
assert user.name == "@username"