python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2025-12-12 10:11:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix compatibility with Python 3.8-3.9 (#1162)

Browse source 
* Try to fix compatibility with Python 3.8-3.9

* Added changelog
This commit is contained in:
Alex Root Junior 2023-04-21 00:17:06 +03:00 • committed by GitHub
parent 399ccb2b00
commit fb3076d40f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 49 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

33
tests/test_dispatcher/test_dispatcher.py
View file

@ -708,10 +708,15 @@ class TestDispatcher:
with pytest.raises(RuntimeError):
await dispatcher.stop_polling()
assert not dispatcher._stop_signal.is_set()
assert not dispatcher._stopped_signal.is_set()
assert not dispatcher._stop_signal
assert not dispatcher._stopped_signal
with patch("asyncio.locks.Event.wait", new_callable=AsyncMock) as mocked_wait:
async with dispatcher._running_lock:
await dispatcher.stop_polling()
assert not dispatcher._stop_signal
dispatcher._stop_signal = Event()
dispatcher._stopped_signal = Event()
await dispatcher.stop_polling()
assert dispatcher._stop_signal.is_set()
mocked_wait.assert_awaited()
@ -723,6 +728,11 @@ class TestDispatcher:
mocked_set.assert_not_called()
async with dispatcher._running_lock:
dispatcher._signal_stop_polling(signal.SIGINT)
mocked_set.assert_not_called()
dispatcher._stop_signal = Event()
dispatcher._stopped_signal = Event()
dispatcher._signal_stop_polling(signal.SIGINT)
mocked_set.assert_called()
@ -764,12 +774,29 @@ class TestDispatcher:
def test_run_polling(self, bot: MockedBot):
dispatcher = Dispatcher()
async def stop():
await asyncio.sleep(0.5)
await dispatcher.stop_polling()
start_called = False
@dispatcher.startup()
async def startup():
nonlocal start_called
start_called = True
asyncio.create_task(stop())
original_start_polling = dispatcher.start_polling
with patch(
"aiogram.dispatcher.dispatcher.Dispatcher.start_polling"
"aiogram.dispatcher.dispatcher.Dispatcher.start_polling",
side_effect=original_start_polling,
) as patched_start_polling:
dispatcher.run_polling(bot)
patched_start_polling.assert_awaited_once()
assert start_called
async def test_feed_webhook_update_fast_process(self, bot: MockedBot):
dispatcher = Dispatcher()
dispatcher.message.register(simple_message_handler)