415 lines
12 KiB
Python
415 lines
12 KiB
Python
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
from bot.services.notification_service import NotificationService
|
|
from config.settings import Settings
|
|
|
|
|
|
def _settings(**overrides):
|
|
data = {
|
|
"BOT_TOKEN": "123456:test",
|
|
"POSTGRES_USER": "app_user",
|
|
"POSTGRES_PASSWORD": "app_password",
|
|
}
|
|
data.update(overrides)
|
|
return Settings(_env_file=None, **data)
|
|
|
|
|
|
def test_support_ticket_url_uses_subscription_mini_app_url():
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.com"),
|
|
)
|
|
|
|
assert service._support_ticket_url(42, admin=True) == "https://app.example.com/admin/support/42"
|
|
assert service._support_ticket_url(42, admin=False) == "https://app.example.com/support/42"
|
|
|
|
|
|
def test_support_ticket_url_falls_back_to_startapp_deeplink():
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(),
|
|
bot_username="demo_bot",
|
|
)
|
|
|
|
assert service._support_ticket_url(42) == "https://t.me/demo_bot?startapp=ticket_42"
|
|
|
|
|
|
def test_admin_support_keyboard_uses_consistent_admin_links():
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.com/app"),
|
|
)
|
|
ticket = SimpleNamespace(ticket_id=42)
|
|
user = SimpleNamespace(user_id=100200300)
|
|
|
|
keyboard = service._support_keyboard(ticket, user, admin=True)
|
|
ticket_button = keyboard.inline_keyboard[0][0]
|
|
user_card_button = keyboard.inline_keyboard[1][1]
|
|
|
|
assert keyboard.inline_keyboard[0][0].text == "Открыть тикет"
|
|
assert ticket_button.url is None
|
|
assert ticket_button.web_app.url == "https://app.example.com/app/admin/support/42"
|
|
assert keyboard.inline_keyboard[1][0].url == "tg://user?id=100200300"
|
|
assert user_card_button.url is None
|
|
assert user_card_button.web_app.url == "https://app.example.com/app/admin/users/100200300"
|
|
|
|
|
|
def test_admin_support_keyboard_falls_back_to_startapp_url():
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(),
|
|
bot_username="demo_bot",
|
|
)
|
|
ticket = SimpleNamespace(ticket_id=42)
|
|
user = SimpleNamespace(user_id=100200300)
|
|
|
|
keyboard = service._support_keyboard(ticket, user, admin=True)
|
|
button = keyboard.inline_keyboard[0][0]
|
|
|
|
assert button.web_app is None
|
|
assert button.url == "https://t.me/demo_bot?startapp=ticket_42"
|
|
|
|
|
|
def test_user_support_keyboard_uses_web_app_button_when_configured():
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.com/app"),
|
|
)
|
|
ticket = SimpleNamespace(ticket_id=42)
|
|
user = SimpleNamespace(language_code="ru")
|
|
|
|
keyboard = service._support_user_keyboard(ticket, user)
|
|
button = keyboard.inline_keyboard[0][0]
|
|
|
|
assert button.text == "Открыть тикет"
|
|
assert button.url is None
|
|
assert button.web_app.url == "https://app.example.com/app/support/42"
|
|
|
|
|
|
def test_user_support_keyboard_falls_back_to_startapp_url():
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(),
|
|
bot_username="demo_bot",
|
|
)
|
|
ticket = SimpleNamespace(ticket_id=42)
|
|
user = SimpleNamespace(language_code="ru")
|
|
|
|
keyboard = service._support_user_keyboard(ticket, user)
|
|
button = keyboard.inline_keyboard[0][0]
|
|
|
|
assert button.text == "Открыть тикет"
|
|
assert button.web_app is None
|
|
assert button.url == "https://t.me/demo_bot?startapp=ticket_42"
|
|
|
|
|
|
def test_admin_support_email_notifications_can_be_disabled():
|
|
sent = []
|
|
|
|
class EmailService:
|
|
async def send_rendered_email(self, *, email, content):
|
|
sent.append((email, content))
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED=False),
|
|
email_auth_service=EmailService(),
|
|
)
|
|
|
|
async def admin_email_users():
|
|
return [SimpleNamespace(user_id=1, email="admin@example.com", language_code="en")]
|
|
|
|
service._admin_email_users = admin_email_users
|
|
|
|
async def run():
|
|
await service._send_admin_support_email(
|
|
lambda *_args, **_kwargs: SimpleNamespace(subject="Ticket", html="Body", text="Body"),
|
|
ticket_id=1,
|
|
)
|
|
|
|
asyncio.run(run())
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_admin_support_email_notifications_default_to_disabled():
|
|
sent = []
|
|
|
|
class EmailService:
|
|
async def send_rendered_email(self, *, email, content):
|
|
sent.append((email, content))
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(),
|
|
email_auth_service=EmailService(),
|
|
)
|
|
|
|
async def admin_email_users():
|
|
return [SimpleNamespace(user_id=1, email="admin@example.com", language_code="en")]
|
|
|
|
service._admin_email_users = admin_email_users
|
|
|
|
async def run():
|
|
await service._send_admin_support_email(
|
|
lambda *_args, **_kwargs: SimpleNamespace(subject="Ticket", html="Body", text="Body"),
|
|
ticket_id=1,
|
|
)
|
|
|
|
asyncio.run(run())
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_persisted_support_email_override_disables_env_enabled(monkeypatch):
|
|
sent = []
|
|
|
|
class EmailService:
|
|
async def send_rendered_email(self, *, email, content):
|
|
sent.append((email, content))
|
|
|
|
class SessionFactory:
|
|
def __call__(self):
|
|
return self
|
|
|
|
async def __aenter__(self):
|
|
return SimpleNamespace()
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return None
|
|
|
|
async def get_override_value(_session, key):
|
|
assert key == "SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED"
|
|
return True, False
|
|
|
|
monkeypatch.setattr(
|
|
"bot.services.notification_service.app_settings_dal.get_override_value",
|
|
get_override_value,
|
|
)
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED=True),
|
|
session_factory=SessionFactory(),
|
|
email_auth_service=EmailService(),
|
|
)
|
|
|
|
async def admin_email_users():
|
|
return [SimpleNamespace(user_id=1, email="admin@example.com", language_code="en")]
|
|
|
|
service._admin_email_users = admin_email_users
|
|
|
|
async def run():
|
|
await service._send_admin_support_email(
|
|
lambda *_args, **_kwargs: SimpleNamespace(subject="Ticket", html="Body", text="Body"),
|
|
ticket_id=1,
|
|
)
|
|
|
|
asyncio.run(run())
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_persisted_support_email_override_enables_default_disabled(monkeypatch):
|
|
sent = []
|
|
|
|
class EmailService:
|
|
async def send_rendered_email(self, *, email, content):
|
|
sent.append((email, content))
|
|
|
|
class SessionFactory:
|
|
def __call__(self):
|
|
return self
|
|
|
|
async def __aenter__(self):
|
|
return SimpleNamespace()
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return None
|
|
|
|
async def get_override_value(_session, key):
|
|
assert key == "SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED"
|
|
return True, True
|
|
|
|
monkeypatch.setattr(
|
|
"bot.services.notification_service.app_settings_dal.get_override_value",
|
|
get_override_value,
|
|
)
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(),
|
|
session_factory=SessionFactory(),
|
|
email_auth_service=EmailService(),
|
|
)
|
|
|
|
async def admin_email_users():
|
|
return [SimpleNamespace(user_id=1, email="admin@example.com", language_code="en")]
|
|
|
|
service._admin_email_users = admin_email_users
|
|
|
|
async def run():
|
|
await service._send_admin_support_email(
|
|
lambda *_args, **_kwargs: SimpleNamespace(subject="Ticket", html="Body", text="Body"),
|
|
ticket_id=1,
|
|
)
|
|
|
|
asyncio.run(run())
|
|
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_disabled_admin_support_email_keeps_telegram_and_log_notifications():
|
|
emails = []
|
|
channels = []
|
|
|
|
class EmailService:
|
|
async def send_rendered_email(self, *, email, content):
|
|
emails.append((email, content))
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(
|
|
SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED=False,
|
|
SUBSCRIPTION_MINI_APP_URL="https://app.example.com",
|
|
),
|
|
email_auth_service=EmailService(),
|
|
)
|
|
|
|
async def send_to_admins(message, reply_markup=None):
|
|
channels.append(("admins", bool(message), bool(reply_markup)))
|
|
|
|
async def send_to_log_channel(message, thread_id=None, reply_markup=None):
|
|
channels.append(("log", bool(message), bool(reply_markup)))
|
|
|
|
service._send_to_admins = send_to_admins
|
|
service._send_to_log_channel = send_to_log_channel
|
|
|
|
ticket = SimpleNamespace(
|
|
ticket_id=7,
|
|
priority="normal",
|
|
category="technical",
|
|
subject="Connection issue",
|
|
)
|
|
user = SimpleNamespace(
|
|
user_id=100200300,
|
|
username="user",
|
|
first_name="User",
|
|
last_name=None,
|
|
email="user@example.com",
|
|
)
|
|
|
|
asyncio.run(
|
|
service.notify_new_support_ticket(
|
|
ticket,
|
|
user,
|
|
"Cannot connect",
|
|
{"tariff": "Standard", "end_date": "2026-06-01"},
|
|
)
|
|
)
|
|
|
|
assert [item[0] for item in channels] == ["admins", "log"]
|
|
assert emails == []
|
|
|
|
|
|
def test_support_user_reply_can_send_email_without_telegram_channels():
|
|
emails = []
|
|
channels = []
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.com"),
|
|
)
|
|
|
|
async def send_to_admins(message, reply_markup=None):
|
|
channels.append(("admins", bool(message), bool(reply_markup)))
|
|
|
|
async def send_to_log_channel(message, thread_id=None, reply_markup=None):
|
|
channels.append(("log", bool(message), bool(reply_markup)))
|
|
|
|
async def send_admin_support_email(renderer, **kwargs):
|
|
emails.append(kwargs)
|
|
|
|
service._send_to_admins = send_to_admins
|
|
service._send_to_log_channel = send_to_log_channel
|
|
service._send_admin_support_email = send_admin_support_email
|
|
|
|
ticket = SimpleNamespace(
|
|
ticket_id=7,
|
|
priority="normal",
|
|
category="technical",
|
|
subject="Connection issue",
|
|
)
|
|
message = SimpleNamespace(body="Still cannot connect")
|
|
user = SimpleNamespace(
|
|
user_id=100200300,
|
|
username="user",
|
|
first_name="User",
|
|
last_name=None,
|
|
email="user@example.com",
|
|
)
|
|
|
|
asyncio.run(
|
|
service.notify_support_user_reply(
|
|
ticket,
|
|
message,
|
|
user,
|
|
{},
|
|
unread_count=3,
|
|
send_telegram=False,
|
|
send_email=True,
|
|
)
|
|
)
|
|
|
|
assert channels == []
|
|
assert emails[0]["ticket_id"] == 7
|
|
|
|
|
|
def test_account_merge_notification_goes_to_log_channel():
|
|
messages = []
|
|
|
|
class I18n:
|
|
def gettext(self, _language, key, **kwargs):
|
|
if key == "log_open_profile_link":
|
|
return "Open profile"
|
|
assert key == "log_account_merged"
|
|
return (
|
|
f"merged primary={kwargs['primary_user_id']} "
|
|
f"removed={kwargs['removed_user_id']} "
|
|
f"email={kwargs['email']} end={kwargs['final_end_date']}"
|
|
)
|
|
|
|
service = NotificationService(
|
|
bot=SimpleNamespace(),
|
|
settings=_settings(LOG_CHAT_ID=-100123, DEFAULT_LANGUAGE="en"),
|
|
i18n=I18n(),
|
|
)
|
|
|
|
async def send_to_log_channel(message, thread_id=None, reply_markup=None):
|
|
messages.append((message, thread_id, reply_markup))
|
|
|
|
service._send_to_log_channel = send_to_log_channel
|
|
|
|
asyncio.run(
|
|
service.notify_account_merged(
|
|
primary_user_id=42,
|
|
removed_user_id=-100,
|
|
email="paid@example.com",
|
|
telegram_id=100200300,
|
|
username="alice",
|
|
first_name="Alice",
|
|
final_end_date_text="2026-06-21 10:00",
|
|
primary_panel_user_uuid="panel-telegram",
|
|
removed_panel_user_uuid="panel-email",
|
|
)
|
|
)
|
|
|
|
assert len(messages) == 1
|
|
message, thread_id, reply_markup = messages[0]
|
|
assert "primary=42" in message
|
|
assert "removed=-100" in message
|
|
assert "paid@example.com" in message
|
|
assert thread_id is None
|
|
assert reply_markup.inline_keyboard[0][0].url == "tg://user?id=100200300"
|