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 07:20:17 +01:00
Code Issues Projects Releases Packages Wiki Activity

Don't Pass Default Values of Optional Parameters to Telegram (#2978)

Browse source
This commit is contained in:
Harshil 2022-05-09 23:00:46 +05:30 committed by GitHub
parent ea8089ad3e
commit be8f4f7aad
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 111 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
telegram/_bot.py
View file

@ -2658,7 +2658,7 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
results: Union[
Sequence["InlineQueryResult"], Callable[[int], Optional[Sequence["InlineQueryResult"]]]
],
cache_time: int = 300,
cache_time: int = None,
is_personal: bool = None,
next_offset: str = None,
switch_pm_text: str = None,
@ -2775,7 +2775,7 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
self,
user_id: Union[str, int],
offset: int = None,
limit: int = 100,
limit: int = None,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3177,7 +3177,7 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
self,
callback_query_id: str,
text: str = None,
show_alert: bool = False,
show_alert: bool = None,
url: str = None,
cache_time: int = None,
read_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3598,8 +3598,8 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
async def get_updates(
self,
offset: int = None,
limit: int = 100,
timeout: int = 0,
limit: int = None,
timeout: float = None,
read_timeout: float = 2,
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,
@ -3678,7 +3678,7 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
await self._post(
"getUpdates",
data,
read_timeout=read_timeout + timeout,
read_timeout=read_timeout + timeout if timeout else read_timeout,
write_timeout=write_timeout,
connect_timeout=connect_timeout,
pool_timeout=pool_timeout,
@ -3702,7 +3702,7 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,
pool_timeout: ODVInput[float] = DEFAULT_NONE,
max_connections: int = 40,
max_connections: int = None,
allowed_updates: List[str] = None,
api_kwargs: JSONDict = None,
ip_address: str = None,
@ -6557,9 +6557,9 @@ class Bot(TelegramObject, AbstractAsyncContextManager):
chat_id: Union[int, str],
question: str,
options: List[str],
is_anonymous: bool = True,
type: str = Poll.REGULAR, # pylint: disable=redefined-builtin
allows_multiple_answers: bool = False,
is_anonymous: bool = None,
type: str = None, # pylint: disable=redefined-builtin
allows_multiple_answers: bool = None,
correct_option_id: int = None,
is_closed: bool = None,
disable_notification: ODVInput[bool] = DEFAULT_NONE,

2
telegram/_callbackquery.py
View file

@ -149,7 +149,7 @@ class CallbackQuery(TelegramObject):
async def answer(
self,
text: str = None,
show_alert: bool = False,
show_alert: bool = None,
url: str = None,
cache_time: int = None,
read_timeout: ODVInput[float] = DEFAULT_NONE,

7
telegram/_chat.py
View file

@ -1701,10 +1701,9 @@ class Chat(TelegramObject):
self,
question: str,
options: List[str],
is_anonymous: bool = True,
# We use constant.PollType.REGULAR instead of Poll.REGULAR here to avoid circular imports
type: str = constants.PollType.REGULAR, # pylint: disable=redefined-builtin
allows_multiple_answers: bool = False,
is_anonymous: bool = None,
type: str = None,
allows_multiple_answers: bool = None,
correct_option_id: int = None,
is_closed: bool = None,
disable_notification: ODVInput[bool] = DEFAULT_NONE,

4
telegram/_forcereply.py
View file

@ -68,12 +68,12 @@ class ForceReply(TelegramObject):
def __init__(
self,
selective: bool = False,
selective: bool = None,
input_field_placeholder: str = None,
**_kwargs: Any,
):
self.force_reply = True
self.selective = bool(selective)
self.selective = selective
self.input_field_placeholder = input_field_placeholder
self._id_attrs = (self.selective,)

2
telegram/_inline/inlinequery.py
View file

@ -118,7 +118,7 @@ class InlineQuery(TelegramObject):
results: Union[
Sequence["InlineQueryResult"], Callable[[int], Optional[Sequence["InlineQueryResult"]]]
],
cache_time: int = 300,
cache_time: int = None,
is_personal: bool = None,
next_offset: str = None,
switch_pm_text: str = None,

14
telegram/_message.py
View file

@ -460,10 +460,10 @@ class Message(TelegramObject):
left_chat_member: User = None,
new_chat_title: str = None,
new_chat_photo: List[PhotoSize] = None,
delete_chat_photo: bool = False,
group_chat_created: bool = False,
supergroup_chat_created: bool = False,
channel_chat_created: bool = False,
delete_chat_photo: bool = None,
group_chat_created: bool = None,
supergroup_chat_created: bool = None,
channel_chat_created: bool = None,
migrate_to_chat_id: int = None,
migrate_from_chat_id: int = None,
pinned_message: "Message" = None,
@ -1671,9 +1671,9 @@ class Message(TelegramObject):
self,
question: str,
options: List[str],
is_anonymous: bool = True,
type: str = Poll.REGULAR, # pylint: disable=redefined-builtin
allows_multiple_answers: bool = False,
is_anonymous: bool = None,
type: str = None, # pylint: disable=redefined-builtin
allows_multiple_answers: bool = None,
correct_option_id: int = None,
is_closed: bool = None,
disable_notification: ODVInput[bool] = DEFAULT_NONE,

12
telegram/_replykeyboardmarkup.py
View file

@ -88,9 +88,9 @@ class ReplyKeyboardMarkup(TelegramObject):
def __init__(
self,
keyboard: Sequence[Sequence[Union[str, KeyboardButton]]],
resize_keyboard: bool = False,
one_time_keyboard: bool = False,
selective: bool = False,
resize_keyboard: bool = None,
one_time_keyboard: bool = None,
selective: bool = None,
input_field_placeholder: str = None,
**_kwargs: Any,
):
@ -112,9 +112,9 @@ class ReplyKeyboardMarkup(TelegramObject):
self.keyboard.append(button_row)
# Optionals
self.resize_keyboard = bool(resize_keyboard)
self.one_time_keyboard = bool(one_time_keyboard)
self.selective = bool(selective)
self.resize_keyboard = resize_keyboard
self.one_time_keyboard = one_time_keyboard
self.selective = selective
self.input_field_placeholder = input_field_placeholder
self._id_attrs = (self.keyboard,)

4
telegram/_replykeyboardremove.py
View file

@ -57,8 +57,8 @@ class ReplyKeyboardRemove(TelegramObject):
__slots__ = ("selective", "remove_keyboard")
def __init__(self, selective: bool = False, **_kwargs: Any):
def __init__(self, selective: bool = None, **_kwargs: Any):
# Required
self.remove_keyboard = True
# Optionals
self.selective = bool(selective)
self.selective = selective

10
telegram/_user.py
View file

@ -21,7 +21,6 @@
from datetime import datetime
from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Union
from telegram import constants
from telegram._inline.inlinekeyboardbutton import InlineKeyboardButton
from telegram._menubutton import MenuButton
from telegram._telegramobject import TelegramObject
@ -166,7 +165,7 @@ class User(TelegramObject):
async def get_profile_photos(
self,
offset: int = None,
limit: int = 100,
limit: int = None,
read_timeout: ODVInput[float] = DEFAULT_NONE,
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,
@ -1179,10 +1178,9 @@ class User(TelegramObject):
self,
question: str,
options: List[str],
is_anonymous: bool = True,
# We use constant.PollType.REGULAR instead of Poll.REGULAR here to avoid circular imports
type: str = constants.PollType.REGULAR, # pylint: disable=redefined-builtin
allows_multiple_answers: bool = False,
is_anonymous: bool = None,
type: str = None,
allows_multiple_answers: bool = None,
correct_option_id: int = None,
is_closed: bool = None,
disable_notification: ODVInput[bool] = DEFAULT_NONE,

4
telegram/ext/_extbot.py
View file

@ -271,8 +271,8 @@ class ExtBot(Bot):
async def get_updates(
self,
offset: int = None,
limit: int = 100,
timeout: int = 0,
limit: int = None,
timeout: float = None,
read_timeout: float = 2,
write_timeout: ODVInput[float] = DEFAULT_NONE,
connect_timeout: ODVInput[float] = DEFAULT_NONE,

6
tests/test_bot.py
View file

@ -259,6 +259,12 @@ class TestBot:
# Second argument makes sure that we ignore logs from e.g. httpx
with caplog.at_level(logging.DEBUG, logger="telegram"):
await bot.get_me()
# Only for stabilizing this test-
if len(caplog.records) == 4:
for idx, record in enumerate(caplog.records):
print(record)
if record.getMessage().startswith("Task was destroyed but it is pending"):
caplog.records.pop(idx)
assert len(caplog.records) == 3
assert caplog.records[0].getMessage().startswith("Entering: get_me")
assert caplog.records[-1].getMessage().startswith("Exiting: get_me")

107
tests/test_official.py
View file

@ -25,6 +25,7 @@ import pytest
from bs4 import BeautifulSoup
import telegram
from telegram._utils.defaultvalue import DefaultValue
from tests.conftest import env_var_2_bool
IGNORED_OBJECTS = ("ResponseParameters", "CallbackGame")
@ -77,15 +78,23 @@ def check_method(h4):
sig = inspect.signature(method, follow_wrapped=True)
checked = []
for parameter in table:
param = sig.parameters.get(parameter[0])
assert param is not None, f"Parameter {parameter[0]} not found in {method.__name__}"
for tg_parameter in table: # Iterates through each row in the table
param = sig.parameters.get(
tg_parameter[0] # parameter[0] is first element (the param name)
)
assert param is not None, f"Parameter {tg_parameter[0]} not found in {method.__name__}"
# TODO: Check type via docstring
assert check_required_param(
parameter, param.name, sig, method.__name__
tg_parameter, param, method.__name__
), f"Param {param.name!r} of method {method.__name__!r} requirement mismatch!"
checked.append(parameter[0])
# Now we will check that we don't pass default values if the parameter is not required.
if param.default is not inspect.Parameter.empty: # If there is a default argument...
default_arg_none = check_defaults_type(param) # check if it's None
assert default_arg_none, f"Param {param.name!r} of {method.__name__!r} should be None"
checked.append(tg_parameter[0])
ignored = IGNORED_PARAMETERS.copy()
if name == "getUpdates":
@ -124,8 +133,8 @@ def check_object(h4):
sig = inspect.signature(obj.__init__, follow_wrapped=True)
checked = set()
for parameter in table:
field = parameter[0]
for tg_parameter in table:
field: str = tg_parameter[0] # From telegram docs
if field == "from":
field = "from_user"
elif (
@ -148,8 +157,13 @@ def check_object(h4):
assert param is not None, f"Attribute {field} not found in {obj.__name__}"
# TODO: Check type via docstring
assert check_required_param(
parameter, field, sig, obj.__name__
tg_parameter, param, obj.__name__
), f"{obj.__name__!r} parameter {param.name!r} requirement mismatch"
if param.default is not inspect.Parameter.empty: # If there is a default argument...
default_arg_none = check_defaults_type(param) # check if its None
assert default_arg_none, f"Param {param.name!r} of {obj.__name__!r} should be `None`"
checked.add(field)
ignored = IGNORED_PARAMETERS.copy()
@ -175,49 +189,62 @@ def check_object(h4):
assert (sig.parameters.keys() ^ checked) - ignored == set()
def is_parameter_required_by_tg(field: str) -> bool:
if field in {"Required", "Yes"}:
return True
if field.split(".", 1)[0] == "Optional": # splits the sentence and extracts first word
return False
else:
return True
def check_required_param(
param_desc: List[str], param_name: str, sig: inspect.Signature, method_or_obj_name: str
param_desc: List[str], param: inspect.Parameter, method_or_obj_name: str
) -> bool:
"""Checks if the method/class parameter is a required/optional param as per Telegram docs."""
if len(param_desc) == 4: # this means that there is a dedicated 'Required' column present.
# Handle cases where we provide convenience intentionally-
if param_name in ignored_param_requirements.get(method_or_obj_name, {}):
return True
is_required = True if param_desc[2] in {"Required", "Yes"} else False
is_ours_required = sig.parameters[param_name].default is inspect.Signature.empty
return is_required is is_ours_required
"""Checks if the method/class parameter is a required/optional param as per Telegram docs.
if len(param_desc) == 3: # The docs mention the requirement in the description for classes...
if param_name in ignored_param_requirements.get(method_or_obj_name, {}):
return True
is_required = False if param_desc[2].split(".", 1)[0] == "Optional" else True
is_ours_required = sig.parameters[param_name].default is inspect.Signature.empty
return is_required is is_ours_required
Returns:
:obj:`bool`: The boolean returned represents whether our parameter's requirement (optional
or required) is the same as Telegram's or not.
"""
is_ours_required = param.default is inspect.Parameter.empty
telegram_requires = is_parameter_required_by_tg(param_desc[2])
# Handle cases where we provide convenience intentionally-
if param.name in ignored_param_requirements.get(method_or_obj_name, {}):
return True
return telegram_requires is is_ours_required
def check_defaults_type(ptb_param: inspect.Parameter) -> bool:
return True if DefaultValue.get_value(ptb_param.default) is None else False
to_run = env_var_2_bool(os.getenv("TEST_OFFICIAL"))
argvalues = []
names = []
request = httpx.get("https://core.telegram.org/bots/api")
soup = BeautifulSoup(request.text, "html.parser")
for thing in soup.select("h4 > a.anchor"):
# Methods and types don't have spaces in them, luckily all other sections of the docs do
# TODO: don't depend on that
if "-" not in thing["name"]:
h4 = thing.parent
if to_run:
argvalues = []
names = []
request = httpx.get("https://core.telegram.org/bots/api")
soup = BeautifulSoup(request.text, "html.parser")
# Is it a method
if h4.text[0].lower() == h4.text[0]:
argvalues.append((check_method, h4))
names.append(h4.text)
elif h4.text not in IGNORED_OBJECTS: # Or a type/object
argvalues.append((check_object, h4))
names.append(h4.text)
for thing in soup.select("h4 > a.anchor"):
# Methods and types don't have spaces in them, luckily all other sections of the docs do
# TODO: don't depend on that
if "-" not in thing["name"]:
h4 = thing.parent
# Is it a method
if h4.text[0].lower() == h4.text[0]:
argvalues.append((check_method, h4))
names.append(h4.text)
elif h4.text not in IGNORED_OBJECTS: # Or a type/object
argvalues.append((check_object, h4))
names.append(h4.text)
@pytest.mark.skipif(not to_run, reason="test_official is not enabled")
@pytest.mark.parametrize(("method", "data"), argvalues=argvalues, ids=names)
@pytest.mark.skipif(
not env_var_2_bool(os.getenv("TEST_OFFICIAL")), reason="test_official is not enabled"
)
def test_official(method, data):
method(data)