Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_trigger(self):
router = Router()
observer = router.message_handler
observer.bind_filter(MyFilter1)
observer.register(my_handler, test="ok")
message = Message(
message_id=42,
date=datetime.datetime.now(),
text="test",
chat=Chat(id=42, type="private"),
from_user=User(id=42, is_bot=False, first_name="Test"),
)
results = [result async for result in observer.trigger(message)]
assert results == [message]
async def test_nested_router_listen_update(self):
router1 = Router()
router2 = Router()
router1.include_router(router2)
observer = router2.message_handler
@observer()
async def my_handler(event: Message, **kwargs: Any):
assert Chat.get_current(False)
assert User.get_current(False)
return kwargs
update = Update(
update_id=42,
message=Message(
message_id=42,
date=datetime.datetime.now(),
text="test",
chat=Chat(id=42, type="private"),
def test_include_router_by_string(self):
router = Router()
router.include_router("tests.test_dispatcher.test_router:importable_router")
async def test_emit_startup(self):
router1 = Router()
router2 = Router()
router1.include_router(router2)
results = []
@router1.startup()
async def startup1():
results.append(1)
@router2.startup()
async def startup2():
results.append(2)
await router2.emit_startup()
assert results == [2]
async def test_listen_unknown_update(self):
router = Router()
with pytest.raises(SkipHandler):
await router._listen_update(Update(update_id=42))
async def test_listen_update(
self, event_type: str, update: Update, has_chat: bool, has_user: bool
):
router = Router()
observer = router.observers[event_type]
@observer()
async def my_handler(event: Any, **kwargs: Any):
assert event == getattr(update, event_type)
if has_chat:
assert Chat.get_current(False)
if has_user:
assert User.get_current(False)
return kwargs
result = await router._listen_update(update, test="PASS")
assert isinstance(result, dict)
assert result["event_update"] == update
assert result["event_router"] == router
assert result["test"] == "PASS"
async def test_emit_shutdown(self):
router1 = Router()
router2 = Router()
router1.include_router(router2)
results = []
@router1.shutdown()
async def shutdown1():
results.append(1)
@router2.shutdown()
async def shutdown2():
results.append(2)
await router2.emit_shutdown()
assert results == [2]
await router1.emit_shutdown()
ChosenInlineResult,
InlineQuery,
Message,
Poll,
PollOption,
PreCheckoutQuery,
ShippingAddress,
ShippingQuery,
Update,
User,
)
from aiogram.dispatcher.event.observer import SkipHandler
from aiogram.dispatcher.router import Router
from aiogram.utils.warnings import CodeHasNoEffect
importable_router = Router()
class TestRouter:
def test_including_routers(self):
router1 = Router()
router2 = Router()
router3 = Router()
assert router1.parent_router is None
assert router2.parent_router is None
assert router3.parent_router is None
with pytest.raises(RuntimeError, match="Self-referencing routers is not allowed"):
router1.include_router(router1)
router1.include_router(router2)
def test_include_router_by_string_bad_type(self):
router = Router()
with pytest.raises(ValueError, match=r"router should be instance of Router"):
router.include_router("tests.test_dispatcher.test_router:TestRouter")
async def test_emit_startup(self):
router1 = Router()
router2 = Router()
router1.include_router(router2)
results = []
@router1.startup()
async def startup1():
results.append(1)
@router2.startup()
async def startup2():
results.append(2)
await router2.emit_startup()
assert results == [2]
await router1.emit_startup()