Add "used in escalation" filter for schedules internal API (#1425)

# What this PR does
Adds a `used` filter on schedules endpoint for internal API.

Usage:
- `?used=true` returns schedules that are referenced by at least one
escalation policy
- `?used=false` returns schedules that are NOT referenced
- `?used=null` or not providing the query param at all will return all
schedules
## Which issue(s) this PR fixes
https://github.com/grafana/oncall/issues/1423

## Checklist

- [x] Tests updated
This commit is contained in:
Vadim Stepanov 2023-03-01 10:09:07 +00:00 committed by GitHub
parent 173a02c7d3
commit 4c31ede558
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 43 additions and 0 deletions

View file

@ -329,6 +329,45 @@ def test_get_list_schedules_by_type(
assert response.json() == expected_payload
@pytest.mark.django_db
@pytest.mark.parametrize(
"query_param, expected_schedule_names",
[
("?used=true", ["test_web_schedule"]),
("?used=false", ["test_calendar_schedule", "test_ical_schedule"]),
("?used=null", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]),
("", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]),
],
)
def test_get_list_schedules_by_used(
schedule_internal_api_setup,
make_escalation_chain,
make_escalation_policy,
make_user_auth_headers,
query_param,
expected_schedule_names,
):
user, token, calendar_schedule, ical_schedule, web_schedule, slack_channel = schedule_internal_api_setup
client = APIClient()
# setup escalation chain linked to web schedule
escalation_chain = make_escalation_chain(user.organization)
make_escalation_policy(
escalation_chain=escalation_chain,
escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE,
notify_schedule=web_schedule,
)
url = reverse("api-internal:schedule-list") + query_param
response = client.get(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
assert response.json()["count"] == len(expected_schedule_names)
schedule_names = [schedule["name"] for schedule in response.json()["results"]]
assert set(schedule_names) == set(expected_schedule_names)
@pytest.mark.django_db
def test_get_detail_calendar_schedule(schedule_internal_api_setup, make_user_auth_headers):
user, token, calendar_schedule, _, _, _ = schedule_internal_api_setup

View file

@ -7,6 +7,7 @@ from django.utils.functional import cached_property
from rest_framework import mixins, status
from rest_framework.decorators import action
from rest_framework.exceptions import NotFound
from rest_framework.fields import BooleanField
from rest_framework.filters import SearchFilter
from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import IsAuthenticated
@ -154,6 +155,7 @@ class ScheduleView(
def get_queryset(self):
is_short_request = self.request.query_params.get("short", "false") == "true"
filter_by_type = self.request.query_params.get("type")
used = BooleanField(allow_null=True).to_internal_value(data=self.request.query_params.get("used"))
organization = self.request.auth.organization
queryset = OnCallSchedule.objects.filter(organization=organization, team=self.request.user.current_team).defer(
# avoid requesting large text fields which are not used when listing schedules
@ -165,6 +167,8 @@ class ScheduleView(
queryset = self.serializer_class.setup_eager_loading(queryset)
if filter_by_type is not None and filter_by_type in SCHEDULE_TYPE_TO_CLASS:
queryset = queryset.filter().instance_of(SCHEDULE_TYPE_TO_CLASS[filter_by_type])
if used is not None:
queryset = queryset.filter(escalation_policies__isnull=not used).distinct()
return queryset
def perform_create(self, serializer):