hermes
12ef2c6b44
/start возвращает в лобби, но не снимал is_active в support_topics: FSM сбрасывался, а обращение в БД оставалось открытым. Пользователь оказывался в лобби, а следующее сообщение всё равно уходило оператору. Дополнительно enter_mode() блокировал вход в режим сообщением «Идёт обращение в поддержку» — после /start это выглядело как необъяснимый отказ. - start() вызывает support.deactivate() перед сбросом FSM. - Блокировка из enter_mode() удалена: выход в любой режим из открытого обращения разрешён — пользователя нельзя запирать. Проверка дублировала middleware иConflictовала с /start. Тесты: 227 OK (3 новых). Регрессия проверена на старом коде: /start оставляет is_active=True (баг), с фиксом — False. Топик при этом сохраняется, поэтому следующее обращение продолжает ту же переписку. VERSION: 0.1.0+54 -> 0.1.0+55
529 lines
24 KiB
Python
529 lines
24 KiB
Python
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("<code>42</code>", 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("<code>42</code>", 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_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)
|
||
|
||
|
||
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("<b>x</b> & y")
|
||
self.assertNotIn("<b>x</b>", 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()
|