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 2025-03-16 12:25:45 +01:00
Code Issues Projects Releases Packages Wiki Activity

Add default_disable_timeout

Browse source
This commit is contained in:
Hinrich mahler 2019-10-27 13:45:50 +00:00
parent 6ef7d763ae
commit d073813fb6
11 changed files with 117 additions and 95 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
telegram/bot.py
View file

@ -89,6 +89,8 @@ class Bot(TelegramObject):
`disable_notification` parameter used if not set explicitly in method call.
default_disable_web_page_preview (:obj:`bool`, optional): Default setting for the
`disable_web_page_preview` parameter used if not set explicitly in method call.
default_timeout (:obj:`int` | :obj:`float`, optional): Default setting for the
`timeout` parameter used if not set explicitly in method call.
"""
@ -114,7 +116,9 @@ class Bot(TelegramObject):
]
# ... make a dict of kwarg name and the default value
default_kwargs = {
kwarg_name: getattr(defaults, kwarg_name) for kwarg_name in needs_default
kwarg_name: getattr(defaults, kwarg_name) for kwarg_name in needs_default if (
getattr(defaults, kwarg_name) is not DEFAULT_NONE
)
}
# ... apply the defaults using a partial
if default_kwargs:
@ -131,13 +135,17 @@ class Bot(TelegramObject):
private_key_password=None,
default_parse_mode=None,
default_disable_notification=None,
default_disable_web_page_preview=None):
default_disable_web_page_preview=None,
# Timeout needs special treatment, since the bot methods have two different
# default values for timeout (None and 20s)
default_timeout=DEFAULT_NONE):
self.token = self._validate_token(token)
# Gather default
self.defaults = Defaults(parse_mode=default_parse_mode,
disable_notification=default_disable_notification,
disable_web_page_preview=None)
disable_web_page_preview=default_disable_web_page_preview,
timeout=default_timeout)
if base_url is None:
base_url = 'https://api.telegram.org/bot'

14
telegram/ext/updater.py
View file

@ -28,7 +28,7 @@ from queue import Queue
from telegram import Bot, TelegramError
from telegram.ext import Dispatcher, JobQueue
from telegram.error import Unauthorized, InvalidToken, RetryAfter, TimedOut
from telegram.utils.helpers import get_signal_name
from telegram.utils.helpers import get_signal_name, DEFAULT_NONE
from telegram.utils.request import Request
from telegram.utils.webhookhandler import (WebhookServer, WebhookAppClass)
@ -58,12 +58,6 @@ class Updater(object):
persistence (:class:`telegram.ext.BasePersistence`): Optional. The persistence class to
store data that should be persistent over restarts.
use_context (:obj:`bool`, optional): ``True`` if using context based callbacks.
default_parse_mode (:obj:`str`): Optional. Default parse mode used if not set explicitly in
method call. See the constants in :class:`telegram.ParseMode` for the available modes.
default_disable_notification (:obj:`bool`): Optional. Default setting for the
`disable_notification` parameter used if not set explicitly in method call.
default_disable_web_page_preview (:obj:`bool`): Optional. Default setting for the
`disable_web_page_preview` parameter used if not set explicitly in method call.
Args:
token (:obj:`str`, optional): The bot's token given by the @BotFather.
@ -93,6 +87,8 @@ class Updater(object):
`disable_notification` parameter used if not set explicitly in method call.
default_disable_web_page_preview (:obj:`bool`, optional): Default setting for the
`disable_web_page_preview` parameter used if not set explicitly in method call.
default_timeout (:obj:`int` | :obj:`float`, optional): Default setting for the
`timeout` parameter used if not set explicitly in method call.
Note:
You must supply either a :attr:`bot` or a :attr:`token` argument.
@ -117,6 +113,7 @@ class Updater(object):
default_parse_mode=None,
default_disable_notification=None,
default_disable_web_page_preview=None,
default_timeout=DEFAULT_NONE,
use_context=False):
if (token is None) and (bot is None):
@ -152,7 +149,8 @@ class Updater(object):
private_key_password=private_key_password,
default_parse_mode=default_parse_mode,
default_disable_notification=default_disable_notification,
default_disable_web_page_preview=default_disable_web_page_preview)
default_disable_web_page_preview=default_disable_web_page_preview,
default_timeout=default_timeout)
self.user_sig_handler = user_sig_handler
self.update_queue = Queue()
self.job_queue = JobQueue()

88
telegram/utils/helpers.py
View file

@ -258,42 +258,6 @@ def decode_user_chat_data_from_json(data):
return tmp
class Defaults:
"""Convenience Class to gather all parameters with a (user defined) default value
Attributes:
parse_mode (:obj:`str`): Optional. Send Markdown or HTML, if you want Telegram apps to show
bold, italic, fixed-width toxt or URLs in your bot's message.
disable_notification (:obj:`bool`): Optional. Sends the message silently. Users will
receive a notification with no sound.
disable_web_page_preview (:obj:`bool`): Optional. Disables link previews for links in this
message.
Parameters:
parse_mode (:obj:`str`, optional): Send Markdown or HTML, if you want Telegram apps to show
bold, italic, fixed-width toxt or URLs in your bot's message.
disable_notification (:obj:`bool`, optional): Sends the message silently. Users will
receive a notification with no sound.
disable_web_page_preview (:obj:`bool`, optional): Disables link previews for links in this
message.
"""
def __init__(self, parse_mode=None, disable_notification=None, disable_web_page_preview=None):
self.parse_mode = parse_mode
self.disable_notification = disable_notification
self.disable_web_page_preview = disable_web_page_preview
def __hash__(self):
return hash((self.parse_mode, self.disable_notification, self.disable_web_page_preview))
def __eq__(self, other):
if isinstance(other, Defaults):
return self.__dict__ == other.__dict__
return False
def __ne__(self, other):
return not self == other
class DefaultValue:
"""Wrapper for immutable default arguments that allows to check, if the default value was set
explicitly. Usage::
@ -349,3 +313,55 @@ class DefaultValue:
DEFAULT_NONE = DefaultValue(None)
""":class:`DefaultValue`: Default `None`"""
class Defaults:
"""Convenience Class to gather all parameters with a (user defined) default value
Attributes:
parse_mode (:obj:`str`): Optional. Send Markdown or HTML, if you want Telegram apps to show
bold, italic, fixed-width toxt or URLs in your bot's message.
disable_notification (:obj:`bool`): Optional. Sends the message silently. Users will
receive a notification with no sound.
disable_web_page_preview (:obj:`bool`): Optional. Disables link previews for links in this
message.
timeout (:obj:`int` | :obj:`float`): Optional. If this value is specified, use it as the
read timeout from the server (instead of the one specified during creation of the
connection pool).
Parameters:
parse_mode (:obj:`str`, optional): Send Markdown or HTML, if you want Telegram apps to show
bold, italic, fixed-width toxt or URLs in your bot's message.
disable_notification (:obj:`bool`, optional): Sends the message silently. Users will
receive a notification with no sound.
disable_web_page_preview (:obj:`bool`, optional): Disables link previews for links in this
message.
timeout (:obj:`int` | :obj:`float`, optional): If this value is specified, use it as the
read timeout from the server (instead of the one specified during creation of the
connection pool).
"""
def __init__(self,
parse_mode=None,
disable_notification=None,
disable_web_page_preview=None,
# Timeout needs special treatment, since the bot methods have two different
# default values for timeout (None and 20s)
timeout=DEFAULT_NONE):
self.parse_mode = parse_mode
self.disable_notification = disable_notification
self.disable_web_page_preview = disable_web_page_preview
self.timeout = timeout
def __hash__(self):
return hash((self.parse_mode,
self.disable_notification,
self.disable_web_page_preview,
self.timeout))
def __eq__(self, other):
if isinstance(other, Defaults):
return self.__dict__ == other.__dict__
return False
def __ne__(self, other):
return not self == other

4
tests/test_callbackquery.py
View file

@ -80,9 +80,9 @@ class TestCallbackQuery(object):
def test_answer(self, monkeypatch, callback_query):
def test(*args, **kwargs):
return args[1] == callback_query.id
return args[0] == callback_query.id
monkeypatch.setattr('telegram.Bot.answerCallbackQuery', test)
monkeypatch.setattr(callback_query.bot, 'answerCallbackQuery', test)
# TODO: PEP8
assert callback_query.answer()

40
tests/test_chat.py
View file

@ -86,69 +86,69 @@ class TestChat(object):
def test_send_action(self, monkeypatch, chat):
def test(*args, **kwargs):
id = args[1] == chat.id
id = args[0] == chat.id
action = kwargs['action'] == ChatAction.TYPING
return id and action
monkeypatch.setattr('telegram.Bot.send_chat_action', test)
monkeypatch.setattr(chat.bot, 'send_chat_action', test)
assert chat.send_action(action=ChatAction.TYPING)
def test_leave(self, monkeypatch, chat):
def test(*args, **kwargs):
return args[1] == chat.id
return args[0] == chat.id
monkeypatch.setattr('telegram.Bot.leave_chat', test)
monkeypatch.setattr(chat.bot, 'leave_chat', test)
assert chat.leave()
def test_get_administrators(self, monkeypatch, chat):
def test(*args, **kwargs):
return args[1] == chat.id
return args[0] == chat.id
monkeypatch.setattr('telegram.Bot.get_chat_administrators', test)
monkeypatch.setattr(chat.bot, 'get_chat_administrators', test)
assert chat.get_administrators()
def test_get_members_count(self, monkeypatch, chat):
def test(*args, **kwargs):
return args[1] == chat.id
return args[0] == chat.id
monkeypatch.setattr('telegram.Bot.get_chat_members_count', test)
monkeypatch.setattr(chat.bot, 'get_chat_members_count', test)
assert chat.get_members_count()
def test_get_member(self, monkeypatch, chat):
def test(*args, **kwargs):
chat_id = args[1] == chat.id
user_id = args[2] == 42
chat_id = args[0] == chat.id
user_id = args[1] == 42
return chat_id and user_id
monkeypatch.setattr('telegram.Bot.get_chat_member', test)
monkeypatch.setattr(chat.bot, 'get_chat_member', test)
assert chat.get_member(42)
def test_kick_member(self, monkeypatch, chat):
def test(*args, **kwargs):
chat_id = args[1] == chat.id
user_id = args[2] == 42
chat_id = args[0] == chat.id
user_id = args[1] == 42
until = kwargs['until_date'] == 43
return chat_id and user_id and until
monkeypatch.setattr('telegram.Bot.kick_chat_member', test)
monkeypatch.setattr(chat.bot, 'kick_chat_member', test)
assert chat.kick_member(42, until_date=43)
def test_unban_member(self, monkeypatch, chat):
def test(*args, **kwargs):
chat_id = args[1] == chat.id
user_id = args[2] == 42
chat_id = args[0] == chat.id
user_id = args[1] == 42
return chat_id and user_id
monkeypatch.setattr('telegram.Bot.unban_chat_member', test)
monkeypatch.setattr(chat.bot, 'unban_chat_member', test)
assert chat.unban_member(42)
def test_set_permissions(self, monkeypatch, chat):
def test(*args, **kwargs):
chat_id = args[1] == chat.id
permissions = args[2] == self.permissions
chat_id = args[0] == chat.id
permissions = args[1] == self.permissions
return chat_id and permissions
monkeypatch.setattr('telegram.Bot.set_chat_permissions', test)
monkeypatch.setattr(chat.bot, 'set_chat_permissions', test)
assert chat.set_permissions(self.permissions)
def test_instance_method_send_message(self, monkeypatch, chat):

4
tests/test_inlinequery.py
View file

@ -63,9 +63,9 @@ class TestInlineQuery(object):
def test_answer(self, monkeypatch, inline_query):
def test(*args, **kwargs):
return args[1] == inline_query.id
return args[0] == inline_query.id
monkeypatch.setattr('telegram.Bot.answer_inline_query', test)
monkeypatch.setattr(inline_query.bot, 'answer_inline_query', test)
assert inline_query.answer()
def test_equality(self):

4
tests/test_passport.py
View file

@ -296,9 +296,9 @@ class TestPassport(object):
selfie = passport_data.decrypted_data[1].selfie
def get_file(*args, **kwargs):
return File(args[1])
return File(args[0])
monkeypatch.setattr('telegram.Bot.get_file', get_file)
monkeypatch.setattr(passport_data.bot, 'get_file', get_file)
file = selfie.get_file()
assert file.file_id == selfie.file_id
assert file._credentials.file_hash == self.driver_license_selfie_credentials_file_hash

4
tests/test_precheckoutquery.py
View file

@ -77,9 +77,9 @@ class TestPreCheckoutQuery(object):
def test_answer(self, monkeypatch, pre_checkout_query):
def test(*args, **kwargs):
return args[1] == pre_checkout_query.id
return args[0] == pre_checkout_query.id
monkeypatch.setattr('telegram.Bot.answer_pre_checkout_query', test)
monkeypatch.setattr(pre_checkout_query.bot, 'answer_pre_checkout_query', test)
assert pre_checkout_query.answer()
def test_equality(self):

4
tests/test_shippingquery.py
View file

@ -63,9 +63,9 @@ class TestShippingQuery(object):
def test_answer(self, monkeypatch, shipping_query):
def test(*args, **kwargs):
return args[1] == shipping_query.id
return args[0] == shipping_query.id
monkeypatch.setattr('telegram.Bot.answer_shipping_query', test)
monkeypatch.setattr(shipping_query.bot, 'answer_shipping_query', test)
assert shipping_query.answer()
def test_equality(self):

32
tests/test_updater.py
View file

@ -79,8 +79,8 @@ class TestUpdater(object):
def test(*args, **kwargs):
raise error
monkeypatch.setattr('telegram.Bot.get_updates', test)
monkeypatch.setattr('telegram.Bot.set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'get_updates', test)
monkeypatch.setattr(updater.bot, 'set_webhook', lambda *args, **kwargs: True)
updater.dispatcher.add_error_handler(self.error_handler)
updater.start_polling(0.01)
@ -99,8 +99,8 @@ class TestUpdater(object):
raise error
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr('telegram.Bot.get_updates', test)
monkeypatch.setattr('telegram.Bot.set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'get_updates', test)
monkeypatch.setattr(updater.bot, 'set_webhook', lambda *args, **kwargs: True)
updater.dispatcher.add_error_handler(self.error_handler)
updater.start_polling(0.01)
assert self.err_handler_called.wait(1) is not True
@ -127,8 +127,8 @@ class TestUpdater(object):
event.set()
raise error
monkeypatch.setattr('telegram.Bot.get_updates', test)
monkeypatch.setattr('telegram.Bot.set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'get_updates', test)
monkeypatch.setattr(updater.bot, 'set_webhook', lambda *args, **kwargs: True)
updater.dispatcher.add_error_handler(self.error_handler)
updater.start_polling(0.01)
@ -144,8 +144,8 @@ class TestUpdater(object):
def test_webhook(self, monkeypatch, updater):
q = Queue()
monkeypatch.setattr('telegram.Bot.set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr('telegram.Bot.delete_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'delete_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr('telegram.ext.Dispatcher.process_update', lambda _, u: q.put(u))
ip = '127.0.0.1'
@ -182,8 +182,8 @@ class TestUpdater(object):
updater.stop()
def test_webhook_ssl(self, monkeypatch, updater):
monkeypatch.setattr('telegram.Bot.set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr('telegram.Bot.delete_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'delete_webhook', lambda *args, **kwargs: True)
ip = '127.0.0.1'
port = randrange(1024, 49152) # Select random port for travis
tg_err = False
@ -204,8 +204,8 @@ class TestUpdater(object):
def test_webhook_no_ssl(self, monkeypatch, updater):
q = Queue()
monkeypatch.setattr('telegram.Bot.set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr('telegram.Bot.delete_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'set_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr(updater.bot, 'delete_webhook', lambda *args, **kwargs: True)
monkeypatch.setattr('telegram.ext.Dispatcher.process_update', lambda _, u: q.put(u))
ip = '127.0.0.1'
@ -227,12 +227,12 @@ class TestUpdater(object):
def test_bootstrap_retries_success(self, monkeypatch, updater, error):
retries = 2
def attempt(_, *args, **kwargs):
def attempt(*args, **kwargs):
if self.attempts < retries:
self.attempts += 1
raise error
monkeypatch.setattr('telegram.Bot.set_webhook', attempt)
monkeypatch.setattr(updater.bot, 'set_webhook', attempt)
updater.running = True
updater._bootstrap(retries, False, 'path', None, bootstrap_interval=0)
@ -246,11 +246,11 @@ class TestUpdater(object):
def test_bootstrap_retries_error(self, monkeypatch, updater, error, attempts):
retries = 1
def attempt(_, *args, **kwargs):
def attempt(*args, **kwargs):
self.attempts += 1
raise error
monkeypatch.setattr('telegram.Bot.set_webhook', attempt)
monkeypatch.setattr(updater.bot, 'set_webhook', attempt)
updater.running = True
with pytest.raises(type(error)):

4
tests/test_user.py
View file

@ -103,10 +103,10 @@ class TestUser(object):
assert user.link is None
def test_get_profile_photos(self, monkeypatch, user):
def test(_, *args, **kwargs):
def test(*args, **kwargs):
return args[0] == user.id
monkeypatch.setattr('telegram.Bot.get_user_profile_photos', test)
monkeypatch.setattr(user.bot, 'get_user_profile_photos', test)
assert user.get_profile_photos()
def test_instance_method_send_message(self, monkeypatch, user):