mirror/pyTelegramBotAPI
1
0
Fork 0
mirror of https://github.com/eternnoir/pyTelegramBotAPI.git synced 2023-08-10 21:12:57 +03:00
Code Releases Activity

Merge remote-tracking branch 'upstream/master'

Browse Source
This commit is contained in:
Badiboy 2020-04-24 18:19:55 +03:00
commit aac9ce45a3
10 changed files with 569 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
README.md
View File

@ -15,6 +15,7 @@
* [General use of the API](#general-use-of-the-api)
* [Message handlers](#message-handlers)
* [Callback Query handlers](#callback-query-handler)
* [Middleware handlers](#middleware-handler)
* [TeleBot](#telebot)
* [Reply markup](#reply-markup)
* [Inline Mode](#inline-mode)
@ -223,6 +224,24 @@ In bot2.0 update. You can get `callback_query` in update object. In telebot use
def test_callback(call):
logger.info(call)
```
#### Middleware Handler
A middleware handler is a function that allows you to modify requests or the bot context as they pass through the
Telegram to the bot. You can imagine middleware as a chain of logic connection handled before any other handlers are
executed.
```python
@bot.middleware_handler(update_types=['message'])
def modify_message(bot_instance, message):
# modifying the message before it reaches any other handler
message.another_text = message.text + ':changed'
@bot.message_handler(commands=['start'])
def start(message):
# the message is already modified when it reaches message handler
assert message.another_text == message.text + ':changed'
```
There are other examples using middleware handler in the [examples/middleware](examples/middleware) directory.
#### TeleBot
```python

1
requirements.txt
View File

@ -3,3 +3,4 @@ pytest==3.0.2
requests==2.20.0
six==1.9.0
wheel==0.24.0
redis==3.4.1

1
setup.py
View File

@ -20,6 +20,7 @@ setup(name='pyTelegramBotAPI',
install_requires=['requests', 'six'],
extras_require={
'json': 'ujson',
'redis': 'redis>=3.4.1'
},
classifiers=[
'Development Status :: 5 - Production/Stable',

207
telebot/__init__.py
View File

@ -2,8 +2,6 @@
from __future__ import print_function
import logging
import os
import pickle
import re
import sys
import threading
@ -23,6 +21,7 @@ logger.addHandler(console_output_handler)
logger.setLevel(logging.ERROR)
from telebot import apihelper, types, util
from telebot.handler_backends import MemoryHandlerBackend, FileHandlerBackend
"""
Module : telebot
@ -43,64 +42,6 @@ class Handler:
return getattr(self, item)
class Saver:
"""
Class for saving (next step|reply) handlers
"""
def __init__(self, handlers, filename, delay):
self.handlers = handlers
self.filename = filename
self.delay = delay
self.timer = threading.Timer(delay, self.save_handlers)
def start_save_timer(self):
if not self.timer.is_alive():
if self.delay <= 0:
self.save_handlers()
else:
self.timer = threading.Timer(self.delay, self.save_handlers)
self.timer.start()
def save_handlers(self):
self.dump_handlers(self.handlers, self.filename)
def load_handlers(self, filename, del_file_after_loading=True):
tmp = self.return_load_handlers(filename, del_file_after_loading=del_file_after_loading)
if tmp is not None:
self.handlers.update(tmp)
@staticmethod
def dump_handlers(handlers, filename, file_mode="wb"):
dirs = filename.rsplit('/', maxsplit=1)[0]
os.makedirs(dirs, exist_ok=True)
with open(filename + ".tmp", file_mode) as file:
if (apihelper.CUSTOM_SERIALIZER is None):
pickle.dump(handlers, file)
else:
apihelper.CUSTOM_SERIALIZER.dump(handlers, file)
if os.path.isfile(filename):
os.remove(filename)
os.rename(filename + ".tmp", filename)
@staticmethod
def return_load_handlers(filename, del_file_after_loading=True):
if os.path.isfile(filename) and os.path.getsize(filename) > 0:
with open(filename, "rb") as file:
if (apihelper.CUSTOM_SERIALIZER is None):
handlers = pickle.load(file)
else:
handlers = apihelper.CUSTOM_SERIALIZER.load(file)
if del_file_after_loading:
os.remove(filename)
return handlers
class TeleBot:
""" This is TeleBot Class
Methods:
@ -117,6 +58,7 @@ class TeleBot:
sendVideoNote
sendLocation
sendChatAction
sendDice
getUserProfilePhotos
getUpdates
getFile
@ -141,7 +83,10 @@ class TeleBot:
answerInlineQuery
"""
def __init__(self, token, threaded=True, skip_pending=False, num_threads=2):
def __init__(
self, token, threaded=True, skip_pending=False, num_threads=2,
next_step_backend=None, reply_backend=None
):
"""
:param token: bot API token
:return: Telebot object.
@ -155,14 +100,13 @@ class TeleBot:
self.last_update_id = 0
self.exc_info = None
# key: message_id, value: handler list
self.reply_handlers = {}
self.next_step_backend = next_step_backend
if not self.next_step_backend:
self.next_step_backend = MemoryHandlerBackend()
# key: chat_id, value: handler list
self.next_step_handlers = {}
self.next_step_saver = None
self.reply_saver = None
self.reply_backend = reply_backend
if not self.reply_backend:
self.reply_backend = MemoryHandlerBackend()
self.message_handlers = []
self.edited_message_handlers = []
@ -196,51 +140,89 @@ class TeleBot:
def enable_save_next_step_handlers(self, delay=120, filename="./.handler-saves/step.save"):
"""
Enable saving next step handlers (by default saving disable)
Enable saving next step handlers (by default saving disabled)
This function explicitly assigns FileHandlerBackend (instead of Saver) just to keep backward
compatibility whose purpose was to enable file saving capability for handlers. And the same
implementation is now available with FileHandlerBackend
Most probably this function should be deprecated in future major releases
:param delay: Delay between changes in handlers and saving
:param filename: Filename of save file
"""
self.next_step_saver = Saver(self.next_step_handlers, filename, delay)
self.next_step_backend = FileHandlerBackend(self.next_step_backend.handlers, filename, delay)
def enable_save_reply_handlers(self, delay=120, filename="./.handler-saves/reply.save"):
"""
Enable saving reply handlers (by default saving disable)
This function explicitly assigns FileHandlerBackend (instead of Saver) just to keep backward
compatibility whose purpose was to enable file saving capability for handlers. And the same
implementation is now available with FileHandlerBackend
Most probably this function should be deprecated in future major releases
:param delay: Delay between changes in handlers and saving
:param filename: Filename of save file
"""
self.reply_saver = Saver(self.reply_handlers, filename, delay)
self.reply_backend = FileHandlerBackend(self.reply_backend.handlers, filename, delay)
def disable_save_next_step_handlers(self):
"""
Disable saving next step handlers (by default saving disable)
This function is left to keep backward compatibility whose purpose was to disable file saving capability
for handlers. For the same purpose, MemoryHandlerBackend is reassigned as a new next_step_backend backend
instead of FileHandlerBackend.
Most probably this function should be deprecated in future major releases
"""
self.next_step_saver = None
self.next_step_backend = MemoryHandlerBackend(self.next_step_backend.handlers)
def disable_save_reply_handlers(self):
"""
Disable saving next step handlers (by default saving disable)
This function is left to keep backward compatibility whose purpose was to disable file saving capability
for handlers. For the same purpose, MemoryHandlerBackend is reassigned as a new reply_backend backend
instead of FileHandlerBackend.
Most probably this function should be deprecated in future major releases
"""
self.reply_saver = None
self.reply_backend = MemoryHandlerBackend(self.reply_backend.handlers)
def load_next_step_handlers(self, filename="./.handler-saves/step.save", del_file_after_loading=True):
"""
Load next step handlers from save file
This function is left to keep backward compatibility whose purpose was to load handlers from file with the
help of FileHandlerBackend and is only recommended to use if next_step_backend was assigned as
FileHandlerBackend before entering this function
Most probably this function should be deprecated in future major releases
:param filename: Filename of the file where handlers was saved
:param del_file_after_loading: Is passed True, after loading save file will be deleted
"""
self.next_step_saver.load_handlers(filename, del_file_after_loading)
self.next_step_backend: FileHandlerBackend
self.next_step_backend.load_handlers(filename, del_file_after_loading)
def load_reply_handlers(self, filename="./.handler-saves/reply.save", del_file_after_loading=True):
"""
Load reply handlers from save file
This function is left to keep backward compatibility whose purpose was to load handlers from file with the
help of FileHandlerBackend and is only recommended to use if reply_backend was assigned as
FileHandlerBackend before entering this function
Most probably this function should be deprecated in future major releases
:param filename: Filename of the file where handlers was saved
:param del_file_after_loading: Is passed True, after loading save file will be deleted
"""
self.reply_saver.load_handlers(filename)
self.reply_backend: FileHandlerBackend
self.reply_backend.load_handlers(filename, del_file_after_loading)
def set_webhook(self, url=None, certificate=None, max_connections=None, allowed_updates=None):
return apihelper.set_webhook(self.token, url, certificate, max_connections, allowed_updates)
@ -397,7 +379,7 @@ class TeleBot:
def process_middlewares(self, update):
for update_type, middlewares in self.typed_middleware_handlers.items():
if hasattr(update, update_type):
if getattr(update, update_type) is not None:
for typed_middleware_handler in middlewares:
typed_middleware_handler(self, getattr(update, update_type))
@ -665,6 +647,19 @@ class TeleBot:
"""
return apihelper.delete_message(self.token, chat_id, message_id)
def send_dice(self, chat_id, disable_notification=None, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send dices.
:param chat_id:
:param disable_notification:
:param reply_to_message_id:
:param reply_markup:
:return: Message
"""
return types.Message.de_json(
apihelper.send_dice(self.token, chat_id, disable_notification, reply_to_message_id, reply_markup)
)
def send_photo(self, chat_id, photo, caption=None, reply_to_message_id=None, reply_markup=None,
parse_mode=None, disable_notification=None):
"""
@ -937,7 +932,7 @@ class TeleBot:
def restrict_chat_member(self, chat_id, user_id, until_date=None, can_send_messages=None,
can_send_media_messages=None, can_send_other_messages=None,
can_add_web_page_previews=None):
can_add_web_page_previews=None, can_invite_users=None):
"""
Use this method to restrict a user in a supergroup.
The bot must be an administrator in the supergroup for this to work and must have
@ -956,11 +951,13 @@ class TeleBot:
use inline bots, implies can_send_media_messages
:param can_add_web_page_previews: Pass True, if the user may add web page previews to their messages,
implies can_send_media_messages
:param can_invite_users: Pass True, if the user is allowed to invite new users to the chat,
implies can_invite_users
:return: types.Message
"""
return apihelper.restrict_chat_member(self.token, chat_id, user_id, until_date, can_send_messages,
can_send_media_messages, can_send_other_messages,
can_add_web_page_previews)
can_add_web_page_previews, can_invite_users)
def promote_chat_member(self, chat_id, user_id, can_change_info=None, can_post_messages=None,
can_edit_messages=None, can_delete_messages=None, can_invite_users=None,
@ -1401,12 +1398,7 @@ class TeleBot:
:param callback: The callback function to be called when a reply arrives. Must accept one `message`
parameter, which will contain the replied message.
"""
if message_id in self.reply_handlers.keys():
self.reply_handlers[message_id].append(Handler(callback, *args, **kwargs))
else:
self.reply_handlers[message_id] = [Handler(callback, *args, **kwargs)]
if self.reply_saver is not None:
self.reply_saver.start_save_timer()
self.reply_backend.register_handler(message_id, Handler(callback, *args, **kwargs))
def _notify_reply_handlers(self, new_messages):
"""
@ -1416,14 +1408,9 @@ class TeleBot:
"""
for message in new_messages:
if hasattr(message, "reply_to_message") and message.reply_to_message is not None:
reply_msg_id = message.reply_to_message.message_id
if reply_msg_id in self.reply_handlers.keys():
handlers = self.reply_handlers[reply_msg_id]
handlers = self.reply_backend.get_handlers(message.reply_to_message.message_id)
for handler in handlers:
self._exec_task(handler["callback"], message, *handler["args"], **handler["kwargs"])
self.reply_handlers.pop(reply_msg_id)
if self.reply_saver is not None:
self.reply_saver.start_save_timer()
def register_next_step_handler(self, message, callback, *args, **kwargs):
"""
@ -1450,13 +1437,7 @@ class TeleBot:
:param args: Args to pass in callback func
:param kwargs: Args to pass in callback func
"""
if chat_id in self.next_step_handlers.keys():
self.next_step_handlers[chat_id].append(Handler(callback, *args, **kwargs))
else:
self.next_step_handlers[chat_id] = [Handler(callback, *args, **kwargs)]
if self.next_step_saver is not None:
self.next_step_saver.start_save_timer()
self.next_step_backend.register_handler(chat_id, Handler(callback, *args, **kwargs))
def clear_step_handler(self, message):
"""
@ -1473,10 +1454,7 @@ class TeleBot:
:param chat_id: The chat for which we want to clear next step handlers
"""
self.next_step_handlers[chat_id] = []
if self.next_step_saver is not None:
self.next_step_saver.start_save_timer()
self.next_step_backend.clear_handlers(chat_id)
def clear_reply_handlers(self, message):
"""
@ -1493,10 +1471,7 @@ class TeleBot:
:param message_id: The message id for which we want to clear reply handlers
"""
self.reply_handlers[message_id] = []
if self.reply_saver is not None:
self.reply_saver.start_save_timer()
self.reply_backend.clear_handlers(message_id)
def _notify_next_handlers(self, new_messages):
"""
@ -1504,22 +1479,14 @@ class TeleBot:
:param new_messages:
:return:
"""
i = 0
while i < len(new_messages):
message = new_messages[i]
chat_id = message.chat.id
was_poped = False
if chat_id in self.next_step_handlers.keys():
handlers = self.next_step_handlers.pop(chat_id, None)
if handlers:
for i, message in enumerate(new_messages):
need_pop = False
handlers = self.next_step_backend.get_handlers(message.chat.id)
for handler in handlers:
need_pop = True
self._exec_task(handler["callback"], message, *handler["args"], **handler["kwargs"])
if need_pop:
new_messages.pop(i) # removing message that detects with next_step_handler
was_poped = True
if self.next_step_saver is not None:
self.next_step_saver.start_save_timer()
if not was_poped:
i += 1
@staticmethod
def _build_handler_dict(handler, **filters):
@ -1994,6 +1961,10 @@ class AsyncTeleBot(TeleBot):
def send_message(self, *args, **kwargs):
return TeleBot.send_message(self, *args, **kwargs)
@util.async_dec()
def send_dice(self, *args, **kwargs):
return TeleBot.send_dice(self, *args, **kwargs)
@util.async_dec()
def forward_message(self, *args, **kwargs):
return TeleBot.forward_message(self, *args, **kwargs)

16
telebot/apihelper.py
View File

@ -259,6 +259,18 @@ def forward_message(token, chat_id, from_chat_id, message_id, disable_notificati
return _make_request(token, method_url, params=payload)
def send_dice(token, chat_id, disable_notification=None, reply_to_message_id=None, reply_markup=None):
method_url = r'sendDice'
payload = {'chat_id': chat_id}
if disable_notification:
payload['disable_notification'] = disable_notification
if reply_to_message_id:
payload['reply_to_message_id'] = reply_to_message_id
if reply_markup:
payload['reply_markup'] = _convert_markup(reply_markup)
return _make_request(token, method_url, params=payload)
def send_photo(token, chat_id, photo, caption=None, reply_to_message_id=None, reply_markup=None,
parse_mode=None, disable_notification=None):
method_url = r'sendPhoto'
@ -557,7 +569,7 @@ def unban_chat_member(token, chat_id, user_id):
def restrict_chat_member(token, chat_id, user_id, until_date=None, can_send_messages=None,
can_send_media_messages=None, can_send_other_messages=None,
can_add_web_page_previews=None):
can_add_web_page_previews=None, can_invite_users=None):
method_url = 'restrictChatMember'
payload = {'chat_id': chat_id, 'user_id': user_id}
if until_date:
@ -570,6 +582,8 @@ def restrict_chat_member(token, chat_id, user_id, until_date=None, can_send_mess
payload['can_send_other_messages'] = can_send_other_messages
if can_add_web_page_previews:
payload['can_add_web_page_previews'] = can_add_web_page_previews
if can_invite_users:
payload['can_invite_users'] = can_invite_users
return _make_request(token, method_url, params=payload, method='post')

145
telebot/handler_backends.py Normal file
View File

@ -0,0 +1,145 @@
import os
import pickle
import threading
from telebot import apihelper
class HandlerBackend:
"""
Class for saving (next step|reply) handlers
"""
def __init__(self, handlers=None):
if handlers is None:
handlers = {}
self.handlers = handlers
def register_handler(self, handler_group_id, handler):
raise NotImplementedError()
def clear_handlers(self, handler_group_id):
raise NotImplementedError()
def get_handlers(self, handler_group_id):
raise NotImplementedError()
class MemoryHandlerBackend(HandlerBackend):
def register_handler(self, handler_group_id, handler):
if handler_group_id in self.handlers:
self.handlers[handler_group_id].append(handler)
else:
self.handlers[handler_group_id] = [handler]
def clear_handlers(self, handler_group_id):
self.handlers.pop(handler_group_id, [])
def get_handlers(self, handler_group_id):
return self.handlers.pop(handler_group_id, [])
class FileHandlerBackend(HandlerBackend):
def __init__(self, handlers=None, filename='./.handler-saves/handlers.save', delay=120):
super().__init__(handlers)
self.filename = filename
self.delay = delay
self.timer = threading.Timer(delay, self.save_handlers)
def register_handler(self, handler_group_id, handler):
if handler_group_id in self.handlers:
self.handlers[handler_group_id].append(handler)
else:
self.handlers[handler_group_id] = [handler]
self.start_save_timer()
def clear_handlers(self, handler_group_id):
self.handlers.pop(handler_group_id, [])
self.start_save_timer()
def get_handlers(self, handler_group_id):
handlers = self.handlers.pop(handler_group_id, [])
self.start_save_timer()
return handlers
def start_save_timer(self):
if not self.timer.is_alive():
if self.delay <= 0:
self.save_handlers()
else:
self.timer = threading.Timer(self.delay, self.save_handlers)
self.timer.start()
def save_handlers(self):
self.dump_handlers(self.handlers, self.filename)
def load_handlers(self, filename=None, del_file_after_loading=True):
if not filename:
filename = self.filename
tmp = self.return_load_handlers(filename, del_file_after_loading=del_file_after_loading)
if tmp is not None:
self.handlers.update(tmp)
@staticmethod
def dump_handlers(handlers, filename, file_mode="wb"):
dirs = filename.rsplit('/', maxsplit=1)[0]
os.makedirs(dirs, exist_ok=True)
with open(filename + ".tmp", file_mode) as file:
if (apihelper.CUSTOM_SERIALIZER is None):
pickle.dump(handlers, file)
else:
apihelper.CUSTOM_SERIALIZER.dump(handlers, file)
if os.path.isfile(filename):
os.remove(filename)
os.rename(filename + ".tmp", filename)
@staticmethod
def return_load_handlers(filename, del_file_after_loading=True):
if os.path.isfile(filename) and os.path.getsize(filename) > 0:
with open(filename, "rb") as file:
if (apihelper.CUSTOM_SERIALIZER is None):
handlers = pickle.load(file)
else:
handlers = apihelper.CUSTOM_SERIALIZER.load(file)
if del_file_after_loading:
os.remove(filename)
return handlers
class RedisHandlerBackend(HandlerBackend):
def __init__(self, handlers=None, host='localhost', port=6379, db=0, prefix='telebot'):
super().__init__(handlers)
from redis import Redis
self.prefix = prefix
self.redis = Redis(host, port, db)
def _key(self, handle_group_id):
return ':'.join((self.prefix, str(handle_group_id)))
def register_handler(self, handler_group_id, handler):
handlers = []
value = self.redis.get(self._key(handler_group_id))
if value:
handlers = pickle.loads(value)
handlers.append(handler)
self.redis.set(self._key(handler_group_id), pickle.dumps(handlers))
def clear_handlers(self, handler_group_id):
self.redis.delete(self._key(handler_group_id))
def get_handlers(self, handler_group_id):
handlers = []
value = self.redis.get(self._key(handler_group_id))
if value:
handlers = pickle.loads(value)
self.clear_handlers(handler_group_id)
return handlers

22
telebot/types.py
View File

@ -293,6 +293,9 @@ class Message(JsonDeserializable):
if 'venue' in obj:
opts['venue'] = Venue.de_json(obj['venue'])
content_type = 'venue'
if 'dice' in obj:
opts['dice'] = Dice.de_json(obj['dice'])
content_type = 'dice'
if 'new_chat_members' in obj:
new_chat_members = []
for member in obj['new_chat_members']:
@ -397,6 +400,7 @@ class Message(JsonDeserializable):
self.location = None
self.venue = None
self.animation = None
self.dice = None
self.new_chat_member = None # Deprecated since Bot API 3.0. Not processed anymore
self.new_chat_members = None
self.left_chat_member = None
@ -511,6 +515,24 @@ class MessageEntity(JsonDeserializable):
self.user = user
class Dice(JsonSerializable, Dictionaryable, JsonDeserializable):
@classmethod
def de_json(cls, json_string):
if (json_string is None): return None
obj = cls.check_json(json_string)
value = obj['value']
return cls(value)
def __init__(self, value):
self.value = value
def to_json(self):
return json.dumps({'value': self.value})
def to_dic(self):
return {'value': self.value}
class PhotoSize(JsonDeserializable):
@classmethod
def de_json(cls, json_string):

257
tests/test_handler_backends.py Normal file
View File

@ -0,0 +1,257 @@
import sys
sys.path.append('../')
import os
import time
import pytest
import telebot
from telebot import types, MemoryHandlerBackend, FileHandlerBackend
from telebot.handler_backends import RedisHandlerBackend
@pytest.fixture()
def telegram_bot():
return telebot.TeleBot('', threaded=False)
@pytest.fixture
def private_chat():
return types.Chat(id=11, type='private')
@pytest.fixture
def user():
return types.User(id=10, is_bot=False, first_name='Some User')
@pytest.fixture()
def message(user, private_chat):
params = {'text': '/start'}
return types.Message(
message_id=1, from_user=user, date=None, chat=private_chat, content_type='text', options=params, json_string=""
)
@pytest.fixture()
def reply_to_message(user, private_chat, message):
params = {'text': '/start'}
reply_message = types.Message(
message_id=2, from_user=user, date=None, chat=private_chat, content_type='text', options=params, json_string=""
)
reply_message.reply_to_message = message
return reply_message
@pytest.fixture()
def update_type(message):
edited_message = None
channel_post = None
edited_channel_post = None
inline_query = None
chosen_inline_result = None
callback_query = None
shipping_query = None
pre_checkout_query = None
poll = None
return types.Update(1001234038283, message, edited_message, channel_post, edited_channel_post, inline_query,
chosen_inline_result, callback_query, shipping_query, pre_checkout_query, poll)
@pytest.fixture()
def reply_to_message_update_type(reply_to_message):
edited_message = None
channel_post = None
edited_channel_post = None
inline_query = None
chosen_inline_result = None
callback_query = None
shipping_query = None
pre_checkout_query = None
poll = None
return types.Update(1001234038284, reply_to_message, edited_message, channel_post, edited_channel_post,
inline_query,
chosen_inline_result, callback_query, shipping_query, pre_checkout_query, poll)
def next_handler(message):
message.text = 'entered next_handler'
def test_memory_handler_backend_default_backend(telegram_bot):
assert telegram_bot.reply_backend.__class__ == MemoryHandlerBackend
assert telegram_bot.next_step_backend.__class__ == MemoryHandlerBackend
def test_memory_handler_backend_register_next_step_handler(telegram_bot, private_chat, update_type):
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_next_step_handler_by_chat_id(message.chat.id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
assert len(telegram_bot.next_step_backend.handlers[private_chat.id]) == 1
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered next_handler'
assert private_chat.id not in telegram_bot.next_step_backend.handlers
def test_memory_handler_backend_clear_next_step_handler(telegram_bot, private_chat, update_type):
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_next_step_handler_by_chat_id(message.chat.id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
assert len(telegram_bot.next_step_backend.handlers[private_chat.id]) == 1
telegram_bot.clear_step_handler_by_chat_id(private_chat.id)
assert private_chat.id not in telegram_bot.next_step_backend.handlers
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
def test_memory_handler_backend_register_reply_handler(telegram_bot, private_chat, update_type,
reply_to_message_update_type):
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_for_reply_by_message_id(message.message_id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
assert len(telegram_bot.reply_backend.handlers[update_type.message.message_id]) == 1
telegram_bot.process_new_updates([reply_to_message_update_type])
assert reply_to_message_update_type.message.text == 'entered next_handler'
assert private_chat.id not in telegram_bot.reply_backend.handlers
def test_memory_handler_backend_clear_reply_handler(telegram_bot, private_chat, update_type,
reply_to_message_update_type):
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_for_reply_by_message_id(message.message_id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
assert len(telegram_bot.reply_backend.handlers[update_type.message.message_id]) == 1
telegram_bot.clear_reply_handlers_by_message_id(update_type.message.message_id)
assert update_type.message.message_id not in telegram_bot.reply_backend.handlers
telegram_bot.process_new_updates([reply_to_message_update_type])
assert reply_to_message_update_type.message.text == 'entered start'
def test_file_handler_backend_register_next_step_handler(telegram_bot, private_chat, update_type):
telegram_bot.next_step_backend=FileHandlerBackend(filename='./.handler-saves/step1.save', delay=0.1)
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_next_step_handler_by_chat_id(message.chat.id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
time.sleep(0.2)
assert os.path.exists(telegram_bot.next_step_backend.filename)
assert len(telegram_bot.next_step_backend.handlers[private_chat.id]) == 1
telegram_bot.next_step_backend.handlers = {}
telegram_bot.next_step_backend.load_handlers()
assert len(telegram_bot.next_step_backend.handlers[private_chat.id]) == 1
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered next_handler'
assert private_chat.id not in telegram_bot.next_step_backend.handlers
time.sleep(0.2)
if os.path.exists(telegram_bot.next_step_backend.filename):
os.remove(telegram_bot.next_step_backend.filename)
def test_file_handler_backend_clear_next_step_handler(telegram_bot, private_chat, update_type):
telegram_bot.next_step_backend=FileHandlerBackend(filename='./.handler-saves/step2.save', delay=0.1)
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_next_step_handler_by_chat_id(message.chat.id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
assert len(telegram_bot.next_step_backend.handlers[private_chat.id]) == 1
time.sleep(0.2)
assert os.path.exists(telegram_bot.next_step_backend.filename)
telegram_bot.clear_step_handler_by_chat_id(private_chat.id)
time.sleep(0.2)
telegram_bot.next_step_backend.load_handlers()
assert private_chat.id not in telegram_bot.next_step_backend.handlers
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
time.sleep(0.2)
if os.path.exists(telegram_bot.next_step_backend.filename):
os.remove(telegram_bot.next_step_backend.filename)
def test_redis_handler_backend_register_next_step_handler(telegram_bot, private_chat, update_type):
telegram_bot.next_step_backend = RedisHandlerBackend(prefix='pyTelegramBotApi:step_backend1')
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_next_step_handler_by_chat_id(message.chat.id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered next_handler'
def test_redis_handler_backend_clear_next_step_handler(telegram_bot, private_chat, update_type):
telegram_bot.next_step_backend = RedisHandlerBackend(prefix='pyTelegramBotApi:step_backend2')
@telegram_bot.message_handler(commands=['start'])
def start(message):
message.text = 'entered start'
telegram_bot.register_next_step_handler_by_chat_id(message.chat.id, next_handler)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'
telegram_bot.clear_step_handler_by_chat_id(private_chat.id)
telegram_bot.process_new_updates([update_type])
assert update_type.message.text == 'entered start'

6
tests/test_telebot.py
View File

@ -241,6 +241,12 @@ class TestTeleBot:
ret_msg = tb.send_message(CHAT_ID, text)
assert ret_msg.message_id
def test_send_dice(self):
tb = telebot.TeleBot(TOKEN)
ret_msg = tb.send_dice(CHAT_ID)
assert ret_msg.message_id
assert ret_msg.content_type == 'dice'
def test_send_message_dis_noti(self):
text = 'CI Test Message'
tb = telebot.TeleBot(TOKEN)

8
tests/test_types.py
View File

@ -17,6 +17,14 @@ def test_json_message():
assert msg.text == 'HIHI'
def test_json_message_with_dice():
jsonstring = r'{"message_id":5560,"from":{"id":879343317,"is_bot":false,"first_name":"George","last_name":"Forse","username":"dr_fxrse","language_code":"ru"},"chat":{"id":879343317,"first_name":"George","last_name":"Forse","username":"dr_fxrse","type":"private"},"date":1586926330,"dice":{"value":4}}'
msg = types.Message.de_json(jsonstring)
assert msg.content_type == 'dice'
assert isinstance(msg.dice, types.Dice)
assert msg.dice.value == 4
def test_json_message_group():
json_string = r'{"message_id":10,"from":{"id":12345,"first_name":"g","last_name":"G","username":"GG","is_bot":true},"chat":{"id":-866,"type":"private","title":"\u4ea4"},"date":1435303157,"text":"HIHI"}'
msg = types.Message.de_json(json_string)