diff --git a/.github/workflows/test_official.yml b/.github/workflows/test_official.yml index 430f63561..9b9726946 100644 --- a/.github/workflows/test_official.yml +++ b/.github/workflows/test_official.yml @@ -33,7 +33,7 @@ jobs: python -W ignore -m pip install -r requirements-dev.txt - name: Compare to official api run: | - pytest -v tests/test_official.py --junit-xml=.test_report_official.xml + pytest -v tests/test_official/test_official.py --junit-xml=.test_report_official.xml exit $? env: TEST_OFFICIAL: "true" diff --git a/pyproject.toml b/pyproject.toml index d796a6a9a..233af8fec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,6 +64,8 @@ markers = [ "req", ] asyncio_mode = "auto" +log_format = "%(funcName)s - Line %(lineno)d - %(message)s" +# log_level = "DEBUG" # uncomment to see DEBUG logs # MYPY: [tool.mypy] diff --git a/telegram/_bot.py b/telegram/_bot.py index 280e47d12..48165bd9a 100644 --- a/telegram/_bot.py +++ b/telegram/_bot.py @@ -6138,8 +6138,8 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified. async def upload_sticker_file( self, user_id: int, - sticker: Optional[FileInput], - sticker_format: Optional[str], + sticker: FileInput, + sticker_format: str, *, read_timeout: ODVInput[float] = DEFAULT_NONE, write_timeout: ODVInput[float] = DEFAULT_NONE, @@ -6180,7 +6180,7 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified. """ data: JSONDict = { "user_id": user_id, - "sticker": self._parse_file_input(sticker), # type: ignore[arg-type] + "sticker": self._parse_file_input(sticker), "sticker_format": sticker_format, } result = await self._post( @@ -6199,7 +6199,7 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified. self, user_id: int, name: str, - sticker: Optional["InputSticker"], + sticker: "InputSticker", *, read_timeout: ODVInput[float] = DEFAULT_NONE, write_timeout: ODVInput[float] = DEFAULT_NONE, @@ -6298,8 +6298,8 @@ CUSTOM_EMOJI_IDENTIFIER_LIMIT` custom emoji identifiers can be specified. user_id: int, name: str, title: str, - stickers: Optional[Sequence["InputSticker"]], - sticker_format: Optional[str], + stickers: Sequence["InputSticker"], + sticker_format: str, sticker_type: Optional[str] = None, needs_repainting: Optional[bool] = None, *, diff --git a/telegram/ext/_extbot.py b/telegram/ext/_extbot.py index a92ea2fab..1b6e4fcdc 100644 --- a/telegram/ext/_extbot.py +++ b/telegram/ext/_extbot.py @@ -862,7 +862,7 @@ class ExtBot(Bot, Generic[RLARGS]): self, user_id: int, name: str, - sticker: Optional["InputSticker"], + sticker: "InputSticker", *, read_timeout: ODVInput[float] = DEFAULT_NONE, write_timeout: ODVInput[float] = DEFAULT_NONE, @@ -1177,8 +1177,8 @@ class ExtBot(Bot, Generic[RLARGS]): user_id: int, name: str, title: str, - stickers: Optional[Sequence["InputSticker"]], - sticker_format: Optional[str], + stickers: Sequence["InputSticker"], + sticker_format: str, sticker_type: Optional[str] = None, needs_repainting: Optional[bool] = None, *, @@ -3673,8 +3673,8 @@ class ExtBot(Bot, Generic[RLARGS]): async def upload_sticker_file( self, user_id: int, - sticker: Optional[FileInput], - sticker_format: Optional[str], + sticker: FileInput, + sticker_format: str, *, read_timeout: ODVInput[float] = DEFAULT_NONE, write_timeout: ODVInput[float] = DEFAULT_NONE, diff --git a/tests/README.rst b/tests/README.rst index d15d745a2..753dd6a16 100644 --- a/tests/README.rst +++ b/tests/README.rst @@ -82,6 +82,17 @@ Use as follows: $ pytest -m dev +Debugging tests +=============== + +Writing tests can be challenging, and fixing failing tests can be even more so. To help with this, +PTB has started to adopt the use of ``logging`` in the test suite. You can insert debug logging +statements in your tests to help you understand what's going on. To enable these logs, you can set +``log_level = DEBUG`` in ``setup.cfg`` or use the ``--log-level=INFO`` flag when running the tests. +If a test is large and complicated, it is recommended to leave the debug logs for others to use as +well. + + Bots used in tests ================== diff --git a/tests/conftest.py b/tests/conftest.py index ce940bdeb..689d816bb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -57,7 +57,7 @@ collect_ignore = [] if sys.version_info < (3, 10): if RUN_TEST_OFFICIAL: logging.warning("Skipping test_official.py since it requires Python 3.10+") - collect_ignore.append("test_official.py") + collect_ignore_glob = ["test_official/*.py"] # This is here instead of in setup.cfg due to https://github.com/pytest-dev/pytest/issues/8343 diff --git a/tests/test_official.py b/tests/test_official.py deleted file mode 100644 index 8aaca8a69..000000000 --- a/tests/test_official.py +++ /dev/null @@ -1,597 +0,0 @@ -#!/usr/bin/env python -# -# A library that provides a Python interface to the Telegram Bot API -# Copyright (C) 2015-2023 -# Leandro Toledo de Souza -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Lesser Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Lesser Public License for more details. -# -# You should have received a copy of the GNU Lesser Public License -# along with this program. If not, see [http://www.gnu.org/licenses/]. -import inspect -import re -from datetime import datetime -from types import FunctionType -from typing import Any, Callable, ForwardRef, Sequence, get_args, get_origin - -import httpx -import pytest -from bs4 import BeautifulSoup, PageElement, Tag - -import telegram -from telegram._utils.defaultvalue import DefaultValue -from telegram._utils.types import FileInput, ODVInput -from telegram.ext import Defaults -from tests.auxil.envvars import RUN_TEST_OFFICIAL - -IGNORED_OBJECTS = ("ResponseParameters", "CallbackGame") -GLOBALLY_IGNORED_PARAMETERS = { - "self", - "read_timeout", - "write_timeout", - "connect_timeout", - "pool_timeout", - "bot", - "api_kwargs", -} - -# Types for certain parameters accepted by PTB but not in the official API -ADDITIONAL_TYPES = { - "photo": ForwardRef("PhotoSize"), - "video": ForwardRef("Video"), - "video_note": ForwardRef("VideoNote"), - "audio": ForwardRef("Audio"), - "document": ForwardRef("Document"), - "animation": ForwardRef("Animation"), - "voice": ForwardRef("Voice"), - "sticker": ForwardRef("Sticker"), -} - -# Exceptions to the "Array of" types, where we accept more types than the official API -# key: parameter name, value: type which must be present in the annotation -ARRAY_OF_EXCEPTIONS = { - "results": "InlineQueryResult", # + Callable - "commands": "BotCommand", # + tuple[str, str] - "keyboard": "KeyboardButton", # + sequence[sequence[str]] - "reaction": "ReactionType", # + str - # TODO: Deprecated and will be corrected (and removed) in next major PTB version: - "file_hashes": "list[str]", -} - -# Special cases for other parameters that accept more types than the official API, and are -# too complex to compare/predict with official API: -COMPLEX_TYPES = { # (param_name, is_class (i.e appears in a class?)): reduced form of annotation - ("correct_option_id", False): int, # actual: Literal - ("file_id", False): str, # actual: Union[str, objs_with_file_id_attr] - ("invite_link", False): str, # actual: Union[str, ChatInviteLink] - ("provider_data", False): str, # actual: Union[str, obj] - ("callback_data", True): str, # actual: Union[str, obj] - ("media", True): str, # actual: Union[str, InputMedia*, FileInput] - ("data", True): str, # actual: Union[IdDocumentData, PersonalDetails, ResidentialAddress] -} - -# These are param names ignored in the param type checking in classes for the `tg.Defaults` case. -IGNORED_DEFAULTS_PARAM_NAMES = { - "quote", - "link_preview_options", -} - -# These classes' params are all ODVInput, so we ignore them in the defaults type checking. -IGNORED_DEFAULTS_CLASSES = {"LinkPreviewOptions"} - - -def _get_params_base(object_name: str, search_dict: dict[str, set[Any]]) -> set[Any]: - """Helper function for the *_params functions below. - Given an object name and a search dict, goes through the keys of the search dict and checks if - the object name matches any of the regexes (keys). The union of all the sets (values) of the - matching regexes is returned. `object_name` may be a CamelCase or snake_case name. - """ - out = set() - for regex, params in search_dict.items(): - if re.fullmatch(regex, object_name): - out.update(params) - # also check the snake_case version - snake_case_name = re.sub(r"(? set[str]: - return _get_params_base(object_name, PTB_EXTRA_PARAMS) - - -# Arguments *removed* from the official API -# Mostly due to the value being fixed anyway -PTB_IGNORED_PARAMS = { - r"InlineQueryResult\w+": {"type"}, - r"ChatMember\w+": {"status"}, - r"PassportElementError\w+": {"source"}, - "ForceReply": {"force_reply"}, - "ReplyKeyboardRemove": {"remove_keyboard"}, - r"BotCommandScope\w+": {"type"}, - r"MenuButton\w+": {"type"}, - r"InputMedia\w+": {"type"}, - "InaccessibleMessage": {"date"}, - r"MessageOrigin\w+": {"type"}, - r"ChatBoostSource\w+": {"source"}, - r"ReactionType\w+": {"type"}, -} - - -def ptb_ignored_params(object_name: str) -> set[str]: - return _get_params_base(object_name, PTB_IGNORED_PARAMS) - - -IGNORED_PARAM_REQUIREMENTS = { - # Ignore these since there's convenience params in them (eg. Venue) - # <---- - "send_location": {"latitude", "longitude"}, - "edit_message_live_location": {"latitude", "longitude"}, - "send_venue": {"latitude", "longitude", "title", "address"}, - "send_contact": {"phone_number", "first_name"}, - # ----> -} - - -def ignored_param_requirements(object_name: str) -> set[str]: - return _get_params_base(object_name, IGNORED_PARAM_REQUIREMENTS) - - -# Arguments that are optional arguments for now for backwards compatibility -BACKWARDS_COMPAT_KWARGS: dict[str, set[str]] = { - # Deprecated by Bot API 7.0, kept for now for bw compat: - "KeyboardButton": {"request_user"}, - "Message": { - "forward_from", - "forward_signature", - "forward_sender_name", - "forward_date", - "forward_from_chat", - "forward_from_message_id", - "user_shared", - }, - "(send_message|edit_message_text)": { - "disable_web_page_preview", - "reply_to_message_id", - "allow_sending_without_reply", - }, - r"copy_message|send_\w+": {"allow_sending_without_reply", "reply_to_message_id"}, -} - - -def backwards_compat_kwargs(object_name: str) -> set[str]: - return _get_params_base(object_name, BACKWARDS_COMPAT_KWARGS) - - -IGNORED_PARAM_REQUIREMENTS.update(BACKWARDS_COMPAT_KWARGS) - - -def find_next_sibling_until(tag: Tag, name: str, until: Tag) -> PageElement | None: - for sibling in tag.next_siblings: - if sibling is until: - return None - if sibling.name == name: - return sibling - return None - - -def parse_table(h4: Tag) -> list[list[str]]: - """Parses the Telegram doc table and has an output of a 2D list.""" - table = find_next_sibling_until(h4, "table", h4.find_next_sibling("h4")) - if not table: - return [] - return [[td.text for td in tr.find_all("td")] for tr in table.find_all("tr")[1:]] - - -def check_method(h4: Tag) -> None: - name = h4.text # name of the method in telegram's docs. - method: FunctionType | None = getattr(telegram.Bot, name, None) # Retrieve our lib method - if not method: - raise AssertionError(f"Method {name} not found in telegram.Bot") - - table = parse_table(h4) - - # Check arguments based on source - sig = inspect.signature(method, follow_wrapped=True) - checked = [] - for tg_parameter in table: # Iterates through each row in the table - # Check if parameter is present in our method - param = sig.parameters.get( - tg_parameter[0] # parameter[0] is first element (the param name) - ) - if param is None: - raise AssertionError(f"Parameter {tg_parameter[0]} not found in {method.__name__}") - - # Check if type annotation is present and correct - if param.annotation is inspect.Parameter.empty: - raise AssertionError( - f"Param {param.name!r} of {method.__name__!r} should have a type annotation" - ) - if not check_param_type(param, tg_parameter, method): - raise AssertionError( - f"Param {param.name!r} of {method.__name__!r} should be {tg_parameter[1]} or " - "something else!" - ) - - # Now check if the parameter is required or not - if not check_required_param(tg_parameter, param, method.__name__): - raise AssertionError( - f"Param {param.name!r} of method {method.__name__!r} requirement mismatch!" - ) - - # 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 - if not default_arg_none: - raise AssertionError(f"Param {param.name!r} of {method.__name__!r} should be None") - checked.append(tg_parameter[0]) - - expected_additional_args = GLOBALLY_IGNORED_PARAMETERS.copy() - expected_additional_args |= ptb_extra_params(name) - expected_additional_args |= backwards_compat_kwargs(name) - - unexpected_args = (sig.parameters.keys() ^ checked) - expected_additional_args - if unexpected_args != set(): - raise AssertionError( - f"In {method.__qualname__}, unexpected args were found: {unexpected_args}." - ) - - kw_or_positional_args = [ - p.name for p in sig.parameters.values() if p.kind != inspect.Parameter.KEYWORD_ONLY - ] - non_kw_only_args = set(kw_or_positional_args).difference(checked).difference(["self"]) - non_kw_only_args -= backwards_compat_kwargs(name) - if non_kw_only_args != set(): - raise AssertionError( - f"In {method.__qualname__}, extra args should be keyword only " - f"(compared to {name} in API)" - ) - - -def check_object(h4: Tag) -> None: - name = h4.text - obj = getattr(telegram, name) - table = parse_table(h4) - - # Check arguments based on source. Makes sure to only check __init__'s signature & nothing else - sig = inspect.signature(obj.__init__, follow_wrapped=True) - - checked = set() - fields_removed_by_ptb = ptb_ignored_params(name) - for tg_parameter in table: - field: str = tg_parameter[0] # From telegram docs - - if field in fields_removed_by_ptb: - continue - - if field == "from": - field = "from_user" - - param = sig.parameters.get(field) - if param is None: - raise AssertionError(f"Attribute {field} not found in {obj.__name__}") - # Check if type annotation is present and correct - if param.annotation is inspect.Parameter.empty: - raise AssertionError( - f"Param {param.name!r} of {obj.__name__!r} should have a type annotation" - ) - if not check_param_type(param, tg_parameter, obj): - raise AssertionError( - f"Param {param.name!r} of {obj.__name__!r} should be {tg_parameter[1]} or " - "something else!" - ) - if not check_required_param(tg_parameter, param, obj.__name__): - raise AssertionError(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 - if not default_arg_none: - raise AssertionError(f"Param {param.name!r} of {obj.__name__!r} should be `None`") - - checked.add(field) - - expected_additional_args = GLOBALLY_IGNORED_PARAMETERS.copy() - expected_additional_args |= ptb_extra_params(name) - expected_additional_args |= backwards_compat_kwargs(name) - - unexpected_args = (sig.parameters.keys() ^ checked) - expected_additional_args - if unexpected_args != set(): - raise AssertionError(f"In {name}, unexpected args were found: {unexpected_args}.") - - -def is_parameter_required_by_tg(field: str) -> bool: - if field in {"Required", "Yes"}: - return True - return field.split(".", 1)[0] != "Optional" # splits the sentence and extracts first word - - -def check_required_param( - 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. - - 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(method_or_obj_name): - return True - return telegram_requires is is_ours_required - - -def check_defaults_type(ptb_param: inspect.Parameter) -> bool: - return DefaultValue.get_value(ptb_param.default) is None - - -def check_param_type( - ptb_param: inspect.Parameter, tg_parameter: list[str], obj: FunctionType | type -) -> bool: - """This function checks whether the type annotation of the parameter is the same as the one - specified in the official API. It also checks for some special cases where we accept more types - - Args: - ptb_param (inspect.Parameter): The parameter object from our methods/classes - tg_parameter (list[str]): The table row corresponding to the parameter from official API. - obj (object): The object (method/class) that we are checking. - - Returns: - :obj:`bool`: The boolean returned represents whether our parameter's type annotation is the - same as Telegram's or not. - """ - # PRE-PROCESSING: - # In order to evaluate the type annotation, we need to first have a mapping of the types - # specified in the official API to our types. The keys are types in the column of official API. - TYPE_MAPPING: dict[str, set[Any]] = { - "Integer or String": {int | str}, - "Integer": {int}, - "String": {str}, - r"Boolean|True": {bool}, - r"Float(?: number)?": {float}, - # Distinguishing 1D and 2D Sequences and finding the inner type is done later. - r"Array of (?:Array of )?[\w\,\s]*": {Sequence}, - r"InputFile(?: or String)?": {FileInput}, - } - - tg_param_type: str = tg_parameter[1] # Type of parameter as specified in the docs - is_class = inspect.isclass(obj) - # Let's check for a match: - mapped: set[type] = _get_params_base(tg_param_type, TYPE_MAPPING) - - # We should have a maximum of one match. - assert len(mapped) <= 1, f"More than one match found for {tg_param_type}" - - if not mapped: # no match found, it's from telegram module - # it could be a list of objects, so let's check that: - objs = _extract_words(tg_param_type) - # We want to store both string version of class and the class obj itself. e.g. "InputMedia" - # and InputMedia because some annotations might be ForwardRefs. - if len(objs) >= 2: # We have to unionize the objects - mapped_type: tuple[Any, ...] = (_unionizer(objs, False), _unionizer(objs, True)) - else: - mapped_type = ( - getattr(telegram, tg_param_type), # This will fail if it's not from telegram mod - ForwardRef(tg_param_type), - tg_param_type, # for some reason, some annotations are just a string. - ) - elif len(mapped) == 1: - mapped_type = mapped.pop() - - # Resolve nested annotations to get inner types. - if (ptb_annotation := list(get_args(ptb_param.annotation))) == []: - ptb_annotation = ptb_param.annotation # if it's not nested, just use the annotation - - if isinstance(ptb_annotation, list): - # Some cleaning: - # Remove 'Optional[...]' from the annotation if it's present. We do it this way since: 1) - # we already check if argument should be optional or not + type checkers will complain. - # 2) we want to check if our `obj` is same as API's `obj`, and since python evaluates - # `Optional[obj] != obj` we have to remove the Optional, so that we can compare the two. - if type(None) in ptb_annotation: - ptb_annotation.remove(type(None)) - - # Cleaning done... now let's put it back together. - # Join all the annotations back (i.e. Union) - ptb_annotation = _unionizer(ptb_annotation, False) - - # Last step, we need to use get_origin to get the original type, since using get_args - # above will strip that out. - wrapped = get_origin(ptb_param.annotation) - if wrapped is not None: - # collections.abc.Sequence -> typing.Sequence - if "collections.abc.Sequence" in str(wrapped): - wrapped = Sequence - ptb_annotation = wrapped[ptb_annotation] - # We have put back our annotation together after removing the NoneType! - - # CHECKING: - # Each branch may have exits in the form of return statements. If the annotation is found to be - # correct, the function will return True. If not, it will return False. - - # 1) HANDLING ARRAY TYPES: - # Now let's do the checking, starting with "Array of ..." types. - if "Array of " in tg_param_type: - assert mapped_type is Sequence - # For exceptions just check if they contain the annotation - if ptb_param.name in ARRAY_OF_EXCEPTIONS: - return ARRAY_OF_EXCEPTIONS[ptb_param.name] in str(ptb_annotation) - - pattern = r"Array of(?: Array of)? ([\w\,\s]*)" - obj_match: re.Match | None = re.search(pattern, tg_param_type) # extract obj from string - if obj_match is None: - raise AssertionError(f"Array of {tg_param_type} not found in {ptb_param.name}") - obj_str: str = obj_match.group(1) - # is obj a regular type like str? - array_of_mapped: set[type] = _get_params_base(obj_str, TYPE_MAPPING) - - if len(array_of_mapped) == 0: # no match found, it's from telegram module - # it could be a list of objects, so let's check that: - objs = _extract_words(obj_str) - # let's unionize all the objects, with and without ForwardRefs. - unionized_objs: list[type] = [_unionizer(objs, True), _unionizer(objs, False)] - else: - unionized_objs = [array_of_mapped.pop()] - - # This means it is Array of Array of [obj] - if "Array of Array of" in tg_param_type: - return any(Sequence[Sequence[o]] == ptb_annotation for o in unionized_objs) - - # This means it is Array of [obj] - return any(mapped_type[o] == ptb_annotation for o in unionized_objs) - - # 2) HANDLING DEFAULTS PARAMETERS: - # Classes whose parameters are all ODVInput should be converted and checked. - if obj.__name__ in IGNORED_DEFAULTS_CLASSES: - parsed = ODVInput[mapped_type] - return (ptb_annotation | None) == parsed # We have to add back None in our annotation - if not ( - # Defaults checking should not be done for: - # 1. Parameters that have name conflict with `Defaults.name` - is_class - and obj.__name__ in ("ReplyParameters", "Message", "ExternalReplyInfo") - and ptb_param.name in IGNORED_DEFAULTS_PARAM_NAMES - ): - # Now let's check if the parameter is a Defaults parameter, it should be - for name, _ in inspect.getmembers(Defaults, lambda x: isinstance(x, property)): - if name == ptb_param.name or "parse_mode" in ptb_param.name: - # mapped_type should not be a tuple since we need to check for equality: - # This can happen when the Defaults parameter is a class, e.g. LinkPreviewOptions - if isinstance(mapped_type, tuple): - mapped_type = mapped_type[1] # We select the ForwardRef - # Assert if it's ODVInput by checking equality: - parsed = ODVInput[mapped_type] - if (ptb_annotation | None) == parsed: # We have to add back None in our annotation - return True - return False - - # 3) HANDLING OTHER TYPES: - # Special case for send_* methods where we accept more types than the official API: - if ( - ptb_param.name in ADDITIONAL_TYPES - and not isinstance(mapped_type, tuple) - and obj.__name__.startswith("send") - ): - mapped_type = mapped_type | ADDITIONAL_TYPES[ptb_param.name] - - # 4) HANDLING DATETIMES: - if ( - re.search( - r"""([_]+|\b) # check for word boundary or underscore - date # check for "date" - [^\w]*\b # optionally check for a word after 'date' - """, - ptb_param.name, - re.VERBOSE, - ) - or "Unix time" in tg_parameter[-1] - ): - # TODO: Remove this in v22 when it becomes a datetime - datetime_exceptions = { - "file_date", - } - if ptb_param.name in datetime_exceptions: - return True - # If it's a class, we only accept datetime as the parameter - mapped_type = datetime if is_class else mapped_type | datetime - - # RESULTS: ALL OTHER BASIC TYPES- - # Some types are too complicated, so we replace them with a simpler type: - for (param_name, expected_class), exception_type in COMPLEX_TYPES.items(): - if ptb_param.name == param_name and is_class is expected_class: - ptb_annotation = exception_type - - # Final check, if the annotation is a tuple, we need to check if any of the types in the tuple - # match the mapped type. - if isinstance(mapped_type, tuple) and any(ptb_annotation == t for t in mapped_type): - return True - - # If the annotation is not a tuple, we can just check if it's equal to the mapped type. - return mapped_type == ptb_annotation - - -def _extract_words(text: str) -> set[str]: - """Extracts all words from a string, removing all punctuation and words like 'and' & 'or'.""" - return set(re.sub(r"[^\w\s]", "", text).split()) - {"and", "or"} - - -def _unionizer(annotation: Sequence[Any] | set[Any], forward_ref: bool) -> Any: - """Returns a union of all the types in the annotation. If forward_ref is True, it wraps the - annotation in a ForwardRef and then unionizes.""" - union = None - for t in annotation: - if forward_ref: - t = ForwardRef(t) # noqa: PLW2901 - elif not forward_ref and isinstance(t, str): # we have to import objects from lib - t = getattr(telegram, t) # noqa: PLW2901 - union = t if union is None else union | t - return union - - -argvalues: list[tuple[Callable[[Tag], None], Tag]] = [] -names: list[str] = [] - -if RUN_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: Tag | None = thing.parent - - if h4 is None: - raise AssertionError("h4 is None") - # 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 RUN_TEST_OFFICIAL, reason="test_official is not enabled") -@pytest.mark.parametrize(("method", "data"), argvalues=argvalues, ids=names) -def test_official(method, data): - method(data) diff --git a/tests/test_official/__init__.py b/tests/test_official/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_official/arg_type_checker.py b/tests/test_official/arg_type_checker.py new file mode 100644 index 000000000..2ccd7808c --- /dev/null +++ b/tests/test_official/arg_type_checker.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2024 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +"""This module contains functions which confirm that the parameters of our methods and classes +match the official API. It also checks if the type annotations are correct and if the parameters +are required or not.""" + +import inspect +import logging +import re +from datetime import datetime +from types import FunctionType +from typing import Any, Sequence + +from telegram._utils.defaultvalue import DefaultValue +from telegram._utils.types import FileInput, ODVInput +from telegram.ext import Defaults +from tests.test_official.exceptions import ParamTypeCheckingExceptions as PTCE +from tests.test_official.exceptions import ignored_param_requirements +from tests.test_official.helpers import ( + _extract_words, + _get_params_base, + _unionizer, + cached_type_hints, + resolve_forward_refs_in_type, + wrap_with_none, +) +from tests.test_official.scraper import TelegramParameter + +ARRAY_OF_PATTERN = r"Array of(?: Array of)? ([\w\,\s]*)" + +# In order to evaluate the type annotation, we need to first have a mapping of the types +# specified in the official API to our types. The keys are types in the column of official API. +TYPE_MAPPING: dict[str, set[Any]] = { + "Integer or String": {int | str}, + "Integer": {int}, + "String": {str}, + r"Boolean|True": {bool}, + r"Float(?: number)?": {float}, + # Distinguishing 1D and 2D Sequences and finding the inner type is done later. + ARRAY_OF_PATTERN: {Sequence}, + r"InputFile(?: or String)?": {resolve_forward_refs_in_type(FileInput)}, +} + +ALL_DEFAULTS = inspect.getmembers(Defaults, lambda x: isinstance(x, property)) + +DATETIME_REGEX = re.compile( + r"""([_]+|\b) # check for word boundary or underscore + date # check for "date" + [^\w]*\b # optionally check for a word after 'date' + """, + re.VERBOSE, +) + +log = logging.debug + + +def check_required_param( + tg_param: TelegramParameter, param: inspect.Parameter, method_or_obj_name: str +) -> bool: + """Checks if the method/class parameter is a required/optional param as per Telegram docs. + + 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 + # Handle cases where we provide convenience intentionally- + if param.name in ignored_param_requirements(method_or_obj_name): + return True + return tg_param.param_required is is_ours_required + + +def check_defaults_type(ptb_param: inspect.Parameter) -> bool: + return DefaultValue.get_value(ptb_param.default) is None + + +def check_param_type( + ptb_param: inspect.Parameter, + tg_parameter: TelegramParameter, + obj: FunctionType | type, +) -> tuple[bool, type]: + """This function checks whether the type annotation of the parameter is the same as the one + specified in the official API. It also checks for some special cases where we accept more types + + Args: + ptb_param: The parameter object from our methods/classes + tg_parameter: The table row corresponding to the parameter from official API. + obj: The object (method/class) that we are checking. + + Returns: + :obj:`tuple`: A tuple containing: + * :obj:`bool`: The boolean returned represents whether our parameter's type annotation + is the same as Telegram's or not. + * :obj:`type`: The expected type annotation of the parameter. + """ + # PRE-PROCESSING: + tg_param_type: str = tg_parameter.param_type + is_class = inspect.isclass(obj) + ptb_annotation = cached_type_hints(obj, is_class).get(ptb_param.name) + + # Let's check for a match: + # In order to evaluate the type annotation, we need to first have a mapping of the types + # (see TYPE_MAPPING comment defined at the top level of this module) + mapped: set[type] = _get_params_base(tg_param_type, TYPE_MAPPING) + + # We should have a maximum of one match. + assert len(mapped) <= 1, f"More than one match found for {tg_param_type}" + + # it may be a list of objects, so let's extract them using _extract_words: + mapped_type = _unionizer(_extract_words(tg_param_type)) if not mapped else mapped.pop() + # If the parameter is not required by TG, `None` should be added to `mapped_type` + mapped_type = wrap_with_none(tg_parameter, mapped_type, obj) + + log( + "At the end of PRE-PROCESSING, the values of variables are:\n" + "Parameter name: %s\n" + "ptb_annotation= %s\n" + "mapped_type= %s\n" + "tg_param_type= %s\n" + "tg_parameter.param_required= %s\n", + ptb_param.name, + ptb_annotation, + mapped_type, + tg_param_type, + tg_parameter.param_required, + ) + + # CHECKING: + # Each branch manipulates the `mapped_type` (except for 4) ) to match the `ptb_annotation`. + + # 1) HANDLING ARRAY TYPES: + # Now let's do the checking, starting with "Array of ..." types. + if "Array of " in tg_param_type: + # For exceptions just check if they contain the annotation + if ptb_param.name in PTCE.ARRAY_OF_EXCEPTIONS: + return PTCE.ARRAY_OF_EXCEPTIONS[ptb_param.name] in str(ptb_annotation), Sequence + + obj_match: re.Match | None = re.search(ARRAY_OF_PATTERN, tg_param_type) + if obj_match is None: + raise AssertionError(f"Array of {tg_param_type} not found in {ptb_param.name}") + obj_str: str = obj_match.group(1) + # is obj a regular type like str? + array_map: set[type] = _get_params_base(obj_str, TYPE_MAPPING) + + mapped_type = _unionizer(_extract_words(obj_str)) if not array_map else array_map.pop() + + if "Array of Array of" in tg_param_type: + log("Array of Array of type found in `%s`\n", tg_param_type) + mapped_type = Sequence[Sequence[mapped_type]] + else: + log("Array of type found in `%s`\n", tg_param_type) + mapped_type = Sequence[mapped_type] + + # 2) HANDLING OTHER TYPES: + # Special case for send_* methods where we accept more types than the official API: + elif ptb_param.name in PTCE.ADDITIONAL_TYPES and obj.__name__.startswith("send"): + log("Checking that `%s` has an additional argument!\n", ptb_param.name) + mapped_type = mapped_type | PTCE.ADDITIONAL_TYPES[ptb_param.name] + + # 3) HANDLING DATETIMES: + elif ( + re.search( + DATETIME_REGEX, + ptb_param.name, + ) + or "Unix time" in tg_parameter.param_description + ): + log("Checking that `%s` is a datetime!\n", ptb_param.name) + if ptb_param.name in PTCE.DATETIME_EXCEPTIONS: + return True, mapped_type + # If it's a class, we only accept datetime as the parameter + mapped_type = datetime if is_class else mapped_type | datetime + + # 4) COMPLEX TYPES: + # Some types are too complicated, so we replace our annotation with a simpler type: + elif any(ptb_param.name in key for key in PTCE.COMPLEX_TYPES): + log("Converting `%s` to a simpler type!\n", ptb_param.name) + for (param_name, is_expected_class), exception_type in PTCE.COMPLEX_TYPES.items(): + if ptb_param.name == param_name and is_class is is_expected_class: + ptb_annotation = wrap_with_none(tg_parameter, exception_type, obj) + + # 5) HANDLING DEFAULTS PARAMETERS: + # Classes whose parameters are all ODVInput should be converted and checked. + elif obj.__name__ in PTCE.IGNORED_DEFAULTS_CLASSES: + log("Checking that `%s`'s param is ODVInput:\n", obj.__name__) + mapped_type = ODVInput[mapped_type] + elif not ( + # Defaults checking should not be done for: + # 1. Parameters that have name conflict with `Defaults.name` + is_class + and obj.__name__ in ("ReplyParameters", "Message", "ExternalReplyInfo") + and ptb_param.name in PTCE.IGNORED_DEFAULTS_PARAM_NAMES + ): + # Now let's check if the parameter is a Defaults parameter, it should be + for name, _ in ALL_DEFAULTS: + if name == ptb_param.name or "parse_mode" in ptb_param.name: + log("Checking that `%s` is a Defaults parameter!\n", ptb_param.name) + mapped_type = ODVInput[mapped_type] + break + + # RESULTS:- + mapped_type = wrap_with_none(tg_parameter, mapped_type, obj) + mapped_type = resolve_forward_refs_in_type(mapped_type) + log( + "At RESULTS, we are comparing:\nptb_annotation= %s\nmapped_type= %s\n", + ptb_annotation, + mapped_type, + ) + return mapped_type == ptb_annotation, mapped_type diff --git a/tests/test_official/exceptions.py b/tests/test_official/exceptions.py new file mode 100644 index 000000000..e554b0888 --- /dev/null +++ b/tests/test_official/exceptions.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2024 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +"""This module contains exceptions to our API compared to the official API.""" + + +from telegram import Animation, Audio, Document, PhotoSize, Sticker, Video, VideoNote, Voice +from tests.test_official.helpers import _get_params_base + +IGNORED_OBJECTS = ("ResponseParameters",) +GLOBALLY_IGNORED_PARAMETERS = { + "self", + "read_timeout", + "write_timeout", + "connect_timeout", + "pool_timeout", + "bot", + "api_kwargs", +} + + +class ParamTypeCheckingExceptions: + # Types for certain parameters accepted by PTB but not in the official API + ADDITIONAL_TYPES = { + "photo": PhotoSize, + "video": Video, + "video_note": VideoNote, + "audio": Audio, + "document": Document, + "animation": Animation, + "voice": Voice, + "sticker": Sticker, + } + + # Exceptions to the "Array of" types, where we accept more types than the official API + # key: parameter name, value: type which must be present in the annotation + ARRAY_OF_EXCEPTIONS = { + "results": "InlineQueryResult", # + Callable + "commands": "BotCommand", # + tuple[str, str] + "keyboard": "KeyboardButton", # + sequence[sequence[str]] + "reaction": "ReactionType", # + str + # TODO: Deprecated and will be corrected (and removed) in next major PTB version: + "file_hashes": "List[str]", + } + + # Special cases for other parameters that accept more types than the official API, and are + # too complex to compare/predict with official API: + COMPLEX_TYPES = ( + { # (param_name, is_class (i.e appears in a class?)): reduced form of annotation + ("correct_option_id", False): int, # actual: Literal + ("file_id", False): str, # actual: Union[str, objs_with_file_id_attr] + ("invite_link", False): str, # actual: Union[str, ChatInviteLink] + ("provider_data", False): str, # actual: Union[str, obj] + ("callback_data", True): str, # actual: Union[str, obj] + ("media", True): str, # actual: Union[str, InputMedia*, FileInput] + ( + "data", + True, + ): str, # actual: Union[IdDocumentData, PersonalDetails, ResidentialAddress] + } + ) + + # param names ignored in the param type checking in classes for the `tg.Defaults` case. + IGNORED_DEFAULTS_PARAM_NAMES = { + "quote", + "link_preview_options", + } + + # These classes' params are all ODVInput, so we ignore them in the defaults type checking. + IGNORED_DEFAULTS_CLASSES = {"LinkPreviewOptions"} + + # TODO: Remove this in v22 when it becomes a datetime (also remove from arg_type_checker.py) + DATETIME_EXCEPTIONS = { + "file_date", + } + + +# Arguments *added* to the official API +PTB_EXTRA_PARAMS = { + "send_contact": {"contact"}, + "send_location": {"location"}, + "edit_message_live_location": {"location"}, + "send_venue": {"venue"}, + "answer_inline_query": {"current_offset"}, + "send_media_group": {"caption", "parse_mode", "caption_entities"}, + "send_(animation|audio|document|photo|video(_note)?|voice)": {"filename"}, + "InlineQueryResult": {"id", "type"}, # attributes common to all subclasses + "ChatMember": {"user", "status"}, # attributes common to all subclasses + "BotCommandScope": {"type"}, # attributes common to all subclasses + "MenuButton": {"type"}, # attributes common to all subclasses + "PassportFile": {"credentials"}, + "EncryptedPassportElement": {"credentials"}, + "PassportElementError": {"source", "type", "message"}, + "InputMedia": {"caption", "caption_entities", "media", "media_type", "parse_mode"}, + "InputMedia(Animation|Audio|Document|Photo|Video|VideoNote|Voice)": {"filename"}, + "InputFile": {"attach", "filename", "obj"}, + "MaybeInaccessibleMessage": {"date", "message_id", "chat"}, # attributes common to all subcls + "ChatBoostSource": {"source"}, # attributes common to all subclasses + "MessageOrigin": {"type", "date"}, # attributes common to all subclasses + "ReactionType": {"type"}, # attributes common to all subclasses + "InputTextMessageContent": {"disable_web_page_preview"}, # convenience arg, here for bw compat +} + + +def ptb_extra_params(object_name: str) -> set[str]: + return _get_params_base(object_name, PTB_EXTRA_PARAMS) + + +# Arguments *removed* from the official API +# Mostly due to the value being fixed anyway +PTB_IGNORED_PARAMS = { + r"InlineQueryResult\w+": {"type"}, + r"ChatMember\w+": {"status"}, + r"PassportElementError\w+": {"source"}, + "ForceReply": {"force_reply"}, + "ReplyKeyboardRemove": {"remove_keyboard"}, + r"BotCommandScope\w+": {"type"}, + r"MenuButton\w+": {"type"}, + r"InputMedia\w+": {"type"}, + "InaccessibleMessage": {"date"}, + r"MessageOrigin\w+": {"type"}, + r"ChatBoostSource\w+": {"source"}, + r"ReactionType\w+": {"type"}, +} + + +def ptb_ignored_params(object_name: str) -> set[str]: + return _get_params_base(object_name, PTB_IGNORED_PARAMS) + + +IGNORED_PARAM_REQUIREMENTS = { + # Ignore these since there's convenience params in them (eg. Venue) + # <---- + "send_location": {"latitude", "longitude"}, + "edit_message_live_location": {"latitude", "longitude"}, + "send_venue": {"latitude", "longitude", "title", "address"}, + "send_contact": {"phone_number", "first_name"}, + # ----> +} + + +def ignored_param_requirements(object_name: str) -> set[str]: + return _get_params_base(object_name, IGNORED_PARAM_REQUIREMENTS) + + +# Arguments that are optional arguments for now for backwards compatibility +BACKWARDS_COMPAT_KWARGS: dict[str, set[str]] = { + # Deprecated by Bot API 7.0, kept for now for bw compat: + "KeyboardButton": {"request_user"}, + "Message": { + "forward_from", + "forward_signature", + "forward_sender_name", + "forward_date", + "forward_from_chat", + "forward_from_message_id", + "user_shared", + }, + "(send_message|edit_message_text)": { + "disable_web_page_preview", + "reply_to_message_id", + "allow_sending_without_reply", + }, + r"copy_message|send_\w+": {"allow_sending_without_reply", "reply_to_message_id"}, +} + + +def backwards_compat_kwargs(object_name: str) -> set[str]: + return _get_params_base(object_name, BACKWARDS_COMPAT_KWARGS) + + +IGNORED_PARAM_REQUIREMENTS.update(BACKWARDS_COMPAT_KWARGS) diff --git a/tests/test_official/helpers.py b/tests/test_official/helpers.py new file mode 100644 index 000000000..6851bf85f --- /dev/null +++ b/tests/test_official/helpers.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2024 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +"""This module contains helper functions for the official API tests used in the other modules.""" + +import functools +import re +from typing import TYPE_CHECKING, Any, Sequence, _eval_type, get_type_hints + +from bs4 import PageElement, Tag + +import telegram +import telegram._utils.defaultvalue +import telegram._utils.types + +if TYPE_CHECKING: + from tests.test_official.scraper import TelegramParameter + + +tg_objects = vars(telegram) +tg_objects.update(vars(telegram._utils.types)) +tg_objects.update(vars(telegram._utils.defaultvalue)) + + +def _get_params_base(object_name: str, search_dict: dict[str, set[Any]]) -> set[Any]: + """Helper function for the *_params functions below. + Given an object name and a search dict, goes through the keys of the search dict and checks if + the object name matches any of the regexes (keys). The union of all the sets (values) of the + matching regexes is returned. `object_name` may be a CamelCase or snake_case name. + """ + out = set() + for regex, params in search_dict.items(): + if re.fullmatch(regex, object_name): + out.update(params) + # also check the snake_case version + snake_case_name = re.sub(r"(? set[str]: + """Extracts all words from a string, removing all punctuation and words like 'and' & 'or'.""" + return set(re.sub(r"[^\w\s]", "", text).split()) - {"and", "or"} + + +def _unionizer(annotation: Sequence[Any] | set[Any]) -> Any: + """Returns a union of all the types in the annotation. Also imports objects from lib.""" + union = None + for t in annotation: + if isinstance(t, str): # we have to import objects from lib + t = getattr(telegram, t) # noqa: PLW2901 + union = t if union is None else union | t + return union + + +def find_next_sibling_until(tag: Tag, name: str, until: Tag) -> PageElement | None: + for sibling in tag.next_siblings: + if sibling is until: + return None + if sibling.name == name: + return sibling + return None + + +def is_pascal_case(s): + "PascalCase. Starts with a capital letter and has no spaces. Useful for identifying classes." + return bool(re.match(r"^[A-Z][a-zA-Z\d]*$", s)) + + +def is_parameter_required_by_tg(field: str) -> bool: + if field in {"Required", "Yes"}: + return True + return field.split(".", 1)[0] != "Optional" # splits the sentence and extracts first word + + +def wrap_with_none(tg_parameter: "TelegramParameter", mapped_type: Any, obj: object) -> type: + """Adds `None` to type annotation if the parameter isn't required. Respects ignored params.""" + # have to import here to avoid circular imports + from tests.test_official.exceptions import ignored_param_requirements + + if tg_parameter.param_name in ignored_param_requirements(obj.__name__): + return mapped_type | type(None) + return mapped_type | type(None) if not tg_parameter.param_required else mapped_type + + +@functools.cache +def cached_type_hints(obj: Any, is_class: bool) -> dict[str, Any]: + """Returns type hints of a class, method, or function, with forward refs evaluated.""" + return get_type_hints(obj.__init__ if is_class else obj, localns=tg_objects) + + +@functools.cache +def resolve_forward_refs_in_type(obj: type) -> type: + """Resolves forward references in a type hint.""" + return _eval_type(obj, localns=tg_objects, globalns=None) diff --git a/tests/test_official/scraper.py b/tests/test_official/scraper.py new file mode 100644 index 000000000..1da83a87a --- /dev/null +++ b/tests/test_official/scraper.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2024 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +"""This module contains functions which are used to scrape the official Bot API documentation.""" + +import asyncio +from dataclasses import dataclass +from typing import Literal, overload + +import httpx +from bs4 import BeautifulSoup, Tag + +from tests.test_official.exceptions import IGNORED_OBJECTS +from tests.test_official.helpers import ( + find_next_sibling_until, + is_parameter_required_by_tg, + is_pascal_case, +) + + +@dataclass(slots=True, frozen=True) +class TelegramParameter: + """Represents the scraped Telegram parameter. Contains all relevant attributes needed for + comparison. Relevant for both TelegramMethod and TelegramClass.""" + + param_name: str + param_type: str + param_required: bool + param_description: str + + +@dataclass(slots=True, frozen=True) +class TelegramClass: + """Represents the scraped Telegram class. Contains all relevant attributes needed for + comparison.""" + + class_name: str + class_parameters: list[TelegramParameter] + # class_description: str + + +@dataclass(slots=True, frozen=True) +class TelegramMethod: + """Represents the scraped Telegram method. Contains all relevant attributes needed for + comparison.""" + + method_name: str + method_parameters: list[TelegramParameter] + # method_description: str + + +@dataclass(slots=True, frozen=False) +class Scraper: + request: httpx.Response | None = None + soup: BeautifulSoup | None = None + + async def make_request(self) -> None: + async with httpx.AsyncClient() as client: + self.request = await client.get("https://core.telegram.org/bots/api", timeout=10) + self.soup = BeautifulSoup(self.request.text, "html.parser") + + @overload + def parse_docs( + self, doc_type: Literal["method"] + ) -> tuple[list[TelegramMethod], list[str]]: ... + + @overload + def parse_docs(self, doc_type: Literal["class"]) -> tuple[list[TelegramClass], list[str]]: ... + + def parse_docs(self, doc_type): + argvalues = [] + names: list[str] = [] + if self.request is None: + asyncio.run(self.make_request()) + + for unparsed in self.soup.select("h4 > a.anchor"): + if "-" not in unparsed["name"]: + h4: Tag | None = unparsed.parent + name = h4.text + if h4 is None: + raise AssertionError("h4 is None") + if doc_type == "method" and name[0].lower() == name[0]: + params = parse_table_for_params(h4) + obj = TelegramMethod(method_name=name, method_parameters=params) + argvalues.append(obj) + names.append(name) + elif doc_type == "class" and is_pascal_case(name) and name not in IGNORED_OBJECTS: + params = parse_table_for_params(h4) + obj = TelegramClass(class_name=name, class_parameters=params) + argvalues.append(obj) + names.append(name) + + return argvalues, names + + def collect_methods(self) -> tuple[list[TelegramMethod], list[str]]: + return self.parse_docs("method") + + def collect_classes(self) -> tuple[list[TelegramClass], list[str]]: + return self.parse_docs("class") + + +def parse_table_for_params(h4: Tag) -> list[TelegramParameter]: + """Parses the Telegram doc table and outputs a list of TelegramParameter objects.""" + table = find_next_sibling_until(h4, "table", h4.find_next_sibling("h4")) + if not table: + return [] + + params = [] + for tr in table.find_all("tr")[1:]: + fields = [] + for td in tr.find_all("td"): + param = td.text + fields.append(param) + + param_name = fields[0] + param_type = fields[1] + param_required = is_parameter_required_by_tg(fields[2]) + param_desc = fields[-1] # since length can be 2 or 3, but desc is always the last + params.append(TelegramParameter(param_name, param_type, param_required, param_desc)) + + return params diff --git a/tests/test_official/test_official.py b/tests/test_official/test_official.py new file mode 100644 index 000000000..5ad1d8b56 --- /dev/null +++ b/tests/test_official/test_official.py @@ -0,0 +1,193 @@ +#!/usr/bin/env python +# +# A library that provides a Python interface to the Telegram Bot API +# Copyright (C) 2015-2024 +# Leandro Toledo de Souza +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser Public License for more details. +# +# You should have received a copy of the GNU Lesser Public License +# along with this program. If not, see [http://www.gnu.org/licenses/]. +import inspect +from typing import TYPE_CHECKING + +import pytest + +import telegram +from tests.auxil.envvars import RUN_TEST_OFFICIAL +from tests.test_official.arg_type_checker import ( + check_defaults_type, + check_param_type, + check_required_param, +) +from tests.test_official.exceptions import ( + GLOBALLY_IGNORED_PARAMETERS, + backwards_compat_kwargs, + ptb_extra_params, + ptb_ignored_params, +) +from tests.test_official.scraper import Scraper, TelegramClass, TelegramMethod + +if TYPE_CHECKING: + from types import FunctionType + +# Will skip all tests in this file if the env var is False +pytestmark = pytest.mark.skipif(not RUN_TEST_OFFICIAL, reason="test_official is not enabled") + +methods, method_ids, classes, class_ids = [], [], [], [] # not needed (just for completeness) + +if RUN_TEST_OFFICIAL: + scraper = Scraper() + methods, method_ids = scraper.collect_methods() + classes, class_ids = scraper.collect_classes() + + +@pytest.mark.parametrize("tg_method", argvalues=methods, ids=method_ids) +def test_check_method(tg_method: TelegramMethod) -> None: + """This function checks for the following things compared to the official API docs: + + - Method existence + - Parameter existence + - Parameter requirement correctness + - Parameter type annotation existence + - Parameter type annotation correctness + - Parameter default value correctness + - No unexpected parameters + - Extra parameters should be keyword only + """ + ptb_method: FunctionType | None = getattr(telegram.Bot, tg_method.method_name, None) + assert ptb_method, f"Method {tg_method.method_name} not found in telegram.Bot" + + # Check arguments based on source + sig = inspect.signature(ptb_method, follow_wrapped=True) + checked = [] + + for tg_parameter in tg_method.method_parameters: + # Check if parameter is present in our method + ptb_param = sig.parameters.get(tg_parameter.param_name) + assert ( + ptb_param is not None + ), f"Parameter {tg_parameter.param_name} not found in {ptb_method.__name__}" + + # Now check if the parameter is required or not + assert check_required_param( + tg_parameter, ptb_param, ptb_method.__name__ + ), f"Param {ptb_param.name!r} of {ptb_method.__name__!r} requirement mismatch" + + # Check if type annotation is present + assert ( + ptb_param.annotation is not inspect.Parameter.empty + ), f"Param {ptb_param.name!r} of {ptb_method.__name__!r} should have a type annotation!" + # Check if type annotation is correct + correct_type_hint, expected_type_hint = check_param_type( + ptb_param, + tg_parameter, + ptb_method, + ) + assert correct_type_hint, ( + f"Type hint of param {ptb_param.name!r} of {ptb_method.__name__!r} should be " + f"{expected_type_hint!r} or something else!" + ) + + # Now we will check that we don't pass default values if the parameter is not required. + if ptb_param.default is not inspect.Parameter.empty: # If there is a default argument... + default_arg_none = check_defaults_type(ptb_param) # check if it's None + assert ( + default_arg_none + ), f"Param {ptb_param.name!r} of {ptb_method.__name__!r} should be `None`" + checked.append(tg_parameter.param_name) + + expected_additional_args = GLOBALLY_IGNORED_PARAMETERS.copy() + expected_additional_args |= ptb_extra_params(tg_method.method_name) + expected_additional_args |= backwards_compat_kwargs(tg_method.method_name) + + unexpected_args = (sig.parameters.keys() ^ checked) - expected_additional_args + assert ( + unexpected_args == set() + ), f"In {ptb_method.__qualname__}, unexpected args were found: {unexpected_args}." + + kw_or_positional_args = [ + p.name for p in sig.parameters.values() if p.kind != inspect.Parameter.KEYWORD_ONLY + ] + non_kw_only_args = set(kw_or_positional_args).difference(checked).difference(["self"]) + non_kw_only_args -= backwards_compat_kwargs(tg_method.method_name) + assert non_kw_only_args == set(), ( + f"In {ptb_method.__qualname__}, extra args should be keyword only (compared to " + f"{tg_method.method_name} in API)" + ) + + +@pytest.mark.parametrize("tg_class", argvalues=classes, ids=class_ids) +def test_check_object(tg_class: TelegramClass) -> None: + """This function checks for the following things compared to the official API docs: + + - Class existence + - Parameter existence + - Parameter requirement correctness + - Parameter type annotation existence + - Parameter type annotation correctness + - Parameter default value correctness + - No unexpected parameters + """ + obj = getattr(telegram, tg_class.class_name) + + # Check arguments based on source. Makes sure to only check __init__'s signature & nothing else + sig = inspect.signature(obj.__init__, follow_wrapped=True) + + checked = set() + fields_removed_by_ptb = ptb_ignored_params(tg_class.class_name) + + for tg_parameter in tg_class.class_parameters: + field: str = tg_parameter.param_name + + if field in fields_removed_by_ptb: + continue + + if field == "from": + field = "from_user" + + ptb_param = sig.parameters.get(field) + assert ptb_param is not None, f"Attribute {field} not found in {obj.__name__}" + + # Now check if the parameter is required or not + assert check_required_param( + tg_parameter, ptb_param, obj.__name__ + ), f"Param {ptb_param.name!r} of {obj.__name__!r} requirement mismatch" + + # Check if type annotation is present + assert ( + ptb_param.annotation is not inspect.Parameter.empty + ), f"Param {ptb_param.name!r} of {obj.__name__!r} should have a type annotation" + + # Check if type annotation is correct + correct_type_hint, expected_type_hint = check_param_type(ptb_param, tg_parameter, obj) + assert correct_type_hint, ( + f"Type hint of param {ptb_param.name!r} of {obj.__name__!r} should be " + f"{expected_type_hint!r} or something else!" + ) + + # Now we will check that we don't pass default values if the parameter is not required. + if ptb_param.default is not inspect.Parameter.empty: # If there is a default argument... + default_arg_none = check_defaults_type(ptb_param) # check if its None + assert ( + default_arg_none + ), f"Param {ptb_param.name!r} of {obj.__name__!r} should be `None`" + + checked.add(field) + + expected_additional_args = GLOBALLY_IGNORED_PARAMETERS.copy() + expected_additional_args |= ptb_extra_params(tg_class.class_name) + expected_additional_args |= backwards_compat_kwargs(tg_class.class_name) + + unexpected_args = (sig.parameters.keys() ^ checked) - expected_additional_args + assert ( + unexpected_args == set() + ), f"In {tg_class.class_name}, unexpected args were found: {unexpected_args}."