centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oncall-engine/engine/apps/alerts/tests/test_notify_user.py
Joey Orlando e115617528
chore: drop usage of SlackMessage.organization + drop orphaned SlackMessages (#5330) 
# What this PR does

- Stops writing `SlackMessage.organization` + removes references to this
field. [As we
discussed](https://raintank-corp.slack.com/archives/C083TU81TCH/p1733315887463279?thread_ts=1733311105.095309&cid=C083TU81TCH),
we do not need this field on this model/table,
`SlackMessage._slack_team_identity` is sufficient (`organization` will
be dropped in a separate PR)
- Adds a data migration script which:
- drops orphaned `SlackMessage` records; ie. ones which, even after the
[`engine/apps/slack/migrations/0007_migrate_slackmessage_channel_id.py`](https://github.com/grafana/oncall/blob/dev/engine/apps/slack/migrations/0007_migrate_slackmessage_channel_id.py)
migration, still don't have a `SlackMessage.channel` id filled in (we
discussed + agreed on dropping these records
[here](https://raintank-corp.slack.com/archives/C083TU81TCH/p1733329914516859?thread_ts=1733311105.095309&cid=C083TU81TCH))
- fills in empty `SlackMessage.slack_team_identity` values (from
`slack_message.channel.slack_team_identity`)

### Other notes

On the `organization` topic.

We store records in `SlackMessage` for two purposes (AFAICT), and in
both cases, we have references back to the `organization`:
- alert groups - `slack_message.alert_group.channel.organization`
- shift swap requests - `shift_swap_request.schedule.organization`

## 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] Added the relevant release notes label (see labels prefixed w/
`release:`). These labels dictate how your PR will
    show up in the autogenerated release notes.
2024-12-06 11:43:40 -05:00

707 lines
29 KiB
Python
Raw Permalink Blame History

from unittest.mock import patch
import pytest
from django.utils import timezone
from telegram.error import RetryAfter
from apps.alerts.models import AlertGroup
from apps.alerts.paging import direct_paging
from apps.alerts.tasks.notify_user import notify_user_task, perform_notification, send_bundled_notification
from apps.api.permissions import LegacyAccessControlRole
from apps.base.models.user_notification_policy import UserNotificationPolicy
from apps.base.models.user_notification_policy_log_record import UserNotificationPolicyLogRecord
from apps.slack.models import SlackMessage
from apps.telegram.models import TelegramToUserConnector
NOTIFICATION_UNAUTHORIZED_MSG = "notification is not allowed for user"
@pytest.mark.django_db
def test_custom_backend_call(
make_organization,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
):
organization = make_organization()
user_1 = make_user(organization=organization)
user_notification_policy = make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user_1,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
with patch("apps.base.tests.messaging_backend.TestOnlyBackend.notify_user") as mock_notify_user:
perform_notification(log_record.pk, False)
mock_notify_user.assert_called_once_with(user_1, alert_group, user_notification_policy)
@pytest.mark.django_db
def test_custom_backend_error(
make_organization,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
):
organization = make_organization()
user_1 = make_user(organization=organization)
user_notification_policy = make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user_1,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
with patch("apps.alerts.tasks.notify_user.get_messaging_backend_from_id") as mock_get_backend:
mock_get_backend.return_value = None
perform_notification(log_record.pk, False)
error_log_record = UserNotificationPolicyLogRecord.objects.last()
assert error_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert error_log_record.reason == "Messaging backend not available"
assert (
error_log_record.notification_error_code
== UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_MESSAGING_BACKEND_ERROR
)
@pytest.mark.django_db
@pytest.mark.parametrize(
"author_set,notification_policy_set",
[
(False, True),
(True, False),
],
)
def test_notify_user_missing_data_errors(
make_organization,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
author_set,
notification_policy_set,
):
organization = make_organization()
user_1 = make_user(organization=organization)
user_notification_policy = make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user_1 if author_set else None,
alert_group=alert_group,
notification_policy=user_notification_policy if notification_policy_set else None,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
with patch("apps.alerts.tasks.notify_user.get_messaging_backend_from_id") as mock_get_backend:
mock_get_backend.return_value = None
perform_notification(log_record.pk, False)
error_log_record = UserNotificationPolicyLogRecord.objects.last()
assert error_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert error_log_record.reason == "Expected data is missing"
assert error_log_record.notification_error_code is None
@pytest.mark.django_db
def test_notify_user_perform_notification_error_if_viewer(
make_organization,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
):
organization = make_organization()
user_1 = make_user(
organization=organization, role=LegacyAccessControlRole.VIEWER, _verified_phone_number="1234567890"
)
user_notification_policy = make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user_1,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
perform_notification(log_record.pk, False)
error_log_record = UserNotificationPolicyLogRecord.objects.last()
assert error_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert error_log_record.reason == NOTIFICATION_UNAUTHORIZED_MSG
assert error_log_record.notification_error_code == UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_FORBIDDEN
@pytest.mark.django_db
def test_notify_user_error_if_viewer(
make_organization,
make_user,
make_alert_receive_channel,
make_alert_group,
):
organization = make_organization()
user_1 = make_user(
organization=organization, role=LegacyAccessControlRole.VIEWER, _verified_phone_number="1234567890"
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
notify_user_task(user_1.pk, alert_group.pk)
error_log_record = UserNotificationPolicyLogRecord.objects.last()
assert error_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert error_log_record.reason == NOTIFICATION_UNAUTHORIZED_MSG
assert error_log_record.notification_error_code == UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_FORBIDDEN
@pytest.mark.django_db
def test_notify_user_perform_notification_skip_if_resolved(
make_organization,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
):
organization = make_organization()
user_1 = make_user(organization=organization, _verified_phone_number="1234567890")
user_notification_policy = make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel, resolved=True)
log_record = make_user_notification_policy_log_record(
author=user_1,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
perform_notification(log_record.pk, False)
error_log_record = UserNotificationPolicyLogRecord.objects.last()
assert error_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert error_log_record.reason == "Skipped notification because alert group is resolved"
@pytest.mark.django_db
@pytest.mark.parametrize(
"reason_to_skip_escalation,error_code",
[
(AlertGroup.RATE_LIMITED, UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_IN_SLACK_RATELIMIT),
(AlertGroup.CHANNEL_ARCHIVED, UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_IN_SLACK_CHANNEL_IS_ARCHIVED),
(AlertGroup.ACCOUNT_INACTIVE, UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_IN_SLACK_TOKEN_ERROR),
(AlertGroup.RESTRICTED_ACTION, UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_IN_SLACK),
(AlertGroup.NO_REASON, None),
],
)
def test_perform_notification_reason_to_skip_escalation_in_slack(
reason_to_skip_escalation,
error_code,
make_organization_with_slack_team_identity,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
make_slack_channel,
make_slack_message,
):
organization, slack_team_identity = make_organization_with_slack_team_identity()
user = make_user(organization=organization)
user_notification_policy = make_user_notification_policy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(
alert_receive_channel=alert_receive_channel,
reason_to_skip_escalation=reason_to_skip_escalation,
)
log_record = make_user_notification_policy_log_record(
author=user,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
if not error_code:
slack_channel = make_slack_channel(slack_team_identity=slack_team_identity)
make_slack_message(slack_channel, alert_group=alert_group)
with patch.object(SlackMessage, "send_slack_notification") as mocked_send_slack_notification:
perform_notification(log_record.pk, False)
last_log_record = UserNotificationPolicyLogRecord.objects.last()
if error_code:
log_reason = f"Skipped escalation in Slack, reason: '{alert_group.get_reason_to_skip_escalation_display()}'"
mocked_send_slack_notification.assert_not_called()
assert last_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert last_log_record.reason == log_reason
assert last_log_record.notification_error_code == error_code
else:
mocked_send_slack_notification.assert_called()
assert last_log_record.type != UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
@pytest.mark.django_db
def test_perform_notification_slack_prevent_posting(
make_organization_with_slack_team_identity,
make_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
make_slack_channel,
make_slack_message,
):
organization, slack_team_identity = make_organization_with_slack_team_identity()
user = make_user(organization=organization)
user_notification_policy = make_user_notification_policy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
slack_prevent_posting=True,
)
slack_channel = make_slack_channel(slack_team_identity=slack_team_identity)
make_slack_message(slack_channel, alert_group=alert_group)
with patch.object(SlackMessage, "send_slack_notification") as mocked_send_slack_notification:
perform_notification(log_record.pk, False)
mocked_send_slack_notification.assert_not_called()
last_log_record = UserNotificationPolicyLogRecord.objects.last()
assert last_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_SUCCESS
assert last_log_record.reason == "Prevented from posting in Slack"
assert (
last_log_record.notification_error_code
== UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_POSTING_TO_SLACK_IS_DISABLED
)
@pytest.mark.django_db
def test_perform_notification_missing_user_notification_policy_log_record(caplog):
invalid_pk = 12345
perform_notification(invalid_pk, False)
assert (
f"perform_notification: log_record {invalid_pk} doesn't exist. Skipping remainder of task. "
"The alert group associated with this log record may have been deleted."
) in caplog.text
assert f"perform_notification: found record for {invalid_pk}" not in caplog.text
@pytest.mark.django_db
def test_perform_notification_telegram_retryafter_error(
make_organization_and_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
):
organization, user = make_organization_and_user()
user_notification_policy = make_user_notification_policy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.TELEGRAM,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user,
alert_group=alert_group,
notification_policy=user_notification_policy,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
countdown = 15
exc = RetryAfter(countdown)
with patch.object(TelegramToUserConnector, "notify_user", side_effect=exc) as mock_notify_user:
with patch.object(perform_notification, "apply_async") as mock_apply_async:
perform_notification(log_record.pk, False)
mock_notify_user.assert_called_once_with(user, alert_group, user_notification_policy)
# task is rescheduled using the countdown value from the exception
mock_apply_async.assert_called_once_with((log_record.pk, False), countdown=countdown)
assert alert_group.personal_log_records.last() == log_record
# but if the log was too old, skip and create a failed log record
log_record.created_at = timezone.now() - timezone.timedelta(minutes=90)
log_record.save()
with patch.object(TelegramToUserConnector, "notify_user", side_effect=exc) as mock_notify_user:
with patch.object(perform_notification, "apply_async") as mock_apply_async:
perform_notification(log_record.pk, False)
mock_notify_user.assert_called_once_with(user, alert_group, user_notification_policy)
assert not mock_apply_async.called
last_log_record = UserNotificationPolicyLogRecord.objects.last()
assert last_log_record.type == UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED
assert last_log_record.reason == "Telegram rate limit exceeded"
assert (
last_log_record.notification_error_code
== UserNotificationPolicyLogRecord.ERROR_NOTIFICATION_IN_TELEGRAM_RATELIMIT
)
@patch("apps.base.models.UserNotificationPolicy.get_default_fallback_policy")
@patch("apps.base.tests.messaging_backend.TestOnlyBackend.notify_user")
@pytest.mark.django_db
def test_perform_notification_use_default_notification_policy_fallback(
mock_notify_user,
mock_get_default_fallback_policy,
make_organization,
make_user,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
):
organization = make_organization()
user = make_user(organization=organization)
fallback_notification_policy = UserNotificationPolicy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
important=False,
order=0,
)
mock_get_default_fallback_policy.return_value = fallback_notification_policy
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
log_record = make_user_notification_policy_log_record(
author=user,
alert_group=alert_group,
notification_policy=None,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
)
perform_notification(log_record.pk, True)
mock_notify_user.assert_called_once_with(user, alert_group, fallback_notification_policy)
@pytest.mark.django_db
def test_notify_user_task_notification_bundle_is_enabled(
make_organization_and_user,
make_user_for_organization,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
settings,
):
settings.FEATURE_NOTIFICATION_BUNDLE_ENABLED = True
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS, # channel is in NOTIFICATION_CHANNELS_TO_BUNDLE
)
make_user_notification_policy(
user=user_1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS,
important=True,
)
make_user_notification_policy(
user=user_2,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK, # channel is not in NOTIFICATION_CHANNELS_TO_BUNDLE
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group_1 = make_alert_group(alert_receive_channel=alert_receive_channel)
alert_group_2 = make_alert_group(alert_receive_channel=alert_receive_channel)
assert not user_1.notification_bundles.exists()
# send 1st notification to user_1, check notification_bundle was created
# without scheduling send_bundled_notification task
notify_user_task(user_1.id, alert_group_1.id)
assert user_1.notification_bundles.count() == 1
notification_bundle = user_1.notification_bundles.first()
assert notification_bundle.notification_task_id is None
assert not notification_bundle.notifications.exists()
# send 2nd notification to user_1, check bundled notification was attached to notification_bundle
# and send_bundled_notification was scheduled
notify_user_task(user_1.id, alert_group_2.id)
notification_bundle.refresh_from_db()
assert notification_bundle.notifications.count() == 1
assert notification_bundle.notification_task_id is not None
# send important notification to user_1, check new notification_bundle was created
notify_user_task(user_1.id, alert_group_1.id, important=True)
assert user_1.notification_bundles.count() == 2
important_notification_bundle = user_1.notification_bundles.get(important=True)
assert important_notification_bundle.notification_task_id is None
assert not important_notification_bundle.notifications.exists()
# send notification to user_2 (notification channel is not in NOTIFICATION_CHANNELS_TO_BUNDLE),
# check notification_bundle was not created
notify_user_task(user_2.id, alert_group_1.id)
assert not user_2.notification_bundles.exists()
@pytest.mark.django_db
def test_notify_user_task_notification_bundle_is_not_enabled(
make_organization_and_user,
make_user_notification_policy,
make_alert_receive_channel,
make_alert_group,
settings,
):
settings.FEATURE_NOTIFICATION_BUNDLE_ENABLED = False
organization, user = make_organization_and_user()
make_user_notification_policy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS, # channel is in NOTIFICATION_CHANNELS_TO_BUNDLE
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
# send notification, check notification_bundle was not created
notify_user_task(user.id, alert_group.id)
assert not user.notification_bundles.exists()
@pytest.mark.django_db
def test_send_bundle_notification(
make_organization_and_user,
make_user_notification_policy,
make_user_notification_bundle,
make_alert_receive_channel,
make_alert_group,
settings,
caplog,
):
settings.FEATURE_NOTIFICATION_BUNDLE_ENABLED = True
organization, user = make_organization_and_user()
notification_policy = make_user_notification_policy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS, # channel is in NOTIFICATION_CHANNELS_TO_BUNDLE
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group_1 = make_alert_group(alert_receive_channel=alert_receive_channel)
alert_group_2 = make_alert_group(alert_receive_channel=alert_receive_channel)
alert_group_3 = make_alert_group(alert_receive_channel=alert_receive_channel)
task_id = "test_task_id"
notification_bundle = make_user_notification_bundle(
user, UserNotificationPolicy.NotificationChannel.SMS, notification_task_id=task_id, eta=timezone.now()
)
notification_bundle.append_notification(alert_group_1, notification_policy)
notification_bundle.append_notification(alert_group_2, notification_policy)
notification_bundle.append_notification(alert_group_3, notification_policy)
assert notification_bundle.notifications.filter(bundle_uuid__isnull=True).count() == 3
alert_group_3.resolve()
# send notification for 2 active alert groups
send_bundled_notification.apply((notification_bundle.id,), task_id=task_id)
assert f"alert_group {alert_group_3.id} is not active, skip notification" in caplog.text
assert "perform bundled notification for alert groups with ids:" in caplog.text
# check bundle_uuid was set, notification for resolved alert group was deleted
assert notification_bundle.notifications.filter(bundle_uuid__isnull=True).count() == 0
assert notification_bundle.notifications.all().count() == 2
assert not notification_bundle.notifications.filter(alert_group=alert_group_3).exists()
# send notification for 1 active alert group
notification_bundle.notifications.update(bundle_uuid=None)
# since we're calling send_bundled_notification several times within this test, we need to reset task_id
# because it gets set to None after the first call
notification_bundle.notification_task_id = task_id
notification_bundle.save()
alert_group_2.resolve()
send_bundled_notification.apply((notification_bundle.id,), task_id=task_id)
assert f"alert_group {alert_group_2.id} is not active, skip notification" in caplog.text
assert (
f"there is only one alert group in bundled notification, perform regular notification. "
f"alert_group {alert_group_1.id}"
) in caplog.text
# check bundle_uuid was set
assert notification_bundle.notifications.filter(bundle_uuid__isnull=True).count() == 0
assert notification_bundle.notifications.all().count() == 1
# cleanup notifications
notification_bundle.notifications.all().delete()
# send notification for 0 active alert group
notification_bundle.append_notification(alert_group_1, notification_policy)
# since we're calling send_bundled_notification several times within this test, we need to reset task_id
# because it gets set to None after the first call
notification_bundle.notification_task_id = task_id
notification_bundle.save()
alert_group_1.resolve()
send_bundled_notification.apply((notification_bundle.id,), task_id=task_id)
assert f"alert_group {alert_group_1.id} is not active, skip notification" in caplog.text
assert f"no alert groups to notify about or notification is not allowed for user {user.id}" in caplog.text
# check all notifications were deleted
assert notification_bundle.notifications.all().count() == 0
@pytest.mark.django_db
def test_send_bundle_notification_task_id_mismatch(
make_organization_and_user,
make_user_notification_bundle,
settings,
caplog,
):
settings.FEATURE_NOTIFICATION_BUNDLE_ENABLED = True
organization, user = make_organization_and_user()
notification_bundle = make_user_notification_bundle(
user, UserNotificationPolicy.NotificationChannel.SMS, notification_task_id="test_task_id", eta=timezone.now()
)
send_bundled_notification(notification_bundle.id)
assert (
f"send_bundled_notification: notification_task_id mismatch. "
f"Duplication or non-active notification triggered. "
f"Active: {notification_bundle.notification_task_id}"
) in caplog.text
@pytest.mark.django_db
def test_notify_user_task_notification_bundle_eta_is_outdated(
make_organization_and_user,
make_user_for_organization,
make_user_notification_policy,
make_user_notification_bundle,
make_alert_receive_channel,
make_alert_group,
settings,
):
settings.FEATURE_NOTIFICATION_BUNDLE_ENABLED = True
organization, user = make_organization_and_user()
notification_policy = make_user_notification_policy(
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SMS, # channel is in NOTIFICATION_CHANNELS_TO_BUNDLE
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group_1 = make_alert_group(alert_receive_channel=alert_receive_channel)
alert_group_2 = make_alert_group(alert_receive_channel=alert_receive_channel)
now = timezone.now()
outdated_eta = now - timezone.timedelta(minutes=5)
test_task_id = "test_task_id"
notification_bundle = make_user_notification_bundle(
user,
UserNotificationPolicy.NotificationChannel.SMS,
eta=outdated_eta,
notification_task_id=test_task_id,
last_notified_at=now,
)
notification_bundle.append_notification(alert_group_1, notification_policy)
assert not notification_bundle.eta_is_valid()
assert notification_bundle.notifications.count() == 1
# call notify_user_task and check that new notification task for notification_bundle was scheduled
notify_user_task(user.id, alert_group_2.id)
notification_bundle.refresh_from_db()
assert notification_bundle.eta_is_valid()
assert notification_bundle.notification_task_id != test_task_id
assert notification_bundle.last_notified_at == now
assert notification_bundle.notifications.count() == 2
@patch.object(perform_notification, "apply_async")
@pytest.mark.django_db
def test_notify_user_task_direct_paging_acknowledged(
mock_perform_notification_apply_async,
make_organization,
make_user,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
make_user_notification_policy,
django_capture_on_commit_callbacks,
):
organization = make_organization()
from_user = make_user(organization=organization)
user1 = make_user(organization=organization)
user2 = make_user(organization=organization)
make_user_notification_policy(
user=user1,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
)
make_user_notification_policy(
user=user2,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
)
alert_receive_channel = make_alert_receive_channel(organization=organization)
alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
direct_paging(organization, from_user, "Test", alert_group=alert_group, users=[(user1, False), (user2, False)])
alert_group.acknowledge_by_user_or_backsync(user1)
# no notification should be sent for user1 because they have acknowledged the alert group
with django_capture_on_commit_callbacks(execute=True):
notify_user_task(user1.pk, alert_group.pk, notify_even_acknowledged=True)
mock_perform_notification_apply_async.assert_not_called()
# user2 should receive a notification because they have not acknowledged the alert group
with django_capture_on_commit_callbacks(execute=True):
notify_user_task(user2.pk, alert_group.pk, notify_even_acknowledged=True)
mock_perform_notification_apply_async.assert_called_once()
Reference in a new issue View git blame Copy permalink