centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oncall-engine/engine/apps/api/tests/test_user_notification_policy.py
Vadim Stepanov d8e1a1dfae
Allow custom wait delays (#4422) 
# What this PR does

Allows custom wait durations for:
* `Wait` escalation policy
* `>X alerts per Y minutes` escalation policy
* `Wait` user notification policy

## Which issue(s) this PR closes

Related to https://github.com/grafana/oncall/issues/2464

## 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.

---------

Co-authored-by: Rares Mardare <rares.mardare@grafana.com>
2024-05-31 13:18:59 +00:00

498 lines
18 KiB
Python
Raw Blame History

import json
from datetime import timedelta
from unittest.mock import patch
import pytest
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from rest_framework.test import APIClient
from apps.api.permissions import LegacyAccessControlRole
from apps.base.models import UserNotificationPolicy
from apps.base.tests.messaging_backend import TestOnlyBackend
DEFAULT_NOTIFICATION_CHANNEL = UserNotificationPolicy.NotificationChannel.SLACK
@pytest.fixture()
def user_notification_policy_internal_api_setup(
make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_notification_policy
):
organization, admin, token = make_organization_and_user_with_plugin_token()
user = make_user_for_organization(organization, role=LegacyAccessControlRole.EDITOR)
wait_notification_step = make_user_notification_policy(
admin, UserNotificationPolicy.Step.WAIT, wait_delay=timezone.timedelta(minutes=15), important=False
)
notify_notification_step = make_user_notification_policy(
admin,
UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
important=False,
)
important_notification_step = make_user_notification_policy(
admin,
UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
important=True,
)
second_user_step = make_user_notification_policy(
user,
UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
important=False,
)
steps = wait_notification_step, notify_notification_step, important_notification_step, second_user_step
users = admin, user
return token, steps, users
@pytest.mark.django_db
def test_create_notification_policy(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, _, users = user_notification_policy_internal_api_setup
admin, _ = users
client = APIClient()
url = reverse("api-internal:notification_policy-list")
data = {
"step": UserNotificationPolicy.Step.NOTIFY,
"notify_by": UserNotificationPolicy.NotificationChannel.SLACK,
"wait_delay": None,
"important": False,
"user": admin.public_primary_key,
}
response = client.post(url, data, format="json", **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_201_CREATED
@pytest.mark.django_db
@pytest.mark.parametrize("wait_delay", (timedelta(seconds=59), timedelta(hours=24, seconds=1)))
def test_create_notification_policy_wait_delay_invalid(
user_notification_policy_internal_api_setup, make_user_auth_headers, wait_delay
):
token, _, users = user_notification_policy_internal_api_setup
admin, _ = users
client = APIClient()
url = reverse("api-internal:notification_policy-list")
data = {
"step": UserNotificationPolicy.Step.WAIT,
"wait_delay": int(wait_delay.total_seconds()),
"important": False,
"user": admin.public_primary_key,
}
response = client.post(url, data, format="json", **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_admin_can_create_notification_policy_for_user(
user_notification_policy_internal_api_setup, make_user_auth_headers
):
token, _, users = user_notification_policy_internal_api_setup
admin, user = users
client = APIClient()
url = reverse("api-internal:notification_policy-list")
data = {
"step": UserNotificationPolicy.Step.NOTIFY,
"notify_by": UserNotificationPolicy.NotificationChannel.SLACK,
"wait_delay": None,
"important": False,
"user": user.public_primary_key,
}
response = client.post(url, data, format="json", **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_201_CREATED
@pytest.mark.django_db
def test_user_cant_create_notification_policy_for_user(
user_notification_policy_internal_api_setup,
make_user_auth_headers,
):
token, _, users = user_notification_policy_internal_api_setup
admin, user = users
client = APIClient()
url = reverse("api-internal:notification_policy-list")
data = {
"step": UserNotificationPolicy.Step.NOTIFY,
"notify_by": UserNotificationPolicy.NotificationChannel.SLACK,
"wait_delay": None,
"important": False,
"user": admin.public_primary_key,
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db
def test_create_notification_policy_order_is_ignored(
user_notification_policy_internal_api_setup,
make_user_auth_headers,
):
token, steps, users = user_notification_policy_internal_api_setup
wait_notification_step, _, _, _ = steps
admin, _ = users
client = APIClient()
url = reverse("api-internal:notification_policy-list")
data = {
"position": 2023,
"step": UserNotificationPolicy.Step.NOTIFY,
"notify_by": UserNotificationPolicy.NotificationChannel.SLACK,
"wait_delay": None,
"important": False,
"user": admin.public_primary_key,
}
response = client.post(url, data, format="json", **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_201_CREATED
assert response.data["order"] == 2
@pytest.mark.django_db
def test_move_to_position_position_error(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, steps, users = user_notification_policy_internal_api_setup
admin, _ = users
step = steps[0]
client = APIClient()
url = reverse("api-internal:notification_policy-move-to-position", kwargs={"pk": step.public_primary_key})
# position value only can be 0 or 1 for this test setup, because there are only 2 steps
response = client.put(f"{url}?position=2", content_type="application/json", **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_update_step(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, steps, users = user_notification_policy_internal_api_setup
admin, _ = users
_, notify_notification_step, _, _ = steps
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": notify_notification_step.public_primary_key})
response = client.patch(
url,
data=json.dumps({"notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL}),
content_type="application/json",
**make_user_auth_headers(admin, token),
)
assert response.status_code == status.HTTP_200_OK
assert response.data["notify_by"] == UserNotificationPolicy.NotificationChannel.PHONE_CALL
@pytest.mark.django_db
def test_admin_can_update_user_step(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, steps, users = user_notification_policy_internal_api_setup
admin, _ = users
_, _, _, second_user_step = steps
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": second_user_step.public_primary_key})
response = client.patch(
url,
data=json.dumps({"notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL}),
content_type="application/json",
**make_user_auth_headers(admin, token),
)
assert response.status_code == status.HTTP_200_OK
assert response.data["notify_by"] == UserNotificationPolicy.NotificationChannel.PHONE_CALL
@pytest.mark.django_db
def test_user_cant_update_admin_step(
user_notification_policy_internal_api_setup,
make_user_auth_headers,
):
token, steps, users = user_notification_policy_internal_api_setup
_, user = users
admin_step, _, _, _ = steps
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": admin_step.public_primary_key})
response = client.patch(
url,
data=json.dumps({"notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL}),
content_type="application/json",
**make_user_auth_headers(user, token),
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db
def test_admin_can_move_user_step(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, steps, users = user_notification_policy_internal_api_setup
admin, _ = users
_, _, _, second_user_step = steps
client = APIClient()
url = reverse(
"api-internal:notification_policy-move-to-position", kwargs={"pk": second_user_step.public_primary_key}
)
response = client.put(f"{url}?position=0", content_type="application/json", **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_200_OK
@pytest.mark.django_db
def test_user_cant_move_admin_step(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, steps, users = user_notification_policy_internal_api_setup
_, user = users
admin_step, _, _, _ = steps
client = APIClient()
url = reverse("api-internal:notification_policy-move-to-position", kwargs={"pk": admin_step.public_primary_key})
response = client.put(f"{url}?position=1", content_type="application/json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.django_db
def test_unable_to_change_importance(user_notification_policy_internal_api_setup, make_user_auth_headers):
token, steps, users = user_notification_policy_internal_api_setup
admin, _ = users
_, notify_notification_step, _, _ = steps
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": notify_notification_step.public_primary_key})
data = {
"important": True,
}
response = client.patch(
url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(admin, token)
)
notify_notification_step.refresh_from_db()
assert response.status_code == status.HTTP_200_OK
assert notify_notification_step.important != data["important"]
@pytest.mark.django_db
@pytest.mark.parametrize("wait_delay, expected_wait_delay", [(None, 300), (900, 900)])
def test_switch_step_type_from_notify_to_wait(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_user_notification_policy,
wait_delay,
expected_wait_delay,
):
organization, admin, token = make_organization_and_user_with_plugin_token()
notify_notification_step = make_user_notification_policy(
admin,
UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
important=False,
)
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": notify_notification_step.public_primary_key})
data = {
"id": notify_notification_step.public_primary_key,
"important": False,
"notify_by": None,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.WAIT,
"wait_delay": wait_delay,
}
expected_response = {
"id": notify_notification_step.public_primary_key,
"important": False,
"notify_by": None,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.WAIT,
"wait_delay": expected_wait_delay,
}
response = client.put(
url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(admin, token)
)
notify_notification_step.refresh_from_db()
assert response.status_code == status.HTTP_200_OK
assert response.json() == expected_response
@pytest.mark.django_db
@pytest.mark.parametrize(
"notification_channel, expected_notification_channel",
[
(None, DEFAULT_NOTIFICATION_CHANNEL),
(
UserNotificationPolicy.NotificationChannel.PHONE_CALL,
UserNotificationPolicy.NotificationChannel.PHONE_CALL,
),
],
)
def test_switch_step_type_wait_to_notify(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_user_notification_policy,
notification_channel,
expected_notification_channel,
):
organization, admin, token = make_organization_and_user_with_plugin_token()
wait_notification_step = make_user_notification_policy(
admin,
UserNotificationPolicy.Step.WAIT,
important=False,
)
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": wait_notification_step.public_primary_key})
data = {
"id": wait_notification_step.public_primary_key,
"important": False,
"notify_by": notification_channel,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.NOTIFY,
"wait_delay": None,
}
expected_response = {
"id": wait_notification_step.public_primary_key,
"important": False,
"notify_by": expected_notification_channel,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.NOTIFY,
"wait_delay": None,
}
response = client.put(
url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(admin, token)
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == expected_response
@pytest.mark.django_db
def test_switch_notification_channel(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_user_notification_policy,
):
organization, admin, token = make_organization_and_user_with_plugin_token()
notify_notification_step = make_user_notification_policy(
admin,
UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.SLACK,
important=False,
)
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": notify_notification_step.public_primary_key})
data = {
"id": notify_notification_step.public_primary_key,
"important": False,
"notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.NOTIFY,
"wait_delay": None,
}
expected_response = {
"id": notify_notification_step.public_primary_key,
"important": False,
"notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.NOTIFY,
"wait_delay": None,
}
response = client.put(
url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(admin, token)
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == expected_response
@pytest.mark.django_db
@pytest.mark.parametrize("from_wait_delay, to_wait_delay", [(None, 300), (timezone.timedelta(seconds=900), 900)])
def test_switch_wait_delay(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_user_notification_policy,
from_wait_delay,
to_wait_delay,
):
organization, admin, token = make_organization_and_user_with_plugin_token()
wait_notification_step = make_user_notification_policy(
admin, UserNotificationPolicy.Step.WAIT, wait_delay=from_wait_delay, important=False
)
client = APIClient()
url = reverse("api-internal:notification_policy-detail", kwargs={"pk": wait_notification_step.public_primary_key})
data = {
"id": wait_notification_step.public_primary_key,
"important": False,
"notify_by": DEFAULT_NOTIFICATION_CHANNEL,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.NOTIFY,
"wait_delay": to_wait_delay,
}
expected_response = {
"id": wait_notification_step.public_primary_key,
"important": False,
"notify_by": DEFAULT_NOTIFICATION_CHANNEL,
"order": 0,
"user": admin.public_primary_key,
"step": UserNotificationPolicy.Step.NOTIFY,
"wait_delay": to_wait_delay,
}
response = client.put(
url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(admin, token)
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == expected_response
@pytest.mark.django_db
def test_notification_policy_backends_enabled(
user_notification_policy_internal_api_setup, settings, make_user_auth_headers
):
token, _, users = user_notification_policy_internal_api_setup
admin, _ = users
client = APIClient()
url = reverse("api-internal:notification_policy-notify-by-options")
response = client.get(url, **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_200_OK
options = [opt["display_name"] for opt in response.json()]
assert "Test Only Backend" in options
@pytest.mark.django_db
def test_notification_policy_backends_disabled_for_organization(
user_notification_policy_internal_api_setup, settings, make_user_auth_headers
):
token, _, users = user_notification_policy_internal_api_setup
admin, _ = users
client = APIClient()
url = reverse("api-internal:notification_policy-notify-by-options")
with patch.object(TestOnlyBackend, "is_enabled_for_organization", return_value=False):
response = client.get(url, **make_user_auth_headers(admin, token))
assert response.status_code == status.HTTP_200_OK
options = [opt["display_name"] for opt in response.json()]
assert "Test Only Backend" not in options
Reference in a new issue View git blame Copy permalink