83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
import asyncio
|
|
import unittest
|
|
from typing import Any, cast
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import bot
|
|
|
|
|
|
class FakePhoto:
|
|
file_size = 10
|
|
|
|
|
|
class FakeBot:
|
|
async def download(self, file_obj, destination):
|
|
destination.write(b"image-bytes")
|
|
|
|
|
|
class FakeMessage:
|
|
def __init__(self, message_id, group_id, image_type="photo"):
|
|
self.message_id = message_id
|
|
self.media_group_id = group_id
|
|
self.chat = type("Chat", (), {"id": 55})()
|
|
self.from_user = type("User", (), {"id": 12})()
|
|
self.photo = [FakePhoto()] if image_type == "photo" else []
|
|
self.document = None
|
|
self.bot = FakeBot()
|
|
self.answer = AsyncMock()
|
|
|
|
|
|
class MediaGroupTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_collects_gallery_and_returns_same_group_once(self):
|
|
first = FakeMessage(101, "album-a")
|
|
second = FakeMessage(102, "album-a")
|
|
with patch.object(bot, "MEDIA_GROUP_COLLECT_SECONDS", 0.02):
|
|
first_task = asyncio.create_task(bot._collect_media_group(first, FakeBot()))
|
|
await asyncio.sleep(0)
|
|
second_task = asyncio.create_task(bot._collect_media_group(second, FakeBot()))
|
|
first_result, second_result = await asyncio.gather(first_task, second_task)
|
|
self.assertEqual([m.message_id for m in first_result], [101, 102])
|
|
self.assertEqual(first_result, second_result)
|
|
|
|
async def test_non_gallery_is_single_item(self):
|
|
message = FakeMessage(201, None)
|
|
result = await bot._collect_media_group(message, FakeBot())
|
|
self.assertEqual(result, [message])
|
|
|
|
|
|
class TypingStatusTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_operation_status_is_edited_and_finished_in_place(self):
|
|
status_message = cast(Any, type("StatusMessage", (), {"edit_text": AsyncMock()})())
|
|
message = cast(Any, type("Message", (), {"answer": AsyncMock(return_value=status_message)})())
|
|
|
|
status = await bot.OperationStatus.start(message, "⏳ Старт")
|
|
await status.update("⏳ Работаю")
|
|
await status.finish("✅ Готово")
|
|
|
|
message.answer.assert_awaited_once_with("⏳ Старт")
|
|
self.assertEqual(
|
|
[call.args[0] for call in status_message.edit_text.await_args_list],
|
|
["⏳ Работаю", "✅ Готово"],
|
|
)
|
|
|
|
async def test_sends_immediately_refreshes_and_cancels(self):
|
|
fake_bot = AsyncMock()
|
|
with patch.object(bot, "TYPING_REFRESH_SECONDS", 0.01):
|
|
async with bot.typing_status(fake_bot, 123):
|
|
await bot.asyncio.sleep(0.025)
|
|
self.assertGreaterEqual(fake_bot.send_chat_action.await_count, 2)
|
|
fake_bot.send_chat_action.assert_any_await(123, "typing")
|
|
|
|
async def test_stops_refresh_after_context(self):
|
|
fake_bot = AsyncMock()
|
|
with patch.object(bot, "TYPING_REFRESH_SECONDS", 0.01):
|
|
async with bot.typing_status(fake_bot, 123):
|
|
await bot.asyncio.sleep(0.015)
|
|
calls_after_exit = fake_bot.send_chat_action.await_count
|
|
await bot.asyncio.sleep(0.03)
|
|
self.assertEqual(fake_bot.send_chat_action.await_count, calls_after_exit)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|