#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2025
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program.  If not, see [http://www.gnu.org/licenses/].
"""The integration of persistence into the application is tested in test_basepersistence."""
import asyncio
import functools
import inspect
import logging
import os
import platform
import signal
import sys
import threading
import time
from collections import defaultdict
from pathlib import Path
from queue import Queue
from random import randrange
from threading import Thread
from typing import Optional

import pytest

from telegram import Bot, Chat, Message, MessageEntity, User
from telegram.error import InvalidToken, TelegramError
from telegram.ext import (
    Application,
    ApplicationBuilder,
    ApplicationHandlerStop,
    BaseHandler,
    CallbackContext,
    CommandHandler,
    ContextTypes,
    Defaults,
    JobQueue,
    MessageHandler,
    PicklePersistence,
    SimpleUpdateProcessor,
    TypeHandler,
    Updater,
    filters,
)
from telegram.warnings import PTBDeprecationWarning, PTBUserWarning
from tests.auxil.asyncio_helpers import call_after
from tests.auxil.build_messages import make_message_update
from tests.auxil.files import SOURCE_ROOT_PATH
from tests.auxil.monkeypatch import empty_get_updates, return_true
from tests.auxil.networking import send_webhook_message
from tests.auxil.pytest_classes import PytestApplication, PytestUpdater, make_bot
from tests.auxil.slots import mro_slots


class CustomContext(CallbackContext):
    pass


class TestApplication:
    """The integration of persistence into the application is tested in
    test_basepersistence.
    """

    message_update = make_message_update(message="Text")
    received = None
    count = 0

    @pytest.fixture(autouse=True, name="reset")
    def _reset_fixture(self):
        self.reset()

    def reset(self):
        self.received = None
        self.count = 0

    async def error_handler_context(self, update, context):
        self.received = context.error.message

    async def error_handler_raise_error(self, update, context):
        raise Exception("Failing bigly")

    async def callback_increase_count(self, update, context):
        self.count += 1

    def callback_set_count(self, count, sleep: Optional[float] = None):
        async def callback(update, context):
            if sleep:
                await asyncio.sleep(sleep)
            self.count = count

        return callback

    def callback_raise_error(self, error_message: str):
        async def callback(update, context):
            raise TelegramError(error_message)

        return callback

    async def callback_received(self, update, context):
        self.received = update.message

    async def callback_context(self, update, context):
        if (
            isinstance(context, CallbackContext)
            and isinstance(context.bot, Bot)
            and isinstance(context.update_queue, Queue)
            and isinstance(context.job_queue, JobQueue)
            and isinstance(context.error, TelegramError)
        ):
            self.received = context.error.message

    async def test_slot_behaviour(self, one_time_bot):
        async with ApplicationBuilder().bot(one_time_bot).build() as app:
            for at in app.__slots__:
                attr = f"_Application{at}" if at.startswith("__") and not at.endswith("__") else at
                assert getattr(app, attr, "err") != "err", f"got extra slot '{attr}'"
            assert len(mro_slots(app)) == len(set(mro_slots(app))), "duplicate slot"

    def test_manual_init_warning(self, recwarn, updater):
        Application(
            bot=None,
            update_queue=None,
            job_queue=None,
            persistence=None,
            context_types=ContextTypes(),
            updater=updater,
            update_processor=False,
            post_init=None,
            post_shutdown=None,
            post_stop=None,
        )
        assert len(recwarn) == 1
        assert (
            str(recwarn[-1].message)
            == "`Application` instances should be built via the `ApplicationBuilder`."
        )
        assert recwarn[0].category is PTBUserWarning
        assert recwarn[0].filename == __file__, "stacklevel is incorrect!"

    @pytest.mark.filterwarnings("ignore: `Application` instances should")
    def test_init(self, one_time_bot):
        update_queue = asyncio.Queue()
        job_queue = JobQueue()
        persistence = PicklePersistence("file_path")
        context_types = ContextTypes()
        update_processor = SimpleUpdateProcessor(1)
        updater = Updater(bot=one_time_bot, update_queue=update_queue)

        async def post_init(application: Application) -> None:
            pass

        async def post_shutdown(application: Application) -> None:
            pass

        async def post_stop(application: Application) -> None:
            pass

        app = Application(
            bot=one_time_bot,
            update_queue=update_queue,
            job_queue=job_queue,
            persistence=persistence,
            context_types=context_types,
            updater=updater,
            update_processor=update_processor,
            post_init=post_init,
            post_shutdown=post_shutdown,
            post_stop=post_stop,
        )
        assert app.bot is one_time_bot
        assert app.update_queue is update_queue
        assert app.job_queue is job_queue
        assert app.persistence is persistence
        assert app.context_types is context_types
        assert app.updater is updater
        assert app.update_queue is updater.update_queue
        assert app.bot is updater.bot
        assert app.update_processor is update_processor
        assert app.post_init is post_init
        assert app.post_shutdown is post_shutdown
        assert app.post_stop is post_stop

        # These should be done by the builder
        assert app.persistence.bot is None
        with pytest.raises(RuntimeError, match="No application was set"):
            app.job_queue.application

        assert isinstance(app.bot_data, dict)
        assert isinstance(app.chat_data[1], dict)
        assert isinstance(app.user_data[1], dict)

    async def test_repr(self, app):
        assert repr(app) == f"PytestApplication[bot={app.bot!r}]"

    def test_job_queue(self, one_time_bot, app, recwarn):
        expected_warning = (
            "No `JobQueue` set up. To use `JobQueue`, you must install PTB via "
            '`pip install "python-telegram-bot[job-queue]"`.'
        )
        assert app.job_queue is app._job_queue
        application = ApplicationBuilder().bot(one_time_bot).job_queue(None).build()
        assert application.job_queue is None
        assert len(recwarn) == 1
        assert str(recwarn[0].message) == expected_warning
        assert recwarn[0].category is PTBUserWarning
        assert recwarn[0].filename == __file__, "wrong stacklevel"

    def test_custom_context_init(self, one_time_bot):
        cc = ContextTypes(
            context=CustomContext,
            user_data=int,
            chat_data=float,
            bot_data=complex,
        )

        application = ApplicationBuilder().bot(one_time_bot).context_types(cc).build()

        assert isinstance(application.user_data[1], int)
        assert isinstance(application.chat_data[1], float)
        assert isinstance(application.bot_data, complex)

    @pytest.mark.parametrize("updater", [True, False])
    async def test_initialize(self, one_time_bot, monkeypatch, updater):
        """Initialization of persistence is tested test_basepersistence"""
        self.test_flag = set()

        async def after_initialize_bot(*args, **kwargs):
            self.test_flag.add("bot")

        async def after_initialize_update_processor(*args, **kwargs):
            self.test_flag.add("update_processor")

        async def after_initialize_updater(*args, **kwargs):
            self.test_flag.add("updater")

        update_processor = SimpleUpdateProcessor(1)
        monkeypatch.setattr(Bot, "initialize", call_after(Bot.initialize, after_initialize_bot))
        monkeypatch.setattr(
            SimpleUpdateProcessor,
            "initialize",
            call_after(SimpleUpdateProcessor.initialize, after_initialize_update_processor),
        )
        monkeypatch.setattr(
            Updater, "initialize", call_after(Updater.initialize, after_initialize_updater)
        )
        if updater:
            app = (
                ApplicationBuilder().bot(one_time_bot).concurrent_updates(update_processor).build()
            )
            await app.initialize()
            assert self.test_flag == {"bot", "update_processor", "updater"}
            await app.shutdown()
        else:
            app = (
                ApplicationBuilder()
                .bot(one_time_bot)
                .updater(None)
                .concurrent_updates(update_processor)
                .build()
            )
            await app.initialize()
            assert self.test_flag == {"bot", "update_processor"}
            await app.shutdown()

    @pytest.mark.parametrize("updater", [True, False])
    async def test_shutdown(self, one_time_bot, monkeypatch, updater):
        """Shutdown of persistence is tested in test_basepersistence"""
        self.test_flag = set()

        def after_bot_shutdown(*args, **kwargs):
            self.test_flag.add("bot")

        def after_shutdown_update_processor(*args, **kwargs):
            self.test_flag.add("update_processor")

        def after_updater_shutdown(*args, **kwargs):
            self.test_flag.add("updater")

        update_processor = SimpleUpdateProcessor(1)
        monkeypatch.setattr(Bot, "shutdown", call_after(Bot.shutdown, after_bot_shutdown))
        monkeypatch.setattr(
            SimpleUpdateProcessor,
            "shutdown",
            call_after(SimpleUpdateProcessor.shutdown, after_shutdown_update_processor),
        )
        monkeypatch.setattr(
            Updater, "shutdown", call_after(Updater.shutdown, after_updater_shutdown)
        )

        if updater:
            async with (
                ApplicationBuilder().bot(one_time_bot).concurrent_updates(update_processor).build()
            ):
                pass
            assert self.test_flag == {"bot", "update_processor", "updater"}
        else:
            async with (
                ApplicationBuilder()
                .bot(one_time_bot)
                .updater(None)
                .concurrent_updates(update_processor)
                .build()
            ):
                pass
            assert self.test_flag == {"bot", "update_processor"}

    async def test_multiple_inits_and_shutdowns(self, app, monkeypatch):
        self.received = defaultdict(int)

        async def after_initialize(*args, **kargs):
            self.received["init"] += 1

        async def after_shutdown(*args, **kwargs):
            self.received["shutdown"] += 1

        monkeypatch.setattr(
            app.bot, "initialize", call_after(app.bot.initialize, after_initialize)
        )
        monkeypatch.setattr(app.bot, "shutdown", call_after(app.bot.shutdown, after_shutdown))

        await app.initialize()
        await app.initialize()
        await app.initialize()
        await app.shutdown()
        await app.shutdown()
        await app.shutdown()

        # 2 instead of 1 since `Updater.initialize` also calls bot.init/shutdown
        assert self.received["init"] == 2
        assert self.received["shutdown"] == 2

    async def test_multiple_init_cycles(self, app):
        # nothing really to assert - this should just not fail
        async with app:
            await app.bot.get_me()
        async with app:
            await app.bot.get_me()

    async def test_start_without_initialize(self, app):
        with pytest.raises(RuntimeError, match="not initialized"):
            await app.start()

    async def test_shutdown_while_running(self, app):
        async with app:
            await app.start()
            with pytest.raises(RuntimeError, match="still running"):
                await app.shutdown()
            await app.stop()

    async def test_start_not_running_after_failure(self, one_time_bot, monkeypatch):
        def start(_):
            raise Exception("Test Exception")

        monkeypatch.setattr(JobQueue, "start", start)
        app = ApplicationBuilder().bot(one_time_bot).job_queue(JobQueue()).build()

        async with app:
            with pytest.raises(Exception, match="Test Exception"):
                await app.start()
            assert app.running is False

    async def test_context_manager(self, monkeypatch, app):
        self.test_flag = set()

        async def after_initialize(*args, **kwargs):
            self.test_flag.add("initialize")

        async def after_shutdown(*args, **kwargs):
            self.test_flag.add("stop")

        monkeypatch.setattr(
            Application, "initialize", call_after(Application.initialize, after_initialize)
        )
        monkeypatch.setattr(
            Application, "shutdown", call_after(Application.shutdown, after_shutdown)
        )

        async with app:
            pass

        assert self.test_flag == {"initialize", "stop"}

    async def test_context_manager_exception_on_init(self, monkeypatch, app):
        async def after_initialize(*args, **kwargs):
            raise RuntimeError("initialize")

        async def after_shutdown(*args):
            self.test_flag = "stop"

        monkeypatch.setattr(
            Application, "initialize", call_after(Application.initialize, after_initialize)
        )
        monkeypatch.setattr(
            Application, "shutdown", call_after(Application.shutdown, after_shutdown)
        )

        with pytest.raises(RuntimeError, match="initialize"):
            async with app:
                pass

        assert self.test_flag == "stop"

    @pytest.mark.parametrize("data", ["chat_data", "user_data"])
    def test_chat_user_data_read_only(self, app, data):
        read_only_data = getattr(app, data)
        writable_data = getattr(app, f"_{data}")
        writable_data[123] = 321
        assert read_only_data == writable_data
        with pytest.raises(TypeError):
            read_only_data[111] = 123

    def test_builder(self, app):
        builder_1 = app.builder()
        builder_2 = app.builder()
        assert isinstance(builder_1, ApplicationBuilder)
        assert isinstance(builder_2, ApplicationBuilder)
        assert builder_1 is not builder_2

        # Make sure that setting a token doesn't raise an exception
        # i.e. check that the builders are "empty"/new
        builder_1.token(app.bot.token)
        builder_2.token(app.bot.token)

    @pytest.mark.parametrize("job_queue", [True, False])
    @pytest.mark.filterwarnings("ignore::telegram.warnings.PTBUserWarning")
    async def test_start_stop_processing_updates(self, one_time_bot, job_queue, monkeypatch):
        # TODO: repeat a similar test for create_task, persistence processing and job queue
        if job_queue:
            app = ApplicationBuilder().bot(one_time_bot).build()
        else:
            app = ApplicationBuilder().bot(one_time_bot).job_queue(None).build()

        async def callback(u, c):
            self.received = u

        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)

        assert not app.running
        assert not app.updater.running
        if job_queue:
            assert not app.job_queue.scheduler.running
        else:
            assert app.job_queue is None
        app.add_handler(TypeHandler(object, callback))

        await app.update_queue.put(1)
        await asyncio.sleep(0.05)
        assert not app.update_queue.empty()
        assert self.received is None

        async with app:
            await app.start()
            assert app.running
            tasks = asyncio.all_tasks()
            assert any(":update_fetcher" in task.get_name() for task in tasks)
            if job_queue:
                assert app.job_queue.scheduler.running
            else:
                assert app.job_queue is None
            # app.start() should not start the updater!
            assert not app.updater.running
            await asyncio.sleep(0.05)
            assert app.update_queue.empty()
            assert self.received == 1
            try:  # just in case start_polling times out
                await app.updater.start_polling()
            except TelegramError:
                pytest.xfail("start_polling timed out")
            else:
                await app.stop()
            assert not app.running
            # app.stop() should not stop the updater!
            assert app.updater.running
            if job_queue:
                assert not app.job_queue.scheduler.running
            else:
                assert app.job_queue is None
            await app.update_queue.put(2)
            await asyncio.sleep(0.05)
            assert not app.update_queue.empty()
            assert self.received != 2
            assert self.received == 1

            await app.updater.stop()

    async def test_error_start_stop_twice(self, app):
        async with app:
            await app.start()
            assert app.running
            with pytest.raises(RuntimeError, match="already running"):
                await app.start()

            await app.stop()
            assert not app.running
            with pytest.raises(RuntimeError, match="not running"):
                await app.stop()

    async def test_one_context_per_update(self, app):
        self.received = None

        async def one(update, context):
            self.received = context

        async def two(update, context):
            if update.message.text == "test":
                if context is not self.received:
                    pytest.fail("Expected same context object, got different")
            elif context is self.received:
                pytest.fail("First handler was wrongly called")

        async with app:
            app.add_handler(MessageHandler(filters.Regex("test"), one), group=1)
            app.add_handler(MessageHandler(filters.ALL, two), group=2)
            u = make_message_update(message="test")
            await app.process_update(u)
            self.received = None
            u = make_message_update(message="something")
            await app.process_update(u)

    def test_add_handler_errors(self, app):
        handler = "not a handler"
        with pytest.raises(TypeError, match="handler is not an instance of"):
            app.add_handler(handler)

        handler = MessageHandler(filters.PHOTO, self.callback_set_count(1))
        with pytest.raises(TypeError, match="group is not int"):
            app.add_handler(handler, "one")

    @pytest.mark.parametrize("group_empty", [True, False])
    async def test_add_remove_handler(self, app, group_empty):
        handler = MessageHandler(filters.ALL, self.callback_increase_count)
        app.add_handler(handler)
        if not group_empty:
            app.add_handler(handler)

        async with app:
            await app.start()
            await app.update_queue.put(self.message_update)
            await asyncio.sleep(0.05)
            assert self.count == 1
            app.remove_handler(handler)
            assert (0 in app.handlers) == (not group_empty)
            await app.update_queue.put(self.message_update)
            assert self.count == 1
            await app.stop()

    async def test_add_remove_handler_non_default_group(self, app):
        handler = MessageHandler(filters.ALL, self.callback_increase_count)
        app.add_handler(handler, group=2)
        with pytest.raises(KeyError):
            app.remove_handler(handler)
        app.remove_handler(handler, group=2)

    async def test_handler_order_in_group(self, app):
        app.add_handler(MessageHandler(filters.PHOTO, self.callback_set_count(1)))
        app.add_handler(MessageHandler(filters.ALL, self.callback_set_count(2)))
        app.add_handler(MessageHandler(filters.TEXT, self.callback_set_count(3)))
        async with app:
            await app.start()
            await app.update_queue.put(self.message_update)
            await asyncio.sleep(0.05)
            assert self.count == 2
            await app.stop()

    async def test_groups(self, app):
        app.add_handler(MessageHandler(filters.ALL, self.callback_increase_count))
        app.add_handler(MessageHandler(filters.ALL, self.callback_increase_count), group=2)
        app.add_handler(MessageHandler(filters.ALL, self.callback_increase_count), group=-1)

        async with app:
            await app.start()
            await app.update_queue.put(self.message_update)
            await asyncio.sleep(0.05)
            assert self.count == 3
            await app.stop()

    async def test_add_handlers(self, app):
        """Tests both add_handler & add_handlers together & confirms the correct insertion
        order"""
        msg_handler_set_count = MessageHandler(filters.TEXT, self.callback_set_count(1))
        msg_handler_inc_count = MessageHandler(filters.PHOTO, self.callback_increase_count)

        app.add_handler(msg_handler_set_count, 1)
        app.add_handlers((msg_handler_inc_count, msg_handler_inc_count), 1)

        photo_update = make_message_update(message=Message(2, None, None, photo=(True,)))

        async with app:
            await app.start()
            # Putting updates in the queue calls the callback
            await app.update_queue.put(self.message_update)
            await app.update_queue.put(photo_update)
            await asyncio.sleep(0.05)  # sleep is required otherwise there is random behaviour

            # Test if handler was added to correct group with correct order-
            assert self.count == 2
            assert len(app.handlers[1]) == 3
            assert app.handlers[1][0] is msg_handler_set_count

            # Now lets test add_handlers when `handlers` is a dict-
            voice_filter_handler_to_check = MessageHandler(
                filters.VOICE, self.callback_increase_count
            )
            app.add_handlers(
                handlers={
                    1: [
                        MessageHandler(filters.USER, self.callback_increase_count),
                        voice_filter_handler_to_check,
                    ],
                    -1: [MessageHandler(filters.CAPTION, self.callback_set_count(2))],
                }
            )

            user_update = make_message_update(
                message=Message(3, None, None, from_user=User(1, "s", True))
            )
            voice_update = make_message_update(message=Message(4, None, None, voice=True))
            await app.update_queue.put(user_update)
            await app.update_queue.put(voice_update)
            await asyncio.sleep(0.05)

            assert self.count == 4
            assert len(app.handlers[1]) == 5
            assert app.handlers[1][-1] is voice_filter_handler_to_check

            await app.update_queue.put(
                make_message_update(message=Message(5, None, None, caption="cap"))
            )
            await asyncio.sleep(0.05)

            assert self.count == 2
            assert len(app.handlers[-1]) == 1

            # Now lets test the errors which can be produced-
            with pytest.raises(TypeError, match="The `group` argument"):
                app.add_handlers({2: [msg_handler_set_count]}, group=0)
            with pytest.raises(TypeError, match="Handlers for group 3"):
                app.add_handlers({3: msg_handler_set_count})
            with pytest.raises(TypeError, match="The `handlers` argument must be a sequence"):
                app.add_handlers({msg_handler_set_count})

            await app.stop()

    async def test_check_update(self, app):
        class TestHandler(BaseHandler):
            def check_update(_, update: object):
                self.received = object()

            def handle_update(
                _,
                update,
                application,
                check_result,
                context,
            ):
                assert application is app
                assert check_result is not self.received

        async with app:
            app.add_handler(TestHandler("callback"))
            await app.start()
            await app.update_queue.put(object())
            await asyncio.sleep(0.05)
            await app.stop()

    async def test_flow_stop(self, app, one_time_bot):
        passed = []

        async def start1(b, u):
            passed.append("start1")
            raise ApplicationHandlerStop

        async def start2(b, u):
            passed.append("start2")

        async def start3(b, u):
            passed.append("start3")

        update = make_message_update(
            message=Message(
                1,
                None,
                None,
                None,
                text="/start",
                entities=[
                    MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start"))
                ],
            ),
        )
        await one_time_bot.initialize()
        update.message.set_bot(one_time_bot)

        async with app:
            # If ApplicationHandlerStop raised handlers in other groups should not be called.
            passed = []
            app.add_handler(CommandHandler("start", start1), 1)
            app.add_handler(CommandHandler("start", start3), 1)
            app.add_handler(CommandHandler("start", start2), 2)
            await app.process_update(update)
            assert passed == ["start1"]

    async def test_flow_stop_by_error_handler(self, app):
        passed = []
        exception = Exception("General exception")

        async def start1(u, c):
            passed.append("start1")
            raise exception

        async def start2(u, c):
            passed.append("start2")

        async def start3(u, c):
            passed.append("start3")

        async def error(u, c):
            passed.append("error")
            passed.append(c.error)
            raise ApplicationHandlerStop

        async with app:
            # If ApplicationHandlerStop raised handlers in other groups should not be called.
            passed = []
            app.add_error_handler(error)
            app.add_handler(TypeHandler(object, start1), 1)
            app.add_handler(TypeHandler(object, start2), 1)
            app.add_handler(TypeHandler(object, start3), 2)
            await app.process_update(1)
            assert passed == ["start1", "error", exception]

    async def test_error_in_handler_part_1(self, app):
        app.add_handler(
            MessageHandler(
                filters.ALL,
                self.callback_raise_error(error_message=self.message_update.message.text),
            )
        )
        app.add_handler(MessageHandler(filters.ALL, self.callback_set_count(42)), group=1)
        app.add_error_handler(self.error_handler_context)

        async with app:
            await app.start()
            await app.update_queue.put(self.message_update)
            await asyncio.sleep(0.05)
            await app.stop()

        assert self.received == self.message_update.message.text
        # Higher groups should still be called
        assert self.count == 42

    async def test_error_in_handler_part_2(self, app, one_time_bot):
        passed = []
        err = Exception("General exception")

        async def start1(u, c):
            passed.append("start1")
            raise err

        async def start2(u, c):
            passed.append("start2")

        async def start3(u, c):
            passed.append("start3")

        async def error(u, c):
            passed.append("error")
            passed.append(c.error)

        update = make_message_update(
            message=Message(
                1,
                None,
                None,
                None,
                text="/start",
                entities=[
                    MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len("/start"))
                ],
            ),
        )
        await one_time_bot.initialize()
        update.message.set_bot(one_time_bot)

        async with app:
            # If an unhandled exception was caught, no further handlers from the same group should
            # be called. Also, the error handler should be called and receive the exception
            passed = []
            app.add_handler(CommandHandler("start", start1), 1)
            app.add_handler(CommandHandler("start", start2), 1)
            app.add_handler(CommandHandler("start", start3), 2)
            app.add_error_handler(error)
            await app.process_update(update)
            assert passed == ["start1", "error", err, "start3"]

    @pytest.mark.parametrize("block", [True, False])
    async def test_error_handler(self, app, block):
        app.add_error_handler(self.error_handler_context)
        app.add_handler(TypeHandler(object, self.callback_raise_error("TestError"), block=block))

        async with app:
            await app.start()
            await app.update_queue.put(1)
            await asyncio.sleep(0.05)
            assert self.received == "TestError"

            # Remove handler
            app.remove_error_handler(self.error_handler_context)
            self.reset()

            await app.update_queue.put(1)
            await asyncio.sleep(0.05)
            assert self.received is None

            await app.stop()

    def test_double_add_error_handler(self, app, caplog):
        app.add_error_handler(self.error_handler_context)
        with caplog.at_level(logging.DEBUG):
            app.add_error_handler(self.error_handler_context)
            assert len(caplog.records) == 1
            assert caplog.records[-1].name == "telegram.ext.Application"
            assert caplog.records[-1].getMessage().startswith("The callback is already registered")

    async def test_error_handler_that_raises_errors(self, app, caplog):
        """Make sure that errors raised in error handlers don't break the main loop of the
        application
        """
        handler_raise_error = TypeHandler(
            int, self.callback_raise_error(error_message="TestError")
        )
        handler_increase_count = TypeHandler(str, self.callback_increase_count)

        app.add_error_handler(self.error_handler_raise_error)
        app.add_handler(handler_raise_error)
        app.add_handler(handler_increase_count)

        with caplog.at_level(logging.ERROR):
            async with app:
                await app.start()
                await app.update_queue.put(1)
                await asyncio.sleep(0.05)
                assert self.count == 0
                assert self.received is None
                assert len(caplog.records) > 0
                assert any(
                    "uncaught error was raised while handling the error with an error_handler"
                    in record.getMessage()
                    and record.name == "telegram.ext.Application"
                    for record in caplog.records
                )

                await app.update_queue.put("1")
                self.received = None
                caplog.clear()
                await asyncio.sleep(0.05)
                assert self.count == 1
                assert self.received is None
                assert not caplog.records

                await app.stop()

    async def test_custom_context_error_handler(self, one_time_bot):
        async def error_handler(_, context):
            self.received = (
                type(context),
                type(context.user_data),
                type(context.chat_data),
                type(context.bot_data),
            )

        application = (
            ApplicationBuilder()
            .bot(one_time_bot)
            .context_types(
                ContextTypes(
                    context=CustomContext, bot_data=int, user_data=float, chat_data=complex
                )
            )
            .build()
        )
        application.add_error_handler(error_handler)
        application.add_handler(
            MessageHandler(filters.ALL, self.callback_raise_error("TestError"))
        )

        async with application:
            await application.process_update(self.message_update)
            await asyncio.sleep(0.05)
            assert self.received == (CustomContext, float, complex, int)

    async def test_custom_context_handler_callback(self, one_time_bot):
        async def callback(_, context):
            self.received = (
                type(context),
                type(context.user_data),
                type(context.chat_data),
                type(context.bot_data),
            )

        application = (
            ApplicationBuilder()
            .bot(one_time_bot)
            .context_types(
                ContextTypes(
                    context=CustomContext, bot_data=int, user_data=float, chat_data=complex
                )
            )
            .build()
        )
        application.add_handler(MessageHandler(filters.ALL, callback))

        async with application:
            await application.process_update(self.message_update)
            await asyncio.sleep(0.05)
            assert self.received == (CustomContext, float, complex, int)

    @pytest.mark.parametrize(
        ("check", "expected"),
        [(True, True), (None, False), (False, False), ({}, True), ("", True), ("check", True)],
    )
    async def test_check_update_handling(self, app, check, expected):
        class MyHandler(BaseHandler):
            def check_update(self, update: object):
                return check

            async def handle_update(
                _,
                update,
                application,
                check_result,
                context,
            ):
                await super().handle_update(
                    update=update,
                    application=application,
                    check_result=check_result,
                    context=context,
                )
                self.received = check_result

        async with app:
            app.add_handler(MyHandler(self.callback_increase_count))
            await app.process_update(1)
            assert self.count == (1 if expected else 0)
            if expected:
                assert self.received == check
            else:
                assert self.received is None

    async def test_non_blocking_handler(self, app):
        event = asyncio.Event()

        async def callback(update, context):
            await event.wait()
            self.count = 42

        app.add_handler(TypeHandler(object, callback, block=False))
        app.add_handler(TypeHandler(object, self.callback_increase_count), group=1)

        async with app:
            await app.start()
            await app.update_queue.put(1)
            task = asyncio.create_task(app.stop())
            await asyncio.sleep(0.05)
            tasks = asyncio.all_tasks()
            assert any(":process_update_non_blocking" in t.get_name() for t in tasks)
            assert self.count == 1
            # Make sure that app stops only once all non blocking callbacks are done
            assert not task.done()
            event.set()
            await asyncio.sleep(0.05)
            assert self.count == 42
            assert task.done()

    async def test_non_blocking_handler_applicationhandlerstop(self, app, recwarn):
        async def callback(update, context):
            raise ApplicationHandlerStop

        app.add_handler(TypeHandler(object, callback, block=False))

        async with app:
            await app.start()
            await app.update_queue.put(1)
            await asyncio.sleep(0.05)
            await app.stop()

        assert len(recwarn) == 1
        assert recwarn[0].category is PTBUserWarning
        assert (
            str(recwarn[0].message)
            == "ApplicationHandlerStop is not supported with handlers running non-blocking."
        )
        assert (
            Path(recwarn[0].filename) == SOURCE_ROOT_PATH / "ext" / "_application.py"
        ), "incorrect stacklevel!"

    async def test_non_blocking_no_error_handler(self, app, caplog):
        app.add_handler(TypeHandler(object, self.callback_raise_error("Test error"), block=False))

        with caplog.at_level(logging.ERROR):
            async with app:
                await app.start()
                await app.update_queue.put(1)
                await asyncio.sleep(0.05)
                assert len(caplog.records) == 1
                assert (
                    caplog.records[-1].getMessage().startswith("No error handlers are registered")
                )
                assert caplog.records[-1].name == "telegram.ext.Application"
                await app.stop()

    @pytest.mark.parametrize("handler_block", [True, False])
    async def test_non_blocking_error_handler(self, app, handler_block):
        event = asyncio.Event()

        async def async_error_handler(update, context):
            await event.wait()
            self.received = "done"

        async def normal_error_handler(update, context):
            self.count = 42

        app.add_error_handler(async_error_handler, block=False)
        app.add_error_handler(normal_error_handler)
        app.add_handler(TypeHandler(object, self.callback_raise_error("err"), block=handler_block))

        async with app:
            await app.start()
            await app.update_queue.put(self.message_update)
            task = asyncio.create_task(app.stop())
            await asyncio.sleep(0.05)
            tasks = asyncio.all_tasks()
            assert any(":process_error:non_blocking" in t.get_name() for t in tasks)
            assert self.count == 42
            assert self.received is None
            event.set()
            await asyncio.sleep(0.05)
            assert self.received == "done"
            assert task.done()

    @pytest.mark.parametrize("handler_block", [True, False])
    async def test_non_blocking_error_handler_applicationhandlerstop(
        self, app, recwarn, handler_block
    ):
        async def callback(update, context):
            raise RuntimeError

        async def error_handler(update, context):
            raise ApplicationHandlerStop

        app.add_handler(TypeHandler(object, callback, block=handler_block))
        app.add_error_handler(error_handler, block=False)

        async with app:
            await app.start()
            await app.update_queue.put(1)
            await asyncio.sleep(0.05)
            await app.stop()

        assert len(recwarn) == 1
        assert recwarn[0].category is PTBUserWarning
        assert (
            str(recwarn[0].message)
            == "ApplicationHandlerStop is not supported with handlers running non-blocking."
        )
        assert (
            Path(recwarn[0].filename) == SOURCE_ROOT_PATH / "ext" / "_application.py"
        ), "incorrect stacklevel!"

    @pytest.mark.parametrize(("block", "expected_output"), [(False, 0), (True, 5)])
    async def test_default_block_error_handler(self, bot_info, block, expected_output):
        async def error_handler(*args, **kwargs):
            await asyncio.sleep(0.1)
            self.count = 5

        bot = make_bot(bot_info, defaults=Defaults(block=block))
        app = Application.builder().bot(bot).build()
        async with app:
            app.add_handler(TypeHandler(object, self.callback_raise_error("error")))
            app.add_error_handler(error_handler)
            await app.process_update(1)
            await asyncio.sleep(0.05)
            assert self.count == expected_output
            await asyncio.sleep(0.1)
            assert self.count == 5

    @pytest.mark.parametrize(("block", "expected_output"), [(False, 0), (True, 5)])
    async def test_default_block_handler(self, bot_info, block, expected_output):
        bot = make_bot(bot_info, defaults=Defaults(block=block))
        app = Application.builder().bot(bot).build()
        async with app:
            app.add_handler(TypeHandler(object, self.callback_set_count(5, sleep=0.1)))
            await app.process_update(1)
            await asyncio.sleep(0.05)
            assert self.count == expected_output
            await asyncio.sleep(0.15)
            assert self.count == 5

    @pytest.mark.parametrize("handler_block", [True, False])
    @pytest.mark.parametrize("error_handler_block", [True, False])
    async def test_nonblocking_handler_raises_and_non_blocking_error_handler_raises(
        self, app, caplog, handler_block, error_handler_block
    ):
        handler = TypeHandler(object, self.callback_raise_error("error"), block=handler_block)
        app.add_handler(handler)
        app.add_error_handler(self.error_handler_raise_error, block=error_handler_block)

        async with app:
            await app.start()
            with caplog.at_level(logging.ERROR):
                await app.update_queue.put(1)
                await asyncio.sleep(0.05)
                assert len(caplog.records) == 1
                assert caplog.records[-1].name == "telegram.ext.Application"
                assert (
                    caplog.records[-1]
                    .getMessage()
                    .startswith("An error was raised and an uncaught")
                )

            # Make sure that the main loop still runs
            app.remove_handler(handler)
            app.add_handler(MessageHandler(filters.ALL, self.callback_increase_count, block=True))
            await app.update_queue.put(self.message_update)
            await asyncio.sleep(0.05)
            assert self.count == 1

            await app.stop()

    @pytest.mark.parametrize(
        "message",
        [
            Message(message_id=1, chat=Chat(id=2, type=None), migrate_from_chat_id=1, date=None),
            Message(message_id=1, chat=Chat(id=1, type=None), migrate_to_chat_id=2, date=None),
            Message(message_id=1, chat=Chat(id=1, type=None), date=None),
            None,
        ],
    )
    @pytest.mark.parametrize("old_chat_id", [None, 1, "1"])
    @pytest.mark.parametrize("new_chat_id", [None, 2, "1"])
    def test_migrate_chat_data(self, app, message: "Message", old_chat_id: int, new_chat_id: int):
        def call(match: str):
            with pytest.raises(ValueError, match=match):
                app.migrate_chat_data(
                    message=message, old_chat_id=old_chat_id, new_chat_id=new_chat_id
                )

        if message and (old_chat_id or new_chat_id):
            call(r"^Message and chat_id pair are mutually exclusive$")
            return

        if not any((message, old_chat_id, new_chat_id)):
            call(r"^chat_id pair or message must be passed$")
            return

        if message:
            if message.migrate_from_chat_id is None and message.migrate_to_chat_id is None:
                call(r"^Invalid message instance")
                return
            effective_old_chat_id = message.migrate_from_chat_id or message.chat.id
            effective_new_chat_id = message.migrate_to_chat_id or message.chat.id

        elif not (isinstance(old_chat_id, int) and isinstance(new_chat_id, int)):
            call(r"^old_chat_id and new_chat_id must be integers$")
            return
        else:
            effective_old_chat_id = old_chat_id
            effective_new_chat_id = new_chat_id

        app.chat_data[effective_old_chat_id]["key"] = "test"
        app.migrate_chat_data(message=message, old_chat_id=old_chat_id, new_chat_id=new_chat_id)
        assert effective_old_chat_id not in app.chat_data
        assert app.chat_data[effective_new_chat_id]["key"] == "test"

    @pytest.mark.parametrize(
        ("c_id", "expected"),
        [(321, {222: "remove_me"}), (111, {321: {"not_empty": "no"}, 222: "remove_me"})],
        ids=["test chat_id removal", "test no key in data (no error)"],
    )
    def test_drop_chat_data(self, app, c_id, expected):
        app._chat_data.update({321: {"not_empty": "no"}, 222: "remove_me"})
        app.drop_chat_data(c_id)
        assert app.chat_data == expected

    @pytest.mark.parametrize(
        ("u_id", "expected"),
        [(321, {222: "remove_me"}), (111, {321: {"not_empty": "no"}, 222: "remove_me"})],
        ids=["test user_id removal", "test no key in data (no error)"],
    )
    def test_drop_user_data(self, app, u_id, expected):
        app._user_data.update({321: {"not_empty": "no"}, 222: "remove_me"})
        app.drop_user_data(u_id)
        assert app.user_data == expected

    async def test_create_task_basic(self, app):
        async def callback():
            await asyncio.sleep(0.05)
            self.count = 42
            return 43

        task = app.create_task(callback(), name="test_task")
        assert task.get_name() == "test_task"
        await asyncio.sleep(0.01)
        assert not task.done()
        out = await task
        assert task.done()
        assert self.count == 42
        assert out == 43

    @pytest.mark.parametrize("running", [True, False])
    async def test_create_task_awaiting_warning(self, app, running, recwarn):
        async def callback():
            await asyncio.sleep(0.1)
            return 43

        async with app:
            if running:
                await app.start()

            task = app.create_task(callback())

            if running:
                assert len(recwarn) == 0
                assert not task.done()
                await app.stop()
                assert task.done()
                assert task.result() == 43
            else:
                assert len(recwarn) == 1
                assert recwarn[0].category is PTBUserWarning
                assert "won't be automatically awaited" in str(recwarn[0].message)
                assert recwarn[0].filename == __file__, "wrong stacklevel!"
                assert not task.done()
                await task

    @pytest.mark.parametrize("update", [None, object()])
    async def test_create_task_error_handling(self, app, update):
        exception = RuntimeError("TestError")

        async def callback():
            raise exception

        async def error(update_arg, context):
            self.received = update_arg, context.error

        app.add_error_handler(error)
        if update:
            task = app.create_task(callback(), update=update)
        else:
            task = app.create_task(callback())

        with pytest.raises(RuntimeError, match="TestError"):
            await task
        assert task.exception() is exception
        assert isinstance(self.received, tuple)
        assert self.received[0] is update
        assert self.received[1] is exception

    async def test_create_task_cancel_task(self, app):
        async def callback():
            await asyncio.sleep(5)

        async def error(update_arg, context):
            self.received = update_arg, context.error

        app.add_error_handler(error)
        async with app:
            await app.start()
            task = app.create_task(callback())
            await asyncio.sleep(0.05)
            task.cancel()

            with pytest.raises(asyncio.CancelledError):
                await task
            with pytest.raises(asyncio.CancelledError):
                assert task.exception()

            # Error handlers should not be called if task was cancelled
            assert self.received is None

            # make sure that the cancelled task doesn't block the stopping of the app
            await app.stop()

    async def test_await_create_task_tasks_on_stop(self, app):
        event_1 = asyncio.Event()
        event_2 = asyncio.Event()

        async def callback_1():
            await event_1.wait()

        async def callback_2():
            await event_2.wait()

        async with app:
            await app.start()
            task_1 = app.create_task(callback_1())
            task_2 = app.create_task(callback_2())
            event_2.set()
            await task_2
            assert not task_1.done()
            stop_task = asyncio.create_task(app.stop())
            assert not stop_task.done()
            await asyncio.sleep(0.1)
            assert not stop_task.done()
            event_1.set()
            await asyncio.sleep(0.05)
            assert stop_task.done()

    async def test_create_task_awaiting_future(self, app):
        async def callback():
            await asyncio.sleep(0.01)
            return 42

        # `asyncio.gather` returns an `asyncio.Future` and not an
        # `asyncio.Task`
        out = await app.create_task(asyncio.gather(callback()))
        assert out == [42]

    @pytest.mark.skipif(sys.version_info >= (3, 12), reason="generator coroutines are deprecated")
    async def test_create_task_awaiting_generator(self, app, recwarn):
        event = asyncio.Event()

        def gen():
            yield
            event.set()

        await app.create_task(gen())
        assert event.is_set()
        assert len(recwarn) == 2  # 1st warning is: tasks not being awaited when app isn't running
        assert recwarn[1].category is PTBDeprecationWarning
        assert "Generator-based coroutines are deprecated" in str(recwarn[1].message)

    async def test_no_update_processor(self, app):
        queue = asyncio.Queue()
        event_1 = asyncio.Event()
        event_2 = asyncio.Event()
        await queue.put(event_1)
        await queue.put(event_2)

        async def callback(u, c):
            await asyncio.sleep(0.1)
            event = await queue.get()
            event.set()

        app.add_handler(TypeHandler(object, callback))
        async with app:
            await app.start()
            await app.update_queue.put(1)
            await app.update_queue.put(2)
            assert not event_1.is_set()
            assert not event_2.is_set()
            await asyncio.sleep(0.15)
            assert event_1.is_set()
            assert not event_2.is_set()
            await asyncio.sleep(0.1)
            assert event_1.is_set()
            assert event_2.is_set()

            await app.stop()

    @pytest.mark.parametrize("update_processor", [15, 50, 100])
    async def test_update_processor(self, one_time_bot, update_processor):
        # We don't test with `True` since the large number of parallel coroutines quickly leads
        # to test instabilities
        app = Application.builder().bot(one_time_bot).concurrent_updates(update_processor).build()
        events = {
            i: asyncio.Event() for i in range(app.update_processor.max_concurrent_updates + 10)
        }
        queue = asyncio.Queue()
        for event in events.values():
            await queue.put(event)

        async def callback(u, c):
            await asyncio.sleep(0.5)
            (await queue.get()).set()

        app.add_handler(TypeHandler(object, callback))
        async with app:
            await app.start()
            for i in range(app.update_processor.max_concurrent_updates + 10):
                await app.update_queue.put(i)

            for i in range(app.update_processor.max_concurrent_updates + 10):
                assert not events[i].is_set()

            await asyncio.sleep(0.9)
            tasks = asyncio.all_tasks()
            assert any(":process_concurrent_update" in task.get_name() for task in tasks)
            for i in range(app.update_processor.max_concurrent_updates):
                assert events[i].is_set()
            for i in range(
                app.update_processor.max_concurrent_updates,
                app.update_processor.max_concurrent_updates + 10,
            ):
                assert not events[i].is_set()

            await asyncio.sleep(0.5)
            for i in range(app.update_processor.max_concurrent_updates + 10):
                assert events[i].is_set()

            await app.stop()

    async def test_update_processor_done_on_shutdown(self, one_time_bot):
        app = Application.builder().bot(one_time_bot).concurrent_updates(True).build()
        event = asyncio.Event()

        async def callback(update, context):
            await event.wait()

        app.add_handler(TypeHandler(object, callback))

        async with app:
            await app.start()
            await app.update_queue.put(1)
            stop_task = asyncio.create_task(app.stop())
            await asyncio.sleep(0.1)
            assert not stop_task.done()
            event.set()
            await asyncio.sleep(0.05)
            assert stop_task.done()

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_polling_basic(self, app, monkeypatch, caplog):
        exception_event = threading.Event()
        exception_testing_done = threading.Event()
        update_event = threading.Event()
        exception = TelegramError("This is a test error")
        assertions = {}

        async def get_updates(*args, **kwargs):
            if exception_event.is_set():
                raise exception

            # This makes sure that other coroutines have a chance of running as well
            if exception_testing_done.is_set() and app.updater.running:
                # the longer sleep makes sure that we can exit also while get_updates is running
                await asyncio.sleep(20)
            else:
                await asyncio.sleep(0.01)

            update_event.set()
            return [self.message_update]

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            # Check that everything's running
            assertions["app_running"] = app.running
            assertions["updater_running"] = app.updater.running
            assertions["job_queue_running"] = app.job_queue.scheduler.running

            # Check that we're getting updates
            update_event.wait()
            time.sleep(0.05)
            assertions["getting_updates"] = self.count == 42

            # Check that errors are properly handled during polling
            exception_event.set()
            time.sleep(0.05)
            assertions["exception_handling"] = self.received == exception.message
            exception_testing_done.set()

            # So that the get_updates call on shutdown doesn't fail
            exception_event.clear()

            time.sleep(1)
            os.kill(os.getpid(), signal.SIGINT)
            time.sleep(0.1)

            # # Assert that everything has stopped running
            assertions["app_not_running"] = not app.running
            assertions["updater_not_running"] = not app.updater.running
            assertions["job_queue_not_running"] = not app.job_queue.scheduler.running

        monkeypatch.setattr(app.bot, "get_updates", get_updates)
        app.add_error_handler(self.error_handler_context)
        app.add_handler(TypeHandler(object, self.callback_set_count(42)))

        thread = Thread(target=thread_target)
        thread.start()
        with caplog.at_level(logging.DEBUG):
            app.run_polling(drop_pending_updates=True, close_loop=False)
            thread.join()

        assert len(assertions) == 8
        for key, value in assertions.items():
            assert value, f"assertion '{key}' failed!"

        found_log = False
        for record in caplog.records:
            if "received stop signal" in record.getMessage() and record.levelno == logging.DEBUG:
                found_log = True
        assert found_log

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_polling_post_init(self, one_time_bot, monkeypatch):
        events = []

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            os.kill(os.getpid(), signal.SIGINT)

        async def post_init(app: Application) -> None:
            events.append("post_init")

        app = (
            Application.builder()
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .post_init(post_init)
            .build()
        )
        app.bot._unfreeze()
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)
        monkeypatch.setattr(
            app, "initialize", call_after(app.initialize, lambda _: events.append("init"))
        )
        monkeypatch.setattr(
            app.updater,
            "start_polling",
            call_after(app.updater.start_polling, lambda _: events.append("start_polling")),
        )
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)

        thread = Thread(target=thread_target)
        thread.start()
        app.run_polling(drop_pending_updates=True, close_loop=False)
        thread.join()
        assert events == ["init", "post_init", "start_polling"], "Wrong order of events detected!"

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_polling_post_shutdown(self, one_time_bot, monkeypatch):
        events = []

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            os.kill(os.getpid(), signal.SIGINT)

        async def post_shutdown(app: Application) -> None:
            events.append("post_shutdown")

        app = (
            Application.builder()
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .post_shutdown(post_shutdown)
            .build()
        )
        app.bot._unfreeze()
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)
        monkeypatch.setattr(
            app, "shutdown", call_after(app.shutdown, lambda _: events.append("shutdown"))
        )
        monkeypatch.setattr(
            app.updater,
            "shutdown",
            call_after(app.updater.shutdown, lambda _: events.append("updater.shutdown")),
        )
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)

        thread = Thread(target=thread_target)
        thread.start()
        app.run_polling(drop_pending_updates=True, close_loop=False)
        thread.join()
        assert events == [
            "updater.shutdown",
            "shutdown",
            "post_shutdown",
        ], "Wrong order of events detected!"

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_polling_post_stop(self, one_time_bot, monkeypatch):
        events = []

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            os.kill(os.getpid(), signal.SIGINT)

        async def post_stop(app: Application) -> None:
            events.append("post_stop")

        app = (
            Application.builder()
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .post_stop(post_stop)
            .build()
        )
        app.bot._unfreeze()
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)
        monkeypatch.setattr(app, "stop", call_after(app.stop, lambda _: events.append("stop")))
        monkeypatch.setattr(
            app.updater,
            "stop",
            call_after(app.updater.stop, lambda _: events.append("updater.stop")),
        )
        monkeypatch.setattr(
            app.updater,
            "shutdown",
            call_after(app.updater.shutdown, lambda _: events.append("updater.shutdown")),
        )
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)

        thread = Thread(target=thread_target)
        thread.start()
        app.run_polling(drop_pending_updates=True, close_loop=False)
        thread.join()
        assert events == [
            "updater.stop",
            "stop",
            "post_stop",
            "updater.shutdown",
        ], "Wrong order of events detected!"

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_polling_parameters_passing(self, app, monkeypatch):
        # First check that the default values match and that we have all arguments there
        updater_signature = inspect.signature(app.updater.start_polling)
        app_signature = inspect.signature(app.run_polling)

        for name, param in updater_signature.parameters.items():
            if name == "error_callback":
                assert name not in app_signature.parameters
                continue
            assert name in app_signature.parameters
            assert param.kind == app_signature.parameters[name].kind
            assert param.default == app_signature.parameters[name].default

        # Check that we pass them correctly
        async def start_polling(_, **kwargs):
            self.received = kwargs
            return True

        async def stop(_, **kwargs):
            return True

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            time.sleep(0.1)
            os.kill(os.getpid(), signal.SIGINT)

        monkeypatch.setattr(Updater, "start_polling", start_polling)
        monkeypatch.setattr(Updater, "stop", stop)
        thread = Thread(target=thread_target)
        thread.start()
        app.run_polling(close_loop=False)
        thread.join()

        assert set(self.received.keys()) == set(updater_signature.parameters.keys())
        for name, param in updater_signature.parameters.items():
            if name == "error_callback":
                assert self.received[name] is not None
            else:
                assert self.received[name] == param.default

        expected = {
            name: name for name in updater_signature.parameters if name != "error_callback"
        }
        thread = Thread(target=thread_target)
        thread.start()
        app.run_polling(close_loop=False, **expected)
        thread.join()

        assert set(self.received.keys()) == set(updater_signature.parameters.keys())
        assert self.received.pop("error_callback", None)
        assert self.received == expected

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_webhook_basic(self, app, monkeypatch, caplog):
        assertions = {}

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            # Check that everything's running
            assertions["app_running"] = app.running
            assertions["updater_running"] = app.updater.running
            assertions["job_queue_running"] = app.job_queue.scheduler.running

            # Check that we're getting updates
            loop = asyncio.new_event_loop()
            loop.run_until_complete(
                send_webhook_message(ip, port, self.message_update.to_json(), "TOKEN")
            )
            loop.close()
            time.sleep(0.05)
            assertions["getting_updates"] = self.count == 42

            os.kill(os.getpid(), signal.SIGINT)
            time.sleep(0.1)

            # # Assert that everything has stopped running
            assertions["app_not_running"] = not app.running
            assertions["updater_not_running"] = not app.updater.running
            assertions["job_queue_not_running"] = not app.job_queue.scheduler.running

        app.add_handler(TypeHandler(object, self.callback_set_count(42)))

        thread = Thread(target=thread_target)
        thread.start()

        ip = "127.0.0.1"
        port = randrange(1024, 49152)

        with caplog.at_level(logging.DEBUG):
            app.run_webhook(
                ip_address=ip,
                port=port,
                url_path="TOKEN",
                drop_pending_updates=True,
                close_loop=False,
            )
            thread.join()

        assert len(assertions) == 7
        for key, value in assertions.items():
            assert value, f"assertion '{key}' failed!"

        found_log = False
        for record in caplog.records:
            if "received stop signal" in record.getMessage() and record.levelno == logging.DEBUG:
                found_log = True
        assert found_log

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_webhook_post_init(self, one_time_bot, monkeypatch):
        events = []

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            os.kill(os.getpid(), signal.SIGINT)

        async def post_init(app: Application) -> None:
            events.append("post_init")

        app = (
            Application.builder()
            .post_init(post_init)
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .build()
        )
        app.bot._unfreeze()

        monkeypatch.setattr(
            app, "initialize", call_after(app.initialize, lambda _: events.append("init"))
        )
        monkeypatch.setattr(
            app.updater,
            "start_webhook",
            call_after(app.updater.start_webhook, lambda _: events.append("start_webhook")),
        )
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)
        monkeypatch.setattr(app.bot, "set_webhook", return_true)

        thread = Thread(target=thread_target)
        thread.start()

        ip = "127.0.0.1"
        port = randrange(1024, 49152)

        app.run_webhook(
            ip_address=ip,
            port=port,
            url_path="TOKEN",
            drop_pending_updates=True,
            close_loop=False,
        )
        thread.join()
        assert events == ["init", "post_init", "start_webhook"], "Wrong order of events detected!"

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_webhook_post_shutdown(self, one_time_bot, monkeypatch):
        events = []

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            os.kill(os.getpid(), signal.SIGINT)

        async def post_shutdown(app: Application) -> None:
            events.append("post_shutdown")

        app = (
            Application.builder()
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .post_shutdown(post_shutdown)
            .build()
        )
        app.bot._unfreeze()

        monkeypatch.setattr(
            app, "shutdown", call_after(app.shutdown, lambda _: events.append("shutdown"))
        )
        monkeypatch.setattr(
            app.updater,
            "shutdown",
            call_after(app.updater.shutdown, lambda _: events.append("updater.shutdown")),
        )
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)
        monkeypatch.setattr(app.bot, "set_webhook", return_true)

        thread = Thread(target=thread_target)
        thread.start()

        ip = "127.0.0.1"
        port = randrange(1024, 49152)

        app.run_webhook(
            ip_address=ip,
            port=port,
            url_path="TOKEN",
            drop_pending_updates=True,
            close_loop=False,
        )
        thread.join()
        assert events == [
            "updater.shutdown",
            "shutdown",
            "post_shutdown",
        ], "Wrong order of events detected!"

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_webhook_post_stop(self, one_time_bot, monkeypatch):
        events = []

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            os.kill(os.getpid(), signal.SIGINT)

        async def post_stop(app: Application) -> None:
            events.append("post_stop")

        app = (
            Application.builder()
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .post_stop(post_stop)
            .build()
        )
        app.bot._unfreeze()

        monkeypatch.setattr(app, "stop", call_after(app.stop, lambda _: events.append("stop")))
        monkeypatch.setattr(
            app.updater,
            "stop",
            call_after(app.updater.stop, lambda _: events.append("updater.stop")),
        )
        monkeypatch.setattr(
            app.updater,
            "shutdown",
            call_after(app.updater.shutdown, lambda _: events.append("updater.shutdown")),
        )
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)
        monkeypatch.setattr(app.bot, "set_webhook", return_true)

        thread = Thread(target=thread_target)
        thread.start()

        ip = "127.0.0.1"
        port = randrange(1024, 49152)

        app.run_webhook(
            ip_address=ip,
            port=port,
            url_path="TOKEN",
            drop_pending_updates=True,
            close_loop=False,
        )
        thread.join()
        assert events == [
            "updater.stop",
            "stop",
            "post_stop",
            "updater.shutdown",
        ], "Wrong order of events detected!"

    @pytest.mark.skipif(
        platform.system() == "Windows",
        reason="Can't send signals without stopping whole process on windows",
    )
    def test_run_webhook_parameters_passing(self, one_time_bot, monkeypatch):
        # Check that we pass them correctly

        async def start_webhook(_, **kwargs):
            self.received = kwargs
            return True

        async def stop(_, **kwargs):
            return True

        # First check that the default values match and that we have all arguments there
        updater_signature = inspect.signature(Updater.start_webhook)

        monkeypatch.setattr(Updater, "start_webhook", start_webhook)
        monkeypatch.setattr(Updater, "stop", stop)
        app = ApplicationBuilder().bot(one_time_bot).build()
        app_signature = inspect.signature(app.run_webhook)

        for name, param in updater_signature.parameters.items():
            if name == "self":
                continue
            assert name in app_signature.parameters
            assert param.kind == app_signature.parameters[name].kind
            assert param.default == app_signature.parameters[name].default

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            time.sleep(0.1)
            os.kill(os.getpid(), signal.SIGINT)

        thread = Thread(target=thread_target)
        thread.start()
        app.run_webhook(close_loop=False)
        thread.join()

        assert set(self.received.keys()) == set(updater_signature.parameters.keys()) - {"self"}
        for name, param in updater_signature.parameters.items():
            if name == "self":
                continue
            assert self.received[name] == param.default

        expected = {name: name for name in updater_signature.parameters if name != "self"}
        thread = Thread(target=thread_target)
        thread.start()
        app.run_webhook(close_loop=False, **expected)
        thread.join()

        assert set(self.received.keys()) == set(expected.keys())
        assert self.received == expected

    @pytest.mark.parametrize("exception", [SystemExit, KeyboardInterrupt])
    def test_raise_system_exit_keyboard_interrupt_post_init(
        self, one_time_bot, monkeypatch, exception
    ):
        async def post_init(application):
            raise exception

        called_callbacks = set()

        async def callback(*args, **kwargs):
            called_callbacks.add(kwargs["name"])

        for cls, method, entry in [
            (Application, "initialize", "app_initialize"),
            (Application, "start", "app_start"),
            (Application, "stop", "app_stop"),
            (Application, "shutdown", "app_shutdown"),
            (Updater, "initialize", "updater_initialize"),
            (Updater, "shutdown", "updater_shutdown"),
            (Updater, "stop", "updater_stop"),
            (Updater, "start_polling", "updater_start_polling"),
        ]:

            def after(_, name):
                called_callbacks.add(name)

            monkeypatch.setattr(
                cls,
                method,
                call_after(getattr(cls, method), functools.partial(after, name=entry)),
            )

        app = (
            ApplicationBuilder()
            .bot(one_time_bot)
            .post_init(post_init)
            .post_stop(functools.partial(callback, name="post_stop"))
            .post_shutdown(functools.partial(callback, name="post_shutdown"))
            .build()
        )

        app.run_polling(close_loop=False)

        # This checks two things:
        # 1. start/stop are *not* called!
        # 2. we do have a graceful shutdown
        assert called_callbacks == {
            "app_initialize",
            "updater_initialize",
            "app_shutdown",
            "post_shutdown",
            "updater_shutdown",
        }

    @pytest.mark.parametrize("exception", [SystemExit("PTBTest"), KeyboardInterrupt("PTBTest")])
    @pytest.mark.parametrize("kind", ["handler", "error_handler", "job"])
    # @pytest.mark.parametrize("block", [True, False])
    # Testing with block=False would be nice but that doesn't work well with pytest for some reason
    # in any case, block=False is the simpler behavior since it is roughly similar to what happens
    # when you hit CTRL+C in the commandline.
    def test_raise_system_exit_keyboard_jobs_handlers(
        self, one_time_bot, monkeypatch, exception, kind, caplog
    ):
        async def queue_and_raise(application):
            await application.update_queue.put("will_not_be_processed")
            raise exception

        async def handler_callback(update, context):
            if kind == "handler":
                await queue_and_raise(context.application)
            elif kind == "error_handler":
                raise TelegramError("Triggering error callback")

        async def error_callback(update, context):
            await queue_and_raise(context.application)

        async def job_callback(context):
            await queue_and_raise(context.application)

        async def enqueue_update():
            await asyncio.sleep(0.5)
            await app.update_queue.put(1)

        async def post_init(application):
            if kind == "job":
                application.job_queue.run_once(when=0.5, callback=job_callback)
            else:
                app.create_task(enqueue_update())

        async def update_logger_callback(update, context):
            context.bot_data.setdefault("processed_updates", set()).add(update)

        called_callbacks = set()

        async def callback(*args, **kwargs):
            called_callbacks.add(kwargs["name"])

        for cls, method, entry in [
            (Application, "initialize", "app_initialize"),
            (Application, "start", "app_start"),
            (Application, "stop", "app_stop"),
            (Application, "shutdown", "app_shutdown"),
            (Updater, "initialize", "updater_initialize"),
            (Updater, "shutdown", "updater_shutdown"),
            (Updater, "stop", "updater_stop"),
            (Updater, "start_polling", "updater_start_polling"),
        ]:

            def after(_, name):
                called_callbacks.add(name)

            monkeypatch.setattr(
                cls,
                method,
                call_after(getattr(cls, method), functools.partial(after, name=entry)),
            )

        app = (
            ApplicationBuilder()
            .bot(one_time_bot)
            .post_init(post_init)
            .post_stop(functools.partial(callback, name="post_stop"))
            .post_shutdown(functools.partial(callback, name="post_shutdown"))
            .build()
        )
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)

        app.add_handler(TypeHandler(object, update_logger_callback), group=-10)
        app.add_handler(TypeHandler(object, handler_callback))
        app.add_error_handler(error_callback)
        with caplog.at_level(logging.DEBUG):
            app.run_polling(close_loop=False)

        # This checks that we have a clean shutdown even when the user raises SystemExit
        # or KeyboardInterrupt in a handler/error handler/job callback
        assert called_callbacks == {
            "app_initialize",
            "app_shutdown",
            "app_start",
            "app_stop",
            "post_shutdown",
            "post_stop",
            "updater_initialize",
            "updater_shutdown",
            "updater_start_polling",
            "updater_stop",
        }

        # These next checks make sure that the update queue is properly cleaned even if there are
        # still pending updates in the queue
        # Unfortunately this is apparently extremely hard to get right with jobs, so we're
        # skipping that case for the sake of simplicity
        if kind == "job":
            return

        found = False
        for record in caplog.records:
            if record.getMessage() != "Dropping pending update: will_not_be_processed":
                continue
            assert record.name == "telegram.ext.Application"
            assert record.levelno == logging.DEBUG
            found = True
        assert found, "`Dropping pending updates` message not found in logs!"
        assert "will_not_be_processed" not in app.bot_data.get("processed_updates", set())

    def test_run_without_updater(self, one_time_bot):
        app = ApplicationBuilder().bot(one_time_bot).updater(None).build()

        with pytest.raises(RuntimeError, match="only available if the application has an Updater"):
            app.run_webhook()

        with pytest.raises(RuntimeError, match="only available if the application has an Updater"):
            app.run_polling()

    @pytest.mark.parametrize("method", ["start", "initialize"])
    @pytest.mark.filterwarnings("ignore::telegram.warnings.PTBUserWarning")
    def test_run_error_in_application(self, one_time_bot, monkeypatch, method):
        shutdowns = []

        async def raise_method(*args, **kwargs):
            raise RuntimeError("Test Exception")

        def after_shutdown(name):
            def _after_shutdown(*args, **kwargs):
                shutdowns.append(name)

            return _after_shutdown

        monkeypatch.setattr(Application, method, raise_method)
        monkeypatch.setattr(
            Application,
            "shutdown",
            call_after(Application.shutdown, after_shutdown("application")),
        )
        monkeypatch.setattr(
            Updater, "shutdown", call_after(Updater.shutdown, after_shutdown("updater"))
        )
        app = ApplicationBuilder().bot(one_time_bot).build()
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)

        with pytest.raises(RuntimeError, match="Test Exception"):
            app.run_polling(close_loop=False)

        assert not app.running
        assert not app.updater.running
        if method == "initialize":
            # If App.initialize fails, then App.shutdown pretty much does nothing, especially
            # doesn't call Updater.shutdown.
            assert set(shutdowns) == {"application"}
        else:
            assert set(shutdowns) == {"application", "updater"}

    @pytest.mark.parametrize("method", ["start_polling", "start_webhook"])
    @pytest.mark.filterwarnings("ignore::telegram.warnings.PTBUserWarning")
    def test_run_error_in_updater(self, one_time_bot, monkeypatch, method):
        shutdowns = []

        async def raise_method(*args, **kwargs):
            raise RuntimeError("Test Exception")

        def after_shutdown(name):
            def _after_shutdown(*args, **kwargs):
                shutdowns.append(name)

            return _after_shutdown

        monkeypatch.setattr(Updater, method, raise_method)
        monkeypatch.setattr(
            Application,
            "shutdown",
            call_after(Application.shutdown, after_shutdown("application")),
        )
        monkeypatch.setattr(
            Updater, "shutdown", call_after(Updater.shutdown, after_shutdown("updater"))
        )
        app = ApplicationBuilder().bot(one_time_bot).build()
        with pytest.raises(RuntimeError, match="Test Exception"):  # noqa: PT012
            if "polling" in method:
                app.run_polling(close_loop=False)
            else:
                app.run_webhook(close_loop=False)

        assert not app.running
        assert not app.updater.running
        assert set(shutdowns) == {"application", "updater"}

    @pytest.mark.skipif(
        platform.system() != "Windows",
        reason="Only really relevant on windows",
    )
    @pytest.mark.parametrize("method", ["start_polling", "start_webhook"])
    def test_run_stop_signal_warning_windows(self, one_time_bot, method, recwarn, monkeypatch):
        async def raise_method(*args, **kwargs):
            raise RuntimeError("Prevent Actually Running")

        monkeypatch.setattr(Application, "initialize", raise_method)
        app = ApplicationBuilder().bot(one_time_bot).build()

        with pytest.raises(RuntimeError, match="Prevent Actually Running"):  # noqa: PT012
            if "polling" in method:
                app.run_polling(close_loop=False, stop_signals=(signal.SIGINT,))
            else:
                app.run_webhook(close_loop=False, stop_signals=(signal.SIGTERM,))

        assert len(recwarn) >= 1
        found = False
        for record in recwarn:
            print(record)
            if str(record.message).startswith("Could not add signal handlers for the stop"):
                assert record.category is PTBUserWarning
                assert record.filename == __file__, "stacklevel is incorrect!"
                found = True
        assert found

        recwarn.clear()
        with pytest.raises(RuntimeError, match="Prevent Actually Running"):  # noqa: PT012
            if "polling" in method:
                app.run_polling(close_loop=False, stop_signals=None)
            else:
                app.run_webhook(close_loop=False, stop_signals=None)

        for record in recwarn:
            assert not str(record.message).startswith("Could not add signal handlers for the stop")

    @pytest.mark.parametrize("exception_class", [InvalidToken, TelegramError])
    @pytest.mark.parametrize("retries", [3, 0])
    @pytest.mark.parametrize("method_name", ["run_polling", "run_webhook"])
    async def test_run_polling_webhook_bootstrap_retries(
        self, monkeypatch, exception_class, retries, offline_bot, method_name
    ):
        """This doesn't test all of the internals of the network retry loop. We do that quite
        intensively for the `Updater` and here we just want to make sure that the `Application`
        does do the retries.
        """

        def thread_target():
            asyncio.set_event_loop(asyncio.new_event_loop())
            app = (
                ApplicationBuilder().bot(offline_bot).application_class(PytestApplication).build()
            )

            async def initialize(*args, **kwargs):
                self.count += 1
                raise exception_class(str(self.count))

            monkeypatch.setattr(app, "initialize", initialize)
            method = functools.partial(
                getattr(app, method_name),
                bootstrap_retries=retries,
                close_loop=False,
                stop_signals=None,
            )

            if exception_class == InvalidToken:
                with pytest.raises(InvalidToken, match="1"):
                    method()
            else:
                with pytest.raises(TelegramError, match=str(retries + 1)):
                    method()

        thread = Thread(target=thread_target)
        thread.start()
        thread.join(timeout=10)
        assert not thread.is_alive(), "Test took to long to run. Aborting"

    @pytest.mark.flaky(3, 1)  # loop.call_later will error the test when a flood error is received
    def test_signal_handlers(self, app, monkeypatch):
        # this test should make sure that signal handlers are set by default on Linux + Mac,
        # and not on Windows.

        received_signals = []

        def signal_handler_test(*args, **kwargs):
            # args[0] is the signal, [1] the callback
            received_signals.append(args[0])

        loop = asyncio.get_event_loop()

        monkeypatch.setattr(loop, "add_signal_handler", signal_handler_test)
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)

        def abort_app():
            raise SystemExit

        loop.call_later(0.6, abort_app)

        app.run_polling(close_loop=False)

        if platform.system() == "Windows":
            assert received_signals == []
        else:
            assert received_signals == [signal.SIGINT, signal.SIGTERM, signal.SIGABRT]

        received_signals.clear()
        loop.call_later(0.8, abort_app)
        app.run_webhook(port=49152, webhook_url="example.com", close_loop=False)

        if platform.system() == "Windows":
            assert received_signals == []
        else:
            assert received_signals == [signal.SIGINT, signal.SIGTERM, signal.SIGABRT]

    def test_stop_running_not_running(self, app, caplog):
        with caplog.at_level(logging.DEBUG):
            app.stop_running()

        assert len(caplog.records) == 1
        assert caplog.records[-1].name == "telegram.ext.Application"
        assert caplog.records[-1].getMessage().endswith("`stop_running()` likely has no effect.")

    def test_stop_running_post_init(self, app, monkeypatch, caplog, one_time_bot):
        async def post_init(app):
            app.stop_running()

        called_callbacks = []

        async def callback(*args, **kwargs):
            called_callbacks.append(kwargs["name"])

        monkeypatch.setattr(Application, "start", functools.partial(callback, name="start"))
        monkeypatch.setattr(
            Updater, "start_polling", functools.partialmethod(callback, name="start_polling")
        )

        app = (
            ApplicationBuilder()
            .bot(one_time_bot)
            .post_init(post_init)
            .post_stop(functools.partial(callback, name="post_stop"))
            .post_shutdown(functools.partial(callback, name="post_shutdown"))
            .build()
        )

        with caplog.at_level(logging.INFO):
            app.run_polling(close_loop=False)

        # The important part here is that start(_polling) are *not* called!
        # post_stop must not be called either, since we never called stop()
        assert called_callbacks == ["post_shutdown"]

        assert len(caplog.records) == 1
        assert caplog.records[-1].name == "telegram.ext.Application"
        assert (
            "Application received stop signal via `stop_running`"
            in caplog.records[-1].getMessage()
        )

    @pytest.mark.parametrize("method", ["polling", "webhook"])
    def test_stop_running(self, one_time_bot, monkeypatch, method):
        # asyncio.Event() seems to be hard to use across different threads (awaiting in main
        # thread, setting in another thread), so we use threading.Event() instead.
        # This requires the use of run_in_executor, but that's fine.
        put_update_event = threading.Event()
        callback_done_event = threading.Event()
        called_stop_running = threading.Event()
        assertions = {}

        async def post_init(app):
            # Simply calling app.update_queue.put_nowait(method) in the thread_target doesn't work
            # for some reason (probably threading magic), so we use an event from the thread_target
            # to put the update into the queue in the main thread.
            async def task(app):
                await asyncio.get_running_loop().run_in_executor(None, put_update_event.wait)
                await app.update_queue.put(method)

            app.create_task(task(app))

        app = (
            ApplicationBuilder()
            .application_class(PytestApplication)
            .updater(PytestUpdater(one_time_bot, asyncio.Queue()))
            .post_init(post_init)
            .build()
        )
        monkeypatch.setattr(app.bot, "get_updates", empty_get_updates)

        events = []
        monkeypatch.setattr(
            app.updater,
            "stop",
            call_after(app.updater.stop, lambda _: events.append("updater.stop")),
        )
        monkeypatch.setattr(
            app,
            "stop",
            call_after(app.stop, lambda _: events.append("app.stop")),
        )
        monkeypatch.setattr(
            app,
            "shutdown",
            call_after(app.shutdown, lambda _: events.append("app.shutdown")),
        )
        monkeypatch.setattr(app.bot, "set_webhook", return_true)
        monkeypatch.setattr(app.bot, "delete_webhook", return_true)

        def thread_target():
            waited = 0
            while not app.running:
                time.sleep(0.05)
                waited += 0.05
                if waited > 5:
                    pytest.fail("App apparently won't start")

            time.sleep(0.1)
            assertions["called_stop_running_not_set"] = not called_stop_running.is_set()

            put_update_event.set()
            time.sleep(0.1)

            assertions["called_stop_running_set"] = called_stop_running.is_set()

            # App should have entered `stop` now but not finished it yet because the callback
            # is still running
            assertions["updater.stop_event"] = events == ["updater.stop"]
            assertions["app.running_False"] = not app.running

            callback_done_event.set()
            time.sleep(0.1)

            # Now that the update is fully handled, we expect the full shutdown
            assertions["events"] = events == ["updater.stop", "app.stop", "app.shutdown"]

        async def callback(update, context):
            context.application.stop_running()
            called_stop_running.set()
            await asyncio.get_running_loop().run_in_executor(None, callback_done_event.wait)

        app.add_handler(TypeHandler(object, callback))

        thread = Thread(target=thread_target)
        thread.start()

        if method == "polling":
            app.run_polling(close_loop=False, drop_pending_updates=True)
        else:
            ip = "127.0.0.1"
            port = randrange(1024, 49152)

            app.run_webhook(
                ip_address=ip,
                port=port,
                url_path="TOKEN",
                drop_pending_updates=False,
                close_loop=False,
            )

        thread.join()

        assert len(assertions) == 5
        for key, value in assertions.items():
            assert value, f"assertion '{key}' failed!"

    async def test_process_update_exception_in_building_context(self, monkeypatch, caplog, app):
        # Makes sure that exceptions in building the context don't stop the application
        exception = ValueError("TestException")
        original_from_update = CallbackContext.from_update

        def raise_exception(update, application):
            if update == 1:
                raise exception
            return original_from_update(update, application)

        monkeypatch.setattr(CallbackContext, "from_update", raise_exception)

        received_updates = set()

        async def callback(update, context):
            received_updates.add(update)

        app.add_handler(TypeHandler(int, callback))

        async with app:
            with caplog.at_level(logging.CRITICAL):
                await app.process_update(1)

            assert received_updates == set()
            assert len(caplog.records) == 1
            record = caplog.records[0]
            assert record.name == "telegram.ext.Application"
            assert record.getMessage().startswith(
                "Error while building CallbackContext for update 1"
            )
            assert record.levelno == logging.CRITICAL

            # Let's also check that no critical log is produced when the exception is not raised
            caplog.clear()
            with caplog.at_level(logging.CRITICAL):
                await app.process_update(2)

            assert received_updates == {2}
            assert len(caplog.records) == 0

    @pytest.mark.parametrize("change_type", ["remove", "add"])
    async def test_process_update_handler_change_groups_during_iteration(self, app, change_type):
        run_groups = set()

        async def dummy_callback(_, __, g: int):
            run_groups.add(g)

        for group in range(10, 20):
            handler = TypeHandler(int, functools.partial(dummy_callback, g=group))
            app.add_handler(handler, group=group)

        async def wait_callback(_, context):
            # Trigger a change of the app.handlers dict during the iteration
            if change_type == "remove":
                context.application.remove_handler(handler, group)
            else:
                context.application.add_handler(
                    TypeHandler(int, functools.partial(dummy_callback, g=42)), group=42
                )

        app.add_handler(TypeHandler(int, wait_callback))

        async with app:
            await app.process_update(1)

        # check that exactly those handlers were called that were configured when
        # process_update was called
        assert run_groups == set(range(10, 20))

    async def test_process_update_handler_change_group_during_iteration(self, app):
        async def dummy_callback(_, __):
            pass

        checked_handlers = set()

        class TrackHandler(TypeHandler):
            def __init__(self, name: str, *args, **kwargs):
                self.name = name
                super().__init__(*args, **kwargs)

            def check_update(self, update: object) -> bool:
                checked_handlers.add(self.name)
                return super().check_update(update)

        remove_handler = TrackHandler("remove", int, dummy_callback)
        add_handler = TrackHandler("add", int, dummy_callback)

        class TriggerHandler(TypeHandler):
            def check_update(self, update: object) -> bool:
                # Trigger a change of the app.handlers *in the same group* during the iteration
                app.remove_handler(remove_handler)
                app.add_handler(add_handler)
                # return False to ensure that additional handlers in the same group are checked
                return False

        app.add_handler(TriggerHandler(str, dummy_callback))
        app.add_handler(remove_handler)
        async with app:
            await app.process_update("string update")

        # check that exactly those handlers were checked that were configured when
        # process_update was called
        assert checked_handlers == {"remove"}

    async def test_process_error_exception_in_building_context(self, monkeypatch, caplog, app):
        # Makes sure that exceptions in building the context don't stop the application
        exception = ValueError("TestException")
        original_from_error = CallbackContext.from_error

        def raise_exception(update, error, application, *args, **kwargs):
            if error == 1:
                raise exception
            return original_from_error(update, error, application, *args, **kwargs)

        monkeypatch.setattr(CallbackContext, "from_error", raise_exception)

        received_errors = set()

        async def callback(update, context):
            received_errors.add(context.error)

        app.add_error_handler(callback)

        async with app:
            with caplog.at_level(logging.CRITICAL):
                await app.process_error(update=None, error=1)

            assert received_errors == set()
            assert len(caplog.records) == 1
            record = caplog.records[0]
            assert record.name == "telegram.ext.Application"
            assert record.getMessage().startswith(
                "Error while building CallbackContext for exception 1"
            )
            assert record.levelno == logging.CRITICAL

            # Let's also check that no critical log is produced when the exception is not raised
            caplog.clear()
            with caplog.at_level(logging.CRITICAL):
                await app.process_error(update=None, error=2)

            assert received_errors == {2}
            assert len(caplog.records) == 0

    @pytest.mark.parametrize("change_type", ["remove", "add"])
    async def test_process_error_change_during_iteration(self, app, change_type):
        called_handlers = set()

        async def dummy_process_error(name: str, *_, **__):
            called_handlers.add(name)

        add_error_handler = functools.partial(dummy_process_error, "add_handler")
        remove_error_handler = functools.partial(dummy_process_error, "remove_handler")

        async def trigger_change(*_, **__):
            if change_type == "remove":
                app.remove_error_handler(remove_error_handler)
            else:
                app.add_error_handler(add_error_handler)

        app.add_error_handler(trigger_change)
        app.add_error_handler(remove_error_handler)
        async with app:
            await app.process_error(update=None, error=None)

        # check that exactly those handlers were checked that were configured when
        # add_error_handler was called
        assert called_handlers == {"remove_handler"}