312 lines
13 KiB
Python
312 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import unittest
|
||
from datetime import datetime
|
||
from typing import Any, cast
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
from aiogram.types import Chat, Message
|
||
|
||
import bot
|
||
|
||
|
||
class _StubState:
|
||
def __init__(self, mode="lobby"):
|
||
self.value = bot.MODE_STATES[mode].state
|
||
|
||
async def get_state(self):
|
||
return self.value
|
||
|
||
async def set_state(self, state):
|
||
self.value = state.state
|
||
|
||
async def clear(self):
|
||
self.value = None
|
||
|
||
|
||
class DeployNoticeTests(unittest.IsolatedAsyncioTestCase):
|
||
async def test_sends_once_per_new_version_only_in_development(self):
|
||
fake_bot = AsyncMock()
|
||
state = _StubState("generation")
|
||
with (
|
||
patch.object(bot, "DEPLOY_ENV", "development"),
|
||
patch.object(bot, "VERSION", "0.1.0+5"),
|
||
patch.dict("os.environ", {"DEV_CHAT_ID": "2201809732"}, clear=False),
|
||
patch.object(bot, "get_deploy_notice_version", return_value="0.1.0+4"),
|
||
patch.object(bot, "set_deploy_notice_version"),
|
||
patch.object(bot.asyncio, "to_thread", new=AsyncMock(side_effect=["0.1.0+4", None, "0.1.0+5"])),
|
||
):
|
||
self.assertTrue(await bot.notify_development_deploy(fake_bot, cast(Any, state), 2201809732))
|
||
self.assertFalse(await bot.notify_development_deploy(fake_bot))
|
||
fake_bot.send_message.assert_awaited_once_with(
|
||
chat_id=2201809732,
|
||
text="🚀 Бот обновлён\nВерсия: 0.1.0+5",
|
||
reply_markup=bot.main_menu_keyboard(),
|
||
)
|
||
self.assertEqual(state.value, bot.BotMode.lobby.state)
|
||
|
||
async def test_does_not_send_outside_development(self):
|
||
fake_bot = AsyncMock()
|
||
for env in ("production", "", "staging", "Development", " development"):
|
||
with self.subTest(env=env), patch.object(bot, "DEPLOY_ENV", env):
|
||
await bot.notify_development_deploy(fake_bot)
|
||
fake_bot.send_message.assert_not_awaited()
|
||
|
||
async def test_deploy_notice_does_not_block_if_test_api_hangs(self):
|
||
fake_bot = AsyncMock()
|
||
fake_bot.send_message.side_effect = asyncio.TimeoutError
|
||
with (
|
||
patch.object(bot, "DEPLOY_ENV", "development"),
|
||
patch.dict("os.environ", {"DEV_CHAT_ID": "2201809732"}, clear=False),
|
||
patch.object(bot, "get_deploy_notice_version", return_value="old"),
|
||
):
|
||
await bot.notify_development_deploy(fake_bot)
|
||
fake_bot.send_message.assert_awaited_once()
|
||
|
||
|
||
class UserFacingErrorTests(unittest.IsolatedAsyncioTestCase):
|
||
async def test_failure_log_has_traceback_and_operation_id(self):
|
||
message = AsyncMock()
|
||
with bot.diagnostic_operation() as operation_id:
|
||
with self.assertLogs(bot.logger, level="ERROR") as captured:
|
||
try:
|
||
raise ValueError("provider failure")
|
||
except ValueError as exc:
|
||
await bot.report_error(message, "генерация поста", exc)
|
||
self.assertIn(f"operation_id={operation_id}", "\n".join(captured.output))
|
||
self.assertIn("Traceback (most recent call last)", "\n".join(captured.output))
|
||
self.assertIn("provider failure", "\n".join(captured.output))
|
||
|
||
async def test_production_error_is_generic_without_exception_details(self):
|
||
message = AsyncMock()
|
||
with patch.object(bot, "DEPLOY_ENV", "production"):
|
||
await bot.report_error(message, "генерация поста", RuntimeError("internal detail"))
|
||
message.answer.assert_awaited_once()
|
||
self.assertNotIn("internal detail", message.answer.await_args.args[0])
|
||
|
||
async def test_development_error_details_are_a_separate_message(self):
|
||
message = AsyncMock()
|
||
with patch.object(bot, "DEPLOY_ENV", "development"):
|
||
await bot.report_error(message, "генерация поста", RuntimeError("internal detail"))
|
||
self.assertEqual(message.answer.await_count, 2)
|
||
self.assertNotIn("internal detail", message.answer.await_args_list[0].args[0])
|
||
diagnostic = message.answer.await_args_list[1]
|
||
self.assertIn("Dev-диагностика", diagnostic.args[0])
|
||
self.assertIn("internal detail", diagnostic.args[0])
|
||
|
||
async def test_unknown_environment_does_not_send_developer_diagnostics(self):
|
||
message = AsyncMock()
|
||
with patch.object(bot, "DEPLOY_ENV", "staging"):
|
||
await bot.report_error(message, "генерация поста", RuntimeError("internal detail"))
|
||
message.answer.assert_awaited_once()
|
||
self.assertNotIn("internal detail", message.answer.await_args.args[0])
|
||
|
||
async def test_missing_environment_disables_developer_diagnostics(self):
|
||
message = AsyncMock()
|
||
with patch.object(bot, "DEPLOY_ENV", ""):
|
||
await bot.report_error(message, "генерация поста", RuntimeError("internal detail"))
|
||
message.answer.assert_awaited_once()
|
||
self.assertNotIn("internal detail", message.answer.await_args.args[0])
|
||
|
||
|
||
|
||
class HelpRoutingTests(unittest.TestCase):
|
||
def test_commands_are_routed_in_command_handler(self):
|
||
self.assertEqual(
|
||
bot.MODE_COMMANDS,
|
||
{
|
||
"/training": "training",
|
||
"/generate": "generation",
|
||
"/subscription": "subscription",
|
||
"/help": "help",
|
||
},
|
||
)
|
||
from pathlib import Path
|
||
|
||
source = Path(bot.__file__).read_text(encoding="utf-8")
|
||
self.assertIn("async def handle_command_text", source)
|
||
self.assertIn('if command == "/start":', source)
|
||
self.assertIn("elif command in MODE_COMMANDS:", source)
|
||
self.assertIn('elif command == "/clear":', source)
|
||
self.assertIn('elif command == "/idea":', source)
|
||
self.assertNotIn("idea_voice", source)
|
||
|
||
def test_help_sections_cover_usage_scenarios(self):
|
||
from help_message import HELP_SECTIONS, help_plain_view, rich_to_plain
|
||
|
||
text = "\n".join(rich_to_plain(body) for _title, body in HELP_SECTIONS.values())
|
||
self.assertIn("/clear", text)
|
||
self.assertIn("/idea", text)
|
||
self.assertNotIn("/idea_voice", text)
|
||
for term in ("пост", "голосовое", "скриншот", "реплаем", "/help"):
|
||
self.assertIn(term, text)
|
||
menu = help_plain_view(None)
|
||
for title, _body in HELP_SECTIONS.values():
|
||
self.assertIn(title, menu)
|
||
|
||
|
||
class HelpNavigationTests(unittest.TestCase):
|
||
def test_menu_packs_buttons_without_truncating_labels(self):
|
||
from help_message import HELP_SECTIONS, HELP_SECTION_ORDER, help_menu_html
|
||
|
||
menu = help_menu_html()
|
||
rows = menu.split("<tg-button-row")[1:]
|
||
self.assertLess(len(rows), len(HELP_SECTION_ORDER))
|
||
for row in rows:
|
||
self.assertTrue(row.startswith(' align="left">'), row)
|
||
self.assertLessEqual(row.count("<tg-button "), 2, row)
|
||
labels = [
|
||
title
|
||
for title, _body in HELP_SECTIONS.values()
|
||
if f">{title}</tg-button>" in row
|
||
]
|
||
self.assertLessEqual(
|
||
sum(len(title) for title in labels) + max(0, len(labels) - 1),
|
||
24,
|
||
row,
|
||
)
|
||
for section, (title, _body) in HELP_SECTIONS.items():
|
||
self.assertIn(f'data="help:{section}"', menu)
|
||
self.assertIn(f">{title}</tg-button>", menu)
|
||
self.assertNotIn('data="help:menu"', menu)
|
||
self.assertNotIn("В меню", menu)
|
||
|
||
def test_section_navigates_to_other_sections_without_menu_button(self):
|
||
from help_message import HELP_SECTIONS, help_section_html
|
||
|
||
for section in HELP_SECTIONS:
|
||
html = help_section_html(section)
|
||
self.assertNotIn('data="help:menu"', html)
|
||
self.assertNotIn("В меню", html)
|
||
for other in HELP_SECTIONS:
|
||
if other == section:
|
||
continue
|
||
self.assertIn(f'data="help:{other}"', html, section)
|
||
|
||
def test_current_section_is_disabled(self):
|
||
from help_message import HELP_SECTIONS, help_section_html
|
||
|
||
html = help_section_html("tone")
|
||
self.assertIn('<tg-button type="disabled">🎨 Tone of Voice</tg-button>', html)
|
||
self.assertNotIn('data="help:tone"', html)
|
||
for section in HELP_SECTIONS:
|
||
if section != "tone":
|
||
self.assertIn(f'data="help:{section}" style="primary"', html)
|
||
|
||
def test_unknown_section_falls_back_to_menu(self):
|
||
from help_message import help_view
|
||
|
||
self.assertEqual(help_view("nope"), help_view(None))
|
||
|
||
def test_plain_fallback_has_menu_and_section_text(self):
|
||
from help_message import help_plain_view
|
||
|
||
self.assertIn("Помощь по боту", help_plain_view(None))
|
||
self.assertIn("Написать пост", help_plain_view("posts"))
|
||
|
||
|
||
class OnboardingTests(unittest.IsolatedAsyncioTestCase):
|
||
async def test_start_sends_only_one_welcome_with_main_menu_keyboard(self):
|
||
message = AsyncMock()
|
||
message.from_user.id = 42
|
||
message.message_thread_id = None
|
||
calls = []
|
||
|
||
async def fake_rich(_message, html, _stage, *, reply_markup=None):
|
||
calls.append((html, reply_markup))
|
||
|
||
with (
|
||
patch.object(bot, "notify_development_deploy", new=AsyncMock()),
|
||
patch.object(bot, "ensure_subscription"),
|
||
patch.object(bot.asyncio, "to_thread", new=AsyncMock()),
|
||
patch.object(bot, "send_rich_message", new=fake_rich),
|
||
):
|
||
state = _StubState("help")
|
||
await bot.start(message, state)
|
||
self.assertEqual(len(calls), 1)
|
||
self.assertEqual(calls[0][0], bot.WELCOME_RICH_HTML)
|
||
markup = calls[0][1]
|
||
self.assertIsNotNone(markup)
|
||
if markup is None:
|
||
return
|
||
labels = {button.text for row in markup.keyboard for button in row}
|
||
expected = {
|
||
button.text
|
||
for row in bot.main_menu_keyboard().keyboard
|
||
for button in row
|
||
}
|
||
self.assertEqual(labels, expected)
|
||
message.answer.assert_not_awaited()
|
||
self.assertEqual(await state.get_state(), bot.BotMode.lobby.state)
|
||
|
||
def _callback(self, data: str, message: Message | None):
|
||
from aiogram.types import CallbackQuery
|
||
|
||
callback = AsyncMock(spec=CallbackQuery)
|
||
callback.data = data
|
||
callback.answer = AsyncMock()
|
||
callback.message = message
|
||
return callback
|
||
|
||
async def test_help_callback_edits_rich_message(self):
|
||
message = Message(
|
||
message_id=77,
|
||
date=datetime.now(),
|
||
chat=Chat(id=5, type="private"),
|
||
)
|
||
callback = self._callback("help:tone", message)
|
||
|
||
with patch.object(bot, "edit_rich_message", new=AsyncMock(return_value=True)) as edit:
|
||
await bot.help_navigation(callback, _StubState("help"))
|
||
|
||
edit.assert_awaited_once_with(message, bot.help_view("tone"))
|
||
callback.answer.assert_awaited_once_with()
|
||
|
||
async def test_help_callback_reports_error_when_edit_fails(self):
|
||
message = Message(
|
||
message_id=78,
|
||
date=datetime.now(),
|
||
chat=Chat(id=5, type="private"),
|
||
)
|
||
callback = self._callback("help:menu", message)
|
||
|
||
with patch.object(bot, "edit_rich_message", new=AsyncMock(return_value=False)):
|
||
await bot.help_navigation(callback, _StubState("help"))
|
||
|
||
callback.answer.assert_awaited_once_with("Не удалось обновить справку, открой /help заново", show_alert=True)
|
||
|
||
async def test_help_callback_ignores_inaccessible_message(self):
|
||
callback = self._callback("help:posts", None)
|
||
|
||
with patch.object(bot, "edit_rich_message", new=AsyncMock()) as edit:
|
||
await bot.help_navigation(callback, _StubState("help"))
|
||
|
||
edit.assert_not_awaited()
|
||
|
||
async def test_help_callback_is_rejected_outside_help_mode(self):
|
||
message = Message(
|
||
message_id=79,
|
||
date=datetime.now(),
|
||
chat=Chat(id=5, type="private"),
|
||
)
|
||
callback = self._callback("help:posts", message)
|
||
|
||
with patch.object(bot, "edit_rich_message", new=AsyncMock()) as edit:
|
||
await bot.help_navigation(callback, _StubState("lobby"))
|
||
|
||
edit.assert_not_awaited()
|
||
callback.answer.assert_awaited_once_with("Сначала открой режим /help.", show_alert=True)
|
||
|
||
def test_welcome_message_has_examples_and_help_handoff(self):
|
||
from help_message import WELCOME_RICH_HTML, rich_to_plain
|
||
|
||
text = rich_to_plain(WELCOME_RICH_HTML)
|
||
for term in ("Привет", "Напиши пост", "скриншот", "Главное меню"):
|
||
self.assertIn(term, text)
|
||
self.assertNotIn("Лобби", text)
|
||
self.assertNotIn("/lobby", text)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main() |