* Fix filtering skip_patterns

* cache skip actions

* fix item' is not defined, add tests
This commit is contained in:
Andrey Tikhonov 2022-08-14 18:29:35 +03:00 committed by GitHub
parent d5654068b1
commit c43ff9b6f9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 59 additions and 5 deletions

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):

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()