How to use the aiogram.dispatcher.router.Router function in aiogram

To help you get started, we’ve selected a few aiogram examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github aiogram / aiogram / tests / test_dispatcher / test_event / test_observer.py View on Github external
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]
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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"),
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
def test_include_router_by_string(self):
        router = Router()
        router.include_router("tests.test_dispatcher.test_router:importable_router")
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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]
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
async def test_listen_unknown_update(self):
        router = Router()

        with pytest.raises(SkipHandler):
            await router._listen_update(Update(update_id=42))
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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"
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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()
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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)
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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")
github aiogram / aiogram / tests / test_dispatcher / test_router.py View on Github external
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()