Handle Unauthorized exception on posting alert group to telegram (#3540)

## Which issue(s) this PR fixes
https://github.com/grafana/oncall-private/issues/2377

## Checklist

- [x] Unit, integration, and e2e (if applicable) tests updated
- [x] Documentation added (or `pr:no public docs` PR label added if not
required)
- [x] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not
required)
This commit is contained in:
Yulya Artyukhina 2023-12-11 17:32:15 +01:00 committed by GitHub
parent 1a22db0c29
commit 58de3f8458
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 57 additions and 1 deletions

View file

@ -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

View file

@ -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