Fix escalation step "Notify if num alerts in time window" (#2965)

# What this PR does
Fix escalation step "Notify if num alerts in time window" when
escalation policy was deleted during the escalation

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

## 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-09-05 12:32:59 +02:00 committed by GitHub
parent 395977d934
commit 6ff61ad172
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 42 additions and 2 deletions

View file

@ -411,11 +411,11 @@ class EscalationPolicySnapshot:
last_alert = alert_group.alerts.last()
time_delta = datetime.timedelta(minutes=self.escalation_policy.num_minutes_in_window)
time_delta = datetime.timedelta(minutes=self.num_minutes_in_window)
num_alerts_in_window = alert_group.alerts.filter(created_at__gte=last_alert.created_at - time_delta).count()
# pause escalation if there are not enough alerts in time window
if num_alerts_in_window <= self.escalation_policy.num_alerts_in_window:
if num_alerts_in_window <= self.num_alerts_in_window:
self.pause_escalation = True
return self._get_result_tuple(pause_escalation=True)
return None

View file

@ -408,6 +408,46 @@ def test_escalation_step_notify_if_num_alerts_in_window(
assert not mocked_execute_tasks.called
@pytest.mark.django_db
def test_escalation_step_notify_if_num_alerts_in_window_deleted_escalation_policy(
escalation_step_test_setup, make_escalation_policy, make_alert
):
_, _, _, channel_filter, alert_group, reason = escalation_step_test_setup
make_alert(alert_group=alert_group, raw_request_data={})
notify_if_2_alerts_per_1_minute = make_escalation_policy(
escalation_chain=channel_filter.escalation_chain,
escalation_policy_step=EscalationPolicy.STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
num_alerts_in_window=2,
num_minutes_in_window=1,
)
escalation_policy_snapshot = get_escalation_policy_snapshot_from_model(notify_if_2_alerts_per_1_minute)
notify_if_2_alerts_per_1_minute.delete()
with pytest.raises(EscalationPolicy.DoesNotExist):
notify_if_2_alerts_per_1_minute.refresh_from_db()
assert not alert_group.log_records.filter(
type=AlertGroupLogRecord.TYPE_ESCALATION_TRIGGERED,
escalation_policy_step=EscalationPolicy.STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
).exists()
result = escalation_policy_snapshot.execute(alert_group, reason)
expected_result = EscalationPolicySnapshot.StepExecutionResultData(
eta=None,
stop_escalation=False,
pause_escalation=True,
start_from_beginning=False,
)
assert result == expected_result
assert alert_group.log_records.filter(
type=AlertGroupLogRecord.TYPE_ESCALATION_TRIGGERED,
escalation_policy_step=EscalationPolicy.STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
).exists()
@patch("apps.alerts.escalation_snapshot.snapshot_classes.EscalationPolicySnapshot._execute_tasks", return_value=None)
@pytest.mark.django_db
def test_escalation_step_trigger_custom_button(