diff --git a/telegram/ext/handler.py b/telegram/ext/handler.py index 2201943ff..3c3374250 100644 --- a/telegram/ext/handler.py +++ b/telegram/ext/handler.py @@ -151,14 +151,11 @@ class Handler(object): optional_args['update_queue'] = dispatcher.update_queue if self.pass_job_queue: optional_args['job_queue'] = dispatcher.job_queue - if self.pass_user_data or self.pass_chat_data: - chat = update.effective_chat + if self.pass_user_data: user = update.effective_user - - if self.pass_user_data: - optional_args['user_data'] = dispatcher.user_data[user.id if user else None] - - if self.pass_chat_data: - optional_args['chat_data'] = dispatcher.chat_data[chat.id if chat else None] + optional_args['user_data'] = dispatcher.user_data[user.id if user else None] + if self.pass_chat_data: + chat = update.effective_chat + optional_args['chat_data'] = dispatcher.chat_data[chat.id if chat else None] return optional_args diff --git a/tests/conftest.py b/tests/conftest.py index b643a5165..c702b0f63 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,8 +16,10 @@ # # 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 datetime import os import sys +import re from collections import defaultdict from queue import Queue from threading import Thread, Event @@ -25,8 +27,9 @@ from time import sleep import pytest -from telegram import Bot -from telegram.ext import Dispatcher, JobQueue, Updater +from telegram import Bot, Message, User, Chat, MessageEntity, Update, \ + InlineQuery, CallbackQuery, ShippingQuery, PreCheckoutQuery, ChosenInlineResult +from telegram.ext import Dispatcher, JobQueue, Updater, BaseFilter from tests.bots import get_bot TRAVIS = os.getenv('TRAVIS', False) @@ -46,7 +49,7 @@ def bot_info(): @pytest.fixture(scope='session') def bot(bot_info): - return Bot(bot_info['token'], private_key=PRIVATE_KEY) + return make_bot(bot_info) @pytest.fixture(scope='session') @@ -146,3 +149,107 @@ def pytest_configure(config): if sys.version_info >= (3,): config.addinivalue_line('filterwarnings', 'ignore::ResourceWarning') # TODO: Write so good code that we don't need to ignore ResourceWarnings anymore + + +def make_bot(bot_info): + return Bot(bot_info['token'], private_key=PRIVATE_KEY) + + +CMD_PATTERN = re.compile(r'/[\da-z_]{1,32}(?:@\w{1,32})?') +DATE = datetime.datetime.now() + + +def make_message(text, **kwargs): + """ + Testing utility factory to create a fake ``telegram.Message`` with + reasonable defaults for mimicking a real message. + :param text: (str) message text + :return: a (fake) ``telegram.Message`` + """ + return Message(message_id=1, + from_user=kwargs.pop('user', User(id=1, first_name='', is_bot=False)), + date=kwargs.pop('date', DATE), + chat=kwargs.pop('chat', Chat(id=1, type='')), + text=text, + bot=kwargs.pop('bot', make_bot(get_bot())), + **kwargs) + + +def make_command_message(text, **kwargs): + """ + Testing utility factory to create a message containing a single telegram + command. + Mimics the Telegram API in that it identifies commands within the message + and tags the returned ``Message`` object with the appropriate ``MessageEntity`` + tag (but it does this only for commands). + + :param text: (str) message text containing (or not) the command + :return: a (fake) ``telegram.Message`` containing only the command + """ + + match = re.search(CMD_PATTERN, text) + entities = [MessageEntity(type=MessageEntity.BOT_COMMAND, + offset=match.start(0), + length=len(match.group(0)))] if match else [] + + return make_message(text, entities=entities, **kwargs) + + +def make_message_update(message, message_factory=make_message, edited=False, **kwargs): + """ + Testing utility factory to create an update from a message, as either a + ``telegram.Message`` or a string. In the latter case ``message_factory`` + is used to convert ``message`` to a ``telegram.Message``. + :param message: either a ``telegram.Message`` or a string with the message text + :param message_factory: function to convert the message text into a ``telegram.Message`` + :param edited: whether the message should be stored as ``edited_message`` (vs. ``message``) + :return: ``telegram.Update`` with the given message + """ + if not isinstance(message, Message): + message = message_factory(message, **kwargs) + update_kwargs = {'message' if not edited else 'edited_message': message} + return Update(0, **update_kwargs) + + +def make_command_update(message, edited=False, **kwargs): + """ + Testing utility factory to create an update from a message that potentially + contains a command. See ``make_command_message`` for more details. + :param message: message potentially containing a command + :param edited: whether the message should be stored as ``edited_message`` (vs. ``message``) + :return: ``telegram.Update`` with the given message + """ + return make_message_update(message, make_command_message, edited, **kwargs) + + +@pytest.fixture(scope='function') +def mock_filter(): + class MockFilter(BaseFilter): + def __init__(self): + self.tested = False + + def filter(self, message): + self.tested = True + + return MockFilter() + + +def get_false_update_fixture_decorator_params(): + message = Message(1, User(1, '', False), DATE, Chat(1, ''), text='test') + params = [ + {'callback_query': CallbackQuery(1, User(1, '', False), 'chat', message=message)}, + {'channel_post': message}, + {'edited_channel_post': message}, + {'inline_query': InlineQuery(1, User(1, '', False), '', '')}, + {'chosen_inline_result': ChosenInlineResult('id', User(1, '', False), '')}, + {'shipping_query': ShippingQuery('id', User(1, '', False), '', None)}, + {'pre_checkout_query': PreCheckoutQuery('id', User(1, '', False), '', 0, '')}, + {'callback_query': CallbackQuery(1, User(1, '', False), 'chat')} + ] + ids = tuple(key for kwargs in params for key in kwargs) + return {'params': params, 'ids': ids} + + +@pytest.fixture(scope='function', **get_false_update_fixture_decorator_params()) +def false_update(request): + return Update(update_id=1, **request.param) diff --git a/tests/test_commandhandler.py b/tests/test_commandhandler.py index b52e383ca..f3c5297d7 100644 --- a/tests/test_commandhandler.py +++ b/tests/test_commandhandler.py @@ -21,50 +21,30 @@ import re from queue import Queue import pytest +import itertools from telegram.utils.deprecate import TelegramDeprecationWarning -from telegram import (Message, Update, Chat, Bot, User, CallbackQuery, InlineQuery, - ChosenInlineResult, ShippingQuery, PreCheckoutQuery, MessageEntity) -from telegram.ext import CommandHandler, Filters, BaseFilter, CallbackContext, JobQueue, \ - PrefixHandler - -message = Message(1, User(1, '', False), None, Chat(1, ''), text='test') - -params = [ - {'callback_query': CallbackQuery(1, User(1, '', False), 'chat', message=message)}, - {'channel_post': message}, - {'edited_channel_post': message}, - {'inline_query': InlineQuery(1, User(1, '', False), '', '')}, - {'chosen_inline_result': ChosenInlineResult('id', User(1, '', False), '')}, - {'shipping_query': ShippingQuery('id', User(1, '', False), '', None)}, - {'pre_checkout_query': PreCheckoutQuery('id', User(1, '', False), '', 0, '')}, - {'callback_query': CallbackQuery(1, User(1, '', False), 'chat')} -] - -ids = ('callback_query', 'channel_post', 'edited_channel_post', 'inline_query', - 'chosen_inline_result', 'shipping_query', 'pre_checkout_query', - 'callback_query_without_message',) +from telegram import Message, Update, Chat, Bot +from telegram.ext import CommandHandler, Filters, CallbackContext, JobQueue, PrefixHandler +from tests.conftest import make_command_message, make_command_update, make_message, \ + make_message_update -@pytest.fixture(scope='class', params=params, ids=ids) -def false_update(request): - return Update(update_id=1, **request.param) +def is_match(handler, update): + """ + Utility function that returns whether an update matched + against a specific handler. + :param handler: ``CommandHandler`` to check against + :param update: update to check + :return: (bool) whether ``update`` matched with ``handler`` + """ + check = handler.check_update(update) + return check is not None and check is not False -@pytest.fixture(scope='function') -def message(bot): - return Message(message_id=1, - from_user=User(id=1, first_name='', is_bot=False), - date=None, - chat=Chat(id=1, type=''), - message='/test', - bot=bot, - entities=[MessageEntity(type=MessageEntity.BOT_COMMAND, - offset=0, - length=len('/test'))]) - - -class TestCommandHandler(object): +class BaseTest(object): + """Base class for command and prefix handler test classes. Contains + utility methods an several callbacks used by both classes.""" test_flag = False SRE_TYPE = type(re.match("", "")) @@ -72,30 +52,33 @@ class TestCommandHandler(object): def reset(self): self.test_flag = False + PASS_KEYWORDS = ('pass_user_data', 'pass_chat_data', 'pass_job_queue', 'pass_update_queue') + + @pytest.fixture(scope='module', params=itertools.combinations(PASS_KEYWORDS, 2)) + def pass_combination(self, request): + return {key: True for key in request.param} + + def response(self, dispatcher, update): + """ + Utility to send an update to a dispatcher and assert + whether the callback was called appropriately. Its purpose is + for repeated usage in the same test function. + """ + self.test_flag = False + dispatcher.process_update(update) + return self.test_flag + def callback_basic(self, bot, update): test_bot = isinstance(bot, Bot) test_update = isinstance(update, Update) self.test_flag = test_bot and test_update - def callback_data_1(self, bot, update, user_data=None, chat_data=None): - self.test_flag = (user_data is not None) or (chat_data is not None) + def make_callback_for(self, pass_keyword): + def callback(bot, update, **kwargs): + self.test_flag = kwargs.get(keyword, None) is not None - def callback_data_2(self, bot, update, user_data=None, chat_data=None): - self.test_flag = (user_data is not None) and (chat_data is not None) - - def callback_queue_1(self, bot, update, job_queue=None, update_queue=None): - self.test_flag = (job_queue is not None) or (update_queue is not None) - - def callback_queue_2(self, bot, update, job_queue=None, update_queue=None): - self.test_flag = (job_queue is not None) and (update_queue is not None) - - def ch_callback_args(self, bot, update, args): - if update.message.text == '/test': - self.test_flag = len(args) == 0 - elif update.message.text == '/test@{}'.format(bot.username): - self.test_flag = len(args) == 0 - else: - self.test_flag = args == ['one', 'two'] + keyword = pass_keyword[5:] + return callback def callback_context(self, update, context): self.test_flag = (isinstance(context, CallbackContext) @@ -122,550 +105,295 @@ class TestCommandHandler(object): num = len(context.matches) == 2 self.test_flag = types and num - def test_basic(self, dp, message): - handler = CommandHandler('test', self.callback_basic) + def _test_context_args_or_regex(self, cdp, handler, text): + cdp.add_handler(handler) + update = make_command_update(text) + assert not self.response(cdp, update) + update.message.text += ' one two' + assert self.response(cdp, update) + + def _test_edited(self, message, handler_edited, handler_not_edited): + """ + Assert whether a handler that should accept edited messages + and a handler that shouldn't work correctly. + :param message: ``telegram.Message`` to check against the handlers + :param handler_edited: handler that should accept edited messages + :param handler_not_edited: handler that should not accept edited messages + """ + update = make_command_update(message) + edited_update = make_command_update(message, edited=True) + + assert is_match(handler_edited, update) + assert is_match(handler_edited, edited_update) + assert is_match(handler_not_edited, update) + assert not is_match(handler_not_edited, edited_update) + + +# ----------------------------- CommandHandler ----------------------------- + +class TestCommandHandler(BaseTest): + CMD = '/test' + + @pytest.fixture(scope='class') + def command(self): + return self.CMD + + @pytest.fixture(scope='class') + def command_message(self, command): + return make_command_message(command) + + @pytest.fixture(scope='class') + def command_update(self, command_message): + return make_command_update(command_message) + + def ch_callback_args(self, bot, update, args): + if update.message.text == self.CMD: + self.test_flag = len(args) == 0 + elif update.message.text == '{}@{}'.format(self.CMD, bot.username): + self.test_flag = len(args) == 0 + else: + self.test_flag = args == ['one', 'two'] + + def make_default_handler(self, callback=None, **kwargs): + callback = callback or self.callback_basic + return CommandHandler(self.CMD[1:], callback, **kwargs) + + def test_basic(self, dp, command): + """Test whether a command handler responds to its command + and not to others, or badly formatted commands""" + handler = self.make_default_handler() dp.add_handler(handler) - message.text = '/test' - dp.process_update(Update(0, message)) - assert self.test_flag + assert self.response(dp, make_command_update(command)) + assert not is_match(handler, make_command_update(command[1:])) + assert not is_match(handler, make_command_update('/not{}'.format(command[1:]))) + assert not is_match(handler, make_command_update('not {} at start'.format(command))) - message.text = '/nottest' - check = handler.check_update(Update(0, message)) - assert check is None or check is False - - message.text = 'test' - check = handler.check_update(Update(0, message)) - assert check is None or check is False - - message.text = 'not /test at start' - check = handler.check_update(Update(0, message)) - assert check is None or check is False - - message.entities = [] - message.text = '/test' - check = handler.check_update(Update(0, message)) - assert check is None or check is False - - @pytest.mark.parametrize('command', + @pytest.mark.parametrize('cmd', ['way_too_longcommand1234567yes_way_toooooooLong', 'ïñválídletters', 'invalid #&* chars'], ids=['too long', 'invalid letter', 'invalid characters']) - def test_invalid_commands(self, command): + def test_invalid_commands(self, cmd): with pytest.raises(ValueError, match='not a valid bot command'): - CommandHandler(command, self.callback_basic) + CommandHandler(cmd, self.callback_basic) - def test_command_list(self, message): + def test_command_list(self): + """A command handler with multiple commands registered should respond to all of them.""" handler = CommandHandler(['test', 'star'], self.callback_basic) - - message.text = '/test' - check = handler.check_update(Update(0, message)) - - message.text = '/star' - check = handler.check_update(Update(0, message)) - - message.text = '/stop' - check = handler.check_update(Update(0, message)) - assert check is None or check is False + assert is_match(handler, make_command_update('/test')) + assert is_match(handler, make_command_update('/star')) + assert not is_match(handler, make_command_update('/stop')) def test_deprecation_warning(self): + """``allow_edited`` deprecated in favor of filters""" with pytest.warns(TelegramDeprecationWarning, match='See https://git.io/fxJuV'): - CommandHandler('test', self.callback_basic, allow_edited=True) + self.make_default_handler(allow_edited=True) - def test_no_edited(self, message): - handler = CommandHandler('test', self.callback_basic) - message.text = '/test' - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False + def test_edited(self, command_message): + """Test that a CH responds to an edited message iff its filters allow it""" + handler_edited = self.make_default_handler() + handler_no_edited = self.make_default_handler(filters=~Filters.update.edited_message) + self._test_edited(command_message, handler_edited, handler_no_edited) - check = handler.check_update(Update(0, edited_message=message)) - assert check is not None and check is not False + def test_edited_deprecated(self, command_message): + """Test that a CH responds to an edited message iff ``allow_edited`` is True""" + handler_edited = self.make_default_handler(allow_edited=True) + handler_no_edited = self.make_default_handler(allow_edited=False) + self._test_edited(command_message, handler_edited, handler_no_edited) - handler = CommandHandler('test', self.callback_basic, - filters=~Filters.update.edited_message) - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False + def test_directed_commands(self, bot, command): + """Test recognition of commands with a mention to the bot""" + handler = self.make_default_handler() + assert is_match(handler, make_command_update(command + '@' + bot.username, bot=bot)) + assert not is_match(handler, make_command_update(command + '@otherbot', bot=bot)) - check = handler.check_update(Update(0, edited_message=message)) - assert check is None or check is False + def test_with_filter(self, command): + """Test that a CH with a (generic) filter responds iff its filters match""" + handler = self.make_default_handler(filters=Filters.group) + assert is_match(handler, make_command_update(command, chat=Chat(-23, Chat.GROUP))) + assert not is_match(handler, make_command_update(command, chat=Chat(23, Chat.PRIVATE))) - def test_edited_deprecated(self, message): - handler = CommandHandler('test', self.callback_basic, - allow_edited=False) - message.text = '/test' - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False - - check = handler.check_update(Update(0, edited_message=message)) - assert check is None or check is False - - handler = CommandHandler('test', self.callback_basic, - allow_edited=True) - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False - - check = handler.check_update(Update(0, edited_message=message)) - assert check is not None and check is not False - - def test_directed_commands(self, message): - handler = CommandHandler('test', self.callback_basic) - - message.text = '/test@{}'.format(message.bot.username) - message.entities[0].length = len(message.text) - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False - - message.text = '/test@otherbot' - check = handler.check_update(Update(0, message)) - assert check is None or check is False - - def test_with_filter(self, message): - handler = CommandHandler('test', self.callback_basic, Filters.group) - - message.chat = Chat(-23, 'group') - message.text = '/test' - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False - - message.chat = Chat(23, 'private') - check = handler.check_update(Update(0, message)) - assert check is None or check is False - - def test_pass_args(self, dp, message): - handler = CommandHandler('test', self.ch_callback_args, pass_args=True) + def test_pass_args(self, dp, bot, command): + """Test the passing of arguments alongside a command""" + handler = self.make_default_handler(self.ch_callback_args, pass_args=True) dp.add_handler(handler) + at_command = '{}@{}'.format(command, bot.username) + assert self.response(dp, make_command_update(command)) + assert self.response(dp, make_command_update(command + ' one two')) + assert self.response(dp, make_command_update(at_command, bot=bot)) + assert self.response(dp, make_command_update(at_command + ' one two', bot=bot)) - message.text = '/test' - dp.process_update(Update(0, message=message)) - assert self.test_flag - - self.test_flag = False - message.text = '/test@{}'.format(message.bot.username) - message.entities[0].length = len(message.text) - dp.process_update(Update(0, message=message)) - assert self.test_flag - - self.test_flag = False - message.text = '/test@{} one two'.format(message.bot.username) - dp.process_update(Update(0, message=message)) - assert self.test_flag - - self.test_flag = False - message.text = '/test one two' - message.entities[0].length = len('/test') - dp.process_update(Update(0, message=message)) - assert self.test_flag - - def test_newline(self, dp, message): - handler = CommandHandler('test', self.callback_basic) + def test_newline(self, dp, command): + """Assert that newlines don't interfere with a command handler matching a message""" + handler = self.make_default_handler() dp.add_handler(handler) + update = make_command_update(command + '\nfoobar') + assert is_match(handler, update) + assert self.response(dp, update) - message.text = '/test\nfoobar' - check = handler.check_update(Update(0, message)) - assert check is not None and check is not False - - dp.process_update(Update(0, message)) - assert self.test_flag - - def test_pass_user_or_chat_data(self, dp, message): - handler = CommandHandler('test', self.callback_data_1, - pass_user_data=True) + @pytest.mark.parametrize('pass_keyword', BaseTest.PASS_KEYWORDS) + def test_pass_data(self, dp, command_update, pass_combination, pass_keyword): + handler = CommandHandler('test', self.make_callback_for(pass_keyword), **pass_combination) dp.add_handler(handler) - - message.text = '/test' - dp.process_update(Update(0, message=message)) - assert self.test_flag - - dp.remove_handler(handler) - handler = CommandHandler('test', self.callback_data_1, - pass_chat_data=True) - dp.add_handler(handler) - - self.test_flag = False - dp.process_update(Update(0, message=message)) - assert self.test_flag - - dp.remove_handler(handler) - handler = CommandHandler('test', self.callback_data_2, - pass_chat_data=True, - pass_user_data=True) - dp.add_handler(handler) - - self.test_flag = False - dp.process_update(Update(0, message=message)) - assert self.test_flag - - def test_pass_job_or_update_queue(self, dp, message): - handler = CommandHandler('test', self.callback_queue_1, - pass_job_queue=True) - dp.add_handler(handler) - - message.text = '/test' - dp.process_update(Update(0, message=message)) - assert self.test_flag - - dp.remove_handler(handler) - handler = CommandHandler('test', self.callback_queue_1, - pass_update_queue=True) - dp.add_handler(handler) - - self.test_flag = False - dp.process_update(Update(0, message=message)) - assert self.test_flag - - dp.remove_handler(handler) - handler = CommandHandler('test', self.callback_queue_2, - pass_job_queue=True, - pass_update_queue=True) - dp.add_handler(handler) - - self.test_flag = False - dp.process_update(Update(0, message=message)) - assert self.test_flag + assert self.response(dp, command_update) == pass_combination.get(pass_keyword, False) def test_other_update_types(self, false_update): - handler = CommandHandler('test', self.callback_basic) - check = handler.check_update(false_update) - assert check is None or check is False + """Test that a command handler doesn't respond to unrelated updates""" + handler = self.make_default_handler() + assert not is_match(handler, false_update) - def test_filters_for_wrong_command(self, message): + def test_filters_for_wrong_command(self, mock_filter): """Filters should not be executed if the command does not match the handler""" + handler = self.make_default_handler(filters=mock_filter) + assert not is_match(handler, make_command_update('/star')) + assert not mock_filter.tested - class TestFilter(BaseFilter): - def __init__(self): - self.tested = False - - def filter(self, message): - self.tested = True - - test_filter = TestFilter() - - handler = CommandHandler('test', self.callback_basic, - filters=test_filter) - message.text = '/star' - - check = handler.check_update(Update(0, message=message)) - assert check is None or check is False - - assert not test_filter.tested - - def test_context(self, cdp, message): - handler = CommandHandler('test', self.callback_context) + def test_context(self, cdp, command_update): + """Test correct behaviour of CHs with context-based callbacks""" + handler = self.make_default_handler(self.callback_context) cdp.add_handler(handler) + assert self.response(cdp, command_update) - message.text = '/test' - cdp.process_update(Update(0, message)) - assert self.test_flag + def test_context_args(self, cdp, command): + """Test CHs that pass arguments through ``context``""" + handler = self.make_default_handler(self.callback_context_args) + self._test_context_args_or_regex(cdp, handler, command) - def test_context_args(self, cdp, message): - handler = CommandHandler('test', self.callback_context_args) - cdp.add_handler(handler) + def test_context_regex(self, cdp, command): + """Test CHs with context-based callbacks and a single filter""" + handler = self.make_default_handler(self.callback_context_regex1, + filters=Filters.regex('one two')) + self._test_context_args_or_regex(cdp, handler, command) - message.text = '/test' - cdp.process_update(Update(0, message)) - assert not self.test_flag - - message.text = '/test one two' - cdp.process_update(Update(0, message)) - assert self.test_flag - - def test_context_regex(self, cdp, message): - handler = CommandHandler('test', self.callback_context_regex1, Filters.regex('one two')) - cdp.add_handler(handler) - - message.text = '/test' - cdp.process_update(Update(0, message)) - assert not self.test_flag - - message.text += ' one two' - cdp.process_update(Update(0, message)) - assert self.test_flag - - def test_context_multiple_regex(self, cdp, message): - handler = CommandHandler('test', self.callback_context_regex2, - Filters.regex('one') & Filters.regex('two')) - cdp.add_handler(handler) - - message.text = '/test' - cdp.process_update(Update(0, message)) - assert not self.test_flag - - message.text += ' one two' - cdp.process_update(Update(0, message)) - assert self.test_flag + def test_context_multiple_regex(self, cdp, command): + """Test CHs with context-based callbacks and filters combined""" + handler = self.make_default_handler(self.callback_context_regex2, + filters=Filters.regex('one') & Filters.regex('two')) + self._test_context_args_or_regex(cdp, handler, command) -par = ['!help', '!test', '#help', '#test', 'mytrig-help', 'mytrig-test'] +# ----------------------------- PrefixHandler ----------------------------- + +def combinations(prefixes, commands): + return (prefix + command for prefix in prefixes for command in commands) -@pytest.fixture(scope='function', params=par) -def prefixmessage(bot, request): - return Message(message_id=1, - from_user=User(id=1, first_name='', is_bot=False), - date=None, - chat=Chat(id=1, type=''), - text=request.param, - bot=bot) +class TestPrefixHandler(BaseTest): + # Prefixes and commands with which to test PrefixHandler: + PREFIXES = ['!', '#', 'mytrig-'] + COMMANDS = ['help', 'test'] + COMBINATIONS = list(combinations(PREFIXES, COMMANDS)) + @pytest.fixture(scope='class', params=PREFIXES) + def prefix(self, request): + return request.param -class TestPrefixHandler(object): - test_flag = False - SRE_TYPE = type(re.match("", "")) + @pytest.fixture(scope='class', params=[1, 2], ids=['single prefix', 'multiple prefixes']) + def prefixes(self, request): + return TestPrefixHandler.PREFIXES[:request.param] - @pytest.fixture(autouse=True) - def reset(self): - self.test_flag = False + @pytest.fixture(scope='class', params=COMMANDS) + def command(self, request): + return request.param - def callback_basic(self, bot, update): - test_bot = isinstance(bot, Bot) - test_update = isinstance(update, Update) - self.test_flag = test_bot and test_update + @pytest.fixture(scope='class', params=[1, 2], ids=['single command', 'multiple commands']) + def commands(self, request): + return TestPrefixHandler.COMMANDS[:request.param] - def callback_data_1(self, bot, update, user_data=None, chat_data=None): - self.test_flag = (user_data is not None) or (chat_data is not None) + @pytest.fixture(scope='class') + def prefix_message_text(self, prefix, command): + return prefix + command - def callback_data_2(self, bot, update, user_data=None, chat_data=None): - self.test_flag = (user_data is not None) and (chat_data is not None) + @pytest.fixture(scope='class') + def prefix_message(self, prefix_message_text): + return make_message(prefix_message_text) - def callback_queue_1(self, bot, update, job_queue=None, update_queue=None): - self.test_flag = (job_queue is not None) or (update_queue is not None) + @pytest.fixture(scope='class') + def prefix_message_update(self, prefix_message): + return make_message_update(prefix_message) - def callback_queue_2(self, bot, update, job_queue=None, update_queue=None): - self.test_flag = (job_queue is not None) and (update_queue is not None) + def make_default_handler(self, callback=None, **kwargs): + callback = callback or self.callback_basic + return PrefixHandler(self.PREFIXES, self.COMMANDS, callback, **kwargs) def ch_callback_args(self, bot, update, args): - if update.message.text in par: + if update.message.text in TestPrefixHandler.COMBINATIONS: self.test_flag = len(args) == 0 else: self.test_flag = args == ['one', 'two'] - def callback_context(self, update, context): - self.test_flag = (isinstance(context, CallbackContext) - and isinstance(context.bot, Bot) - and isinstance(update, Update) - and isinstance(context.update_queue, Queue) - and isinstance(context.job_queue, JobQueue) - and isinstance(context.user_data, dict) - and isinstance(context.chat_data, dict) - and isinstance(update.message, Message)) - - def callback_context_args(self, update, context): - self.test_flag = context.args == ['one', 'two'] - - def callback_context_regex1(self, update, context): - if context.matches: - types = all([type(res) == self.SRE_TYPE for res in context.matches]) - num = len(context.matches) == 1 - self.test_flag = types and num - - def callback_context_regex2(self, update, context): - if context.matches: - types = all([type(res) == self.SRE_TYPE for res in context.matches]) - num = len(context.matches) == 2 - self.test_flag = types and num - - def test_basic(self, dp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_basic) + def test_basic(self, dp, prefix, command): + """Test the basic expected response from a prefix handler""" + handler = self.make_default_handler() dp.add_handler(handler) + text = prefix + command - dp.process_update(Update(0, prefixmessage)) - assert self.test_flag + assert self.response(dp, make_message_update(text)) + assert not is_match(handler, make_message_update(command)) + assert not is_match(handler, make_message_update(prefix + 'notacommand')) + assert not is_match(handler, make_command_update('not {} at start'.format(text))) - prefixmessage.text = 'test' - check = handler.check_update(Update(0, prefixmessage)) - assert check is None or check is False + def test_single_multi_prefixes_commands(self, prefixes, commands, prefix_message_update): + """Test various combinations of prefixes and commands""" + handler = self.make_default_handler() + result = is_match(handler, prefix_message_update) + expected = prefix_message_update.message.text in combinations(prefixes, commands) + return result == expected - prefixmessage.text = '#nocom' - check = handler.check_update(Update(0, prefixmessage)) - assert check is None or check is False + def test_edited(self, prefix_message): + handler_edited = self.make_default_handler() + handler_no_edited = self.make_default_handler(filters=~Filters.update.edited_message) + self._test_edited(prefix_message, handler_edited, handler_no_edited) - message.text = 'not !test at start' - check = handler.check_update(Update(0, message)) - assert check is None or check is False + def test_with_filter(self, prefix_message_text): + handler = self.make_default_handler(filters=Filters.group) + text = prefix_message_text + assert is_match(handler, make_message_update(text, chat=Chat(-23, Chat.GROUP))) + assert not is_match(handler, make_message_update(text, chat=Chat(23, Chat.PRIVATE))) - def test_single_prefix_single_command(self, prefixmessage): - handler = PrefixHandler('!', 'test', self.callback_basic) - - check = handler.check_update(Update(0, prefixmessage)) - if prefixmessage.text in ['!test']: - assert check is not None and check is not False - else: - assert check is None or check is False - - def test_single_prefix_multi_command(self, prefixmessage): - handler = PrefixHandler('!', ['test', 'help'], self.callback_basic) - - check = handler.check_update(Update(0, prefixmessage)) - if prefixmessage.text in ['!test', '!help']: - assert check is not None and check is not False - else: - assert check is None or check is False - - def test_multi_prefix_single_command(self, prefixmessage): - handler = PrefixHandler(['!', '#'], 'test', self.callback_basic) - - check = handler.check_update(Update(0, prefixmessage)) - if prefixmessage.text in ['!test', '#test']: - assert check is not None and check is not False - else: - assert check is None or check is False - - def test_no_edited(self, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_basic) - check = handler.check_update(Update(0, prefixmessage)) - assert check is not None and check is not False - - check = handler.check_update(Update(0, edited_message=prefixmessage)) - assert check is not None and check is not False - - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_basic, - filters=~Filters.update.edited_message) - check = handler.check_update(Update(0, prefixmessage)) - assert check is not None and check is not False - - check = handler.check_update(Update(0, edited_message=prefixmessage)) - assert check is None or check is False - - def test_with_filter(self, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_basic, - filters=Filters.group) - - prefixmessage.chat = Chat(-23, 'group') - check = handler.check_update(Update(0, prefixmessage)) - assert check is not None and check is not False - - prefixmessage.chat = Chat(23, 'private') - check = handler.check_update(Update(0, prefixmessage)) - assert check is None or check is False - - def test_pass_args(self, dp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.ch_callback_args, - pass_args=True) + def test_pass_args(self, dp, prefix_message): + handler = self.make_default_handler(self.ch_callback_args, pass_args=True) dp.add_handler(handler) + assert self.response(dp, make_message_update(prefix_message)) - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag + update_with_args = make_message_update(prefix_message.text + ' one two') + assert self.response(dp, update_with_args) - self.test_flag = False - prefixmessage.text += ' one two' - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag - - def test_pass_user_or_chat_data(self, dp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_data_1, - pass_user_data=True) + @pytest.mark.parametrize('pass_keyword', BaseTest.PASS_KEYWORDS) + def test_pass_data(self, dp, pass_combination, prefix_message_update, pass_keyword): + """Assert that callbacks receive data iff its corresponding ``pass_*`` kwarg is enabled""" + handler = self.make_default_handler(self.make_callback_for(pass_keyword), + **pass_combination) dp.add_handler(handler) - - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag - - dp.remove_handler(handler) - self.test_flag = False - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_data_1, - pass_chat_data=True) - dp.add_handler(handler) - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag - - dp.remove_handler(handler) - self.test_flag = False - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_data_2, - pass_chat_data=True, pass_user_data=True) - dp.add_handler(handler) - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag - - def test_pass_job_or_update_queue(self, dp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_queue_1, - pass_job_queue=True) - dp.add_handler(handler) - - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag - - dp.remove_handler(handler) - self.test_flag = False - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_queue_1, - pass_update_queue=True) - dp.add_handler(handler) - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag - - dp.remove_handler(handler) - self.test_flag = False - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_queue_2, - pass_job_queue=True, pass_update_queue=True) - dp.add_handler(handler) - dp.process_update(Update(0, message=prefixmessage)) - assert self.test_flag + assert self.response(dp, prefix_message_update) \ + == pass_combination.get(pass_keyword, False) def test_other_update_types(self, false_update): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_basic) - check = handler.check_update(false_update) - assert check is None or check is False + handler = self.make_default_handler() + assert not is_match(handler, false_update) - def test_filters_for_wrong_command(self, prefixmessage): + def test_filters_for_wrong_command(self, mock_filter): """Filters should not be executed if the command does not match the handler""" + handler = self.make_default_handler(filters=mock_filter) + assert not is_match(handler, make_message_update('/test')) + assert not mock_filter.tested - class TestFilter(BaseFilter): - def __init__(self): - self.tested = False - - def filter(self, message): - self.tested = True - - test_filter = TestFilter() - - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_basic, - filters=test_filter) - - prefixmessage.text = '/star' - - check = handler.check_update(Update(0, message=prefixmessage)) - assert check is None or check is False - - assert not test_filter.tested - - def test_context(self, cdp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], self.callback_context) + def test_context(self, cdp, prefix_message_update): + handler = self.make_default_handler(self.callback_context) cdp.add_handler(handler) + assert self.response(cdp, prefix_message_update) - cdp.process_update(Update(0, prefixmessage)) - assert self.test_flag + def test_context_args(self, cdp, prefix_message_text): + handler = self.make_default_handler(self.callback_context_args) + self._test_context_args_or_regex(cdp, handler, prefix_message_text) - def test_context_args(self, cdp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], - self.callback_context_args) - cdp.add_handler(handler) + def test_context_regex(self, cdp, prefix_message_text): + handler = self.make_default_handler(self.callback_context_regex1, + filters=Filters.regex('one two')) + self._test_context_args_or_regex(cdp, handler, prefix_message_text) - cdp.process_update(Update(0, prefixmessage)) - assert not self.test_flag - - prefixmessage.text += ' one two' - cdp.process_update(Update(0, prefixmessage)) - assert self.test_flag - - def test_context_regex(self, cdp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], - self.callback_context_regex1, Filters.regex('one two')) - cdp.add_handler(handler) - - cdp.process_update(Update(0, prefixmessage)) - assert not self.test_flag - - prefixmessage.text += ' one two' - cdp.process_update(Update(0, prefixmessage)) - assert self.test_flag - - def test_context_multiple_regex(self, cdp, prefixmessage): - handler = PrefixHandler(['!', '#', 'mytrig-'], ['help', 'test'], - self.callback_context_regex2, - Filters.regex('one') & Filters.regex('two')) - cdp.add_handler(handler) - - cdp.process_update(Update(0, prefixmessage)) - assert not self.test_flag - - prefixmessage.text += ' one two' - cdp.process_update(Update(0, prefixmessage)) - assert self.test_flag + def test_context_multiple_regex(self, cdp, prefix_message_text): + handler = self.make_default_handler(self.callback_context_regex2, + filters=Filters.regex('one') & Filters.regex( + 'two')) + self._test_context_args_or_regex(cdp, handler, prefix_message_text)