330 lines
13 KiB
Python
330 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_contains_no_embedded_buttons(self):
|
|
from help_message import help_menu_html
|
|
|
|
self.assertNotIn("<tg-button", help_menu_html())
|
|
|
|
def test_section_contains_no_embedded_buttons(self):
|
|
from help_message import HELP_SECTIONS, help_section_html
|
|
|
|
for section in HELP_SECTIONS:
|
|
with self.subTest(section=section):
|
|
self.assertNotIn("<tg-button", help_section_html(section))
|
|
|
|
def test_help_keyboard_has_one_button_per_section_with_stable_callback_data(self):
|
|
from help_message import HELP_SECTIONS, HELP_SECTION_ORDER
|
|
from keyboards import help_inline_keyboard
|
|
|
|
rows = help_inline_keyboard().inline_keyboard
|
|
self.assertEqual(len(rows), len(HELP_SECTION_ORDER))
|
|
for section, row in zip(HELP_SECTION_ORDER, rows, strict=True):
|
|
self.assertEqual(len(row), 1)
|
|
self.assertEqual(row[0].text, HELP_SECTIONS[section][0])
|
|
self.assertEqual(row[0].callback_data, f"help:{section}")
|
|
|
|
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_edit_rich_message_attaches_standard_keyboard(self):
|
|
message_bot = AsyncMock()
|
|
message = Message(
|
|
message_id=80,
|
|
date=datetime.now(),
|
|
chat=Chat(id=5, type="private"),
|
|
).as_(message_bot)
|
|
markup = bot.help_inline_keyboard()
|
|
|
|
result = await bot.edit_rich_message(
|
|
message,
|
|
bot.help_view("tone"),
|
|
reply_markup=markup,
|
|
)
|
|
|
|
self.assertTrue(result)
|
|
message_bot.edit_message_text.assert_awaited_once()
|
|
call = message_bot.edit_message_text.await_args.kwargs
|
|
self.assertEqual(call["chat_id"], 5)
|
|
self.assertEqual(call["message_id"], 80)
|
|
self.assertEqual(call["reply_markup"], markup)
|
|
self.assertEqual(call["rich_message"], bot.InputRichMessage(html=bot.help_view("tone")))
|
|
|
|
async def test_edit_rich_message_treats_unchanged_content_as_success(self):
|
|
message_bot = AsyncMock()
|
|
message_bot.edit_message_text.side_effect = RuntimeError("Bad Request: message is not modified")
|
|
message = Message(
|
|
message_id=81,
|
|
date=datetime.now(),
|
|
chat=Chat(id=5, type="private"),
|
|
).as_(message_bot)
|
|
|
|
self.assertTrue(await bot.edit_rich_message(message, bot.help_view("tone")))
|
|
|
|
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"),
|
|
reply_markup=bot.help_inline_keyboard(),
|
|
)
|
|
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_edits_message_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(return_value=True)) as edit:
|
|
await bot.help_navigation(callback, _StubState("lobby"))
|
|
|
|
edit.assert_awaited_once_with(
|
|
message,
|
|
bot.help_view("posts"),
|
|
reply_markup=bot.help_inline_keyboard(),
|
|
)
|
|
callback.answer.assert_awaited_once_with()
|
|
|
|
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() |