-
Notifications
You must be signed in to change notification settings - Fork 6k
Expand file tree
/
Copy pathtest_commandhandler.py
More file actions
459 lines (376 loc) · 19.6 KB
/
test_commandhandler.py
File metadata and controls
459 lines (376 loc) · 19.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
#!/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/].
import re
from queue import Queue
import pytest
import itertools
from telegram.utils.deprecate import TelegramDeprecationWarning
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,
)
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
class BaseTest:
"""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("", ""))
@pytest.fixture(autouse=True)
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 make_callback_for(self, pass_keyword):
def callback(bot, update, **kwargs):
self.test_flag = kwargs.get(keyword) is not None
keyword = pass_keyword[5:]
return callback
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(context.bot_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) is 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) is self.SRE_TYPE for res in context.matches)
num = len(context.matches) == 2
self.test_flag = types and num
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'
def test_slot_behaviour(self, recwarn, mro_slots):
handler = self.make_default_handler()
for attr in handler.__slots__:
assert getattr(handler, attr, 'err') != 'err', f"got extra slot '{attr}'"
assert not handler.__dict__, f"got missing slot(s): {handler.__dict__}"
assert len(mro_slots(handler)) == len(set(mro_slots(handler))), "duplicate slot"
handler.custom, handler.command = 'should give warning', self.CMD
assert len(recwarn) == 1 and 'custom' in str(recwarn[0].message), recwarn.list
@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 == f'{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)
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(f'/not{command[1:]}'))
assert not is_match(handler, make_command_update(f'not {command} at start'))
@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, cmd):
with pytest.raises(ValueError, match='not a valid bot command'):
CommandHandler(cmd, self.callback_basic)
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)
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://github.com/python-telegram-bot/python-telegram-bot/wiki'
'/Transition-guide-to-Version-12.0',
):
self.make_default_handler(allow_edited=True)
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)
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)
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))
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_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 = f'{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))
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)
@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)
assert self.response(dp, command_update) == pass_combination.get(pass_keyword, False)
def test_other_update_types(self, false_update):
"""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, 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
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)
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_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)
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)
# ----------------------------- 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, recwarn):
handler = self.make_default_handler()
for attr in handler.__slots__:
assert getattr(handler, attr, 'err') != 'err', f"got extra slot '{attr}'"
assert not handler.__dict__, f"got missing slot(s): {handler.__dict__}"
assert len(mro_slots(handler)) == len(set(mro_slots(handler))), "duplicate slot"
handler.custom, handler.command = 'should give warning', self.COMMANDS
assert len(recwarn) == 1 and 'custom' in str(recwarn[0].message), recwarn.list
@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)
def ch_callback_args(self, bot, update, args):
if update.message.text in TestPrefixHandler.COMBINATIONS:
self.test_flag = len(args) == 0
else:
self.test_flag = args == ['one', 'two']
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
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(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.update.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.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_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))
update_with_args = make_message_update(prefix_message.text + ' one two')
assert self.response(dp, update_with_args)
@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)
assert self.response(dp, prefix_message_update) == pass_combination.get(
pass_keyword, False
)
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']))
def test_basic_after_editing(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
assert self.response(dp, make_message_update(text))
handler.command = 'foo'
text = prefix + 'foo'
assert self.response(dp, make_message_update(text))
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)
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_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)
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)