OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2024-12-22 06:25:12 +01:00
Code Issues Projects Releases Packages Wiki Activity

Split {Command, Prefix}Handler And Make Attributes Immutable (#3045)

Browse source
This commit is contained in:
Bibo-Joshi 2022-05-26 11:10:00 +02:00 committed by GitHub
parent 349baa0202
commit dc13b69dac
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 365 additions and 315 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
telegram/ext/__init__.py
View file

@ -65,7 +65,7 @@ from ._callbackqueryhandler import CallbackQueryHandler
from ._chatjoinrequesthandler import ChatJoinRequestHandler
from ._chatmemberhandler import ChatMemberHandler
from ._choseninlineresulthandler import ChosenInlineResultHandler
from ._commandhandler import CommandHandler, PrefixHandler
from ._commandhandler import CommandHandler
from ._contexttypes import ContextTypes
from ._conversationhandler import ConversationHandler
from ._defaults import Defaults
@ -79,6 +79,7 @@ from ._picklepersistence import PicklePersistence
from ._pollanswerhandler import PollAnswerHandler
from ._pollhandler import PollHandler
from ._precheckoutqueryhandler import PreCheckoutQueryHandler
from ._prefixhandler import PrefixHandler
from ._shippingqueryhandler import ShippingQueryHandler
from ._stringcommandhandler import StringCommandHandler
from ._stringregexhandler import StringRegexHandler

188
telegram/ext/_commandhandler.py
View file

@ -52,9 +52,15 @@ class CommandHandler(BaseHandler[Update, CCT]):
When setting :paramref:`block` to :obj:`False`, you cannot rely on adding custom
attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
.. versionchanged:: 20.0
* Renamed the attribute ``command`` to :attr:`commands`, which now is always a
:class:`frozenset`
* Updating the commands this handler listens to is no longer possible.
Args:
command (:obj:`str` | Collection[:obj:`str`]):
The command or list of commands this handler should listen for.
The command or list of commands this handler should listen for. Case-insensitive.
Limitations are the same as described `here <https://core.telegram.org/bots#commands>`_
callback (:term:`coroutine function`): The callback function for this handler. Will be
called when :meth:`check_update` has determined that an update should be processed by
@ -76,8 +82,7 @@ class CommandHandler(BaseHandler[Update, CCT]):
:exc:`ValueError`: When the command is too long or has illegal chars.
Attributes:
command (List[:obj:`str`]): The list of commands this handler should listen for.
Limitations are the same as described `here <https://core.telegram.org/bots#commands>`_
commands (FrozenSet[:obj:`str`]): The set of commands this handler should listen for.
callback (:term:`coroutine function`): The callback function for this handler.
filters (:class:`telegram.ext.filters.BaseFilter`): Optional. Only allow updates with these
Filters.
@ -86,7 +91,7 @@ class CommandHandler(BaseHandler[Update, CCT]):
:meth:`telegram.ext.Application.process_update`.
"""
__slots__ = ("command", "filters")
__slots__ = ("commands", "filters")
def __init__(
self,
@ -98,12 +103,13 @@ class CommandHandler(BaseHandler[Update, CCT]):
super().__init__(callback, block=block)
if isinstance(command, str):
self.command = [command.lower()]
commands = frozenset({command.lower()})
else:
self.command = [x.lower() for x in command]
for comm in self.command:
commands = frozenset(x.lower() for x in command)
for comm in commands:
if not re.match(r"^[\da-z_]{1,32}$", comm):
raise ValueError(f"Command `{comm}` is not a valid bot command")
self.commands = commands
self.filters = filters if filters is not None else filters_module.UpdateType.MESSAGES
@ -135,7 +141,7 @@ class CommandHandler(BaseHandler[Update, CCT]):
command_parts.append(message.get_bot().username)
if not (
command_parts[0].lower() in self.command
command_parts[0].lower() in self.commands
and command_parts[1].lower() == message.get_bot().username.lower()
):
return None
@ -160,169 +166,3 @@ class CommandHandler(BaseHandler[Update, CCT]):
context.args = check_result[0]
if isinstance(check_result[1], dict):
context.update(check_result[1])
class PrefixHandler(CommandHandler):
"""BaseHandler class to handle custom prefix commands.
This is an intermediate handler between :class:`MessageHandler` and :class:`CommandHandler`.
It supports configurable commands with the same options as :class:`CommandHandler`. It will
respond to every combination of :attr:`prefix` and :attr:`command`. It will add a :obj:`list`
to the :class:`CallbackContext` named :attr:`CallbackContext.args`. It will contain a list of
strings, which is the text following the command split on single or consecutive whitespace
characters.
Examples:
Single prefix and command:
.. code:: python
PrefixHandler("!", "test", callback) # will respond to '!test'.
Multiple prefixes, single command:
.. code:: python
PrefixHandler(["!", "#"], "test", callback) # will respond to '!test' and '#test'.
Multiple prefixes and commands:
.. code:: python
PrefixHandler(
["!", "#"], ["test", "help"], callback
) # will respond to '!test', '#test', '!help' and '#help'.
By default, the handler listens to messages as well as edited messages. To change this behavior
use :attr:`~filters.UpdateType.EDITED_MESSAGE <telegram.ext.filters.UpdateType.EDITED_MESSAGE>`
Note:
* :class:`PrefixHandler` does *not* handle (edited) channel posts.
Warning:
When setting :paramref:`block` to :obj:`False`, you cannot rely on adding custom
attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
Args:
prefix (:obj:`str` | Collection[:obj:`str`]):
The prefix(es) that will precede :attr:`command`.
command (:obj:`str` | Collection[:obj:`str`]):
The command or list of commands this handler should listen for.
callback (:term:`coroutine function`): The callback function for this handler. Will be
called when :meth:`check_update` has determined that an update should be processed by
this handler. Callback signature::
async def callback(update: Update, context: CallbackContext)
The return value of the callback is usually ignored except for the special case of
:class:`telegram.ext.ConversationHandler`.
filters (:class:`telegram.ext.filters.BaseFilter`, optional): A filter inheriting from
:class:`telegram.ext.filters.BaseFilter`. Standard filters can be found in
:mod:`telegram.ext.filters`. Filters can be combined using bitwise
operators (``&`` for :keyword:`and`, ``|`` for :keyword:`or`, ``~`` for :keyword:`not`)
block (:obj:`bool`, optional): Determines whether the return value of the callback should
be awaited before processing the next handler in
:meth:`telegram.ext.Application.process_update`. Defaults to :obj:`True`.
Attributes:
callback (:term:`coroutine function`): The callback function for this handler.
filters (:class:`telegram.ext.filters.BaseFilter`): Optional. Only allow updates with these
Filters.
block (:obj:`bool`): Determines whether the return value of the callback should be
awaited before processing the next handler in
:meth:`telegram.ext.Application.process_update`.
"""
# 'prefix' is a class property, & 'command' is included in the superclass, so they're left out.
__slots__ = ("_prefix", "_command", "_commands")
def __init__(
self,
prefix: SCT[str],
command: SCT[str],
callback: HandlerCallback[Update, CCT, RT],
filters: filters_module.BaseFilter = None,
block: DVInput[bool] = DEFAULT_TRUE,
):
self._prefix: List[str] = []
self._command: List[str] = []
self._commands: List[str] = []
super().__init__(
"nocommand",
callback,
filters=filters,
block=block,
)
self.prefix = prefix # type: ignore[assignment]
self.command = command # type: ignore[assignment]
self._build_commands()
@property
def prefix(self) -> List[str]:
"""
The prefixes that will precede :attr:`command`.
Returns:
List[:obj:`str`]
"""
return self._prefix
@prefix.setter
def prefix(self, prefix: Union[str, List[str]]) -> None:
if isinstance(prefix, str):
self._prefix = [prefix.lower()]
else:
self._prefix = prefix
self._build_commands()
@property # type: ignore[override]
def command(self) -> List[str]: # type: ignore[override]
"""
The list of commands this handler should listen for.
Returns:
List[:obj:`str`]
"""
return self._command
@command.setter
def command(self, command: Union[str, List[str]]) -> None:
if isinstance(command, str):
self._command = [command.lower()]
else:
self._command = command
self._build_commands()
def _build_commands(self) -> None:
self._commands = [x.lower() + y.lower() for x in self.prefix for y in self.command]
def check_update(
self, update: object
) -> Optional[Union[bool, Tuple[List[str], Optional[Union[bool, Dict]]]]]:
"""Determines whether an update should be passed to this handler's :attr:`callback`.
Args:
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`list`: The list of args for the handler.
"""
if isinstance(update, Update) and update.effective_message:
message = update.effective_message
if message.text:
text_list = message.text.split()
if text_list[0].lower() not in self._commands:
return None
filter_result = self.filters.check_update(update)
if filter_result:
return text_list[1:], filter_result
return False
return None

185
telegram/ext/_prefixhandler.py Normal file
View file

@ -0,0 +1,185 @@
#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2022
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains the PrefixHandler class."""
import itertools
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, TypeVar, Union
from telegram import Update
from telegram._utils.defaultvalue import DEFAULT_TRUE
from telegram._utils.types import SCT, DVInput
from telegram.ext import filters as filters_module
from telegram.ext._handler import BaseHandler
from telegram.ext._utils.types import CCT, HandlerCallback
if TYPE_CHECKING:
from telegram.ext import Application
RT = TypeVar("RT")
class PrefixHandler(BaseHandler[Update, CCT]):
"""BaseHandler class to handle custom prefix commands.
This is an intermediate handler between :class:`MessageHandler` and :class:`CommandHandler`.
It supports configurable commands with the same options as :class:`CommandHandler`. It will
respond to every combination of :paramref:`prefix` and :paramref:`command`.
It will add a :obj:`list` to the :class:`CallbackContext` named :attr:`CallbackContext.args`,
containing a list of strings, which is the text following the command split on single or
consecutive whitespace characters.
Examples:
Single prefix and command:
.. code:: python
PrefixHandler("!", "test", callback) # will respond to '!test'.
Multiple prefixes, single command:
.. code:: python
PrefixHandler(["!", "#"], "test", callback) # will respond to '!test' and '#test'.
Multiple prefixes and commands:
.. code:: python
PrefixHandler(
["!", "#"], ["test", "help"], callback
) # will respond to '!test', '#test', '!help' and '#help'.
By default, the handler listens to messages as well as edited messages. To change this behavior
use :attr:`~filters.UpdateType.EDITED_MESSAGE <telegram.ext.filters.UpdateType.EDITED_MESSAGE>`
Note:
* :class:`PrefixHandler` does *not* handle (edited) channel posts.
Warning:
When setting :paramref:`block` to :obj:`False`, you cannot rely on adding custom
attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
.. versionchanged:: 20.0
* :class:`PrefixHandler` is no longer a subclass of :class:`CommandHandler`.
* Removed the attributes ``command`` and ``prefix``. Instead, the new :attr:`commands`
contains all commands that this handler listens to as a :class:`frozenset`, which
includes the prefixes.
* Updating the prefixes and commands this handler listens to is no longer possible.
Args:
prefix (:obj:`str` | Collection[:obj:`str`]):
The prefix(es) that will precede :paramref:`command`.
command (:obj:`str` | Collection[:obj:`str`]):
The command or list of commands this handler should listen for. Case-insensitive.
callback (:term:`coroutine function`): The callback function for this handler. Will be
called when :meth:`check_update` has determined that an update should be processed by
this handler. Callback signature::
async def callback(update: Update, context: CallbackContext)
The return value of the callback is usually ignored except for the special case of
:class:`telegram.ext.ConversationHandler`.
filters (:class:`telegram.ext.filters.BaseFilter`, optional): A filter inheriting from
:class:`telegram.ext.filters.BaseFilter`. Standard filters can be found in
:mod:`telegram.ext.filters`. Filters can be combined using bitwise
operators (``&`` for :keyword:`and`, ``|`` for :keyword:`or`, ``~`` for :keyword:`not`)
block (:obj:`bool`, optional): Determines whether the return value of the callback should
be awaited before processing the next handler in
:meth:`telegram.ext.Application.process_update`. Defaults to :obj:`True`.
Attributes:
commands (FrozenSet[:obj:`str`]): The commands that this handler will listen for, i.e. the
combinations of :paramref:`prefix` and :paramref:`command`.
callback (:term:`coroutine function`): The callback function for this handler.
filters (:class:`telegram.ext.filters.BaseFilter`): Optional. Only allow updates with these
Filters.
block (:obj:`bool`): Determines whether the return value of the callback should be
awaited before processing the next handler in
:meth:`telegram.ext.Application.process_update`.
"""
# 'prefix' is a class property, & 'command' is included in the superclass, so they're left out.
__slots__ = ("commands", "filters")
def __init__(
self,
prefix: SCT[str],
command: SCT[str],
callback: HandlerCallback[Update, CCT, RT],
filters: filters_module.BaseFilter = None,
block: DVInput[bool] = DEFAULT_TRUE,
):
super().__init__(callback=callback, block=block)
if isinstance(prefix, str):
prefixes = {prefix.lower()}
else:
prefixes = {x.lower() for x in prefix}
if isinstance(command, str):
commands = {command.lower()}
else:
commands = {x.lower() for x in command}
self.commands = frozenset(p + c for p, c in itertools.product(prefixes, commands))
self.filters = filters if filters is not None else filters_module.UpdateType.MESSAGES
def check_update(
self, update: object
) -> Optional[Union[bool, Tuple[List[str], Optional[Union[bool, Dict]]]]]:
"""Determines whether an update should be passed to this handler's :attr:`callback`.
Args:
update (:class:`telegram.Update` | :obj:`object`): Incoming update.
Returns:
:obj:`list`: The list of args for the handler.
"""
if isinstance(update, Update) and update.effective_message:
message = update.effective_message
if message.text:
text_list = message.text.split()
if text_list[0].lower() not in self.commands:
return None
filter_result = self.filters.check_update(update)
if filter_result:
return text_list[1:], filter_result
return False
return None
def collect_additional_context(
self,
context: CCT,
update: Update,
application: "Application",
check_result: Optional[Union[bool, Tuple[List[str], Optional[bool]]]],
) -> None:
"""Add text after the command to :attr:`CallbackContext.args` as list, split on single
whitespaces and add output of data filters to :attr:`CallbackContext` as well.
"""
if isinstance(check_result, tuple):
context.args = check_result[0]
if isinstance(check_result[1], dict):
context.update(check_result[1])

153
tests/test_commandhandler.py
View file

@ -22,13 +22,8 @@ import re
import pytest
from telegram import Bot, Chat, Message, Update
from telegram.ext import CallbackContext, CommandHandler, JobQueue, PrefixHandler, filters
from tests.conftest import (
make_command_message,
make_command_update,
make_message,
make_message_update,
)
from telegram.ext import CallbackContext, CommandHandler, JobQueue, filters
from tests.conftest import make_command_message, make_command_update, make_message_update
def is_match(handler, update):
@ -167,6 +162,17 @@ class TestCommandHandler(BaseTest):
assert not is_match(handler, make_command_update(command[1:], bot=app.bot))
assert not is_match(handler, make_command_update(f"/not{command[1:]}", bot=app.bot))
assert not is_match(handler, make_command_update(f"not {command} at start", bot=app.bot))
assert not is_match(
handler, make_message_update(bot=app.bot, message=None, caption="caption")
)
handler = CommandHandler(["FOO", "bAR"], callback=self.callback)
assert isinstance(handler.commands, frozenset)
assert handler.commands == {"foo", "bar"}
handler = CommandHandler(["FOO"], callback=self.callback)
assert isinstance(handler.commands, frozenset)
assert handler.commands == {"foo"}
@pytest.mark.parametrize(
"cmd",
@ -248,136 +254,3 @@ class TestCommandHandler(BaseTest):
self.callback_regex2, filters=filters.Regex("one") & filters.Regex("two")
)
await self._test_context_args_or_regex(app, handler, command)
# ----------------------------- PrefixHandler -----------------------------
def combinations(prefixes, commands):
return (prefix + command for prefix in prefixes for command in commands)
class TestPrefixHandler(BaseTest):
# Prefixes and commands with which to test PrefixHandler:
PREFIXES = ["!", "#", "mytrig-"]
COMMANDS = ["help", "test"]
COMBINATIONS = list(combinations(PREFIXES, COMMANDS))
def test_slot_behaviour(self, mro_slots):
handler = self.make_default_handler()
for attr in handler.__slots__:
assert getattr(handler, attr, "err") != "err", f"got extra slot '{attr}'"
assert len(mro_slots(handler)) == len(set(mro_slots(handler))), "duplicate slot"
@pytest.fixture(scope="class", params=PREFIXES)
def prefix(self, request):
return request.param
@pytest.fixture(scope="class", params=[1, 2], ids=["single prefix", "multiple prefixes"])
def prefixes(self, request):
return TestPrefixHandler.PREFIXES[: request.param]
@pytest.fixture(scope="class", params=COMMANDS)
def command(self, request):
return request.param
@pytest.fixture(scope="class", params=[1, 2], ids=["single command", "multiple commands"])
def commands(self, request):
return TestPrefixHandler.COMMANDS[: request.param]
@pytest.fixture(scope="class")
def prefix_message_text(self, prefix, command):
return prefix + command
@pytest.fixture(scope="class")
def prefix_message(self, prefix_message_text):
return make_message(prefix_message_text)
@pytest.fixture(scope="class")
def prefix_message_update(self, prefix_message):
return make_message_update(prefix_message)
def make_default_handler(self, callback=None, **kwargs):
callback = callback or self.callback_basic
return PrefixHandler(self.PREFIXES, self.COMMANDS, callback, **kwargs)
async def test_basic(self, app, prefix, command):
"""Test the basic expected response from a prefix handler"""
handler = self.make_default_handler()
app.add_handler(handler)
text = prefix + command
assert await self.response(app, 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(f"not {text} at start"))
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
def test_edited(self, prefix_message):
handler_edited = self.make_default_handler()
handler_no_edited = self.make_default_handler(filters=~filters.UpdateType.EDITED_MESSAGE)
self._test_edited(prefix_message, handler_edited, handler_no_edited)
def test_with_filter(self, prefix_message_text):
handler = self.make_default_handler(filters=filters.ChatType.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_other_update_types(self, false_update):
handler = self.make_default_handler()
assert not is_match(handler, false_update)
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
def test_edit_prefix(self):
handler = self.make_default_handler()
handler.prefix = ["?", "§"]
assert handler._commands == list(combinations(["?", "§"], self.COMMANDS))
handler.prefix = "+"
assert handler._commands == list(combinations(["+"], self.COMMANDS))
def test_edit_command(self):
handler = self.make_default_handler()
handler.command = "foo"
assert handler._commands == list(combinations(self.PREFIXES, ["foo"]))
async def test_basic_after_editing(self, app, prefix, command):
"""Test the basic expected response from a prefix handler"""
handler = self.make_default_handler()
app.add_handler(handler)
text = prefix + command
assert await self.response(app, make_message_update(text))
handler.command = "foo"
text = prefix + "foo"
assert await self.response(app, make_message_update(text))
async def test_context(self, app, prefix_message_update):
handler = self.make_default_handler(self.callback)
app.add_handler(handler)
assert await self.response(app, prefix_message_update)
async def test_context_args(self, app, prefix_message_text):
handler = self.make_default_handler(self.callback_args)
await self._test_context_args_or_regex(app, handler, prefix_message_text)
async def test_context_regex(self, app, prefix_message_text):
handler = self.make_default_handler(self.callback_regex1, filters=filters.Regex("one two"))
await self._test_context_args_or_regex(app, handler, prefix_message_text)
async def test_context_multiple_regex(self, app, prefix_message_text):
handler = self.make_default_handler(
self.callback_regex2, filters=filters.Regex("one") & filters.Regex("two")
)
await self._test_context_args_or_regex(app, handler, prefix_message_text)

151
tests/test_prefixhandler.py Normal file
View file

@ -0,0 +1,151 @@
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2022
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# 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 pytest
from telegram import Chat
from telegram.ext import CallbackContext, PrefixHandler, filters
from tests.conftest import make_command_update, make_message, make_message_update
from tests.test_commandhandler import BaseTest, is_match
def combinations(prefixes, commands):
return (prefix + command for prefix in prefixes for command in commands)
class TestPrefixHandler(BaseTest):
# Prefixes and commands with which to test PrefixHandler:
PREFIXES = ["!", "#", "mytrig-"]
COMMANDS = ["help", "test"]
COMBINATIONS = list(combinations(PREFIXES, COMMANDS))
def test_slot_behaviour(self, mro_slots):
handler = self.make_default_handler()
for attr in handler.__slots__:
assert getattr(handler, attr, "err") != "err", f"got extra slot '{attr}'"
assert len(mro_slots(handler)) == len(set(mro_slots(handler))), "duplicate slot"
@pytest.fixture(scope="class", params=PREFIXES)
def prefix(self, request):
return request.param
@pytest.fixture(scope="class", params=[1, 2], ids=["single prefix", "multiple prefixes"])
def prefixes(self, request):
return TestPrefixHandler.PREFIXES[: request.param]
@pytest.fixture(scope="class", params=COMMANDS)
def command(self, request):
return request.param
@pytest.fixture(scope="class", params=[1, 2], ids=["single command", "multiple commands"])
def commands(self, request):
return TestPrefixHandler.COMMANDS[: request.param]
@pytest.fixture(scope="class")
def prefix_message_text(self, prefix, command):
return prefix + command
@pytest.fixture(scope="class")
def prefix_message(self, prefix_message_text):
return make_message(prefix_message_text)
@pytest.fixture(scope="class")
def prefix_message_update(self, prefix_message):
return make_message_update(prefix_message)
def make_default_handler(self, callback=None, **kwargs):
callback = callback or self.callback_basic
return PrefixHandler(self.PREFIXES, self.COMMANDS, callback, **kwargs)
async def test_basic(self, app, prefix, command):
"""Test the basic expected response from a prefix handler"""
handler = self.make_default_handler()
app.add_handler(handler)
text = prefix + command
assert await self.response(app, 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(f"not {text} at start"))
assert not is_match(
handler, make_message_update(bot=app.bot, message=None, caption="caption")
)
handler = PrefixHandler(prefix=["!", "#"], command="cmd", callback=self.callback)
assert isinstance(handler.commands, frozenset)
assert handler.commands == {"!cmd", "#cmd"}
handler = PrefixHandler(prefix="#", command={"cmd", "bmd"}, callback=self.callback)
assert isinstance(handler.commands, frozenset)
assert handler.commands == {"#cmd", "#bmd"}
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
def test_edited(self, prefix_message):
handler_edited = self.make_default_handler()
handler_no_edited = self.make_default_handler(filters=~filters.UpdateType.EDITED_MESSAGE)
self._test_edited(prefix_message, handler_edited, handler_no_edited)
def test_with_filter(self, prefix_message_text):
handler = self.make_default_handler(filters=filters.ChatType.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_other_update_types(self, false_update):
handler = self.make_default_handler()
assert not is_match(handler, false_update)
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
async def test_context(self, app, prefix_message_update):
handler = self.make_default_handler(self.callback)
app.add_handler(handler)
assert await self.response(app, prefix_message_update)
async def test_context_args(self, app, prefix_message_text):
handler = self.make_default_handler(self.callback_args)
await self._test_context_args_or_regex(app, handler, prefix_message_text)
async def test_context_regex(self, app, prefix_message_text):
handler = self.make_default_handler(self.callback_regex1, filters=filters.Regex("one two"))
await self._test_context_args_or_regex(app, handler, prefix_message_text)
async def test_context_multiple_regex(self, app, prefix_message_text):
handler = self.make_default_handler(
self.callback_regex2, filters=filters.Regex("one") & filters.Regex("two")
)
await self._test_context_args_or_regex(app, handler, prefix_message_text)
def test_collect_additional_context(self, app):
handler = self.make_default_handler(
self.callback_regex2, filters=filters.Regex("one") & filters.Regex("two")
)
context = CallbackContext(application=app)
handler.collect_additional_context(
context=context, update=None, application=app, check_result=None
)
assert context.args is None