Aiogram Testing

aiogram testing

Test at the dispatcher boundary: feed an Update in, assert on the API calls that come out. No network, no token, no polling.

the thing to know first

aiogram.test_utils is not part of the distributed package. Verified on aiogram 3.31.0:

$ python -c "import aiogram.test_utils"
ModuleNotFoundError: No module named 'aiogram.test_utils'

mocked_bot.py exists in the aiogram repository under tests/, but the wheel ships only client, dispatcher, enums, exceptions.py, filters, fsm, handlers, loggers.py, methods, types, utils, and webhook. Any guide telling you to from aiogram.test_utils.mocked_bot import MockedBot is describing the repository, not an installed release.

So you write a fake session. It is about thirty lines, and you own it.

when to use this skill

• unit-testing handlers, filters, or middleware
• asserting what the bot sent
• testing an FSM flow end to end
• testing without a token or network
• setting up pytest for an async bot

the recording session

BaseSession has exactly three abstract methods: make_request, stream_content, close. Implement them and Bot never touches the network.

import datetime
from collections.abc import AsyncGenerator
from typing import Any

from aiogram import Bot
from aiogram.client.session.base import BaseSession
from aiogram.methods import SendMessage, TelegramMethod
from aiogram.types import Chat, Message


class RecordingSession(BaseSession):
    """Records every outgoing API call and returns canned results."""

    def __init__(self) -> None:
        super().__init__()
        self.calls: list[TelegramMethod[Any]] = []
        self.results: dict[str, Any] = {}

    def will_return(self, method_name: str, value: Any) -> None:
        """Override the canned result for one API method."""
        self.results[method_name] = value

    async def make_request(
        self,
        bot: Bot,
        method: TelegramMethod[Any],
        timeout: int | None = None,
    ) -> Any:
        self.calls.append(method)
        name = type(method).__name__
        if name in self.results:
            return self.results[name]
        if not isinstance(method, SendMessage):
            raise AssertionError(f"Configure a result for {name}")
        return Message(
            message_id=len(self.calls),
            date=datetime.datetime.now(datetime.timezone.utc),
            chat=Chat(id=getattr(method, "chat_id", 1) or 1, type="private"),
            text=getattr(method, "text", None),
        )

    async def stream_content(
        self,
        url: str,
        headers: dict[str, Any] | None = None,
        timeout: int = 30,
        chunk_size: int = 65536,
        raise_for_status: bool = True,
    ) -> AsyncGenerator[bytes, None]:
        yield b""

    async def close(self) -> None:
        return None

    def calls_to(self, method_name: str) -> list[TelegramMethod[Any]]:
        return [call for call in self.calls if type(call).__name__ == method_name]


session = RecordingSession()
bot = Bot(token="123456:TEST", session=session)
assert bot.id == 123456

123456:TEST passes aiogram.utils.token.validate_token, which only requires <digits>:<non-empty>. No real token is needed anywhere in a test suite.

update fixtures

import datetime

from aiogram.types import CallbackQuery, Chat, Message, Update, User

USER = User(id=42, is_bot=False, first_name="Tester", username="tester")
CHAT = Chat(id=42, type="private")


def now() -> datetime.datetime:
    return datetime.datetime.now(datetime.timezone.utc)


def message_update(text: str, update_id: int = 1) -> Update:
    return Update(
        update_id=update_id,
        message=Message(
            message_id=update_id,
            date=now(),
            chat=CHAT,
            from_user=USER,
            text=text,
        ),
    )


def callback_update(data: str, update_id: int = 1) -> Update:
    return Update(
        update_id=update_id,
        callback_query=CallbackQuery(
            id=str(update_id),
            from_user=USER,
            chat_instance="test",
            data=data,
            message=Message(message_id=1, date=now(), chat=CHAT, text="menu"),
        ),
    )


assert message_update("/start").message is not None
assert callback_update("ping").callback_query is not None

Build these once in conftest.py. Hand-writing an Update in every test is the fastest way to a brittle suite.

feed_raw_update(bot, {...}) takes the raw JSON dict instead, which is useful for replaying a payload captured from production.

a complete test

import asyncio
import datetime
from collections.abc import AsyncGenerator
from typing import Any

from aiogram import Bot, Dispatcher, Router
from aiogram.client.session.base import BaseSession
from aiogram.filters import CommandStart
from aiogram.methods import SendMessage, TelegramMethod
from aiogram.types import Chat, Message, Update, User

router = Router(name="under-test")


@router.message(CommandStart())
async def start(message: Message) -> None:
    await message.answer("Welcome!")


class FakeSession(BaseSession):
    def __init__(self) -> None:
        super().__init__()
        self.calls: list[TelegramMethod[Any]] = []

    async def make_request(
        self, bot: Bot, method: TelegramMethod[Any], timeout: int | None = None
    ) -> Any:
        self.calls.append(method)
        assert isinstance(method, SendMessage), type(method).__name__
        return Message(
            message_id=1,
            date=datetime.datetime.now(datetime.timezone.utc),
            chat=Chat(id=1, type="private"),
        )

    async def stream_content(
        self,
        url: str,
        headers: dict[str, Any] | None = None,
        timeout: int = 30,
        chunk_size: int = 65536,
        raise_for_status: bool = True,
    ) -> AsyncGenerator[bytes, None]:
        yield b""

    async def close(self) -> None:
        return None


async def test_start_greets() -> None:
    session = FakeSession()
    bot = Bot(token="123456:TEST", session=session)
    dispatcher = Dispatcher()
    dispatcher.include_router(router)

    update = Update(
        update_id=1,
        message=Message(
            message_id=1,
            date=datetime.datetime.now(datetime.timezone.utc),
            chat=Chat(id=1, type="private"),
            from_user=User(id=1, is_bot=False, first_name="T"),
            text="/start",
        ),
    )
    await dispatcher.feed_update(bot, update)

    assert len(session.calls) == 1
    sent = session.calls[0]
    assert isinstance(sent, SendMessage)
    assert sent.text == "Welcome!"
    assert sent.chat_id == 1


asyncio.run(test_start_greets())

Assert on the method object, not on a string. isinstance(sent, SendMessage) plus field checks survives refactors that a str(call) comparison would not.

injecting fakes

Anything a handler pulls from workflow data can be replaced per test by passing it to feed_update:

from aiogram import Bot, Dispatcher, Router
from aiogram.types import Message, Update

router = Router(name="di-test")


class Repository:
    async def name_for(self, user_id: int) -> str:
        raise NotImplementedError


@router.message()
async def show_name(message: Message, repository: Repository) -> None:
    if message.from_user is not None:
        await message.answer(await repository.name_for(message.from_user.id))


class FakeRepository(Repository):
    async def name_for(self, user_id: int) -> str:
        return "fake name"


async def run_case(bot: Bot, dispatcher: Dispatcher, update: Update) -> None:
    """Extra kwargs join workflow data for this update only."""
    await dispatcher.feed_update(bot, update, repository=FakeRepository())

This is why handlers should depend on injected services rather than reaching for module globals — the seam already exists.

FSM assertions

import asyncio

from aiogram import Bot
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.base import StorageKey
from aiogram.fsm.storage.memory import MemoryStorage


class Form(StatesGroup):
    name = State()


async def check_state() -> None:
    storage = MemoryStorage()
    bot = Bot(token="123456:TEST")
    key = StorageKey(bot_id=bot.id, chat_id=42, user_id=42)

    await storage.set_state(key, Form.name)
    await storage.set_data(key, {"name": "Ada"})

    assert await storage.get_state(key) == "Form:name"
    assert await storage.get_data(key) == {"name": "Ada"}


asyncio.run(check_state())

Build the StorageKey with the same fields your FSMStrategy uses — with the default USER_IN_CHAT, that is bot_id, chat_id, and user_id. Pass the storage to Dispatcher(storage=storage) so a test can drive a flow update by update and inspect the state between steps.

testing filters and middleware directly

Filters and middleware are plain callables — no dispatcher required:

import asyncio
import datetime

from aiogram.filters import Command
from aiogram.types import Chat, Message, User


async def check_filter() -> None:
    message = Message(
        message_id=1,
        date=datetime.datetime.now(datetime.timezone.utc),
        chat=Chat(id=1, type="private"),
        from_user=User(id=1, is_bot=False, first_name="T"),
        text="/help now",
    )
    # No @mention is present, so this filter path does not consult Bot.get_me().
    result = await Command("help")(message, bot=None)
    assert isinstance(result, dict)
    assert result["command"].args == "now"

    assert await Command("other")(message, bot=None) is False


asyncio.run(check_filter())

A filter returning a dict is what injection looks like from the inside.

pytest setup

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = ["error"]

asyncio_mode = "auto" removes the need for @pytest.mark.asyncio on every test. Setting asyncio_default_fixture_loop_scope explicitly silences a pytest-asyncio deprecation warning that filterwarnings = ["error"] would otherwise turn into a failure.

Use the project's existing fixtures. When a bot fixture is needed, create a fresh RecordingSession and Bot per test and close the bot session in fixture teardown. Do not share a router instance between dispatcher fixtures: a router can have only one parent. Construct the router per test or use the application's router factory.

what to test

For import-side-effect regressions, import the application's handler module with its network/loop-start entry points replaced by fail-fast sentinels. Importing aiogram itself does not verify that the application's module starts nothing.

what not to do

• Do not hit the Telegram API. No real tokens in tests, not even in "integration" ones that CI will run on every push.
• Use feed_update for dispatch behaviour. It does not test HTTP authentication, webhook acknowledgement, polling offsets, or live Telegram rendering. Test transport separately with a local HTTP fixture when that is the changed contract.
• Do not assert on log output when you can assert on the recorded method object.
• Do not share one MemoryStorage across tests — state leaks between them.
• Do not mock aiohttp. Replace the whole session; it is a smaller, more stable surface.

checklist

• a fake BaseSession records calls; no network anywhere in the suite
• fake tokens only
• fixtures for Update, Message, CallbackQuery in conftest.py
• assertions on TelegramMethod instances and their fields
• dependencies injected per test through feed_update kwargs
• fresh storage per test
• FSM transitions covered including invalid input and cancel
• import side effects tested when that is the behaviour being changed

see also

• using-aiogram — the feed_update overview
• aiogram-routing — feed_update / feed_raw_update
• aiogram-fsm — storage and keys
• aiogram-middlewares — injecting fakes through workflow data