diff --git a/engine/apps/telegram/models/connectors/personal.py b/engine/apps/telegram/models/connectors/personal.py index 8dfdc6f7..cad4fe64 100644 --- a/engine/apps/telegram/models/connectors/personal.py +++ b/engine/apps/telegram/models/connectors/personal.py @@ -167,6 +167,13 @@ class TelegramToUserConnector(models.Model): notification_policy, UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_TELEGRAM_TOKEN_ERROR, ) + elif e.message == "Forbidden: user is deactivated": + TelegramToUserConnector.create_telegram_notification_error( + alert_group, + self.user, + notification_policy, + UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_TELEGRAM_USER_IS_DEACTIVATED, + ) else: raise e diff --git a/engine/apps/telegram/tests/test_personal_connector.py b/engine/apps/telegram/tests/test_personal_connector.py index b1ea07a0..add403cb 100644 --- a/engine/apps/telegram/tests/test_personal_connector.py +++ b/engine/apps/telegram/tests/test_personal_connector.py @@ -3,7 +3,7 @@ from unittest.mock import patch import pytest from telegram import error -from apps.base.models import UserNotificationPolicy +from apps.base.models import UserNotificationPolicy, UserNotificationPolicyLogRecord from apps.telegram.client import TelegramClient from apps.telegram.models import TelegramMessage @@ -47,3 +47,52 @@ def test_personal_connector_replied_message_not_found( text="One more notification about this 👆", reply_to_message_id=telegram_message.message_id, ) + + +@pytest.mark.parametrize( + "side_effect,notification_error_code", + [ + ( + error.Unauthorized("Forbidden: bot was blocked by the user"), + UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_TELEGRAM_BOT_IS_DELETED, + ), + (error.Unauthorized("Invalid token"), UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_TELEGRAM_TOKEN_ERROR), + ( + error.Unauthorized("Forbidden: user is deactivated"), + UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_TELEGRAM_USER_IS_DEACTIVATED, + ), + ], +) +@pytest.mark.django_db +def test_personal_connector_send_link_to_channel_message_handle_exceptions( + side_effect, + notification_error_code, + make_organization_and_user, + make_telegram_user_connector, + make_user_notification_policy, + make_alert_receive_channel, + make_alert_group, +): + # set up a user with Telegram account connected + organization, user = make_organization_and_user() + user_connector = make_telegram_user_connector(user) + notification_policy = make_user_notification_policy( + user, + UserNotificationPolicy.Step.NOTIFY, + notify_by=UserNotificationPolicy.NotificationChannel.TELEGRAM, + important=False, + ) + + # create an alert group with an existing Telegram message in user's DM + alert_receive_channel = make_alert_receive_channel(organization) + alert_group = make_alert_group(alert_receive_channel) + + assert not user.personal_log_records.exists() + + with patch.object(TelegramClient, "send_message", side_effect=side_effect) as mock_send_message: + user_connector.send_link_to_channel_message(alert_group, notification_policy) + + mock_send_message.assert_called_once() + log_records = user.personal_log_records.filter(alert_group=alert_group) + assert log_records.count() == 1 + assert log_records.first().notification_error_code == notification_error_code