mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-10-23 17:36:26 +02:00
Remove message decorator to fix default timeouts. (#1156)
* Remove message decorator to fix default timeouts. * Make message wrapper method private. * Make tests pass * Fix callbackquery shortcuts Closes #1180 * Fix callbackquery shortcut tests * Fix merge * Address CR * Add missing default timeout=20 for some bot file uploads * Fix wrong return value in convhandler Probably stems from a combination of bad merge plus quickly merged hacktoberfest PR.
This commit is contained in:
Jasmin Bom
GitHub
parent
d33e1d9913
commit
950ec35970
5 changed files with 136 additions and 105 deletions
154
telegram/bot.py
154
telegram/bot.py
|
|
@ -21,6 +21,7 @@
|
|||
"""This module contains an object that represents a Telegram Bot."""
|
||||
|
||||
import functools
|
||||
|
||||
try:
|
||||
import ujson as json
|
||||
except ImportError:
|
||||
|
|
@ -70,33 +71,6 @@ def log(func):
|
|||
return decorator
|
||||
|
||||
|
||||
def message(func):
|
||||
@functools.wraps(func)
|
||||
def decorator(self, *args, **kwargs):
|
||||
url, data = func(self, *args, **kwargs)
|
||||
if kwargs.get('reply_to_message_id'):
|
||||
data['reply_to_message_id'] = kwargs.get('reply_to_message_id')
|
||||
|
||||
if kwargs.get('disable_notification'):
|
||||
data['disable_notification'] = kwargs.get('disable_notification')
|
||||
|
||||
if kwargs.get('reply_markup'):
|
||||
reply_markup = kwargs.get('reply_markup')
|
||||
if isinstance(reply_markup, ReplyMarkup):
|
||||
data['reply_markup'] = reply_markup.to_json()
|
||||
else:
|
||||
data['reply_markup'] = reply_markup
|
||||
|
||||
result = self._request.post(url, data, timeout=kwargs.get('timeout'))
|
||||
|
||||
if result is True:
|
||||
return result
|
||||
|
||||
return Message.de_json(result, self)
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
class Bot(TelegramObject):
|
||||
"""This object represents a Telegram Bot.
|
||||
|
||||
|
|
@ -132,6 +106,27 @@ class Bot(TelegramObject):
|
|||
password=private_key_password,
|
||||
backend=default_backend())
|
||||
|
||||
def _message(self, url, data, reply_to_message_id=None, disable_notification=None,
|
||||
reply_markup=None, timeout=None, **kwargs):
|
||||
if reply_to_message_id is not None:
|
||||
data['reply_to_message_id'] = reply_to_message_id
|
||||
|
||||
if disable_notification is not None:
|
||||
data['disable_notification'] = disable_notification
|
||||
|
||||
if reply_markup is not None:
|
||||
if isinstance(reply_markup, ReplyMarkup):
|
||||
data['reply_markup'] = reply_markup.to_json()
|
||||
else:
|
||||
data['reply_markup'] = reply_markup
|
||||
|
||||
result = self._request.post(url, data, timeout=timeout)
|
||||
|
||||
if result is True:
|
||||
return result
|
||||
|
||||
return Message.de_json(result, self)
|
||||
|
||||
@property
|
||||
def request(self):
|
||||
return self._request
|
||||
|
|
@ -208,7 +203,6 @@ class Bot(TelegramObject):
|
|||
return self.bot
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_message(self,
|
||||
chat_id,
|
||||
text,
|
||||
|
|
@ -259,7 +253,9 @@ class Bot(TelegramObject):
|
|||
if disable_web_page_preview:
|
||||
data['disable_web_page_preview'] = disable_web_page_preview
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
timeout=timeout, **kwargs)
|
||||
|
||||
@log
|
||||
def delete_message(self, chat_id, message_id, timeout=None, **kwargs):
|
||||
|
|
@ -298,7 +294,6 @@ class Bot(TelegramObject):
|
|||
return result
|
||||
|
||||
@log
|
||||
@message
|
||||
def forward_message(self,
|
||||
chat_id,
|
||||
from_chat_id,
|
||||
|
|
@ -340,10 +335,10 @@ class Bot(TelegramObject):
|
|||
if message_id:
|
||||
data['message_id'] = message_id
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, disable_notification=disable_notification,
|
||||
timeout=timeout, **kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_photo(self,
|
||||
chat_id,
|
||||
photo,
|
||||
|
|
@ -404,10 +399,11 @@ class Bot(TelegramObject):
|
|||
if parse_mode:
|
||||
data['parse_mode'] = parse_mode
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_audio(self,
|
||||
chat_id,
|
||||
audio,
|
||||
|
|
@ -494,10 +490,11 @@ class Bot(TelegramObject):
|
|||
thumb = InputFile(thumb, attach=True)
|
||||
data['thumb'] = thumb
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_document(self,
|
||||
chat_id,
|
||||
document,
|
||||
|
|
@ -570,10 +567,11 @@ class Bot(TelegramObject):
|
|||
thumb = InputFile(thumb, attach=True)
|
||||
data['thumb'] = thumb
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_sticker(self,
|
||||
chat_id,
|
||||
sticker,
|
||||
|
|
@ -622,10 +620,11 @@ class Bot(TelegramObject):
|
|||
|
||||
data = {'chat_id': chat_id, 'sticker': sticker}
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_video(self,
|
||||
chat_id,
|
||||
video,
|
||||
|
|
@ -714,10 +713,11 @@ class Bot(TelegramObject):
|
|||
thumb = InputFile(thumb, attach=True)
|
||||
data['thumb'] = thumb
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_video_note(self,
|
||||
chat_id,
|
||||
video_note,
|
||||
|
|
@ -784,10 +784,11 @@ class Bot(TelegramObject):
|
|||
thumb = InputFile(thumb, attach=True)
|
||||
data['thumb'] = thumb
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_animation(self,
|
||||
chat_id,
|
||||
animation,
|
||||
|
|
@ -866,10 +867,11 @@ class Bot(TelegramObject):
|
|||
if parse_mode:
|
||||
data['parse_mode'] = parse_mode
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_voice(self,
|
||||
chat_id,
|
||||
voice,
|
||||
|
|
@ -936,7 +938,9 @@ class Bot(TelegramObject):
|
|||
if parse_mode:
|
||||
data['parse_mode'] = parse_mode
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
def send_media_group(self,
|
||||
|
|
@ -981,7 +985,6 @@ class Bot(TelegramObject):
|
|||
return [Message.de_json(res, self) for res in result]
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_location(self,
|
||||
chat_id,
|
||||
latitude=None,
|
||||
|
|
@ -1044,10 +1047,11 @@ class Bot(TelegramObject):
|
|||
if live_period:
|
||||
data['live_period'] = live_period
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def edit_message_live_location(self,
|
||||
chat_id=None,
|
||||
message_id=None,
|
||||
|
|
@ -1056,6 +1060,7 @@ class Bot(TelegramObject):
|
|||
longitude=None,
|
||||
location=None,
|
||||
reply_markup=None,
|
||||
timeout=None,
|
||||
**kwargs):
|
||||
"""Use this method to edit live location messages sent by the bot or via the bot
|
||||
(for inline bots). A location can be edited until its :attr:`live_period` expires or
|
||||
|
|
@ -1107,15 +1112,15 @@ class Bot(TelegramObject):
|
|||
if inline_message_id:
|
||||
data['inline_message_id'] = inline_message_id
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, reply_markup=reply_markup, **kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def stop_message_live_location(self,
|
||||
chat_id=None,
|
||||
message_id=None,
|
||||
inline_message_id=None,
|
||||
reply_markup=None,
|
||||
timeout=None,
|
||||
**kwargs):
|
||||
"""Use this method to stop updating a live location message sent by the bot or via the bot
|
||||
(for inline bots) before live_period expires.
|
||||
|
|
@ -1149,10 +1154,9 @@ class Bot(TelegramObject):
|
|||
if inline_message_id:
|
||||
data['inline_message_id'] = inline_message_id
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, reply_markup=reply_markup, **kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_venue(self,
|
||||
chat_id,
|
||||
latitude=None,
|
||||
|
|
@ -1232,10 +1236,11 @@ class Bot(TelegramObject):
|
|||
if foursquare_type:
|
||||
data['foursquare_type'] = foursquare_type
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_contact(self,
|
||||
chat_id,
|
||||
phone_number=None,
|
||||
|
|
@ -1301,10 +1306,11 @@ class Bot(TelegramObject):
|
|||
if vcard:
|
||||
data['vcard'] = vcard
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_game(self,
|
||||
chat_id,
|
||||
game_short_name,
|
||||
|
|
@ -1343,7 +1349,9 @@ class Bot(TelegramObject):
|
|||
|
||||
data = {'chat_id': chat_id, 'game_short_name': game_short_name}
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
def send_chat_action(self, chat_id, action, timeout=None, **kwargs):
|
||||
|
|
@ -1677,7 +1685,6 @@ class Bot(TelegramObject):
|
|||
return result
|
||||
|
||||
@log
|
||||
@message
|
||||
def edit_message_text(self,
|
||||
text,
|
||||
chat_id=None,
|
||||
|
|
@ -1736,10 +1743,9 @@ class Bot(TelegramObject):
|
|||
if disable_web_page_preview:
|
||||
data['disable_web_page_preview'] = disable_web_page_preview
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, reply_markup=reply_markup, **kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def edit_message_caption(self,
|
||||
chat_id=None,
|
||||
message_id=None,
|
||||
|
|
@ -1800,10 +1806,9 @@ class Bot(TelegramObject):
|
|||
if inline_message_id:
|
||||
data['inline_message_id'] = inline_message_id
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, reply_markup=reply_markup, **kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def edit_message_media(self,
|
||||
chat_id=None,
|
||||
message_id=None,
|
||||
|
|
@ -1853,10 +1858,9 @@ class Bot(TelegramObject):
|
|||
if inline_message_id:
|
||||
data['inline_message_id'] = inline_message_id
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, reply_markup=reply_markup, **kwargs)
|
||||
|
||||
@log
|
||||
@message
|
||||
def edit_message_reply_markup(self,
|
||||
chat_id=None,
|
||||
message_id=None,
|
||||
|
|
@ -1907,7 +1911,7 @@ class Bot(TelegramObject):
|
|||
if inline_message_id:
|
||||
data['inline_message_id'] = inline_message_id
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, reply_markup=reply_markup, **kwargs)
|
||||
|
||||
@log
|
||||
def get_updates(self,
|
||||
|
|
@ -2326,7 +2330,6 @@ class Bot(TelegramObject):
|
|||
return WebhookInfo.de_json(result, self)
|
||||
|
||||
@log
|
||||
@message
|
||||
def set_game_score(self,
|
||||
user_id,
|
||||
score,
|
||||
|
|
@ -2385,7 +2388,7 @@ class Bot(TelegramObject):
|
|||
if disable_edit_message is not None:
|
||||
data['disable_edit_message'] = disable_edit_message
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, **kwargs)
|
||||
|
||||
@log
|
||||
def get_game_high_scores(self,
|
||||
|
|
@ -2436,7 +2439,6 @@ class Bot(TelegramObject):
|
|||
return [GameHighScore.de_json(hs, self) for hs in result]
|
||||
|
||||
@log
|
||||
@message
|
||||
def send_invoice(self,
|
||||
chat_id,
|
||||
title,
|
||||
|
|
@ -2560,7 +2562,9 @@ class Bot(TelegramObject):
|
|||
if send_email_to_provider is not None:
|
||||
data['send_email_to_provider'] = send_email_to_provider
|
||||
|
||||
return url, data
|
||||
return self._message(url, data, timeout=timeout, disable_notification=disable_notification,
|
||||
reply_to_message_id=reply_to_message_id, reply_markup=reply_markup,
|
||||
**kwargs)
|
||||
|
||||
@log
|
||||
def answer_shipping_query(self,
|
||||
|
|
@ -2839,7 +2843,7 @@ class Bot(TelegramObject):
|
|||
return result
|
||||
|
||||
@log
|
||||
def set_chat_photo(self, chat_id, photo, timeout=None, **kwargs):
|
||||
def set_chat_photo(self, chat_id, photo, timeout=20, **kwargs):
|
||||
"""Use this method to set a new profile photo for the chat.
|
||||
|
||||
Photos can't be changed for private chats. The bot must be an administrator in the chat
|
||||
|
|
@ -3075,7 +3079,7 @@ class Bot(TelegramObject):
|
|||
return StickerSet.de_json(result, self)
|
||||
|
||||
@log
|
||||
def upload_sticker_file(self, user_id, png_sticker, timeout=None, **kwargs):
|
||||
def upload_sticker_file(self, user_id, png_sticker, timeout=20, **kwargs):
|
||||
"""
|
||||
Use this method to upload a .png file with a sticker for later use in
|
||||
:attr:`create_new_sticker_set` and :attr:`add_sticker_to_set` methods (can be used multiple
|
||||
|
|
@ -3116,7 +3120,7 @@ class Bot(TelegramObject):
|
|||
|
||||
@log
|
||||
def create_new_sticker_set(self, user_id, name, title, png_sticker, emojis,
|
||||
contains_masks=None, mask_position=None, timeout=None, **kwargs):
|
||||
contains_masks=None, mask_position=None, timeout=20, **kwargs):
|
||||
"""Use this method to create new sticker set owned by a user.
|
||||
|
||||
The bot will be able to edit the created sticker set.
|
||||
|
|
@ -3176,7 +3180,7 @@ class Bot(TelegramObject):
|
|||
|
||||
@log
|
||||
def add_sticker_to_set(self, user_id, name, png_sticker, emojis, mask_position=None,
|
||||
timeout=None, **kwargs):
|
||||
timeout=20, **kwargs):
|
||||
"""Use this method to add a new sticker to a set created by the bot.
|
||||
|
||||
Note:
|
||||
|
|
|
|||
|
|
@ -116,16 +116,16 @@ class CallbackQuery(TelegramObject):
|
|||
"""
|
||||
return self.bot.answerCallbackQuery(self.id, *args, **kwargs)
|
||||
|
||||
def edit_message_text(self, *args, **kwargs):
|
||||
def edit_message_text(self, text, *args, **kwargs):
|
||||
"""Shortcut for either::
|
||||
|
||||
bot.edit_message_text(chat_id=update.callback_query.message.chat_id,
|
||||
bot.edit_message_text(text, chat_id=update.callback_query.message.chat_id,
|
||||
message_id=update.callback_query.message.message_id,
|
||||
*args, **kwargs)
|
||||
|
||||
or::
|
||||
|
||||
bot.edit_message_text(inline_message_id=update.callback_query.inline_message_id,
|
||||
bot.edit_message_text(text, inline_message_id=update.callback_query.inline_message_id,
|
||||
*args, **kwargs)
|
||||
|
||||
Returns:
|
||||
|
|
@ -134,22 +134,24 @@ class CallbackQuery(TelegramObject):
|
|||
|
||||
"""
|
||||
if self.inline_message_id:
|
||||
return self.bot.edit_message_text(
|
||||
inline_message_id=self.inline_message_id, *args, **kwargs)
|
||||
return self.bot.edit_message_text(text, inline_message_id=self.inline_message_id,
|
||||
*args, **kwargs)
|
||||
else:
|
||||
return self.bot.edit_message_text(
|
||||
chat_id=self.message.chat_id, message_id=self.message.message_id, *args, **kwargs)
|
||||
return self.bot.edit_message_text(text, chat_id=self.message.chat_id,
|
||||
message_id=self.message.message_id, *args, **kwargs)
|
||||
|
||||
def edit_message_caption(self, *args, **kwargs):
|
||||
def edit_message_caption(self, caption, *args, **kwargs):
|
||||
"""Shortcut for either::
|
||||
|
||||
bot.edit_message_caption(chat_id=update.callback_query.message.chat_id,
|
||||
bot.edit_message_caption(caption=caption,
|
||||
chat_id=update.callback_query.message.chat_id,
|
||||
message_id=update.callback_query.message.message_id,
|
||||
*args, **kwargs)
|
||||
|
||||
or::
|
||||
|
||||
bot.edit_message_caption(inline_message_id=update.callback_query.inline_message_id,
|
||||
bot.edit_message_caption(caption=caption
|
||||
inline_message_id=update.callback_query.inline_message_id,
|
||||
*args, **kwargs)
|
||||
|
||||
Returns:
|
||||
|
|
@ -158,22 +160,26 @@ class CallbackQuery(TelegramObject):
|
|||
|
||||
"""
|
||||
if self.inline_message_id:
|
||||
return self.bot.edit_message_caption(
|
||||
inline_message_id=self.inline_message_id, *args, **kwargs)
|
||||
return self.bot.edit_message_caption(caption=caption,
|
||||
inline_message_id=self.inline_message_id,
|
||||
*args, **kwargs)
|
||||
else:
|
||||
return self.bot.edit_message_caption(
|
||||
chat_id=self.message.chat_id, message_id=self.message.message_id, *args, **kwargs)
|
||||
return self.bot.edit_message_caption(caption=caption, chat_id=self.message.chat_id,
|
||||
message_id=self.message.message_id,
|
||||
*args, **kwargs)
|
||||
|
||||
def edit_message_reply_markup(self, *args, **kwargs):
|
||||
def edit_message_reply_markup(self, reply_markup, *args, **kwargs):
|
||||
"""Shortcut for either::
|
||||
|
||||
bot.edit_message_replyMarkup(chat_id=update.callback_query.message.chat_id,
|
||||
message_id=update.callback_query.message.message_id,
|
||||
reply_markup=reply_markup,
|
||||
*args, **kwargs)
|
||||
|
||||
or::
|
||||
|
||||
bot.edit_message_reply_markup(inline_message_id=update.callback_query.inline_message_id,
|
||||
reply_markup=reply_markup,
|
||||
*args, **kwargs)
|
||||
|
||||
Returns:
|
||||
|
|
@ -182,8 +188,11 @@ class CallbackQuery(TelegramObject):
|
|||
|
||||
"""
|
||||
if self.inline_message_id:
|
||||
return self.bot.edit_message_reply_markup(
|
||||
inline_message_id=self.inline_message_id, *args, **kwargs)
|
||||
return self.bot.edit_message_reply_markup(reply_markup=reply_markup,
|
||||
inline_message_id=self.inline_message_id,
|
||||
*args, **kwargs)
|
||||
else:
|
||||
return self.bot.edit_message_reply_markup(
|
||||
chat_id=self.message.chat_id, message_id=self.message.message_id, *args, **kwargs)
|
||||
return self.bot.edit_message_reply_markup(reply_markup=reply_markup,
|
||||
chat_id=self.message.chat_id,
|
||||
message_id=self.message.message_id,
|
||||
*args, **kwargs)
|
||||
|
|
|
|||
|
|
@ -276,9 +276,7 @@ class ConversationHandler(Handler):
|
|||
if check is not None and check is not False:
|
||||
# Save the current user and the selected handler for handle_update
|
||||
return key, candidate, check
|
||||
|
||||
else:
|
||||
return None
|
||||
return None
|
||||
|
||||
self.logger.debug('selecting conversation %s with state %s' % (str(key), str(state)))
|
||||
|
||||
|
|
|
|||
|
|
@ -616,8 +616,8 @@ class TestBot(object):
|
|||
# get_sticker_set, upload_sticker_file, create_new_sticker_set, add_sticker_to_set,
|
||||
# set_sticker_position_in_set and delete_sticker_from_set are tested in the
|
||||
# test_sticker module.
|
||||
|
||||
def test_timeout_propagation(self, monkeypatch, bot, chat_id):
|
||||
|
||||
def test_timeout_propagation_explicit(self, monkeypatch, bot, chat_id):
|
||||
|
||||
from telegram.vendor.ptb_urllib3.urllib3.util.timeout import Timeout
|
||||
|
||||
|
|
@ -642,3 +642,23 @@ class TestBot(object):
|
|||
# Test JSON submition
|
||||
with pytest.raises(OkException):
|
||||
bot.get_chat_administrators(chat_id, timeout=TIMEOUT)
|
||||
|
||||
def test_timeout_propagation_implicit(self, monkeypatch, bot, chat_id):
|
||||
|
||||
from telegram.vendor.ptb_urllib3.urllib3.util.timeout import Timeout
|
||||
|
||||
class OkException(Exception):
|
||||
pass
|
||||
|
||||
def request_wrapper(*args, **kwargs):
|
||||
obj = kwargs.get('timeout')
|
||||
if isinstance(obj, Timeout) and obj._read == 20:
|
||||
raise OkException
|
||||
|
||||
return b'{"ok": true, "result": []}'
|
||||
|
||||
monkeypatch.setattr('telegram.utils.request.Request._request_wrapper', request_wrapper)
|
||||
|
||||
# Test file uploading
|
||||
with pytest.raises(OkException):
|
||||
bot.send_photo(chat_id, open('tests/data/telegram.jpg', 'rb'))
|
||||
|
|
|
|||
|
|
@ -88,48 +88,48 @@ class TestCallbackQuery(object):
|
|||
|
||||
def test_edit_message_text(self, monkeypatch, callback_query):
|
||||
def test(*args, **kwargs):
|
||||
text = args[1] == 'test'
|
||||
try:
|
||||
id = kwargs['inline_message_id'] == callback_query.inline_message_id
|
||||
text = kwargs['text'] == 'test'
|
||||
return id and text
|
||||
except KeyError:
|
||||
chat_id = kwargs['chat_id'] == callback_query.message.chat_id
|
||||
message_id = kwargs['message_id'] == callback_query.message.message_id
|
||||
text = kwargs['text'] == 'test'
|
||||
return chat_id and message_id and text
|
||||
|
||||
monkeypatch.setattr('telegram.Bot.edit_message_text', test)
|
||||
assert callback_query.edit_message_text(text='test')
|
||||
assert callback_query.edit_message_text('test')
|
||||
|
||||
def test_edit_message_caption(self, monkeypatch, callback_query):
|
||||
def test(*args, **kwargs):
|
||||
caption = kwargs['caption'] == 'new caption'
|
||||
try:
|
||||
id = kwargs['inline_message_id'] == callback_query.inline_message_id
|
||||
caption = kwargs['caption'] == 'new caption'
|
||||
return id and caption
|
||||
except KeyError:
|
||||
id = kwargs['chat_id'] == callback_query.message.chat_id
|
||||
message = kwargs['message_id'] == callback_query.message.message_id
|
||||
caption = kwargs['caption'] == 'new caption'
|
||||
return id and message and caption
|
||||
|
||||
monkeypatch.setattr('telegram.Bot.edit_message_caption', test)
|
||||
assert callback_query.edit_message_caption(caption='new caption')
|
||||
assert callback_query.edit_message_caption('new caption')
|
||||
|
||||
def test_edit_message_reply_markup(self, monkeypatch, callback_query):
|
||||
def test(*args, **kwargs):
|
||||
reply_markup = kwargs['reply_markup'] == [['1', '2']]
|
||||
try:
|
||||
id = kwargs['inline_message_id'] == callback_query.inline_message_id
|
||||
reply_markup = kwargs['reply_markup'] == [['1', '2']]
|
||||
return id and reply_markup
|
||||
except KeyError:
|
||||
id = kwargs['chat_id'] == callback_query.message.chat_id
|
||||
message = kwargs['message_id'] == callback_query.message.message_id
|
||||
reply_markup = kwargs['reply_markup'] == [['1', '2']]
|
||||
return id and message and reply_markup
|
||||
|
||||
monkeypatch.setattr('telegram.Bot.edit_message_reply_markup', test)
|
||||
assert callback_query.edit_message_reply_markup(reply_markup=[['1', '2']])
|
||||
assert callback_query.edit_message_reply_markup([['1', '2']])
|
||||
|
||||
def test_equality(self):
|
||||
a = CallbackQuery(self.id, self.from_user, 'chat')
|
||||
|
|
|
|||
Loading…
Reference in a new issue