social-post-writer-bot/tests/test_activity_phrases.py

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_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)