import asyncio import unittest from typing import Any, cast from unittest.mock import AsyncMock, patch import activity_phrases import bot def make_message(chat_type: str = "private", chat_id: int = 55): """Минимальный Message с нужными OperationStatus полями.""" fake_bot = AsyncMock() fake_bot.send_message_draft = AsyncMock(return_value=True) message = cast( Any, type( "Message", (), { "chat": type("Chat", (), {"id": chat_id, "type": chat_type})(), "bot": fake_bot, "answer": AsyncMock(return_value=AsyncMock()), }, )(), ) return message class PhrasePoolTests(unittest.TestCase): def test_known_kinds_have_non_empty_pools(self): for kind, pool in activity_phrases.ACTIVITY_PHRASES.items(): self.assertTrue(pool, kind) for phrase in pool: self.assertIsInstance(phrase, str) self.assertTrue(phrase.strip()) def test_next_phrase_respects_kind(self): for kind, pool in activity_phrases.ACTIVITY_PHRASES.items(): self.assertIn(activity_phrases.next_phrase(kind), pool, kind) def test_unknown_kind_falls_back_to_defaults(self): self.assertIn(activity_phrases.next_phrase("no-such-kind"), activity_phrases.DEFAULT_PHRASES) def test_next_phrase_never_repeats_excluded(self): pool = activity_phrases.ACTIVITY_PHRASES["generation"] excluded = pool[0] for _ in range(200): self.assertNotEqual(activity_phrases.next_phrase("generation", excluded), excluded) def test_random_interval_within_bounds(self): low, high = activity_phrases.ACTIVITY_MIN_SECONDS, activity_phrases.ACTIVITY_MAX_SECONDS self.assertLessEqual(low, high) for _ in range(500): value = activity_phrases.random_interval() self.assertGreaterEqual(value, low) self.assertLessEqual(value, high) class DraftStatusTests(unittest.IsolatedAsyncioTestCase): async def test_private_chat_streams_draft_and_never_posts_a_status_message(self): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await status.cancel() message.answer.assert_not_awaited() message.bot.send_message_draft.assert_awaited_once() call = message.bot.send_message_draft.await_args self.assertEqual(call.kwargs["chat_id"], 55) self.assertEqual(call.kwargs["text"], "⏳ Старт") self.assertNotEqual(call.kwargs["draft_id"], 0) async def test_draft_id_is_stable_across_updates(self): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await asyncio.sleep(0) await status.update("⏳ Работаю") await status.cancel() ids = {c.kwargs["draft_id"] for c in message.bot.send_message_draft.await_args_list} self.assertEqual(len(ids), 1) self.assertEqual( [c.kwargs["text"] for c in message.bot.send_message_draft.await_args_list][:2], ["⏳ Старт", "⏳ Работаю"], ) async def test_finish_posts_real_message_and_stops_rotation(self): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await status.finish("✅ Готово") message.answer.assert_awaited_once_with("✅ Готово") self.assertIsNone(status._rotation_task) # после finish ротация не должна больше дёргать драфт calls_before = message.bot.send_message_draft.await_count with patch.object(bot, "random_interval", return_value=0.01): await asyncio.sleep(0.05) self.assertEqual(message.bot.send_message_draft.await_count, calls_before) async def test_rotation_emits_activity_phrases(self): message = make_message("private") with patch.object(bot, "random_interval", return_value=0.01): status = await bot.OperationStatus.start(message, "⏳ Старт", kind="training") await asyncio.sleep(0.09) await status.cancel() texts = [c.kwargs["text"] for c in message.bot.send_message_draft.await_args_list] self.assertGreater(len(texts), 1) self.assertTrue( any(text in activity_phrases.TRAINING_PHRASES for text in texts[1:]), texts, ) async def test_no_kind_means_no_rotation(self): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт") with patch.object(bot, "random_interval", return_value=0.01): await asyncio.sleep(0.05) self.assertEqual(message.bot.send_message_draft.await_count, 1) async def test_retarget_switches_pool(self): message = make_message("private") with patch.object(bot, "random_interval", return_value=0.01): status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") status.retarget("images") await asyncio.sleep(0.09) await status.cancel() texts = [c.kwargs["text"] for c in message.bot.send_message_draft.await_args_list][1:] self.assertTrue(any(text in activity_phrases.IMAGES_PHRASES for text in texts), texts) class GroupStatusTests(unittest.IsolatedAsyncioTestCase): async def test_group_chat_uses_legacy_edit_path(self): for chat_type in ("group", "supergroup", "channel"): message = make_message(chat_type) status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await asyncio.sleep(0) await status.update("⏳ Работаю") await status.finish("✅ Готово") message.bot.send_message_draft.assert_not_awaited() message.answer.assert_awaited_once_with("⏳ Старт") self.assertEqual( [c.args[0] for c in status.status_message.edit_text.await_args_list], ["⏳ Работаю", "✅ Готово"], ) async def test_no_rotation_task_in_groups(self): message = make_message("group") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") self.assertIsNone(status._rotation_task) async def test_finish_pushes_final_text_into_draft_before_sending(self): """Драфт приводится к финальному тексту, иначе он висит ещё несколько секунд.""" message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await status.finish("✅ Обучение завершено") drafts = message.bot.send_message_draft.await_args_list # Последний пуш в драфт — тем же текстом, что ушёл реальным сообщением. self.assertEqual(drafts[-1].kwargs["text"], "✅ Обучение завершено") self.assertEqual(drafts[-1].kwargs["draft_id"], status._finished_draft_id) message.answer.assert_awaited_once_with("✅ Обучение завершено") async def test_finish_without_text_clears_draft(self): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await status.finish() # Никакого сообщения, но драфт погашен и ротация остановлена. message.answer.assert_not_awaited() self.assertEqual(message.bot.send_message_draft.await_args_list[-1].kwargs["text"], "") self.assertTrue(status._closed) self.assertIsNone(status._rotation_task) async def test_finish_is_idempotent(self): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await status.finish("✅ Готово") await status.finish_if_open("✅ Ещё раз") await status.finish_if_open("✅ И ещё") # Повторные вызовы не должны слать дубли. self.assertEqual(message.answer.await_count, 1) message.answer.assert_awaited_once_with("✅ Готово") async def test_finish_status_helper_tolerates_none(self): await bot.finish_status(None, "✅ Готово") # не должно падать async def test_image_and_voice_paths_close_the_status(self): """Успешный выход без явного finish не оставляет драфт висеть.""" for kind in ("images", "voice"): message = make_message("private") status = await bot.OperationStatus.start(message, "⏳ Старт", kind=kind) await bot.finish_status(status, "✅ Готово") self.assertTrue(status._closed) self.assertIsNone(status._rotation_task) self.assertIsNone(status.draft_id) message.answer.assert_awaited_once_with("✅ Готово") async def test_draft_failure_falls_back_to_real_message(self): message = make_message("private") message.bot.send_message_draft = AsyncMock(side_effect=RuntimeError("draft unsupported")) status = await bot.OperationStatus.start(message, "⏳ Старт", kind="generation") await status.cancel() message.answer.assert_awaited_once_with("⏳ Старт") self.assertIsNone(status.draft_id)