From 4c31ede5584ea54dce9e49bb79e22038c405494f Mon Sep 17 00:00:00 2001
From: Vadim Stepanov <vadimkerr@gmail.com>
Date: Wed, 1 Mar 2023 10:09:07 +0000
Subject: [PATCH] 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
---
 engine/apps/api/tests/test_schedules.py | 39 +++++++++++++++++++++++++
 engine/apps/api/views/schedule.py       |  4 +++
 2 files changed, 43 insertions(+)

diff --git a/engine/apps/api/tests/test_schedules.py b/engine/apps/api/tests/test_schedules.py
index 79f6759a..5f552c6a 100644
--- a/engine/apps/api/tests/test_schedules.py
+++ b/engine/apps/api/tests/test_schedules.py
@@ -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
diff --git a/engine/apps/api/views/schedule.py b/engine/apps/api/views/schedule.py
index 44fa0024..ad875ef3 100644
--- a/engine/apps/api/views/schedule.py
+++ b/engine/apps/api/views/schedule.py
@@ -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):