mirror of
https://github.com/aiogram/aiogram.git
synced 2025-12-12 02:03:04 +00:00
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
import datetime
|
|
import time
|
|
|
|
import pytest
|
|
from asynctest import MagicMock, patch
|
|
|
|
from aiogram import Bot
|
|
from aiogram.api.types import Chat, Message, Update, User
|
|
from aiogram.dispatcher.dispatcher import Dispatcher
|
|
from aiogram.dispatcher.router import Router
|
|
|
|
|
|
class TestDispatcher:
|
|
def test_parent_router(self):
|
|
dp = Dispatcher()
|
|
with pytest.raises(RuntimeError):
|
|
dp.parent_router = Router()
|
|
assert dp.parent_router is None
|
|
dp._parent_router = Router()
|
|
assert dp.parent_router is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feed_update(self):
|
|
dp = Dispatcher()
|
|
bot = Bot("42:TEST")
|
|
|
|
@dp.message_handler()
|
|
async def my_handler(message: Message, **kwargs):
|
|
assert "bot" in kwargs
|
|
assert isinstance(kwargs["bot"], Bot)
|
|
assert kwargs["bot"] == bot
|
|
return message.text
|
|
|
|
results_count = 0
|
|
async for result in dp.feed_update(
|
|
bot=bot,
|
|
update=Update(
|
|
update_id=42,
|
|
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_count += 1
|
|
assert result == "test"
|
|
|
|
assert results_count == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feed_raw_update(self):
|
|
dp = Dispatcher()
|
|
bot = Bot("42:TEST")
|
|
|
|
with patch(
|
|
"aiogram.dispatcher.dispatcher.Dispatcher.feed_update", new_callable=MagicMock
|
|
) as patched_feed_update:
|
|
patched_feed_update.__aiter__.return_value = ["test"]
|
|
async for result in dp.feed_raw_update(
|
|
bot=bot,
|
|
update={
|
|
"update_id": 42,
|
|
"message": {
|
|
"message_id": 42,
|
|
"date": int(time.time()),
|
|
"text": "test",
|
|
"chat": {"id": 42, "type": "private"},
|
|
"user": {"id": 42, "is_bot": False, "first_name": "Test"},
|
|
},
|
|
},
|
|
):
|
|
assert result == "test"
|
|
|
|
@pytest.mark.skip
|
|
@pytest.mark.asyncio
|
|
async def test_listen_updates(self):
|
|
pass
|
|
|
|
@pytest.mark.skip
|
|
@pytest.mark.asyncio
|
|
async def test_polling(self):
|
|
pass
|