GeezerVison

tests.ext.test_application.TestApplication::test_stop_running[polling]

Stack Traces | 0.522s run time

self = <tests.ext.test_application.TestApplication object at 0x7fd8a93a18b0>
one_time_bot = PytestExtBot[token=696188732:AAGVwEKfHHlNjscxEDNKAwkhGstZ_ko1mC0]
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7fd8a4d57310>
method = 'polling'

    @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!"
E           AssertionError: assertion 'updater.stop_event' failed!
E           assert False

tests/ext/test_application.py:2547: AssertionError