mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-11-21 22:56:38 +01:00
tests: refactor `test_commandhandler.py` (#1408)
- Improved usage of fixtures
- Replaced fixtures for directly callable factories where
multiple mock objects were needed in the same test function
- Extracted fixtures where possible (in place of literals or
global constants)
- Moved some fixtures to ``conftest.py`` to be used by other
modules
- Made a common base class for both ``TestCommandHandler`` and
``TestPrefixHandler``, extracting common methods, patterns and
signatures
- The extracted patterns in test methods have been named with
leading ``_test``
- Extracted other repeatedly used test utilities into functions
(e.g. ``is_match``) and methods (e.g. ``make_default_handler``)
This commit is contained in:
Paolo Lammens
Noam Meltzer
parent
aadb6df271
commit
3318239cf6
3 changed files with 378 additions and 546 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in a new issue