#!/usr/bin/env python # # A library that provides a Python interface to the Telegram Bot API # Copyright (C) 2015-2020 # 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 time import datetime as dtm from platform import python_implementation import pytest from flaky import flaky from telegram import ( Bot, Update, ChatAction, TelegramError, User, InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResultArticle, InputTextMessageContent, ShippingOption, LabeledPrice, ChatPermissions, Poll, BotCommand, InlineQueryResultDocument, Dice, MessageEntity, ParseMode, ) from telegram.constants import MAX_INLINE_QUERY_RESULTS from telegram.error import BadRequest, InvalidToken, NetworkError, RetryAfter from telegram.utils.helpers import from_timestamp, escape_markdown, to_timestamp from tests.conftest import expect_bad_request BASE_TIME = time.time() HIGHSCORE_DELTA = 1450000000 @pytest.fixture(scope='class') def message(bot, chat_id): to_reply_to = bot.send_message( chat_id, 'Text', disable_web_page_preview=True, disable_notification=True ) return bot.send_message( chat_id, 'Text', reply_to_message_id=to_reply_to.message_id, disable_web_page_preview=True, disable_notification=True, ) @pytest.fixture(scope='class') def media_message(bot, chat_id): with open('tests/data/telegram.ogg', 'rb') as f: return bot.send_voice(chat_id, voice=f, caption='my caption', timeout=10) @pytest.fixture(scope='class') def chat_permissions(): return ChatPermissions(can_send_messages=False, can_change_info=False, can_invite_users=False) def inline_results_callback(page=None): if not page: return [InlineQueryResultArticle(i, str(i), None) for i in range(1, 254)] elif page <= 5: return [ InlineQueryResultArticle(i, str(i), None) for i in range(page * 5 + 1, (page + 1) * 5 + 1) ] return None @pytest.fixture(scope='class') def inline_results(): return inline_results_callback() class TestBot: @pytest.mark.parametrize( 'token', argvalues=[ '123', '12a:abcd1234', '12:abcd1234', '1234:abcd1234\n', ' 1234:abcd1234', ' 1234:abcd1234\r', '1234:abcd 1234', ], ) def test_invalid_token(self, token): with pytest.raises(InvalidToken, match='Invalid token'): Bot(token) @flaky(3, 1) @pytest.mark.timeout(10) def test_invalid_token_server_response(self, monkeypatch): monkeypatch.setattr('telegram.Bot._validate_token', lambda x, y: True) bot = Bot('12') with pytest.raises(InvalidToken): bot.get_me() def test_unknown_kwargs(self, bot, monkeypatch): def post(url, data, timeout): assert data['unknown_kwarg_1'] == 7 assert data['unknown_kwarg_2'] == 5 monkeypatch.setattr(bot.request, 'post', post) bot.send_message(123, 'text', api_kwargs={'unknown_kwarg_1': 7, 'unknown_kwarg_2': 5}) @flaky(3, 1) @pytest.mark.timeout(10) def test_get_me_and_properties(self, bot): get_me_bot = bot.get_me() commands = bot.get_my_commands() assert isinstance(get_me_bot, User) assert get_me_bot.id == bot.id assert get_me_bot.username == bot.username assert get_me_bot.first_name == bot.first_name assert get_me_bot.last_name == bot.last_name assert get_me_bot.name == bot.name assert get_me_bot.can_join_groups == bot.can_join_groups assert get_me_bot.can_read_all_group_messages == bot.can_read_all_group_messages assert get_me_bot.supports_inline_queries == bot.supports_inline_queries assert 'https://t.me/{}'.format(get_me_bot.username) == bot.link assert commands == bot.commands @flaky(3, 1) @pytest.mark.timeout(10) def test_to_dict(self, bot): to_dict_bot = bot.to_dict() assert isinstance(to_dict_bot, dict) assert to_dict_bot["id"] == bot.id assert to_dict_bot["username"] == bot.username assert to_dict_bot["first_name"] == bot.first_name if bot.last_name: assert to_dict_bot["last_name"] == bot.last_name @flaky(3, 1) @pytest.mark.timeout(10) def test_forward_message(self, bot, chat_id, message): message = bot.forward_message(chat_id, from_chat_id=chat_id, message_id=message.message_id) assert message.text == message.text assert message.forward_from.username == message.from_user.username assert isinstance(message.forward_date, dtm.datetime) @flaky(3, 1) @pytest.mark.timeout(10) def test_delete_message(self, bot, chat_id): message = bot.send_message(chat_id, text='will be deleted') time.sleep(2) assert bot.delete_message(chat_id=chat_id, message_id=message.message_id) is True @flaky(3, 1) @pytest.mark.timeout(10) def test_delete_message_old_message(self, bot, chat_id): with pytest.raises(BadRequest): # Considering that the first message is old enough bot.delete_message(chat_id=chat_id, message_id=1) # send_photo, send_audio, send_document, send_sticker, send_video, send_voice, send_video_note, # send_media_group and send_animation are tested in their respective test modules. No need to # duplicate here. @flaky(3, 1) @pytest.mark.timeout(10) def test_send_venue(self, bot, chat_id): longitude = -46.788279 latitude = -23.691288 title = 'title' address = 'address' foursquare_id = 'foursquare id' foursquare_type = 'foursquare type' message = bot.send_venue( chat_id=chat_id, title=title, address=address, latitude=latitude, longitude=longitude, foursquare_id=foursquare_id, foursquare_type=foursquare_type, ) assert message.venue assert message.venue.title == title assert message.venue.address == address assert message.venue.location.latitude == latitude assert message.venue.location.longitude == longitude assert message.venue.foursquare_id == foursquare_id assert message.venue.foursquare_type == foursquare_type @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.xfail(raises=RetryAfter) @pytest.mark.skipif( python_implementation() == 'PyPy', reason='Unstable on pypy for some reason' ) def test_send_contact(self, bot, chat_id): phone_number = '+11234567890' first_name = 'Leandro' last_name = 'Toledo' message = bot.send_contact( chat_id=chat_id, phone_number=phone_number, first_name=first_name, last_name=last_name ) assert message.contact assert message.contact.phone_number == phone_number assert message.contact.first_name == first_name assert message.contact.last_name == last_name # TODO: Add bot to group to test polls too @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize( 'reply_markup', [ None, InlineKeyboardMarkup.from_button( InlineKeyboardButton(text='text', callback_data='data') ), InlineKeyboardMarkup.from_button( InlineKeyboardButton(text='text', callback_data='data') ).to_dict(), ], ) def test_send_and_stop_poll(self, bot, super_group_id, reply_markup): question = 'Is this a test?' answers = ['Yes', 'No', 'Maybe'] message = bot.send_poll( chat_id=super_group_id, question=question, options=answers, is_anonymous=False, allows_multiple_answers=True, timeout=60, ) assert message.poll assert message.poll.question == question assert message.poll.options[0].text == answers[0] assert message.poll.options[1].text == answers[1] assert message.poll.options[2].text == answers[2] assert not message.poll.is_anonymous assert message.poll.allows_multiple_answers assert not message.poll.is_closed assert message.poll.type == Poll.REGULAR # Since only the poll and not the complete message is returned, we can't check that the # reply_markup is correct. So we just test that sending doesn't give an error. poll = bot.stop_poll( chat_id=super_group_id, message_id=message.message_id, reply_markup=reply_markup, timeout=60, ) assert isinstance(poll, Poll) assert poll.is_closed assert poll.options[0].text == answers[0] assert poll.options[0].voter_count == 0 assert poll.options[1].text == answers[1] assert poll.options[1].voter_count == 0 assert poll.options[2].text == answers[2] assert poll.options[2].voter_count == 0 assert poll.question == question assert poll.total_voter_count == 0 explanation = '[Here is a link](https://google.com)' explanation_entities = [ MessageEntity(MessageEntity.TEXT_LINK, 0, 14, url='https://google.com') ] message_quiz = bot.send_poll( chat_id=super_group_id, question=question, options=answers, type=Poll.QUIZ, correct_option_id=2, is_closed=True, explanation=explanation, explanation_parse_mode=ParseMode.MARKDOWN_V2, ) assert message_quiz.poll.correct_option_id == 2 assert message_quiz.poll.type == Poll.QUIZ assert message_quiz.poll.is_closed assert message_quiz.poll.explanation == 'Here is a link' assert message_quiz.poll.explanation_entities == explanation_entities @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize(['open_period', 'close_date'], [(5, None), (None, True)]) def test_send_open_period(self, bot, super_group_id, open_period, close_date): question = 'Is this a test?' answers = ['Yes', 'No', 'Maybe'] reply_markup = InlineKeyboardMarkup.from_button( InlineKeyboardButton(text='text', callback_data='data') ) if close_date: close_date = dtm.datetime.utcnow() + dtm.timedelta(seconds=5) message = bot.send_poll( chat_id=super_group_id, question=question, options=answers, is_anonymous=False, allows_multiple_answers=True, timeout=60, open_period=open_period, close_date=close_date, ) time.sleep(5.1) new_message = bot.edit_message_reply_markup( chat_id=super_group_id, message_id=message.message_id, reply_markup=reply_markup, timeout=60, ) assert new_message.poll.id == message.poll.id assert new_message.poll.is_closed @flaky(5, 1) @pytest.mark.timeout(10) def test_send_close_date_default_tz(self, tz_bot, super_group_id): question = 'Is this a test?' answers = ['Yes', 'No', 'Maybe'] reply_markup = InlineKeyboardMarkup.from_button( InlineKeyboardButton(text='text', callback_data='data') ) aware_close_date = dtm.datetime.now(tz=tz_bot.defaults.tzinfo) + dtm.timedelta(seconds=5) close_date = aware_close_date.replace(tzinfo=None) message = tz_bot.send_poll( chat_id=super_group_id, question=question, options=answers, close_date=close_date, timeout=60, ) assert message.poll.close_date == aware_close_date.replace(microsecond=0) time.sleep(5.1) new_message = tz_bot.edit_message_reply_markup( chat_id=super_group_id, message_id=message.message_id, reply_markup=reply_markup, timeout=60, ) assert new_message.poll.id == message.poll.id assert new_message.poll.is_closed @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize('default_bot', [{'parse_mode': 'Markdown'}], indirect=True) def test_send_poll_default_parse_mode(self, default_bot, super_group_id): explanation = 'Italic Bold Code' explanation_markdown = '_Italic_ *Bold* `Code`' question = 'Is this a test?' answers = ['Yes', 'No', 'Maybe'] message = default_bot.send_poll( chat_id=super_group_id, question=question, options=answers, type=Poll.QUIZ, correct_option_id=2, is_closed=True, explanation=explanation_markdown, ) assert message.poll.explanation == explanation assert message.poll.explanation_entities == [ MessageEntity(MessageEntity.ITALIC, 0, 6), MessageEntity(MessageEntity.BOLD, 7, 4), MessageEntity(MessageEntity.CODE, 12, 4), ] message = default_bot.send_poll( chat_id=super_group_id, question=question, options=answers, type=Poll.QUIZ, correct_option_id=2, is_closed=True, explanation=explanation_markdown, explanation_parse_mode=None, ) assert message.poll.explanation == explanation_markdown assert message.poll.explanation_entities == [] message = default_bot.send_poll( chat_id=super_group_id, question=question, options=answers, type=Poll.QUIZ, correct_option_id=2, is_closed=True, explanation=explanation_markdown, explanation_parse_mode='HTML', ) assert message.poll.explanation == explanation_markdown assert message.poll.explanation_entities == [] @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize('emoji', Dice.ALL_EMOJI + [None]) def test_send_dice(self, bot, chat_id, emoji): message = bot.send_dice(chat_id, emoji=emoji) assert message.dice if emoji is None: assert message.dice.emoji == Dice.DICE else: assert message.dice.emoji == emoji @flaky(3, 1) @pytest.mark.timeout(10) def test_send_chat_action(self, bot, chat_id): assert bot.send_chat_action(chat_id, ChatAction.TYPING) # TODO: Needs improvement. We need incoming inline query to test answer. def test_answer_inline_query(self, monkeypatch, bot): # For now just test that our internals pass the correct data def test(url, data, *args, **kwargs): return data == { 'cache_time': 300, 'results': [ { 'title': 'first', 'id': '11', 'type': 'article', 'input_message_content': {'message_text': 'first'}, }, { 'title': 'second', 'id': '12', 'type': 'article', 'input_message_content': {'message_text': 'second'}, }, ], 'next_offset': '42', 'switch_pm_parameter': 'start_pm', 'inline_query_id': 1234, 'is_personal': True, 'switch_pm_text': 'switch pm', } monkeypatch.setattr(bot.request, 'post', test) results = [ InlineQueryResultArticle('11', 'first', InputTextMessageContent('first')), InlineQueryResultArticle('12', 'second', InputTextMessageContent('second')), ] assert bot.answer_inline_query( 1234, results=results, cache_time=300, is_personal=True, next_offset='42', switch_pm_text='switch pm', switch_pm_parameter='start_pm', ) def test_answer_inline_query_no_default_parse_mode(self, monkeypatch, bot): def test(url, data, *args, **kwargs): return data == { 'cache_time': 300, 'results': [ { 'title': 'test_result', 'id': '123', 'type': 'document', 'document_url': 'https://raw.githubusercontent.com/' 'python-telegram-bot/logos/master/logo/png/' 'ptb-logo_240.png', 'mime_type': 'image/png', 'caption': 'ptb_logo', } ], 'next_offset': '42', 'switch_pm_parameter': 'start_pm', 'inline_query_id': 1234, 'is_personal': True, 'switch_pm_text': 'switch pm', } monkeypatch.setattr(bot.request, 'post', test) results = [ InlineQueryResultDocument( id='123', document_url='https://raw.githubusercontent.com/python-telegram-bot/logos/master/' 'logo/png/ptb-logo_240.png', title='test_result', mime_type='image/png', caption='ptb_logo', ) ] assert bot.answer_inline_query( 1234, results=results, cache_time=300, is_personal=True, next_offset='42', switch_pm_text='switch pm', switch_pm_parameter='start_pm', ) @pytest.mark.parametrize('default_bot', [{'parse_mode': 'Markdown'}], indirect=True) def test_answer_inline_query_default_parse_mode(self, monkeypatch, default_bot): def test(url, data, *args, **kwargs): return data == { 'cache_time': 300, 'results': [ { 'title': 'test_result', 'id': '123', 'type': 'document', 'document_url': 'https://raw.githubusercontent.com/' 'python-telegram-bot/logos/master/logo/png/' 'ptb-logo_240.png', 'mime_type': 'image/png', 'caption': 'ptb_logo', 'parse_mode': 'Markdown', } ], 'next_offset': '42', 'switch_pm_parameter': 'start_pm', 'inline_query_id': 1234, 'is_personal': True, 'switch_pm_text': 'switch pm', } monkeypatch.setattr(default_bot.request, 'post', test) results = [ InlineQueryResultDocument( id='123', document_url='https://raw.githubusercontent.com/python-telegram-bot/logos/master/' 'logo/png/ptb-logo_240.png', title='test_result', mime_type='image/png', caption='ptb_logo', ) ] assert default_bot.answer_inline_query( 1234, results=results, cache_time=300, is_personal=True, next_offset='42', switch_pm_text='switch pm', switch_pm_parameter='start_pm', ) def test_answer_inline_query_current_offset_error(self, bot, inline_results): with pytest.raises(ValueError, match=('`current_offset` and `next_offset`')): bot.answer_inline_query( 1234, results=inline_results, next_offset=42, current_offset=51 ) @pytest.mark.parametrize( 'current_offset,num_results,id_offset,expected_next_offset', [ ('', MAX_INLINE_QUERY_RESULTS, 1, 1), (1, MAX_INLINE_QUERY_RESULTS, 51, 2), (5, 3, 251, ''), ], ) def test_answer_inline_query_current_offset_1( self, monkeypatch, bot, inline_results, current_offset, num_results, id_offset, expected_next_offset, ): # For now just test that our internals pass the correct data def make_assertion(url, data, *args, **kwargs): results = data['results'] length_matches = len(results) == num_results ids_match = all([int(res['id']) == id_offset + i for i, res in enumerate(results)]) next_offset_matches = data['next_offset'] == str(expected_next_offset) return length_matches and ids_match and next_offset_matches monkeypatch.setattr(bot.request, 'post', make_assertion) assert bot.answer_inline_query(1234, results=inline_results, current_offset=current_offset) def test_answer_inline_query_current_offset_2(self, monkeypatch, bot, inline_results): # For now just test that our internals pass the correct data def make_assertion(url, data, *args, **kwargs): results = data['results'] length_matches = len(results) == MAX_INLINE_QUERY_RESULTS ids_match = all([int(res['id']) == 1 + i for i, res in enumerate(results)]) next_offset_matches = data['next_offset'] == '1' return length_matches and ids_match and next_offset_matches monkeypatch.setattr(bot.request, 'post', make_assertion) assert bot.answer_inline_query(1234, results=inline_results, current_offset=0) inline_results = inline_results[:30] def make_assertion(url, data, *args, **kwargs): results = data['results'] length_matches = len(results) == 30 ids_match = all([int(res['id']) == 1 + i for i, res in enumerate(results)]) next_offset_matches = data['next_offset'] == '' return length_matches and ids_match and next_offset_matches monkeypatch.setattr(bot.request, 'post', make_assertion) assert bot.answer_inline_query(1234, results=inline_results, current_offset=0) def test_answer_inline_query_current_offset_callback(self, monkeypatch, bot, caplog): # For now just test that our internals pass the correct data def make_assertion(url, data, *args, **kwargs): results = data['results'] length = len(results) == 5 ids = all([int(res['id']) == 6 + i for i, res in enumerate(results)]) next_offset = data['next_offset'] == '2' return length and ids and next_offset monkeypatch.setattr(bot.request, 'post', make_assertion) assert bot.answer_inline_query(1234, results=inline_results_callback, current_offset=1) def make_assertion(url, data, *args, **kwargs): results = data['results'] length = results == [] next_offset = data['next_offset'] == '' return length and next_offset monkeypatch.setattr(bot.request, 'post', make_assertion) assert bot.answer_inline_query(1234, results=inline_results_callback, current_offset=6) @flaky(3, 1) @pytest.mark.timeout(10) def test_get_user_profile_photos(self, bot, chat_id): user_profile_photos = bot.get_user_profile_photos(chat_id) assert user_profile_photos.photos[0][0].file_size == 5403 @flaky(3, 1) @pytest.mark.timeout(10) def test_get_one_user_profile_photo(self, bot, chat_id): user_profile_photos = bot.get_user_profile_photos(chat_id, offset=0, limit=1) assert user_profile_photos.photos[0][0].file_size == 5403 # get_file is tested multiple times in the test_*media* modules. # TODO: Needs improvement. No feasable way to test until bots can add members. def test_kick_chat_member(self, monkeypatch, bot): def test(url, data, *args, **kwargs): chat_id = data['chat_id'] == 2 user_id = data['user_id'] == 32 until_date = data.get('until_date', 1577887200) == 1577887200 return chat_id and user_id and until_date monkeypatch.setattr(bot.request, 'post', test) until = from_timestamp(1577887200) assert bot.kick_chat_member(2, 32) assert bot.kick_chat_member(2, 32, until_date=until) assert bot.kick_chat_member(2, 32, until_date=1577887200) def test_kick_chat_member_default_tz(self, monkeypatch, tz_bot): until = dtm.datetime(2020, 1, 11, 16, 13) until_timestamp = to_timestamp(until, tzinfo=tz_bot.defaults.tzinfo) def test(url, data, *args, **kwargs): chat_id = data['chat_id'] == 2 user_id = data['user_id'] == 32 until_date = data.get('until_date', until_timestamp) == until_timestamp return chat_id and user_id and until_date monkeypatch.setattr(tz_bot.request, 'post', test) assert tz_bot.kick_chat_member(2, 32) assert tz_bot.kick_chat_member(2, 32, until_date=until) assert tz_bot.kick_chat_member(2, 32, until_date=until_timestamp) # TODO: Needs improvement. def test_unban_chat_member(self, monkeypatch, bot): def test(url, data, *args, **kwargs): chat_id = data['chat_id'] == 2 user_id = data['user_id'] == 32 return chat_id and user_id monkeypatch.setattr(bot.request, 'post', test) assert bot.unban_chat_member(2, 32) def test_set_chat_permissions(self, monkeypatch, bot, chat_permissions): def test(url, data, *args, **kwargs): chat_id = data['chat_id'] == 2 permissions = data['permissions'] == chat_permissions.to_dict() return chat_id and permissions monkeypatch.setattr(bot.request, 'post', test) assert bot.set_chat_permissions(2, chat_permissions) def test_set_chat_administrator_custom_title(self, monkeypatch, bot): def test(url, data, *args, **kwargs): chat_id = data['chat_id'] == 2 user_id = data['user_id'] == 32 custom_title = data['custom_title'] == 'custom_title' return chat_id and user_id and custom_title monkeypatch.setattr(bot.request, 'post', test) assert bot.set_chat_administrator_custom_title(2, 32, 'custom_title') # TODO: Needs improvement. Need an incoming callbackquery to test def test_answer_callback_query(self, monkeypatch, bot): # For now just test that our internals pass the correct data def test(url, data, *args, **kwargs): return data == { 'callback_query_id': 23, 'show_alert': True, 'url': 'no_url', 'cache_time': 1, 'text': 'answer', } monkeypatch.setattr(bot.request, 'post', test) assert bot.answer_callback_query( 23, text='answer', show_alert=True, url='no_url', cache_time=1 ) @flaky(3, 1) @pytest.mark.timeout(10) def test_edit_message_text(self, bot, message): message = bot.edit_message_text( text='new_text', chat_id=message.chat_id, message_id=message.message_id, parse_mode='HTML', disable_web_page_preview=True, ) assert message.text == 'new_text' @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize('default_bot', [{'parse_mode': 'Markdown'}], indirect=True) def test_edit_message_text_default_parse_mode(self, default_bot, message): test_string = 'Italic Bold Code' test_markdown_string = '_Italic_ *Bold* `Code`' message = default_bot.edit_message_text( text=test_markdown_string, chat_id=message.chat_id, message_id=message.message_id, disable_web_page_preview=True, ) assert message.text_markdown == test_markdown_string assert message.text == test_string message = default_bot.edit_message_text( text=test_markdown_string, chat_id=message.chat_id, message_id=message.message_id, parse_mode=None, disable_web_page_preview=True, ) assert message.text == test_markdown_string assert message.text_markdown == escape_markdown(test_markdown_string) message = default_bot.edit_message_text( text=test_markdown_string, chat_id=message.chat_id, message_id=message.message_id, disable_web_page_preview=True, ) message = default_bot.edit_message_text( text=test_markdown_string, chat_id=message.chat_id, message_id=message.message_id, parse_mode='HTML', disable_web_page_preview=True, ) assert message.text == test_markdown_string assert message.text_markdown == escape_markdown(test_markdown_string) @pytest.mark.skip(reason='need reference to an inline message') def test_edit_message_text_inline(self): pass @flaky(3, 1) @pytest.mark.timeout(10) def test_edit_message_caption(self, bot, media_message): message = bot.edit_message_caption( caption='new_caption', chat_id=media_message.chat_id, message_id=media_message.message_id, ) assert message.caption == 'new_caption' # edit_message_media is tested in test_inputmedia @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize('default_bot', [{'parse_mode': 'Markdown'}], indirect=True) def test_edit_message_caption_default_parse_mode(self, default_bot, media_message): test_string = 'Italic Bold Code' test_markdown_string = '_Italic_ *Bold* `Code`' message = default_bot.edit_message_caption( caption=test_markdown_string, chat_id=media_message.chat_id, message_id=media_message.message_id, ) assert message.caption_markdown == test_markdown_string assert message.caption == test_string message = default_bot.edit_message_caption( caption=test_markdown_string, chat_id=media_message.chat_id, message_id=media_message.message_id, parse_mode=None, ) assert message.caption == test_markdown_string assert message.caption_markdown == escape_markdown(test_markdown_string) message = default_bot.edit_message_caption( caption=test_markdown_string, chat_id=media_message.chat_id, message_id=media_message.message_id, ) message = default_bot.edit_message_caption( caption=test_markdown_string, chat_id=media_message.chat_id, message_id=media_message.message_id, parse_mode='HTML', ) assert message.caption == test_markdown_string assert message.caption_markdown == escape_markdown(test_markdown_string) @flaky(3, 1) @pytest.mark.timeout(10) def test_edit_message_caption_with_parse_mode(self, bot, media_message): message = bot.edit_message_caption( caption='new *caption*', parse_mode='Markdown', chat_id=media_message.chat_id, message_id=media_message.message_id, ) assert message.caption == 'new caption' def test_edit_message_caption_without_required(self, bot): with pytest.raises(ValueError, match='Both chat_id and message_id are required when'): bot.edit_message_caption(caption='new_caption') @pytest.mark.skip(reason='need reference to an inline message') def test_edit_message_caption_inline(self): pass @flaky(3, 1) @pytest.mark.timeout(10) def test_edit_reply_markup(self, bot, message): new_markup = InlineKeyboardMarkup([[InlineKeyboardButton(text='test', callback_data='1')]]) message = bot.edit_message_reply_markup( chat_id=message.chat_id, message_id=message.message_id, reply_markup=new_markup ) assert message is not True def test_edit_message_reply_markup_without_required(self, bot): new_markup = InlineKeyboardMarkup([[InlineKeyboardButton(text='test', callback_data='1')]]) with pytest.raises(ValueError, match='Both chat_id and message_id are required when'): bot.edit_message_reply_markup(reply_markup=new_markup) @pytest.mark.skip(reason='need reference to an inline message') def test_edit_reply_markup_inline(self): pass # TODO: Actually send updates to the test bot so this can be tested properly @flaky(3, 1) @pytest.mark.timeout(10) def test_get_updates(self, bot): bot.delete_webhook() # make sure there is no webhook set if webhook tests failed updates = bot.get_updates(timeout=1) assert isinstance(updates, list) if updates: assert isinstance(updates[0], Update) @flaky(3, 1) @pytest.mark.timeout(15) @pytest.mark.xfail def test_set_webhook_get_webhook_info_and_delete_webhook(self, bot): url = 'https://python-telegram-bot.org/test/webhook' max_connections = 7 allowed_updates = ['message'] bot.set_webhook(url, max_connections=max_connections, allowed_updates=allowed_updates) time.sleep(2) live_info = bot.get_webhook_info() time.sleep(6) bot.delete_webhook() time.sleep(2) info = bot.get_webhook_info() assert info.url == '' assert live_info.url == url assert live_info.max_connections == max_connections assert live_info.allowed_updates == allowed_updates @flaky(3, 1) @pytest.mark.timeout(10) def test_leave_chat(self, bot): with pytest.raises(BadRequest, match='Chat not found'): bot.leave_chat(-123456) with pytest.raises(NetworkError, match='Chat not found'): bot.leave_chat(-123456) @flaky(3, 1) @pytest.mark.timeout(10) def test_get_chat(self, bot, super_group_id): chat = bot.get_chat(super_group_id) assert chat.type == 'supergroup' assert chat.title == '>>> telegram.Bot(test) @{}'.format(bot.username) assert chat.id == int(super_group_id) @flaky(3, 1) @pytest.mark.timeout(10) def test_get_chat_administrators(self, bot, channel_id): admins = bot.get_chat_administrators(channel_id) assert isinstance(admins, list) for a in admins: assert a.status in ('administrator', 'creator') @flaky(3, 1) @pytest.mark.timeout(10) def test_get_chat_members_count(self, bot, channel_id): count = bot.get_chat_members_count(channel_id) assert isinstance(count, int) assert count > 3 @flaky(3, 1) @pytest.mark.timeout(10) def test_get_chat_member(self, bot, channel_id, chat_id): chat_member = bot.get_chat_member(channel_id, chat_id) assert chat_member.status == 'administrator' assert chat_member.user.first_name == 'PTB' assert chat_member.user.last_name == 'Test user' @pytest.mark.skip(reason="Not implemented yet.") def test_set_chat_sticker_set(self): pass @pytest.mark.skip(reason="Not implemented yet.") def test_delete_chat_sticker_set(self): pass @flaky(3, 1) @pytest.mark.timeout(10) def test_send_game(self, bot, chat_id): game_short_name = 'test_game' message = bot.send_game(chat_id, game_short_name) assert message.game assert message.game.description == ( 'A no-op test game, for python-telegram-bot ' 'bot framework testing.' ) assert message.game.animation.file_id != '' assert message.game.photo[0].file_size == 851 @flaky(3, 1) @pytest.mark.timeout(10) def test_set_game_score_1(self, bot, chat_id): # NOTE: numbering of methods assures proper order between test_set_game_scoreX methods def func(): game_short_name = 'test_game' game = bot.send_game(chat_id, game_short_name) message = bot.set_game_score( user_id=chat_id, score=int(BASE_TIME) - HIGHSCORE_DELTA, chat_id=game.chat_id, message_id=game.message_id, ) assert message.game.description == game.game.description assert message.game.animation.file_id == game.game.animation.file_id assert message.game.photo[0].file_size == game.game.photo[0].file_size assert message.game.text != game.game.text expect_bad_request(func, 'Bot_score_not_modified', 'This test is a diva for some reason.') @flaky(3, 1) @pytest.mark.timeout(10) def test_set_game_score_2(self, bot, chat_id): # NOTE: numbering of methods assures proper order between test_set_game_scoreX methods game_short_name = 'test_game' game = bot.send_game(chat_id, game_short_name) score = int(BASE_TIME) - HIGHSCORE_DELTA + 1 message = bot.set_game_score( user_id=chat_id, score=score, chat_id=game.chat_id, message_id=game.message_id, disable_edit_message=True, ) assert message.game.description == game.game.description assert message.game.animation.file_id == game.game.animation.file_id assert message.game.photo[0].file_size == game.game.photo[0].file_size assert message.game.text == game.game.text @flaky(3, 1) @pytest.mark.timeout(10) def test_set_game_score_3(self, bot, chat_id): # NOTE: numbering of methods assures proper order between test_set_game_scoreX methods game_short_name = 'test_game' game = bot.send_game(chat_id, game_short_name) score = int(BASE_TIME) - HIGHSCORE_DELTA - 1 with pytest.raises(BadRequest, match='Bot_score_not_modified'): bot.set_game_score( user_id=chat_id, score=score, chat_id=game.chat_id, message_id=game.message_id ) @flaky(3, 1) @pytest.mark.timeout(10) def test_set_game_score_4(self, bot, chat_id): # NOTE: numbering of methods assures proper order between test_set_game_scoreX methods game_short_name = 'test_game' game = bot.send_game(chat_id, game_short_name) score = int(BASE_TIME) - HIGHSCORE_DELTA - 2 message = bot.set_game_score( user_id=chat_id, score=score, chat_id=game.chat_id, message_id=game.message_id, force=True, ) assert message.game.description == game.game.description assert message.game.animation.file_id == game.game.animation.file_id assert message.game.photo[0].file_size == game.game.photo[0].file_size # For some reason the returned message does not contain the updated score. need to fetch # the game again... game2 = bot.send_game(chat_id, game_short_name) assert str(score) in game2.game.text @flaky(3, 1) @pytest.mark.timeout(10) def test_set_game_score_too_low_score(self, bot, chat_id): # We need a game to set the score for game_short_name = 'test_game' game = bot.send_game(chat_id, game_short_name) with pytest.raises(BadRequest): bot.set_game_score( user_id=chat_id, score=100, chat_id=game.chat_id, message_id=game.message_id ) @flaky(3, 1) @pytest.mark.timeout(10) def test_get_game_high_scores(self, bot, chat_id): # We need a game to get the scores for game_short_name = 'test_game' game = bot.send_game(chat_id, game_short_name) high_scores = bot.get_game_high_scores(chat_id, game.chat_id, game.message_id) # We assume that the other game score tests ran within 20 sec assert pytest.approx(high_scores[0].score, abs=20) == int(BASE_TIME) - HIGHSCORE_DELTA # send_invoice is tested in test_invoice # TODO: Needs improvement. Need incoming shippping queries to test def test_answer_shipping_query_ok(self, monkeypatch, bot): # For now just test that our internals pass the correct data def test(url, data, *args, **kwargs): return data == { 'shipping_query_id': 1, 'ok': True, 'shipping_options': [ {'title': 'option1', 'prices': [{'label': 'price', 'amount': 100}], 'id': 1} ], } monkeypatch.setattr(bot.request, 'post', test) shipping_options = ShippingOption(1, 'option1', [LabeledPrice('price', 100)]) assert bot.answer_shipping_query(1, True, shipping_options=[shipping_options]) def test_answer_shipping_query_error_message(self, monkeypatch, bot): # For now just test that our internals pass the correct data def test(url, data, *args, **kwargs): return data == { 'shipping_query_id': 1, 'error_message': 'Not enough fish', 'ok': False, } monkeypatch.setattr(bot.request, 'post', test) assert bot.answer_shipping_query(1, False, error_message='Not enough fish') def test_answer_shipping_query_errors(self, monkeypatch, bot): shipping_options = ShippingOption(1, 'option1', [LabeledPrice('price', 100)]) with pytest.raises(TelegramError, match='should not be empty and there should not be'): bot.answer_shipping_query(1, True, error_message='Not enough fish') with pytest.raises(TelegramError, match='should not be empty and there should not be'): bot.answer_shipping_query(1, False) with pytest.raises(TelegramError, match='should not be empty and there should not be'): bot.answer_shipping_query(1, False, shipping_options=shipping_options) with pytest.raises(TelegramError, match='should not be empty and there should not be'): bot.answer_shipping_query(1, True) # TODO: Needs improvement. Need incoming pre checkout queries to test def test_answer_pre_checkout_query_ok(self, monkeypatch, bot): # For now just test that our internals pass the correct data def test(url, data, *args, **kwargs): return data == {'pre_checkout_query_id': 1, 'ok': True} monkeypatch.setattr(bot.request, 'post', test) assert bot.answer_pre_checkout_query(1, True) def test_answer_pre_checkout_query_error_message(self, monkeypatch, bot): # For now just test that our internals pass the correct data def test(url, data, *args, **kwargs): return data == { 'pre_checkout_query_id': 1, 'error_message': 'Not enough fish', 'ok': False, } monkeypatch.setattr(bot.request, 'post', test) assert bot.answer_pre_checkout_query(1, False, error_message='Not enough fish') def test_answer_pre_checkout_query_errors(self, monkeypatch, bot): with pytest.raises(TelegramError, match='should not be'): bot.answer_pre_checkout_query(1, True, error_message='Not enough fish') with pytest.raises(TelegramError, match='should not be empty'): bot.answer_pre_checkout_query(1, False) @flaky(3, 1) @pytest.mark.timeout(10) def test_restrict_chat_member(self, bot, channel_id, chat_permissions): # TODO: Add bot to supergroup so this can be tested properly with pytest.raises(BadRequest, match='Method is available only for supergroups'): assert bot.restrict_chat_member( channel_id, 95205500, chat_permissions, until_date=dtm.datetime.utcnow() ) def test_restrict_chat_member_default_tz( self, monkeypatch, tz_bot, channel_id, chat_permissions ): until = dtm.datetime(2020, 1, 11, 16, 13) until_timestamp = to_timestamp(until, tzinfo=tz_bot.defaults.tzinfo) def test(url, data, *args, **kwargs): return data.get('until_date', until_timestamp) == until_timestamp monkeypatch.setattr(tz_bot.request, 'post', test) assert tz_bot.restrict_chat_member(channel_id, 95205500, chat_permissions) assert tz_bot.restrict_chat_member( channel_id, 95205500, chat_permissions, until_date=until ) assert tz_bot.restrict_chat_member( channel_id, 95205500, chat_permissions, until_date=until_timestamp ) @flaky(3, 1) @pytest.mark.timeout(10) def test_promote_chat_member(self, bot, channel_id): # TODO: Add bot to supergroup so this can be tested properly / give bot perms with pytest.raises(BadRequest, match='Not enough rights'): assert bot.promote_chat_member( channel_id, 95205500, can_change_info=True, can_post_messages=True, can_edit_messages=True, can_delete_messages=True, can_invite_users=True, can_restrict_members=True, can_pin_messages=True, can_promote_members=True, ) @flaky(3, 1) @pytest.mark.timeout(10) def test_export_chat_invite_link(self, bot, channel_id): # Each link is unique apparently invite_link = bot.export_chat_invite_link(channel_id) assert isinstance(invite_link, str) assert invite_link != '' @flaky(3, 1) @pytest.mark.timeout(10) def test_set_chat_photo(self, bot, channel_id): def func(): assert bot.set_chat_photo(channel_id, f) with open('tests/data/telegram_test_channel.jpg', 'rb') as f: expect_bad_request(func, 'Type of file mismatch', 'Telegram did not accept the file.') @flaky(3, 1) @pytest.mark.timeout(10) def test_delete_chat_photo(self, bot, channel_id): def func(): assert bot.delete_chat_photo(channel_id) expect_bad_request(func, 'Chat_not_modified', 'Chat photo was not set.') @flaky(3, 1) @pytest.mark.timeout(10) def test_set_chat_title(self, bot, channel_id): assert bot.set_chat_title(channel_id, '>>> telegram.Bot() - Tests') @flaky(3, 1) @pytest.mark.timeout(10) def test_set_chat_description(self, bot, channel_id): assert bot.set_chat_description(channel_id, 'Time: ' + str(time.time())) # TODO: Add bot to group to test there too @flaky(3, 1) @pytest.mark.timeout(10) def test_pin_and_unpin_message(self, bot, super_group_id): message = bot.send_message(super_group_id, text="test_pin_message") assert bot.pin_chat_message( chat_id=super_group_id, message_id=message.message_id, disable_notification=True ) chat = bot.get_chat(super_group_id) assert chat.pinned_message == message assert bot.unpinChatMessage(super_group_id) # get_sticker_set, upload_sticker_file, create_new_sticker_set, add_sticker_to_set, # set_sticker_position_in_set and delete_sticker_from_set are tested in the # test_sticker module. def test_timeout_propagation_explicit(self, monkeypatch, bot, chat_id): from telegram.vendor.ptb_urllib3.urllib3.util.timeout import Timeout class OkException(Exception): pass TIMEOUT = 500 def request_wrapper(*args, **kwargs): obj = kwargs.get('timeout') if isinstance(obj, Timeout) and obj._read == TIMEOUT: raise OkException return b'{"ok": true, "result": []}' monkeypatch.setattr('telegram.utils.request.Request._request_wrapper', request_wrapper) # Test file uploading with pytest.raises(OkException): bot.send_photo(chat_id, open('tests/data/telegram.jpg', 'rb'), timeout=TIMEOUT) # Test JSON submition with pytest.raises(OkException): bot.get_chat_administrators(chat_id, timeout=TIMEOUT) def test_timeout_propagation_implicit(self, monkeypatch, bot, chat_id): from telegram.vendor.ptb_urllib3.urllib3.util.timeout import Timeout class OkException(Exception): pass def request_wrapper(*args, **kwargs): obj = kwargs.get('timeout') if isinstance(obj, Timeout) and obj._read == 20: raise OkException return b'{"ok": true, "result": []}' monkeypatch.setattr('telegram.utils.request.Request._request_wrapper', request_wrapper) # Test file uploading with pytest.raises(OkException): bot.send_photo(chat_id, open('tests/data/telegram.jpg', 'rb')) @flaky(3, 1) @pytest.mark.timeout(10) @pytest.mark.parametrize('default_bot', [{'parse_mode': 'Markdown'}], indirect=True) def test_send_message_default_parse_mode(self, default_bot, chat_id): test_string = 'Italic Bold Code' test_markdown_string = '_Italic_ *Bold* `Code`' message = default_bot.send_message(chat_id, test_markdown_string) assert message.text_markdown == test_markdown_string assert message.text == test_string message = default_bot.send_message(chat_id, test_markdown_string, parse_mode=None) assert message.text == test_markdown_string assert message.text_markdown == escape_markdown(test_markdown_string) message = default_bot.send_message(chat_id, test_markdown_string, parse_mode='HTML') assert message.text == test_markdown_string assert message.text_markdown == escape_markdown(test_markdown_string) @flaky(3, 1) @pytest.mark.timeout(10) def test_set_and_get_my_commands(self, bot): commands = [ BotCommand('cmd1', 'descr1'), BotCommand('cmd2', 'descr2'), ] bot.set_my_commands([]) assert bot.get_my_commands() == [] assert bot.commands == [] assert bot.set_my_commands(commands) for bc in [bot.get_my_commands(), bot.commands]: assert len(bc) == 2 assert bc[0].command == 'cmd1' assert bc[0].description == 'descr1' assert bc[1].command == 'cmd2' assert bc[1].description == 'descr2' @flaky(3, 1) @pytest.mark.timeout(10) def test_set_and_get_my_commands_strings(self, bot): commands = [ ['cmd1', 'descr1'], ['cmd2', 'descr2'], ] bot.set_my_commands([]) assert bot.get_my_commands() == [] assert bot.commands == [] assert bot.set_my_commands(commands) for bc in [bot.get_my_commands(), bot.commands]: assert len(bc) == 2 assert bc[0].command == 'cmd1' assert bc[0].description == 'descr1' assert bc[1].command == 'cmd2' assert bc[1].description == 'descr2'