hermes
21db563f28
Обращение открывается кнопкой «🆘 Поддержка» или /support в режиме «Помощь».
Отдельного FSM-режима нет: пока обращение активно, внизу остаётся только
«Главное меню», а переключение режимов отклоняется.
- support.py: личка пользователя ↔ персональный топик группы SUPPORT_GROUP_ID.
Топик создаётся один раз на пользователя и переиспользуется, поэтому
повторные обращения продолжают ту же переписку. Пересылаются все типы
контента через copy_message; ответ оператора из топика уходит в личку;
сообщение с префиксом // остаётся внутренней заметкой. Функция выключена
без SUPPORT_GROUP_ID.
- Карточка профиля уходит первым сообщением топика: аватарка, имя, username,
id и сводка — сколько дней пользуется ботом, сколько продлений подписки,
сколько дней осталось, обучен ли бот (user_store.support_profile_summary).
- Бот не объясняет пользователю, что «передаёт оператору»: в ответ приходит
случайная фраза из пула SUPPORT_GREETINGS (activity_phrases.support_greeting).
Формулировки безличные, без «я/бот/оператор»; пул переопределяется через
SUPPORT_GREETINGS без деплоя.
Проверено: 207 тестов OK (30 новых в tests/test_support.py).
VERSION: 0.1.0+51 -> 0.1.0+52
293 lines
13 KiB
Python
293 lines
13 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 support
|
||
import user_store
|
||
|
||
|
||
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 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 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, 777)
|
||
self.assertEqual(second, 777)
|
||
bot.create_forum_topic.assert_awaited_once()
|
||
|
||
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()
|