Get defaults handling tests to run

This commit is contained in:
Hinrich Mahler 2024-12-30 13:57:17 +01:00
parent 04de3eff7a
commit ae639fbe87
6 changed files with 190 additions and 37 deletions

1
.gitignore vendored
View file

@ -67,6 +67,7 @@ docs/_build/
# PyBuilder
target/
.idea/
.run/
# Sublime Text 2
*.sublime*

View file

@ -466,7 +466,7 @@ class Poll(TelegramObject):
# Get the local timezone from the bot if it has defaults
loc_tzinfo = extract_tzinfo_from_defaults(bot)
data["options"] = [PollOption.de_json(option, bot) for option in data["options"]]
data["options"] = PollOption.de_list(data["options"], bot)
data["explanation_entities"] = de_list_wo(
data.get("explanation_entities"), MessageEntity, bot
)

View file

@ -22,7 +22,8 @@ import functools
import inspect
import re
from collections.abc import Collection, Iterable
from typing import Any, Callable, Optional
from types import GenericAlias
from typing import Any, Callable, ForwardRef, Optional, Union
import pytest
@ -30,7 +31,6 @@ import telegram # for ForwardRef resolution
from telegram import (
Bot,
ChatPermissions,
File,
InlineQueryResultArticle,
InlineQueryResultCachedPhoto,
InputMediaPhoto,
@ -44,6 +44,7 @@ from telegram._utils.defaultvalue import DEFAULT_NONE, DefaultValue
from telegram.constants import InputMediaType
from telegram.ext import Defaults, ExtBot
from telegram.request import RequestData
from tests.auxil.dummy_objects import get_dummy_object_json_dict
from tests.auxil.envvars import TEST_WITH_OPT_DEPS
if TEST_WITH_OPT_DEPS:
@ -261,10 +262,6 @@ async def check_shortcut_call(
f"{expected_args - received_kwargs}"
)
if bot_method_name == "get_file":
# This is here mainly for PassportFile.get_file, which calls .set_credentials on the
# return value
return File(file_id="result", file_unique_id="result")
return True
setattr(bot, bot_method_name, make_assertion)
@ -395,6 +392,34 @@ def make_assertion_for_link_preview_options(
)
def _check_forward_ref(obj: object) -> Union[str, object]:
if isinstance(obj, ForwardRef):
return obj.__forward_arg__
return obj
def guess_return_type_name(method: Callable[[...], Any]) -> tuple[Union[str, object], bool]:
# Using typing.get_type_hints(method) would be the nicer as it also resolves ForwardRefs
# and string annotations. But it also wants to resolve the parameter annotations, which
# need additional namespaces and that's not worth the struggle for now …
return_annotation = _check_forward_ref(inspect.signature(method).return_annotation)
print(return_annotation, type(return_annotation))
as_tuple = False
if isinstance(return_annotation, GenericAlias):
if return_annotation.__origin__ is tuple:
as_tuple = True
else:
raise ValueError(
f"Return type of {method.__name__} is a GenericAlias. This can not be handled yet."
)
# For tuples and Unions, we simply take the first element
if hasattr(return_annotation, "__args__"):
return _check_forward_ref(return_annotation.__args__[0]), as_tuple
return return_annotation, as_tuple
async def make_assertion(
url,
request_data: RequestData,
@ -539,15 +564,6 @@ async def make_assertion(
if default_value_expected and until_date != 946702800:
pytest.fail("Naive until_date should have been interpreted as America/New_York")
if method_name in ["get_file", "get_small_file", "get_big_file"]:
# This is here mainly for PassportFile.get_file, which calls .set_credentials on the
# return value
out = File(file_id="result", file_unique_id="result")
return out.to_dict()
# Otherwise return None by default, as TGObject.de_json/list(None) in [None, []]
# That way we can check what gets passed to Request.post without having to actually
# make a request
# Some methods expect specific output, so we allow to customize that
if isinstance(return_value, TelegramObject):
return return_value.to_dict()
return return_value
@ -556,7 +572,6 @@ async def make_assertion(
async def check_defaults_handling(
method: Callable,
bot: Bot,
return_value=None,
no_default_kwargs: Collection[str] = frozenset(),
) -> bool:
"""
@ -566,13 +581,12 @@ async def check_defaults_handling(
method: The shortcut/bot_method
bot: The bot. May be a telegram.Bot or a telegram.ext.ExtBot. In the former case, all
default values will be converted to None.
return_value: Optional. The return value of Bot._post that the method expects. Defaults to
None. get_file is automatically handled. If this is a `TelegramObject`, Bot._post will
return the `to_dict` representation of it.
no_default_kwargs: Optional. A collection of keyword arguments that should not have default
values. Defaults to an empty frozenset.
"""
guess_return_type_name(method)
raw_bot = not isinstance(bot, ExtBot)
get_updates = method.__name__.lower().replace("_", "") == "getupdates"
@ -604,12 +618,10 @@ async def check_defaults_handling(
)
defaults_custom_defaults = Defaults(**kwargs)
expected_return_values = [None, ()] if return_value is None else [return_value]
if method.__name__ in ["get_file", "get_small_file", "get_big_file"]:
expected_return_values = [File(file_id="result", file_unique_id="result")]
request = bot._request[0] if get_updates else bot.request
orig_post = request.post
return_value = get_dummy_object_json_dict(*guess_return_type_name(method))
try:
if raw_bot:
combinations = [(None, None)]
@ -633,7 +645,7 @@ async def check_defaults_handling(
expected_defaults_value=expected_defaults_value,
)
request.post = assertion_callback
assert await method(**kwargs) in expected_return_values
await method(**kwargs)
# 2: test that we get the manually passed non-None value
kwargs = build_kwargs(
@ -648,7 +660,7 @@ async def check_defaults_handling(
expected_defaults_value=expected_defaults_value,
)
request.post = assertion_callback
assert await method(**kwargs) in expected_return_values
await method(**kwargs)
# 3: test that we get the manually passed None value
kwargs = build_kwargs(
@ -663,7 +675,7 @@ async def check_defaults_handling(
expected_defaults_value=expected_defaults_value,
)
request.post = assertion_callback
assert await method(**kwargs) in expected_return_values
await method(**kwargs)
except Exception as exc:
raise exc
finally:

View file

@ -0,0 +1,145 @@
import datetime as dtm
from collections.abc import Sequence
from typing import Union
from telegram import (
BotCommand,
BotDescription,
BotName,
BotShortDescription,
BusinessConnection,
ChatAdministratorRights,
ChatFullInfo,
ChatInviteLink,
ChatMember,
File,
ForumTopic,
GameHighScore,
Gifts,
MenuButton,
MessageId,
Poll,
PollOption,
PreparedInlineMessage,
SentWebAppMessage,
StarTransactions,
Sticker,
StickerSet,
TelegramObject,
Update,
User,
UserChatBoosts,
UserProfilePhotos,
WebhookInfo,
)
from tests.auxil.build_messages import make_message
_DUMMY_USER = User(
id=123456, is_bot=False, first_name="Dummy", last_name="User", username="dummy_user"
)
_DUMMY_DATE = dtm.datetime(1970, 1, 1, 0, 0, 0, 0, tzinfo=dtm.timezone.utc)
_PREPARED_DUMMY_OBJECTS: dict[str, object] = {
"bool": True,
"BotCommand": BotCommand(command="dummy_command", description="dummy_description"),
"BotDescription": BotDescription(description="dummy_description"),
"BotName": BotName(name="dummy_name"),
"BotShortDescription": BotShortDescription(short_description="dummy_short_description"),
"BusinessConnection": BusinessConnection(
user=_DUMMY_USER,
id="123",
user_chat_id=123456,
date=_DUMMY_DATE,
can_reply=True,
is_enabled=True,
),
"ChatAdministratorRights": ChatAdministratorRights.all_rights(),
"ChatFullInfo": ChatFullInfo(
id=123456,
type="dummy_type",
accent_color_id=1,
max_reaction_count=1,
),
"ChatInviteLink": ChatInviteLink(
"dummy_invite_link",
creator=_DUMMY_USER,
is_primary=True,
is_revoked=False,
creates_join_request=False,
),
"ChatMember": ChatMember(user=_DUMMY_USER, status="dummy_status"),
"File": File(file_id="dummy_file_id", file_unique_id="dummy_file_unique_id"),
"ForumTopic": ForumTopic(message_thread_id=2, name="dummy_name", icon_color=1),
"Gifts": Gifts(gifts=[]),
"GameHighScore": GameHighScore(position=1, user=_DUMMY_USER, score=1),
"int": 123456,
"MenuButton": MenuButton(type="dummy_type"),
"Message": make_message("dummy_text"),
"MessageId": MessageId(123456),
"Poll": Poll(
id="dummy_id",
question="dummy_question",
options=[PollOption(text="dummy_text", voter_count=1)],
is_closed=False,
is_anonymous=False,
total_voter_count=1,
type="dummy_type",
allows_multiple_answers=False,
),
"PreparedInlineMessage": PreparedInlineMessage(id="dummy_id", expiration_date=_DUMMY_DATE),
"SentWebAppMessage": SentWebAppMessage(inline_message_id="dummy_inline_message_id"),
"StarTransactions": StarTransactions(transactions=[]),
"Sticker": Sticker(
file_id="dummy_file_id",
file_unique_id="dummy_file_unique_id",
width=1,
height=1,
is_animated=False,
is_video=False,
type="dummy_type",
),
"StickerSet": StickerSet(
name="dummy_name", title="dummy_title", stickers=[], sticker_type="dummy_type"
),
"str": "dummy_string",
"Update": Update(update_id=123456),
"User": _DUMMY_USER,
"UserChatBoosts": UserChatBoosts(boosts=[]),
"UserProfilePhotos": UserProfilePhotos(total_count=1, photos=[[]]),
"WebhookInfo": WebhookInfo(
url="dummy_url",
has_custom_certificate=False,
pending_update_count=1,
),
}
def get_dummy_object(obj_type: Union[type, str], as_tuple: bool = False) -> object:
obj_type_name = obj_type.__name__ if isinstance(obj_type, type) else obj_type
if (return_value := _PREPARED_DUMMY_OBJECTS.get(obj_type_name)) is None:
raise ValueError(
f"Dummy object of type '{obj_type_name}' not found. Please add it manually."
)
if as_tuple:
return (return_value,)
return return_value
_RETURN_TYPES = Union[bool, int, str, dict[str, object]]
_RETURN_TYPE = Union[_RETURN_TYPES, tuple[_RETURN_TYPES, ...]]
def _serialize_dummy_object(obj: object) -> _RETURN_TYPE:
if isinstance(obj, Sequence) and not isinstance(obj, str):
return tuple(_serialize_dummy_object(item) for item in obj)
if isinstance(obj, (str, int, bool)):
return obj
if isinstance(obj, TelegramObject):
return obj.to_dict()
raise ValueError(f"Serialization of object of type '{type(obj)}' is not supported yet.")
def get_dummy_object_json_dict(obj_type: Union[type, str], as_tuple: bool = False) -> _RETURN_TYPE:
return _serialize_dummy_object(get_dummy_object(obj_type, as_tuple=as_tuple))

View file

@ -66,7 +66,7 @@ class PytestExtBot(ExtBot):
self._unfreeze()
# Here we override get_me for caching because we don't want to call the API repeatedly in tests
async def get_me(self, *args, **kwargs):
async def get_me(self, *args, **kwargs) -> User:
return await _mocked_get_me(self)
@ -77,7 +77,7 @@ class PytestBot(Bot):
self._unfreeze()
# Here we override get_me for caching because we don't want to call the API repeatedly in tests
async def get_me(self, *args, **kwargs):
async def get_me(self, *args, **kwargs) -> User:
return await _mocked_get_me(self)

View file

@ -488,17 +488,11 @@ class TestBotWithoutRequest:
Finally, there are some tests for Defaults.{parse_mode, quote, allow_sending_without_reply}
at the appropriate places, as those are the only things we can actually check.
"""
# Mocking get_me within check_defaults_handling messes with the cached values like
# Bot.{bot, username, id, …}` unless we return the expected User object.
return_value = (
offline_bot.bot if bot_method_name.lower().replace("_", "") == "getme" else None
)
# Check that ExtBot does the right thing
bot_method = getattr(offline_bot, bot_method_name)
raw_bot_method = getattr(raw_bot, bot_method_name)
assert await check_defaults_handling(bot_method, offline_bot, return_value=return_value)
assert await check_defaults_handling(raw_bot_method, raw_bot, return_value=return_value)
assert await check_defaults_handling(bot_method, offline_bot)
assert await check_defaults_handling(raw_bot_method, raw_bot)
@pytest.mark.parametrize(
("name", "method"), inspect.getmembers(Bot, predicate=inspect.isfunction)
@ -2248,6 +2242,7 @@ class TestBotWithoutRequest:
is_closed=True,
is_anonymous=True,
type="regular",
allows_multiple_answers=False,
).to_dict()
)
await return_values.put(True)