diff --git a/backend/bot/services/telegram_notifications.py b/backend/bot/services/telegram_notifications.py index 8d59b83..326b822 100644 --- a/backend/bot/services/telegram_notifications.py +++ b/backend/bot/services/telegram_notifications.py @@ -4,7 +4,6 @@ from typing import Any, Optional from aiogram import Bot from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError -from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, WebAppInfo from sqlalchemy.ext.asyncio import AsyncSession from bot.middlewares.i18n import JsonI18n @@ -128,37 +127,6 @@ async def mark_telegram_notifications_enabled_for_telegram_user( ) -def _translate( - i18n: Optional[JsonI18n], - language: str, - key: str, - fallback: str, - **kwargs: Any, -) -> str: - if not i18n: - return fallback.format(**kwargs) if kwargs else fallback - return i18n.gettext(language, key, **kwargs) or fallback - - -def _probe_keyboard( - settings: Settings, - i18n: Optional[JsonI18n], - language: str, -) -> Optional[InlineKeyboardMarkup]: - app_url = str(getattr(settings, "SUBSCRIPTION_MINI_APP_URL", "") or "").strip() - if not app_url: - return None - text = _translate( - i18n, - language, - "telegram_notifications_open_app_button", - "Open app", - ) - return InlineKeyboardMarkup( - inline_keyboard=[[InlineKeyboardButton(text=text, web_app=WebAppInfo(url=app_url))]] - ) - - async def probe_telegram_notifications( *, session: AsyncSession, @@ -187,20 +155,8 @@ async def probe_telegram_notifications( "start_link": telegram_notifications_start_link(bot_username), } - language = str(getattr(user, "language_code", "") or settings.DEFAULT_LANGUAGE) - text = _translate( - i18n, - language, - "telegram_notifications_enabled_message", - "Telegram notifications are enabled.", - ) try: - await bot.send_message( - int(telegram_id), - text, - reply_markup=_probe_keyboard(settings, i18n, language), - disable_web_page_preview=True, - ) + await bot.get_chat(int(telegram_id)) except Exception as exc: status = telegram_notification_status_from_error(exc) if status: @@ -211,7 +167,7 @@ async def probe_telegram_notifications( "start_link": telegram_notifications_start_link(bot_username), } logger.warning( - "Telegram notification probe failed for user %s / telegram %s: %s", + "Telegram notification chat probe failed for user %s / telegram %s: %s", user.user_id, telegram_id, exc, diff --git a/tests/test_telegram_notifications.py b/tests/test_telegram_notifications.py index 0ca675e..a1ab52b 100644 --- a/tests/test_telegram_notifications.py +++ b/tests/test_telegram_notifications.py @@ -1,6 +1,12 @@ +import asyncio from types import SimpleNamespace +from unittest.mock import AsyncMock -from bot.services.telegram_notifications import telegram_notifications_need_prompt +from bot.services import telegram_notifications as module +from bot.services.telegram_notifications import ( + TELEGRAM_NOTIFICATIONS_ENABLED, + telegram_notifications_need_prompt, +) def _user(status: str): @@ -22,3 +28,39 @@ def test_telegram_notifications_prompt_requires_linked_telegram(): ) assert telegram_notifications_need_prompt(user) is False + + +def test_probe_telegram_notifications_uses_silent_chat_check(monkeypatch): + user = SimpleNamespace( + user_id=42, + telegram_id=123, + telegram_notifications_status="unknown", + ) + bot = SimpleNamespace( + get_chat=AsyncMock(return_value=SimpleNamespace(id=123)), + send_message=AsyncMock(), + ) + recorded = [] + + async def fake_mark_status(session, user_id, status, *, telegram_id=None, checked_at=None): + recorded.append((session, user_id, status, telegram_id, checked_at)) + return user + + monkeypatch.setattr(module, "mark_telegram_notifications_status", fake_mark_status) + + result = asyncio.run( + module.probe_telegram_notifications( + session="session", + bot=bot, + settings=SimpleNamespace(DEFAULT_LANGUAGE="ru"), + i18n=None, + user=user, + bot_username="preview_bot", + ) + ) + + bot.get_chat.assert_awaited_once_with(123) + bot.send_message.assert_not_called() + assert result["ok"] is True + assert result["status"] == TELEGRAM_NOTIFICATIONS_ENABLED + assert recorded == [("session", 42, TELEGRAM_NOTIFICATIONS_ENABLED, 123, None)]