import asyncio import datetime import sqlite3 import tempfile import unittest from contextlib import closing from pathlib import Path from types import SimpleNamespace from typing import Any, cast from unittest.mock import AsyncMock, patch import activity_phrases import bot import support import user_store from aiogram.types import Chat, Message, User class _StubState: """Минимальная FSM-заглушка (полная версия — в tests/test_writer_features.py).""" def __init__(self, mode="lobby"): self.value = bot.MODE_STATES[mode].state self.data = {} 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 self.data = {} async def get_data(self): return dict(self.data) async def update_data(self, **kwargs): self.data.update(kwargs) return dict(self.data) class _TempDbCase(unittest.IsolatedAsyncioTestCase): def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() self.addCleanup(self.temp_dir.cleanup) self.db_path = str(Path(self.temp_dir.name) / "support_test.sqlite3") env = patch.dict("os.environ", {"BOT_DB_PATH": self.db_path}, clear=False) env.start() self.addCleanup(env.stop) user_store.init_db(self.db_path) support.init_support_table(self.db_path) with closing(sqlite3.connect(self.db_path)) as conn, conn: conn.execute( "CREATE TABLE IF NOT EXISTS payments (" "telegram_payment_charge_id TEXT PRIMARY KEY," "telegram_user_id INTEGER NOT NULL," "invoice_payload TEXT NOT NULL," "currency TEXT NOT NULL," "total_amount INTEGER NOT NULL," "created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)" ) def _user(self, user_id=42, trained=0, created_at=None): with closing(sqlite3.connect(self.db_path)) as conn, conn: conn.execute( "INSERT OR IGNORE INTO users (telegram_user_id, is_trained, created_at) " "VALUES (?, ?, COALESCE(?, CURRENT_TIMESTAMP))", (user_id, trained, created_at), ) def _payments(self, user_id, count): with closing(sqlite3.connect(self.db_path)) as conn, conn: for index in range(count): conn.execute( "INSERT OR IGNORE INTO payments " "(telegram_payment_charge_id, telegram_user_id, invoice_payload, " "currency, total_amount) VALUES (?, ?, 'p', 'XTR', 100)", (f"charge-{user_id}-{index}", user_id), ) class SupportProfileSummaryTests(_TempDbCase): def test_summary_reports_defaults_for_unknown_user(self): summary = user_store.support_profile_summary(999, self.db_path) self.assertEqual(summary["user_id"], 999) self.assertEqual(summary["days_used"], 0) self.assertEqual(summary["renewals"], 0) self.assertEqual(summary["days_left"], -1) self.assertEqual(summary["period"], "none") self.assertFalse(summary["is_trained"]) def test_summary_reports_trial_days_left_and_no_renewals(self): self._user(42) user_store.ensure_subscription(42, self.db_path) summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["period"], "trial") self.assertGreater(cast(int, summary["days_left"]), 0) self.assertEqual(summary["renewals"], 0) def test_summary_counts_payments_as_renewals(self): self._user(42) user_store.ensure_subscription(42, self.db_path) self._payments(42, 2) summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["renewals"], 2) def test_summary_reports_paid_period_after_activation(self): self._user(42) user_store.ensure_subscription(42, self.db_path) self._payments(42, 1) user_store.activate_subscription(42, months=1, db_path=self.db_path) summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["period"], "paid") self.assertGreater(cast(int, summary["days_left"]), 0) def test_summary_reports_trained_flag(self): self._user(42, trained=1) summary = user_store.support_profile_summary(42, self.db_path) self.assertTrue(summary["is_trained"]) def test_summary_reports_days_used_from_created_at(self): created = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta(days=9) self._user(42, created_at=created.isoformat()) summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["days_used"], 9) def test_summary_reports_expired_period_as_negative(self): self._user(42) user_store.ensure_subscription(42, self.db_path) with closing(sqlite3.connect(self.db_path)) as conn, conn: conn.execute( "UPDATE subscriptions SET trial_started_at = ? WHERE telegram_user_id = 42", ((datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta(days=10)).isoformat(),), ) summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["days_left"], -1) def test_summary_tolerates_broken_created_at(self): self._user(42, created_at="not-a-date") summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["days_used"], 0) def test_summary_survives_missing_payments_table(self): self._user(42) user_store.ensure_subscription(42, self.db_path) with closing(sqlite3.connect(self.db_path)) as conn, conn: conn.execute("DROP TABLE IF EXISTS payments") summary = user_store.support_profile_summary(42, self.db_path) self.assertEqual(summary["renewals"], 0) class SupportProfileCardTests(unittest.TestCase): def _tg_user(self, **kwargs): defaults = dict(id=42, first_name="Иван", last_name=None, username="ivan", is_bot=False) defaults.update(kwargs) return SimpleNamespace(**defaults) def test_card_without_summary_keeps_identity_only(self): card = support.profile_card(self._tg_user()) self.assertIn("Иван", card) self.assertIn("@ivan", card) self.assertIn("42", card) self.assertNotIn("Пользуется ботом", card) def test_card_with_summary_shows_all_profile_fields(self): summary = { "user_id": 42, "days_used": 9, "renewals": 2, "days_left": 12, "period": "paid", "is_trained": True, } card = support.profile_card(self._tg_user(), summary) self.assertIn("Пользуется ботом", card) self.assertIn("9 дн.", card) self.assertIn("Продлений подписки", card) self.assertIn("Осталось дней", card) self.assertIn("12 дн.", card) self.assertIn("Бот обучен", card) self.assertIn("да", card) def test_card_marks_untrained_bot(self): summary = {"days_used": 1, "renewals": 0, "days_left": 2, "is_trained": False} card = support.profile_card(self._tg_user(), summary) self.assertIn("Бот обучен", card) self.assertIn("нет", card) def test_card_marks_expired_period(self): summary = {"days_used": 40, "renewals": 0, "days_left": -1, "is_trained": False} card = support.profile_card(self._tg_user(), summary) self.assertIn("нет активного периода", card) def test_card_falls_back_to_id_when_no_username(self): card = support.profile_card(self._tg_user(username=None)) self.assertNotIn("@", card) self.assertIn("42", card) def test_card_tolerates_garbage_summary_values(self): summary = {"days_used": None, "renewals": "x", "days_left": None, "is_trained": None} card = support.profile_card(self._tg_user(), summary) self.assertIn("0 дн.", card) self.assertIn("нет активного периода", card) class SupportGreetingTests(unittest.TestCase): def test_greeting_comes_from_the_pool(self): for _ in range(30): self.assertIn(support.support_greeting(), activity_phrases.SUPPORT_GREETINGS) def test_greeting_never_mentions_relaying(self): import re for phrase in activity_phrases.SUPPORT_GREETINGS: lowered = phrase.lower() # Слова целиком: простое вхождение "бот" ловит "добрый". words = set(re.findall(r"[а-яё]+", lowered)) for banned in ("передам", "передаю", "передаёт", "оператор", "бот", "боту", "ботом"): self.assertNotIn(banned, words, phrase) def test_greeting_asks_to_describe_the_problem(self): for phrase in activity_phrases.SUPPORT_GREETINGS: self.assertTrue( any(marker in phrase.lower() for marker in ("опиш", "опис", "расскаж", "напиш", "случилось")), phrase, ) def test_pool_has_enough_variety(self): self.assertGreaterEqual(len(activity_phrases.SUPPORT_GREETINGS), 5) self.assertEqual(len(set(activity_phrases.SUPPORT_GREETINGS)), len(activity_phrases.SUPPORT_GREETINGS)) def test_env_overrides_the_pool(self): custom = "Первая фраза|Вторая фраза" with patch.dict("os.environ", {"SUPPORT_GREETINGS": custom}, clear=False): for _ in range(10): self.assertIn(support.support_greeting(), ("Первая фраза", "Вторая фраза")) def test_env_pool_ignores_empty_entries(self): with patch.dict("os.environ", {"SUPPORT_GREETINGS": " |Одна фраза| "}, clear=False): self.assertEqual(support.support_greeting(), "Одна фраза") def test_env_pool_of_separators_only_falls_back(self): with patch.dict("os.environ", {"SUPPORT_GREETINGS": "|||"}, clear=False): self.assertIn(support.support_greeting(), activity_phrases.SUPPORT_GREETINGS) class SupportBotControlTests(unittest.IsolatedAsyncioTestCase): """При активном обращении кнопки и команды бота не должны уходить оператору.""" def _msg(self, text, chat_type="private", chat_id=555, user_id=42, is_bot=False): # Нужен настоящий Message: middleware проверяет isinstance(event, Message), # иначе все проверки проходят вхолостую. return Message( message_id=1, date=datetime.datetime.now(), chat=Chat(id=chat_id, type=chat_type), from_user=User(id=user_id, is_bot=is_bot, first_name="Test"), text=text, ) def test_mode_buttons_are_bot_control(self): for text in ("📖 Помощь", "🎓 Обучение", "✍️ Генерация", "💳 Подписка"): self.assertTrue(support._is_bot_control(self._msg(text)), text) def test_main_menu_and_support_buttons_are_bot_control(self): self.assertTrue(support._is_bot_control(self._msg("🏠 Главное меню"))) self.assertTrue(support._is_bot_control(self._msg("🆘 Поддержка"))) self.assertTrue(support._is_bot_control(self._msg("🗑 Сбросить данные"))) self.assertTrue(support._is_bot_control(self._msg("✍️ Придумать пост"))) def test_commands_are_bot_control(self): for text in ("/start", "/help", "/training", "/generate", "/post", "/idea", "/balance"): self.assertTrue(support._is_bot_control(self._msg(text)), text) def test_plain_text_is_not_bot_control(self): for text in ("не работает оплата", "ау", "помогите", "почему списали деньги"): self.assertFalse(support._is_bot_control(self._msg(text)), text) def test_unknown_slash_command_is_not_bot_control(self): self.assertFalse(support._is_bot_control(self._msg("/что-то-неизвестное"))) def test_empty_text_is_not_bot_control(self): self.assertFalse(support._is_bot_control(self._msg(""))) self.assertFalse(support._is_bot_control(self._msg(None))) def test_middleware_passes_mode_buttons_to_the_bot(self): # Регрессия: «Помощь»/«Обучение» уезжали оператору, бот в личке молчал. with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): for text in ("📖 Помощь", "🎓 Обучение", "🏠 Главное меню", "🆘 Поддержка"): event = self._msg(text) # is_support_active вызывается через asyncio.to_thread — нужна # синхронная заглушка: AsyncMock вернул бы корутину (всегда truthy). with patch.object(support, "is_support_active", new=lambda uid, db_path=None: True): with patch.object(support, "relay_user_message", new=AsyncMock()) as relay: handled = asyncio.run(self._call_middleware(event)) self.assertEqual(handled, "handled", text) relay.assert_not_awaited() def test_middleware_relays_plain_text_to_operator(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): event = self._msg("не работает оплата") with patch.object(support, "is_support_active", new=lambda uid, db_path=None: True): with patch.object(support, "relay_user_message", new=AsyncMock()) as relay: handled = asyncio.run(self._call_middleware(event)) self.assertIsNone(handled) relay.assert_awaited_once_with(event) def test_middleware_passes_text_to_bot_when_no_active_case(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): event = self._msg("напиши пост про запуск продукта") with patch.object(support, "is_support_active", new=lambda uid, db_path=None: False): with patch.object(support, "relay_user_message", new=AsyncMock()) as relay: handled = asyncio.run(self._call_middleware(event)) self.assertEqual(handled, "handled") relay.assert_not_awaited() def test_middleware_ignores_group_chats(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): event = self._msg("текст", chat_type="supergroup", chat_id=-100123) with patch.object(support, "relay_user_message", new=AsyncMock()) as relay: asyncio.run(self._call_middleware(event)) relay.assert_not_awaited() async def _call_middleware(self, event): async def handler(_event, _data): return "handled" middleware = support.SupportRelayMiddleware() return await middleware(handler, event, {}) class SupportStartTextTests(unittest.TestCase): def test_start_text_embeds_greeting(self): rendered = support.SUPPORT_START_TEXT.format(greeting="Привет!") self.assertIn("Привет!", rendered) self.assertIn("Главное меню", rendered) def test_start_text_has_no_relay_wording(self): rendered = support.SUPPORT_START_TEXT.format(greeting="Привет!") self.assertNotIn("передам", rendered.lower()) self.assertNotIn("оператор", rendered.lower()) class StartClosesSupportCaseTests(_TempDbCase): """/start — возврат в лобби, поэтому обращение поддержки должно закрываться. Регрессия: /start сбрасывал FSM, но не снимал is_active в БД. Пользователь оказывался в лобби, а следующее сообщение всё равно уходило оператору; вход в режим дополнительно блокировался проверкой в enter_mode(). """ def _tg_user(self, user_id=42): return SimpleNamespace(id=user_id, is_bot=False, first_name="Test", username="test") def _message(self, text="/start", user_id=42): message = SimpleNamespace( text=text, from_user=self._tg_user(user_id), chat=SimpleNamespace(id=555, type="private"), message_thread_id=None, bot=AsyncMock(), answers=[], ) async def answer(text, **kwargs): message.answers.append((text, kwargs)) message.answer = answer return message async def test_start_deactivates_active_support_case(self): # is_support_active() и deactivate() работают только при заданной группе. with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): support.set_topic(42, 777, self.db_path) support.set_active(42, True, self.db_path) self.assertTrue(support.is_support_active(42, self.db_path)) message = self._message() with ( patch.object(bot, "notify_development_deploy", new=AsyncMock()), patch.object(bot, "send_rich_message", new=AsyncMock()), ): await bot.start(message, _StubState("help")) self.assertFalse(support.is_support_active(42, self.db_path)) # Топик сохраняется: при следующем обращении переписка продолжается. self.assertEqual(support.get_topic(42, self.db_path), 777) async def test_start_without_case_does_not_fail(self): message = self._message() with ( patch.object(bot, "notify_development_deploy", new=AsyncMock()), patch.object(bot, "send_rich_message", new=AsyncMock()), ): await bot.start(message, _StubState("lobby")) self.assertFalse(support.is_support_active(42, self.db_path)) async def test_enter_any_mode_closes_support_case(self): """Вход в любой режим = уход из поддержки, флаг снимается. Регрессия: деактивация была только в /start и «Главное меню», поэтому после перехода в «Обучение» текст уезжал оператору, а не в обучение. """ for mode in ("training", "generation", "subscription", "help"): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): support.set_topic(42, 777, self.db_path) support.set_active(42, True, self.db_path) state = _StubState("help") message = self._message("/training") with patch.object(bot, "send_rich_message", new=AsyncMock()): await bot.enter_mode(message, state, mode) self.assertFalse( support.is_support_active(42, self.db_path), f"режим {mode} не закрыл обращение", ) # Топик сохраняется: следующее обращение продолжает переписку. self.assertEqual(support.get_topic(42, self.db_path), 777, mode) async def test_mode_switch_then_text_goes_to_bot_not_operator(self): """После перехода в режим обычный текст не должен уходить оператору.""" with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): support.set_topic(42, 777, self.db_path) support.set_active(42, True, self.db_path) state = _StubState("help") message = self._message("/training") with patch.object(bot, "send_rich_message", new=AsyncMock()): await bot.enter_mode(message, state, "training") # Middleware теперь должен пропустить текст в обычные обработчики. event = Message( message_id=2, date=datetime.datetime.now(), chat=Chat(id=555, type="private"), from_user=User(id=42, is_bot=False, first_name="Test"), text="кофейня в Калининграде для семей с детьми", ) with patch.object(support, "relay_user_message", new=AsyncMock()) as relay: handled = await self._call_middleware(event) self.assertEqual(handled, "handled") relay.assert_not_awaited() async def _call_middleware(self, event): async def handler(_event, _data): return "handled" return await support.SupportRelayMiddleware()(handler, event, {}) async def test_enter_mode_no_longer_blocks_other_modes(self): """Вход в режим из открытого обращения разрешён — пользователя нельзя запирать.""" support.set_topic(42, 777, self.db_path) support.set_active(42, True, self.db_path) state = _StubState("help") message = self._message("/training") with patch.object(bot, "send_rich_message", new=AsyncMock()): await bot.enter_mode(message, state, "training") self.assertEqual(await state.get_state(), bot.BotMode.training.state) texts = " ".join(text for text, _ in message.answers) self.assertNotIn("Идёт обращение в поддержку", texts) async def test_main_menu_closes_support_case(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): support.set_topic(42, 777, self.db_path) support.set_active(42, True, self.db_path) await bot.show_main_menu(self._message("🏠 Главное меню"), _StubState("help")) self.assertFalse(support.is_support_active(42, self.db_path)) class SupportTopicTests(_TempDbCase): def _tg_user(self, **kwargs): defaults = dict(id=42, first_name="Иван", last_name=None, username="ivan", is_bot=False) defaults.update(kwargs) return SimpleNamespace(**defaults) def test_topic_is_created_once_and_reused(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): bot = AsyncMock() topic = SimpleNamespace(message_thread_id=777) bot.create_forum_topic = AsyncMock(return_value=topic) bot.get_user_profile_photos = AsyncMock(side_effect=RuntimeError("no photos")) user = self._tg_user() first = asyncio.run(support.ensure_topic(bot, user)) second = asyncio.run(support.ensure_topic(bot, user)) self.assertEqual(first[0], 777) self.assertIsNone(first[1]) self.assertEqual(second[0], 777) bot.create_forum_topic.assert_awaited_once() def test_topic_creation_failure_returns_reason(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): bot = AsyncMock() bot.create_forum_topic = AsyncMock( side_effect=RuntimeError("Bad Request: not enough rights to create a topic") ) thread_id, reason = asyncio.run(support.ensure_topic(bot, self._tg_user())) self.assertIsNone(thread_id) self.assertIn("not enough rights", reason) def test_topic_creation_failure_logs_telegram_reason(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): bot = AsyncMock() bot.create_forum_topic = AsyncMock( side_effect=RuntimeError("Bad Request: not enough rights to create a topic") ) with self.assertLogs("support", level="ERROR") as captured: asyncio.run(support.ensure_topic(bot, self._tg_user())) self.assertIn("not enough rights", "\n".join(captured.output)) def test_unavailable_text_hides_reason_outside_development(self): with patch.dict("os.environ", {"DEPLOY_ENV": "production"}, clear=False): text = support._unavailable_text("not enough rights to create a topic") self.assertNotIn("not enough rights", text) self.assertIn("временно недоступна", text) def test_unavailable_text_shows_reason_in_development(self): with patch.dict("os.environ", {"DEPLOY_ENV": "development"}, clear=False): text = support._unavailable_text("not enough rights to create a topic") self.assertIn("not enough rights", text) self.assertIn("Dev-диагностика", text) def test_unavailable_text_without_reason_is_plain(self): with patch.dict("os.environ", {"DEPLOY_ENV": "development"}, clear=False): text = support._unavailable_text(None) self.assertEqual(text, support.SUPPORT_UNAVAILABLE_TEXT) def test_unavailable_text_escapes_html(self): with patch.dict("os.environ", {"DEPLOY_ENV": "development"}, clear=False): text = support._unavailable_text("x & y") self.assertNotIn("x", text) self.assertIn("<b>x</b>", text) def test_topic_creation_is_skipped_when_disabled(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": ""}, clear=False): bot = AsyncMock() thread_id, reason = asyncio.run(support.ensure_topic(bot, self._tg_user())) self.assertIsNone(thread_id) self.assertIsNone(reason) bot.create_forum_topic.assert_not_awaited() def test_profile_card_is_sent_only_on_topic_creation(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False): bot = AsyncMock() bot.create_forum_topic = AsyncMock( return_value=SimpleNamespace(message_thread_id=778) ) bot.get_user_profile_photos = AsyncMock(side_effect=RuntimeError("no photos")) user = self._tg_user() asyncio.run(support.ensure_topic(bot, user)) sends_after_first = bot.send_message.await_count asyncio.run(support.ensure_topic(bot, user)) self.assertEqual(sends_after_first, 1) self.assertEqual(bot.send_message.await_count, 1) self.assertIn("Пользуется ботом", bot.send_message.await_args.kwargs["text"]) def test_topic_title_uses_username_when_present(self): user = self._tg_user(first_name="Иван", last_name="Петров", username="ivan") self.assertEqual(support.topic_title(user), "Иван Петров · @ivan") def test_topic_title_falls_back_to_id(self): self.assertEqual(support.topic_title(self._tg_user(username=None)), "Иван · 42") def test_topic_title_is_capped(self): long_name = "А" * 200 title = support.topic_title(self._tg_user(first_name=long_name, username="ivan")) self.assertLessEqual(len(title), support.TOPIC_TITLE_LIMIT) def test_support_is_disabled_without_group_id(self): with patch.dict("os.environ", {"SUPPORT_GROUP_ID": ""}, clear=False): self.assertFalse(support.is_enabled()) self.assertFalse(support.is_support_active(42, self.db_path)) if __name__ == "__main__": unittest.main()