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

Merge branch 'dev-2.x'

Browse source
This commit is contained in:
Alex Root Junior 2022-08-14 18:30:42 +03:00
commit a0b5aa299d
No known key found for this signature in database
GPG key ID: 074C1D455EBEA4AC
2 changed files with 59 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

11
aiogram/dispatcher/middlewares.py
View file

@ -117,11 +117,12 @@ class LifetimeControllerMiddleware(BaseMiddleware):
if self._skip_actions is None:
self._skip_actions = []
if self.skip_patterns:
self._skip_actions.extend([
f"pre_process_{item}",
f"process_{item}",
f"post_process_{item}",
])
for item in self.skip_patterns:
self._skip_actions.extend([
f"pre_process_{item}",
f"process_{item}",
f"post_process_{item}",
])
return self._skip_actions
async def pre_process(self, obj, data, *args):

53
tests/test_dispatcher/test_middlewares/test_lifetimecontroller.py Normal file
View file

@ -0,0 +1,53 @@
from unittest.mock import AsyncMock
import pytest
from aiogram.dispatcher.middlewares import LifetimeControllerMiddleware
pytestmark = pytest.mark.asyncio
async def test_no_skip():
class Middleware(LifetimeControllerMiddleware):
pre_process = AsyncMock()
post_process = AsyncMock()
m = Middleware()
await m.trigger("pre_process_update_xxx", [1, 2, 3])
m.pre_process.assert_called_once_with(1, 3, 2)
m.post_process.assert_not_called()
await m.trigger("post_process_update_xxx", [1, 2, 3])
m.pre_process.reset_mock()
m.pre_process.assert_not_called()
m.post_process.assert_called_once_with(1, 3, 2)
async def test_skip_prefix():
class Middleware(LifetimeControllerMiddleware):
skip_patterns = ["update"]
pre_process = AsyncMock()
post_process = AsyncMock()
m = Middleware()
await m.trigger("pre_process_update_xxx", [1, 2, 3])
m.pre_process.assert_called_once_with(1, 3, 2)
m.post_process.assert_not_called()
await m.trigger("post_process_update_xxx", [1, 2, 3])
m.pre_process.reset_mock()
m.pre_process.assert_not_called()
m.post_process.assert_called_once_with(1, 3, 2)
async def test_skip():
class Middleware(LifetimeControllerMiddleware):
skip_patterns = ["update_xxx"]
pre_process = AsyncMock()
post_process = AsyncMock()
m = Middleware()
await m.trigger("pre_process_update_xxx", [1, 2, 3])
m.pre_process.assert_not_called()
m.post_process.assert_not_called()
await m.trigger("post_process_update_xxx", [1, 2, 3])
m.pre_process.reset_mock()
m.pre_process.assert_not_called()
m.post_process.assert_not_called()