#!/usr/bin/env python # # A library that provides a Python interface to the Telegram Bot API # Copyright (C) 2015-2025 # Leandro Toledo de Souza # # 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/]. """Persistence of conversations is tested in test_basepersistence.py""" import asyncio import functools import logging from copy import copy from pathlib import Path from warnings import filterwarnings import pytest from telegram import ( CallbackQuery, Chat, ChosenInlineResult, InlineQuery, Message, MessageEntity, PreCheckoutQuery, ShippingQuery, Update, User, ) from telegram.ext import ( ApplicationBuilder, ApplicationHandlerStop, CallbackContext, CallbackQueryHandler, ChosenInlineResultHandler, CommandHandler, ConversationHandler, Defaults, InlineQueryHandler, JobQueue, MessageHandler, PollAnswerHandler, PollHandler, PreCheckoutQueryHandler, ShippingQueryHandler, StringCommandHandler, StringRegexHandler, TypeHandler, filters, ) from telegram.warnings import PTBUserWarning from tests.auxil.build_messages import make_command_message from tests.auxil.files import SOURCE_ROOT_PATH from tests.auxil.pytest_classes import PytestBot, make_bot from tests.auxil.slots import mro_slots @pytest.fixture(scope="class") def user1(): return User(first_name="Misses Test", id=123, is_bot=False) @pytest.fixture(scope="class") def user2(): return User(first_name="Mister Test", id=124, is_bot=False) def raise_ahs(func): @functools.wraps(func) # for checking __repr__ async def decorator(self, *args, **kwargs): result = await func(self, *args, **kwargs) if self.raise_app_handler_stop: raise ApplicationHandlerStop(result) return result return decorator class TestConversationHandler: """Persistence of conversations is tested in test_basepersistence.py""" # State definitions # At first we're thirsty. Then we brew coffee, we drink it # and then we can start coding! END, THIRSTY, BREWING, DRINKING, CODING = range(-1, 4) # Drinking state definitions (nested) # At first we're holding the cup. Then we sip coffee, and last we swallow it HOLDING, SIPPING, SWALLOWING, REPLENISHING, STOPPING = map(chr, range(ord("a"), ord("f"))) current_state, entry_points, states, fallbacks = None, None, None, None group = Chat(0, Chat.GROUP) second_group = Chat(1, Chat.GROUP) raise_app_handler_stop = False test_flag = False # Test related @pytest.fixture(autouse=True) def _reset(self): self.raise_app_handler_stop = False self.test_flag = False self.current_state = {} self.entry_points = [CommandHandler("start", self.start)] self.states = { self.THIRSTY: [CommandHandler("brew", self.brew), CommandHandler("wait", self.start)], self.BREWING: [CommandHandler("pourCoffee", self.drink)], self.DRINKING: [ CommandHandler("startCoding", self.code), CommandHandler("drinkMore", self.drink), CommandHandler("end", self.end), ], self.CODING: [ CommandHandler("keepCoding", self.code), CommandHandler("gettingThirsty", self.start), CommandHandler("drinkMore", self.drink), ], } self.fallbacks = [CommandHandler("eat", self.start)] self.is_timeout = False # for nesting tests self.nested_states = { self.THIRSTY: [CommandHandler("brew", self.brew), CommandHandler("wait", self.start)], self.BREWING: [CommandHandler("pourCoffee", self.drink)], self.CODING: [ CommandHandler("keepCoding", self.code), CommandHandler("gettingThirsty", self.start), CommandHandler("drinkMore", self.drink), ], } self.drinking_entry_points = [CommandHandler("hold", self.hold)] self.drinking_states = { self.HOLDING: [CommandHandler("sip", self.sip)], self.SIPPING: [CommandHandler("swallow", self.swallow)], self.SWALLOWING: [CommandHandler("hold", self.hold)], } self.drinking_fallbacks = [ CommandHandler("replenish", self.replenish), CommandHandler("stop", self.stop), CommandHandler("end", self.end), CommandHandler("startCoding", self.code), CommandHandler("drinkMore", self.drink), ] self.drinking_entry_points.extend(self.drinking_fallbacks) # Map nested states to parent states: self.drinking_map_to_parent = { # Option 1 - Map a fictional internal state to an external parent state self.REPLENISHING: self.BREWING, # Option 2 - Map a fictional internal state to the END state on the parent self.STOPPING: self.END, # Option 3 - Map the internal END state to an external parent state self.END: self.CODING, # Option 4 - Map an external state to the same external parent state self.CODING: self.CODING, # Option 5 - Map an external state to the internal entry point self.DRINKING: self.DRINKING, } # State handlers def _set_state(self, update, state): self.current_state[update.message.from_user.id] = state return state # Actions @raise_ahs async def start(self, update, context): if isinstance(update, Update): return self._set_state(update, self.THIRSTY) return self._set_state(context.bot, self.THIRSTY) @raise_ahs async def end(self, update, context): return self._set_state(update, self.END) @raise_ahs async def start_end(self, update, context): return self._set_state(update, self.END) @raise_ahs async def start_none(self, update, context): return self._set_state(update, None) @raise_ahs async def brew(self, update, context): if isinstance(update, Update): return self._set_state(update, self.BREWING) return self._set_state(context.bot, self.BREWING) @raise_ahs async def drink(self, update, context): return self._set_state(update, self.DRINKING) @raise_ahs async def code(self, update, context): return self._set_state(update, self.CODING) @raise_ahs async def passout(self, update, context): assert update.message.text == "/brew" assert isinstance(update, Update) self.is_timeout = True @raise_ahs async def passout2(self, update, context): assert isinstance(update, Update) self.is_timeout = True @raise_ahs async def passout_context(self, update, context): assert update.message.text == "/brew" assert isinstance(context, CallbackContext) self.is_timeout = True @raise_ahs async def passout2_context(self, update, context): assert isinstance(context, CallbackContext) self.is_timeout = True # Drinking actions (nested) @raise_ahs async def hold(self, update, context): return self._set_state(update, self.HOLDING) @raise_ahs async def sip(self, update, context): return self._set_state(update, self.SIPPING) @raise_ahs async def swallow(self, update, context): return self._set_state(update, self.SWALLOWING) @raise_ahs async def replenish(self, update, context): return self._set_state(update, self.REPLENISHING) @raise_ahs async def stop(self, update, context): return self._set_state(update, self.STOPPING) def test_slot_behaviour(self): handler = ConversationHandler(entry_points=[], states={}, fallbacks=[]) 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" def test_init(self): entry_points = [] states = {} fallbacks = [] map_to_parent = {} ch = ConversationHandler( entry_points=entry_points, states=states, fallbacks=fallbacks, per_chat="per_chat", per_user="per_user", per_message="per_message", persistent="persistent", name="name", allow_reentry="allow_reentry", conversation_timeout=42, map_to_parent=map_to_parent, ) assert ch.entry_points is entry_points assert ch.states is states assert ch.fallbacks is fallbacks assert ch.map_to_parent is map_to_parent assert ch.per_chat == "per_chat" assert ch.per_user == "per_user" assert ch.per_message == "per_message" assert ch.persistent == "persistent" assert ch.name == "name" assert ch.allow_reentry == "allow_reentry" def test_init_persistent_no_name(self): with pytest.raises(ValueError, match="can't be persistent when handler is unnamed"): ConversationHandler( self.entry_points, states=self.states, fallbacks=[], persistent=True ) def test_repr_no_truncation(self): # ConversationHandler's __repr__ is not inherited from BaseHandler. ch = ConversationHandler( name="test_handler", entry_points=[], states=self.drinking_states, fallbacks=[], ) assert repr(ch) == ( "ConversationHandler[name=test_handler, " "states={'a': [CommandHandler[callback=TestConversationHandler.sip]], " "'b': [CommandHandler[callback=TestConversationHandler.swallow]], " "'c': [CommandHandler[callback=TestConversationHandler.hold]]}]" ) def test_repr_with_truncation(self): states = copy(self.drinking_states) # there are exactly 3 drinking states. adding one more to make sure it's truncated states["extra_to_be_truncated"] = [CommandHandler("foo", self.start)] ch = ConversationHandler( name="test_handler", entry_points=[], states=states, fallbacks=[], ) assert repr(ch) == ( "ConversationHandler[name=test_handler, " "states={'a': [CommandHandler[callback=TestConversationHandler.sip]], " "'b': [CommandHandler[callback=TestConversationHandler.swallow]], " "'c': [CommandHandler[callback=TestConversationHandler.hold]], ...}]" ) async def test_check_update_returns_non(self, app, user1): """checks some cases where updates should not be handled""" conv_handler = ConversationHandler([], {}, [], per_message=True, per_chat=True) assert not conv_handler.check_update("not an Update") assert not conv_handler.check_update(Update(0)) assert not conv_handler.check_update( Update(0, callback_query=CallbackQuery("1", from_user=user1, chat_instance="1")) ) async def test_handlers_generate_warning(self, recwarn): """this function tests all handler + per_* setting combinations.""" # the warning message action needs to be set to always, # otherwise only the first occurrence will be issued filterwarnings(action="always", category=PTBUserWarning) # this class doesn't do anything, its just not the Update class class NotUpdate: pass recwarn.clear() # this conversation handler has the string, string_regex, Pollhandler and TypeHandler # which should all generate a warning no matter the per_* setting. TypeHandler should # not when the class is Update ConversationHandler( entry_points=[StringCommandHandler("code", self.code)], states={ self.BREWING: [ StringRegexHandler("code", self.code), PollHandler(self.code), TypeHandler(NotUpdate, self.code), ], }, fallbacks=[TypeHandler(Update, self.code)], ) # these handlers should all raise a warning when per_chat is True ConversationHandler( entry_points=[ShippingQueryHandler(self.code)], states={ self.BREWING: [ InlineQueryHandler(self.code), PreCheckoutQueryHandler(self.code), PollAnswerHandler(self.code), ], }, fallbacks=[ChosenInlineResultHandler(self.code)], per_chat=True, ) # the CallbackQueryHandler should *not* raise when per_message is True, # but any other one should ConversationHandler( entry_points=[CallbackQueryHandler(self.code)], states={ self.BREWING: [CommandHandler("code", self.code)], }, fallbacks=[CallbackQueryHandler(self.code)], per_message=True, ) # the CallbackQueryHandler should raise when per_message is False ConversationHandler( entry_points=[CommandHandler("code", self.code)], states={ self.BREWING: [CommandHandler("code", self.code)], }, fallbacks=[CallbackQueryHandler(self.code)], per_message=False, ) # adding a nested conv to a conversation with timeout should warn child = ConversationHandler( entry_points=[CommandHandler("code", self.code)], states={ self.BREWING: [CommandHandler("code", self.code)], }, fallbacks=[CommandHandler("code", self.code)], ) ConversationHandler( entry_points=[CommandHandler("code", self.code)], states={ self.BREWING: [child], }, fallbacks=[CommandHandler("code", self.code)], conversation_timeout=42, ) # If per_message is True, per_chat should also be True, since msg ids are not unique ConversationHandler( entry_points=[CallbackQueryHandler(self.code, "code")], states={ self.BREWING: [CallbackQueryHandler(self.code, "code")], }, fallbacks=[CallbackQueryHandler(self.code, "code")], per_message=True, per_chat=False, ) # the overall number of handlers throwing a warning is 13 assert len(recwarn) == 13 # now we test the messages, they are raised in the order they are inserted # into the conversation handler assert ( str(recwarn[0].message) == "The `ConversationHandler` only handles updates of type `telegram.Update`. " "StringCommandHandler handles updates of type `str`." ) assert ( str(recwarn[1].message) == "The `ConversationHandler` only handles updates of type `telegram.Update`. " "StringRegexHandler handles updates of type `str`." ) assert ( str(recwarn[2].message) == "PollHandler will never trigger in a conversation since it has no information " "about the chat or the user who voted in it. Do you mean the " "`PollAnswerHandler`?" ) assert ( str(recwarn[3].message) == "The `ConversationHandler` only handles updates of type `telegram.Update`. " "The TypeHandler is set to handle NotUpdate." ) per_faq_link = ( " Read this FAQ entry to learn more about the per_* settings: " "https://github.com/python-telegram-bot/python-telegram-bot/wiki" "/Frequently-Asked-Questions#what-do-the-per_-settings-in-conversationhandler-do." ) assert str(recwarn[4].message) == ( "Updates handled by ShippingQueryHandler only have information about the user," " so this handler won't ever be triggered if `per_chat=True`." + per_faq_link ) assert str(recwarn[5].message) == ( "Updates handled by ChosenInlineResultHandler only have information about the user," " so this handler won't ever be triggered if `per_chat=True`." + per_faq_link ) assert str(recwarn[6].message) == ( "Updates handled by InlineQueryHandler only have information about the user," " so this handler won't ever be triggered if `per_chat=True`." + per_faq_link ) assert str(recwarn[7].message) == ( "Updates handled by PreCheckoutQueryHandler only have information about the user," " so this handler won't ever be triggered if `per_chat=True`." + per_faq_link ) assert str(recwarn[8].message) == ( "Updates handled by PollAnswerHandler only have information about the user," " so this handler won't ever be triggered if `per_chat=True`." + per_faq_link ) assert str(recwarn[9].message) == ( "If 'per_message=True', all entry points, state handlers, and fallbacks must be " "'CallbackQueryHandler', since no other handlers have a message context." + per_faq_link ) assert str(recwarn[10].message) == ( "If 'per_message=False', 'CallbackQueryHandler' will not be tracked for every message." + per_faq_link ) assert ( str(recwarn[11].message) == "Using `conversation_timeout` with nested conversations is currently not " "supported. You can still try to use it, but it will likely behave differently" " from what you expect." ) assert ( str(recwarn[12].message) == "If 'per_message=True' is used, 'per_chat=True' should also be used, " "since message IDs are not globally unique." ) # this for loop checks if the correct stacklevel is used when generating the warning for warning in recwarn: assert warning.category is PTBUserWarning assert warning.filename == __file__, "incorrect stacklevel!" @pytest.mark.parametrize( "attr", [ "entry_points", "states", "fallbacks", "per_chat", "per_user", "per_message", "name", "persistent", "allow_reentry", "conversation_timeout", "map_to_parent", ], indirect=False, ) def test_immutable(self, attr): ch = ConversationHandler(entry_points=[], states={}, fallbacks=[]) with pytest.raises(AttributeError, match=f"You can not assign a new value to {attr}"): setattr(ch, attr, True) def test_per_all_false(self): with pytest.raises(ValueError, match="can't all be 'False'"): ConversationHandler( entry_points=[], states={}, fallbacks=[], per_chat=False, per_user=False, per_message=False, ) @pytest.mark.parametrize("raise_ahs", [True, False]) async def test_basic_and_app_handler_stop(self, app, bot, user1, user2, raise_ahs): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks ) app.add_handler(handler) async def callback(_, __): self.test_flag = True app.add_handler(TypeHandler(object, callback), group=100) self.raise_app_handler_stop = raise_ahs # User one, starts the state machine. message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.THIRSTY assert self.test_flag == (not raise_ahs) # The user is thirsty and wants to brew coffee. message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.BREWING assert self.test_flag == (not raise_ahs) # Lets see if an invalid command makes sure, no state is changed. message.text = "/nothing" message.entities[0].length = len("/nothing") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.BREWING assert self.test_flag is True self.test_flag = False # Lets see if the state machine still works by pouring coffee. message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING assert self.test_flag == (not raise_ahs) # Let's now verify that for another user, who did not start yet, # the state has not been changed. message.from_user = user2 await app.process_update(Update(update_id=0, message=message)) with pytest.raises(KeyError): self.current_state[user2.id] async def test_conversation_handler_end(self, caplog, app, bot, user1): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks ) app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=0, message=message)) message.text = "/end" message.entities[0].length = len("/end") caplog.clear() with caplog.at_level(logging.ERROR): await app.process_update(Update(update_id=0, message=message)) assert len(caplog.records) == 0 assert self.current_state[user1.id] == self.END # make sure that the conversation has ended by checking that the start command is # accepted again message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(update_id=0, message=message)) async def test_conversation_handler_fallback(self, app, bot, user1, user2): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks ) app.add_handler(handler) # first check if fallback will not trigger start when not started message = Message( 0, None, self.group, from_user=user1, text="/eat", entities=[MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/eat"))], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) with pytest.raises(KeyError): self.current_state[user1.id] # User starts the state machine. message.text = "/start" message.entities[0].length = len("/start") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.THIRSTY # The user is thirsty and wants to brew coffee. message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.BREWING # Now a fallback command is issued message.text = "/eat" message.entities[0].length = len("/eat") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.THIRSTY async def test_unknown_state_warning(self, app, bot, user1, recwarn): def build_callback(state): async def callback(_, __): return state return callback handler = ConversationHandler( entry_points=[CommandHandler("start", build_callback(1))], states={ 1: [TypeHandler(Update, build_callback(69))], 2: [TypeHandler(Update, build_callback(42))], }, fallbacks=self.fallbacks, name="xyz", ) app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) try: await app.process_update(Update(update_id=1, message=message)) except Exception as exc: print(exc) raise exc assert len(recwarn) == 1 assert recwarn[0].category is PTBUserWarning assert ( Path(recwarn[0].filename) == SOURCE_ROOT_PATH / "ext" / "_handlers" / "conversationhandler.py" ), "wrong stacklevel!" assert ( str(recwarn[0].message) == "'callback' returned state 69 which is unknown to the ConversationHandler xyz." ) async def test_conversation_handler_per_chat(self, app, bot, user1, user2): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, per_user=False, ) app.add_handler(handler) # User one, starts the state machine. message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) # The user is thirsty and wants to brew coffee. message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) # Let's now verify that for another user, who did not start yet, # the state will be changed because they are in the same group. message.from_user = user2 message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=0, message=message)) # Check that we're in the DRINKING state by checking that the corresponding command # is accepted message.from_user = user1 message.text = "/startCoding" message.entities[0].length = len("/startCoding") assert handler.check_update(Update(update_id=0, message=message)) message.from_user = user2 assert handler.check_update(Update(update_id=0, message=message)) async def test_conversation_handler_per_user(self, app, bot, user1): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, per_chat=False, ) app.add_handler(handler) # User one, starts the state machine. message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() # First check that updates without user won't be handled message.from_user = None assert not handler.check_update(Update(update_id=0, message=message)) message.from_user = user1 async with app: await app.process_update(Update(update_id=0, message=message)) # The user is thirsty and wants to brew coffee. message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) # Let's now verify that for the same user in a different group, the state will still be # updated message.chat = self.second_group message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=0, message=message)) # Check that we're in the DRINKING state by checking that the corresponding command # is accepted message.chat = self.group message.text = "/startCoding" message.entities[0].length = len("/startCoding") assert handler.check_update(Update(update_id=0, message=message)) message.chat = self.second_group assert handler.check_update(Update(update_id=0, message=message)) @pytest.mark.parametrize("inline", [True, False]) @pytest.mark.filterwarnings("ignore: If 'per_message=True' is used, 'per_chat=True'") async def test_conversation_handler_per_message(self, app, bot, user1, user2, inline): async def entry(update, context): return 1 async def one(update, context): return 2 async def two(update, context): return ConversationHandler.END handler = ConversationHandler( entry_points=[CallbackQueryHandler(entry)], states={ 1: [CallbackQueryHandler(one, pattern="^1$")], 2: [CallbackQueryHandler(two, pattern="^2$")], }, fallbacks=[], per_message=True, per_chat=not inline, ) app.add_handler(handler) # User one, starts the state machine. message = ( Message(0, None, self.group, from_user=user1, text="msg w/ inlinekeyboard") if not inline else None ) if message: message.set_bot(bot) message._unfreeze() inline_message_id = "42" if inline else None async with app: cbq_1 = CallbackQuery( 0, user1, None, message=message, data="1", inline_message_id=inline_message_id, ) cbq_1.set_bot(bot) cbq_2 = CallbackQuery( 0, user1, None, message=message, data="2", inline_message_id=inline_message_id, ) cbq_2.set_bot(bot) cbq_2._unfreeze() await app.process_update(Update(update_id=0, callback_query=cbq_1)) # Make sure that we're in the correct state assert handler.check_update(Update(0, callback_query=cbq_1)) assert not handler.check_update(Update(0, callback_query=cbq_2)) await app.process_update(Update(update_id=0, callback_query=cbq_1)) # Make sure that we're in the correct state assert not handler.check_update(Update(0, callback_query=cbq_1)) assert handler.check_update(Update(0, callback_query=cbq_2)) # Let's now verify that for a different user in the same group, the state will not be # updated cbq_2.from_user = user2 await app.process_update(Update(update_id=0, callback_query=cbq_2)) cbq_2.from_user = user1 assert not handler.check_update(Update(0, callback_query=cbq_1)) assert handler.check_update(Update(0, callback_query=cbq_2)) async def test_end_on_first_message(self, app, bot, user1): handler = ConversationHandler( entry_points=[CommandHandler("start", self.start_end)], states={}, fallbacks=[] ) app.add_handler(handler) # User starts the state machine and immediately ends it. message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) assert handler.check_update(Update(update_id=0, message=message)) async def test_end_on_first_message_non_blocking_handler(self, app, bot, user1): handler = ConversationHandler( entry_points=[CommandHandler("start", callback=self.start_end, block=False)], states={}, fallbacks=[], ) app.add_handler(handler) # User starts the state machine with a non-blocking function that immediately ends the # conversation. non-blocking results are resolved when the users state is queried next # time. message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) # give the task a chance to finish await asyncio.sleep(0.1) # Let's check that processing the same update again is accepted. this confirms that # a) the pending state is correctly resolved # b) the conversation has ended assert handler.check_update(Update(0, message=message)) async def test_none_on_first_message(self, app, bot, user1): handler = ConversationHandler( entry_points=[MessageHandler(filters.ALL, self.start_none)], states={}, fallbacks=[] ) app.add_handler(handler) # User starts the state machine and a callback function returns None message = Message(0, None, self.group, from_user=user1, text="/start") message.set_bot(bot) message._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) # Check that the same message is accepted again, i.e. the conversation immediately # ended assert handler.check_update(Update(0, message=message)) async def test_none_on_first_message_non_blocking_handler(self, app, bot, user1): handler = ConversationHandler( entry_points=[CommandHandler("start", self.start_none, block=False)], states={}, fallbacks=[], ) app.add_handler(handler) # User starts the state machine with a non-blocking handler that returns None # non-blocking results are resolved when the users state is queried next time. message = Message( 0, None, self.group, text="/start", from_user=user1, entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) # Give the task a chance to finish await asyncio.sleep(0.1) # Let's check that processing the same update again is accepted. this confirms that # a) the pending state is correctly resolved # b) the conversation has ended assert handler.check_update(Update(0, message=message)) async def test_per_chat_message_without_chat(self, bot, user1): handler = ConversationHandler( entry_points=[CommandHandler("start", self.start_end)], states={}, fallbacks=[] ) cbq = CallbackQuery(0, user1, None, None) cbq.set_bot(bot) update = Update(0, callback_query=cbq) assert not handler.check_update(update) async def test_channel_message_without_chat(self, bot): handler = ConversationHandler( entry_points=[MessageHandler(filters.ALL, self.start_end)], states={}, fallbacks=[] ) message = Message(0, date=None, chat=Chat(0, Chat.CHANNEL, "Misses Test")) message.set_bot(bot) message._unfreeze() update = Update(0, channel_post=message) assert not handler.check_update(update) update = Update(0, edited_channel_post=message) assert not handler.check_update(update) async def test_all_update_types(self, app, bot, user1): handler = ConversationHandler( entry_points=[CommandHandler("start", self.start_end)], states={}, fallbacks=[] ) message = Message(0, None, self.group, from_user=user1, text="ignore") message.set_bot(bot) message._unfreeze() callback_query = CallbackQuery(0, user1, None, message=message, data="data") callback_query.set_bot(bot) chosen_inline_result = ChosenInlineResult(0, user1, "query") chosen_inline_result.set_bot(bot) inline_query = InlineQuery(0, user1, "query", offset="") inline_query.set_bot(bot) pre_checkout_query = PreCheckoutQuery(0, user1, "USD", 100, []) pre_checkout_query.set_bot(bot) shipping_query = ShippingQuery(0, user1, [], None) shipping_query.set_bot(bot) assert not handler.check_update(Update(0, callback_query=callback_query)) assert not handler.check_update(Update(0, chosen_inline_result=chosen_inline_result)) assert not handler.check_update(Update(0, inline_query=inline_query)) assert not handler.check_update(Update(0, message=message)) assert not handler.check_update(Update(0, pre_checkout_query=pre_checkout_query)) assert not handler.check_update(Update(0, shipping_query=shipping_query)) @pytest.mark.parametrize("jq", [True, False]) async def test_no_running_job_queue_warning(self, app, bot, user1, recwarn, jq): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) if not jq: app = ApplicationBuilder().token(bot.token).job_queue(None).build() app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.5) if jq: assert len(recwarn) == 1 else: assert len(recwarn) == 2 assert str(recwarn[0].message if jq else recwarn[1].message).startswith( "Ignoring `conversation_timeout`" ) assert ("is not running" if jq else "No `JobQueue` set up.") in str(recwarn[0].message) for warning in recwarn: assert warning.category is PTBUserWarning assert ( Path(warning.filename) == SOURCE_ROOT_PATH / "ext" / "_handlers" / "conversationhandler.py" ), "wrong stacklevel!" # now set app.job_queue back to it's original value async def test_schedule_job_exception(self, app, bot, user1, monkeypatch, caplog): def mocked_run_once(*a, **kw): raise Exception("job error") class DictJB(JobQueue): pass app = ApplicationBuilder().token(bot.token).job_queue(DictJB()).build() monkeypatch.setattr(app.job_queue, "run_once", mocked_run_once) handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=100, ) app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.start() with caplog.at_level(logging.ERROR): await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.5) assert len(caplog.records) == 1 assert caplog.records[0].message == "Failed to schedule timeout." assert caplog.records[0].name == "telegram.ext.ConversationHandler" assert str(caplog.records[0].exc_info[1]) == "job error" await app.stop() @pytest.mark.parametrize(argnames="test_type", argvalues=["none", "exception"]) async def test_non_blocking_exception_or_none(self, app, bot, user1, caplog, test_type): """Here we make sure that when a non-blocking handler raises an exception or returns None, the state isn't changed. """ error = Exception("task exception") async def conv_entry(*a, **kw): return 1 async def raise_error(*a, **kw): if test_type == "none": return raise error handler = ConversationHandler( entry_points=[CommandHandler("start", conv_entry)], states={1: [MessageHandler(filters.Text(["error"]), raise_error)]}, fallbacks=self.fallbacks, block=False, ) app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() # start the conversation async with app: await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.1) message.text = "error" await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.1) caplog.clear() with caplog.at_level(logging.ERROR): # This also makes sure that we're still in the same state assert handler.check_update(Update(0, message=message)) if test_type == "exception": assert len(caplog.records) == 1 assert caplog.records[0].name == "telegram.ext.ConversationHandler" assert ( caplog.records[0].message == "Task function raised exception. Falling back to old state 1" ) assert caplog.records[0].exc_info[1] is None else: assert len(caplog.records) == 0 async def test_non_blocking_entry_point_exception(self, app, bot, user1, caplog): """Here we make sure that when a non-blocking entry point raises an exception, the state isn't changed. """ error = Exception("task exception") async def raise_error(*a, **kw): raise error handler = ConversationHandler( entry_points=[CommandHandler("start", raise_error, block=False)], states={}, fallbacks=self.fallbacks, ) app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() # start the conversation async with app: await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.1) caplog.clear() with caplog.at_level(logging.ERROR): # This also makes sure that we're still in the same state assert handler.check_update(Update(0, message=message)) assert len(caplog.records) == 1 assert caplog.records[0].name == "telegram.ext.ConversationHandler" assert ( caplog.records[0].message == "Task function raised exception. Falling back to old state None" ) assert caplog.records[0].exc_info[1] is None async def test_conversation_timeout(self, app, bot, user1): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) # Start state machine, then reach timeout start_message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) start_message.set_bot(bot) brew_message = Message( 0, None, self.group, from_user=user1, text="/brew", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/brew")) ], ) brew_message.set_bot(bot) pour_coffee_message = Message( 0, None, self.group, from_user=user1, text="/pourCoffee", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/pourCoffee")) ], ) pour_coffee_message.set_bot(bot) async with app: await app.start() await app.process_update(Update(update_id=0, message=start_message)) assert handler.check_update(Update(0, message=brew_message)) await asyncio.sleep(0.75) assert handler.check_update(Update(0, message=start_message)) # Start state machine, do something, then reach timeout await app.process_update(Update(update_id=1, message=start_message)) assert handler.check_update(Update(0, message=brew_message)) # assert handler.conversations.get((self.group.id, user1.id)) == self.THIRSTY # start_message.text = '/brew' # start_message.entities[0].length = len('/brew') await app.process_update(Update(update_id=2, message=brew_message)) assert handler.check_update(Update(0, message=pour_coffee_message)) # assert handler.conversations.get((self.group.id, user1.id)) == self.BREWING await asyncio.sleep(0.75) assert handler.check_update(Update(0, message=start_message)) # assert handler.conversations.get((self.group.id, user1.id)) is None await app.stop() async def test_timeout_not_triggered_on_conv_end_non_blocking(self, bot, app, user1): def timeout(*a, **kw): self.test_flag = True self.states.update({ConversationHandler.TIMEOUT: [TypeHandler(Update, timeout)]}) handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=0.5, block=False, ) app.add_handler(handler) message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: # start the conversation await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.1) message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=1, message=message)) await asyncio.sleep(0.1) message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=2, message=message)) await asyncio.sleep(0.1) message.text = "/end" message.entities[0].length = len("/end") await app.process_update(Update(update_id=3, message=message)) await asyncio.sleep(1) # assert timeout handler didn't get called assert self.test_flag is False async def test_conversation_timeout_application_handler_stop(self, app, bot, user1, recwarn): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) def timeout(*args, **kwargs): raise ApplicationHandlerStop self.states.update({ConversationHandler.TIMEOUT: [TypeHandler(Update, timeout)]}) app.add_handler(handler) # Start state machine, then reach timeout message = Message( 0, None, self.group, text="/start", from_user=user1, entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() brew_message = Message( 0, None, self.group, from_user=user1, text="/brew", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/brew")) ], ) brew_message.set_bot(bot) async with app: await app.start() await app.process_update(Update(update_id=0, message=message)) # Make sure that we're in the next state assert handler.check_update(Update(0, message=brew_message)) await app.process_update(Update(0, message=brew_message)) await asyncio.sleep(0.9) # Check that conversation has ended by checking that the start messages is accepted # again assert handler.check_update(Update(0, message=message)) assert len(recwarn) == 1 assert str(recwarn[0].message).startswith("ApplicationHandlerStop in TIMEOUT") assert recwarn[0].category is PTBUserWarning assert ( Path(recwarn[0].filename) == SOURCE_ROOT_PATH / "ext" / "_jobqueue.py" ), "wrong stacklevel!" await app.stop() async def test_conversation_handler_timeout_update_and_context(self, app, bot, user1): context = None async def start_callback(u, c): nonlocal context context = c return await self.start(u, c) # Start state machine, then reach timeout message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() update = Update(update_id=0, message=message) async def timeout_callback(u, c): assert u is update assert c is context self.is_timeout = (u is update) and (c is context) states = self.states timeout_handler = CommandHandler("start", timeout_callback) states.update({ConversationHandler.TIMEOUT: [timeout_handler]}) handler = ConversationHandler( entry_points=[CommandHandler("start", start_callback)], states=states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) async with app: await app.start() await app.process_update(update) await asyncio.sleep(0.9) # check that the conversation has ended by checking that the start message is accepted assert handler.check_update(Update(0, message=message)) assert self.is_timeout await app.stop() @pytest.mark.flaky(3, 1) async def test_conversation_timeout_keeps_extending(self, app, bot, user1): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) # Start state machine, wait, do something, verify the timeout is extended. # t=0 /start (timeout=.5) # t=.35 /brew (timeout=.85) # t=.5 original timeout # t=.6 /pourCoffee (timeout=1.1) # t=.85 second timeout # t=1.1 actual timeout message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.start() await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") assert handler.check_update(Update(0, message=message)) await asyncio.sleep(0.35) # t=.35 assert handler.check_update(Update(0, message=message)) await app.process_update(Update(update_id=0, message=message)) message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") assert handler.check_update(Update(0, message=message)) await asyncio.sleep(0.25) # t=.6 assert handler.check_update(Update(0, message=message)) await app.process_update(Update(update_id=0, message=message)) message.text = "/startCoding" message.entities[0].length = len("/startCoding") assert handler.check_update(Update(0, message=message)) await asyncio.sleep(0.4) # t=1.0 assert handler.check_update(Update(0, message=message)) await asyncio.sleep(0.3) # t=1.3 assert not handler.check_update(Update(0, message=message)) message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) await app.stop() async def test_conversation_timeout_two_users(self, app, bot, user1, user2): handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) # Start state machine, do something as second user, then reach timeout message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.start() await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") assert handler.check_update(Update(0, message=message)) message.from_user = user2 await app.process_update(Update(update_id=0, message=message)) message.text = "/start" message.entities[0].length = len("/start") # Make sure that user2s conversation has not yet started assert handler.check_update(Update(0, message=message)) await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") assert handler.check_update(Update(0, message=message)) await asyncio.sleep(0.7) # check that both conversations have ended by checking that the start message is # accepted again message.text = "/start" message.entities[0].length = len("/start") message.from_user = user1 assert handler.check_update(Update(0, message=message)) message.from_user = user2 assert handler.check_update(Update(0, message=message)) await app.stop() async def test_conversation_handler_timeout_state(self, app, bot, user1): states = self.states states.update( { ConversationHandler.TIMEOUT: [ CommandHandler("brew", self.passout), MessageHandler(~filters.Regex("oding"), self.passout2), ] } ) handler = ConversationHandler( entry_points=self.entry_points, states=states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) # CommandHandler timeout message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.start() await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.7) # check that conversation has ended by checking that start cmd is accepted again message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) assert self.is_timeout # MessageHandler timeout self.is_timeout = False message.text = "/start" message.entities[0].length = len("/start") await app.process_update(Update(update_id=1, message=message)) await asyncio.sleep(0.7) # check that conversation has ended by checking that start cmd is accepted again assert handler.check_update(Update(0, message=message)) assert self.is_timeout # Timeout but no valid handler self.is_timeout = False await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) message.text = "/startCoding" message.entities[0].length = len("/startCoding") await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.7) # check that conversation has ended by checking that start cmd is accepted again message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) assert not self.is_timeout await app.stop() async def test_conversation_handler_timeout_state_context(self, app, bot, user1): states = self.states states.update( { ConversationHandler.TIMEOUT: [ CommandHandler("brew", self.passout_context), MessageHandler(~filters.Regex("oding"), self.passout2_context), ] } ) handler = ConversationHandler( entry_points=self.entry_points, states=states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) # CommandHandler timeout message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.start() await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.7) # check that conversation has ended by checking that start cmd is accepted again message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) assert self.is_timeout # MessageHandler timeout self.is_timeout = False message.text = "/start" message.entities[0].length = len("/start") await app.process_update(Update(update_id=1, message=message)) await asyncio.sleep(0.7) # check that conversation has ended by checking that start cmd is accepted again assert handler.check_update(Update(0, message=message)) assert self.is_timeout # Timeout but no valid handler self.is_timeout = False await app.process_update(Update(update_id=0, message=message)) message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) message.text = "/startCoding" message.entities[0].length = len("/startCoding") await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.7) # check that conversation has ended by checking that start cmd is accepted again message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) assert not self.is_timeout await app.stop() async def test_conversation_timeout_cancel_conflict(self, app, bot, user1): # Start state machine, wait half the timeout, # then call a callback that takes more than the timeout # t=0 /start (timeout=.5) # t=.25 /slowbrew (sleep .5) # | t=.5 original timeout (should not execute) # | t=.75 /slowbrew returns (timeout=1.25) # t=1.25 timeout async def slowbrew(_update, context): await asyncio.sleep(0.25) # Let's give to the original timeout a chance to execute await asyncio.sleep(0.25) # By returning None we do not override the conversation state so # we can see if the timeout has been executed states = self.states states[self.THIRSTY].append(CommandHandler("slowbrew", slowbrew)) states.update({ConversationHandler.TIMEOUT: [MessageHandler(None, self.passout2)]}) handler = ConversationHandler( entry_points=self.entry_points, states=states, fallbacks=self.fallbacks, conversation_timeout=0.5, ) app.add_handler(handler) # CommandHandler timeout message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.start() await app.process_update(Update(update_id=0, message=message)) await asyncio.sleep(0.25) message.text = "/slowbrew" message.entities[0].length = len("/slowbrew") await app.process_update(Update(update_id=0, message=message)) # Check that conversation has not ended by checking that start cmd is not accepted message.text = "/start" message.entities[0].length = len("/start") assert not handler.check_update(Update(0, message=message)) assert not self.is_timeout await asyncio.sleep(0.7) # Check that conversation has ended by checking that start cmd is accepted again message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) assert self.is_timeout await app.stop() async def test_nested_conversation_handler(self, app, bot, user1, user2): self.nested_states[self.DRINKING] = [ ConversationHandler( entry_points=self.drinking_entry_points, states=self.drinking_states, fallbacks=self.drinking_fallbacks, map_to_parent=self.drinking_map_to_parent, ) ] handler = ConversationHandler( entry_points=self.entry_points, states=self.nested_states, fallbacks=self.fallbacks ) app.add_handler(handler) # User one, starts the state machine. message = Message( 0, None, self.group, from_user=user1, text="/start", entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.THIRSTY # The user is thirsty and wants to brew coffee. message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.BREWING # Lets pour some coffee. message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING # The user is holding the cup message.text = "/hold" message.entities[0].length = len("/hold") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.HOLDING # The user is sipping coffee message.text = "/sip" message.entities[0].length = len("/sip") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.SIPPING # The user is swallowing message.text = "/swallow" message.entities[0].length = len("/swallow") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.SWALLOWING # The user is holding the cup again message.text = "/hold" message.entities[0].length = len("/hold") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.HOLDING # The user wants to replenish the coffee supply message.text = "/replenish" message.entities[0].length = len("/replenish") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.REPLENISHING # check that we're in the right state now by checking that the update is accepted message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") assert handler.check_update(Update(0, message=message)) # The user wants to drink their coffee again) await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING # The user is now ready to start coding message.text = "/startCoding" message.entities[0].length = len("/startCoding") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.CODING # The user decides it's time to drink again message.text = "/drinkMore" message.entities[0].length = len("/drinkMore") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING # The user is holding their cup message.text = "/hold" message.entities[0].length = len("/hold") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.HOLDING # The user wants to end with the drinking and go back to coding message.text = "/end" message.entities[0].length = len("/end") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.END # check that we're in the right state now by checking that the update is accepted message.text = "/drinkMore" message.entities[0].length = len("/drinkMore") assert handler.check_update(Update(0, message=message)) # The user wants to drink once more await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING # The user wants to stop altogether message.text = "/stop" message.entities[0].length = len("/stop") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.STOPPING # check that the conversation has ended by checking that the start cmd is accepted message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) async def test_nested_conversation_application_handler_stop(self, app, bot, user1, user2): self.nested_states[self.DRINKING] = [ ConversationHandler( entry_points=self.drinking_entry_points, states=self.drinking_states, fallbacks=self.drinking_fallbacks, map_to_parent=self.drinking_map_to_parent, ) ] handler = ConversationHandler( entry_points=self.entry_points, states=self.nested_states, fallbacks=self.fallbacks ) def test_callback(u, c): self.test_flag = True app.add_handler(handler) app.add_handler(TypeHandler(Update, test_callback), group=1) self.raise_app_handler_stop = True # User one, starts the state machine. message = Message( 0, None, self.group, text="/start", from_user=user1, entities=[ MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start")) ], ) message.set_bot(bot) message._unfreeze() message.entities[0]._unfreeze() async with app: await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.THIRSTY assert not self.test_flag # The user is thirsty and wants to brew coffee. message.text = "/brew" message.entities[0].length = len("/brew") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.BREWING assert not self.test_flag # Lets pour some coffee. message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING assert not self.test_flag # The user is holding the cup message.text = "/hold" message.entities[0].length = len("/hold") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.HOLDING assert not self.test_flag # The user is sipping coffee message.text = "/sip" message.entities[0].length = len("/sip") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.SIPPING assert not self.test_flag # The user is swallowing message.text = "/swallow" message.entities[0].length = len("/swallow") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.SWALLOWING assert not self.test_flag # The user is holding the cup again message.text = "/hold" message.entities[0].length = len("/hold") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.HOLDING assert not self.test_flag # The user wants to replenish the coffee supply message.text = "/replenish" message.entities[0].length = len("/replenish") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.REPLENISHING # check that we're in the right state now by checking that the update is accepted message.text = "/pourCoffee" message.entities[0].length = len("/pourCoffee") assert handler.check_update(Update(0, message=message)) assert not self.test_flag # The user wants to drink their coffee again await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING assert not self.test_flag # The user is now ready to start coding message.text = "/startCoding" message.entities[0].length = len("/startCoding") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.CODING assert not self.test_flag # The user decides it's time to drink again message.text = "/drinkMore" message.entities[0].length = len("/drinkMore") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING assert not self.test_flag # The user is holding their cup message.text = "/hold" message.entities[0].length = len("/hold") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.HOLDING assert not self.test_flag # The user wants to end with the drinking and go back to coding message.text = "/end" message.entities[0].length = len("/end") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.END # check that we're in the right state now by checking that the update is accepted message.text = "/drinkMore" message.entities[0].length = len("/drinkMore") assert handler.check_update(Update(0, message=message)) assert not self.test_flag # The user wants to drink once more message.text = "/drinkMore" message.entities[0].length = len("/drinkMore") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.DRINKING assert not self.test_flag # The user wants to stop altogether message.text = "/stop" message.entities[0].length = len("/stop") await app.process_update(Update(update_id=0, message=message)) assert self.current_state[user1.id] == self.STOPPING # check that the conv has ended by checking that the start cmd is accepted message.text = "/start" message.entities[0].length = len("/start") assert handler.check_update(Update(0, message=message)) assert not self.test_flag @pytest.mark.parametrize("callback_raises", [True, False]) async def test_timeout_non_block(self, app, user1, callback_raises): event = asyncio.Event() async def callback(_, __): await event.wait() if callback_raises: raise RuntimeError return 1 conv_handler = ConversationHandler( entry_points=[MessageHandler(filters.ALL, callback=callback, block=False)], states={ConversationHandler.TIMEOUT: [TypeHandler(Update, self.passout2)]}, fallbacks=[], conversation_timeout=0.5, ) app.add_handler(conv_handler) async with app: await app.start() message = Message( 0, None, self.group, text="/start", from_user=user1, ) assert conv_handler.check_update(Update(0, message=message)) await app.process_update(Update(0, message=message)) await asyncio.sleep(0.7) tasks = asyncio.all_tasks() assert any(":handle_update:non_blocking_cb" in t.get_name() for t in tasks) assert any(":handle_update:timeout_job" in t.get_name() for t in tasks) assert not self.is_timeout event.set() await asyncio.sleep(0.7) assert self.is_timeout == (not callback_raises) await app.stop() async def test_no_timeout_on_end(self, app, user1): conv_handler = ConversationHandler( entry_points=[MessageHandler(filters.ALL, callback=self.start_end)], states={ConversationHandler.TIMEOUT: [TypeHandler(Update, self.passout2)]}, fallbacks=[], conversation_timeout=0.5, ) app.add_handler(conv_handler) async with app: await app.start() message = Message( 0, None, self.group, text="/start", from_user=user1, ) assert conv_handler.check_update(Update(0, message=message)) await app.process_update(Update(0, message=message)) await asyncio.sleep(0.7) assert not self.is_timeout await app.stop() async def test_conversation_handler_block_dont_override(self, app): """This just makes sure that we don't change any attributes of the handlers of the conv""" conv_handler = ConversationHandler( entry_points=self.entry_points, states=self.states, fallbacks=self.fallbacks, block=False, ) all_handlers = conv_handler.entry_points + conv_handler.fallbacks for state_handlers in conv_handler.states.values(): all_handlers += state_handlers for handler in all_handlers: assert handler.block conv_handler = ConversationHandler( entry_points=[CommandHandler("start", self.start_end, block=False)], states={1: [CommandHandler("start", self.start_end, block=False)]}, fallbacks=[CommandHandler("start", self.start_end, block=False)], block=True, ) all_handlers = conv_handler.entry_points + conv_handler.fallbacks for state_handlers in conv_handler.states.values(): all_handlers += state_handlers for handler in all_handlers: assert handler.block is False @pytest.mark.parametrize("default_block", [True, False, None]) @pytest.mark.parametrize("ch_block", [True, False, None]) @pytest.mark.parametrize("handler_block", [True, False, None]) @pytest.mark.parametrize("ext_bot", [True, False], ids=["ExtBot", "Bot"]) async def test_blocking_resolution_order( self, bot_info, default_block, ch_block, handler_block, ext_bot ): event = asyncio.Event() async def callback(_, __): await event.wait() event.clear() self.test_flag = True return 1 if handler_block is not None: handler = CommandHandler("start", callback=callback, block=handler_block) fallback = MessageHandler(filters.ALL, callback, block=handler_block) else: handler = CommandHandler("start", callback=callback) fallback = MessageHandler(filters.ALL, callback, block=handler_block) defaults = Defaults(block=default_block) if default_block is not None else None if ch_block is not None: conv_handler = ConversationHandler( entry_points=[handler], states={1: [handler]}, fallbacks=[fallback], block=ch_block, ) else: conv_handler = ConversationHandler( entry_points=[handler], states={1: [handler]}, fallbacks=[fallback], ) bot = make_bot(bot_info, defaults=defaults) if ext_bot else PytestBot(bot_info["token"]) app = ApplicationBuilder().bot(bot).build() app.add_handler(conv_handler) async with app: start_message = make_command_message("/start") start_message.set_bot(bot) fallback_message = make_command_message("/fallback") fallback_message.set_bot(bot) # This loop makes sure that we test all of entry points, states handler & fallbacks for message in [start_message, fallback_message]: process_update_task = asyncio.create_task( app.process_update(Update(0, message=message)) ) if ( # resolution order is handler_block -> ch_block -> default_block # setting block=True/False on a lower priority setting may only have an effect # if it wasn't set for the higher priority settings (handler_block is False) or ((handler_block is None) and (ch_block is False)) or ( (handler_block is None) and (ch_block is None) and ext_bot and (default_block is False) ) ): # check that the handler was called non-blocking by checking that # `process_update` has finished await asyncio.sleep(0.01) assert process_update_task.done() else: # the opposite assert not process_update_task.done() # In any case, the callback must not have finished assert not self.test_flag # After setting the event, the callback must have finished and in the blocking # case this leads to `process_update` finishing. event.set() await asyncio.sleep(0.01) assert process_update_task.done() assert self.test_flag self.test_flag = False async def test_waiting_state(self, app, user1): event = asyncio.Event() async def callback_1(_, __): self.test_flag = 1 async def callback_2(_, __): self.test_flag = 2 async def callback_3(_, __): self.test_flag = 3 async def blocking(_, __): await event.wait() return 1 conv_handler = ConversationHandler( entry_points=[MessageHandler(filters.ALL, callback=blocking, block=False)], states={ ConversationHandler.WAITING: [ MessageHandler(filters.Regex("1"), callback_1), MessageHandler(filters.Regex("2"), callback_2), ], 1: [MessageHandler(filters.Regex("2"), callback_3)], }, fallbacks=[], ) app.add_handler(conv_handler) message = Message( 0, None, self.group, text="/start", from_user=user1, ) message._unfreeze() async with app: await app.process_update(Update(0, message=message)) assert not self.test_flag message.text = "1" await app.process_update(Update(0, message=message)) assert self.test_flag == 1 message.text = "2" await app.process_update(Update(0, message=message)) assert self.test_flag == 2 event.set() await asyncio.sleep(0.05) self.test_flag = None await app.process_update(Update(0, message=message)) assert self.test_flag == 3