Add list shifts for swap request endpoint

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Add filter_shift_swaps endpoint to schedules API ([#2684](https://github.com/grafana/oncall/pull/2684))
+- Add shifts endpoint to shift swap API ([#2697](https://github.com/grafana/oncall/pull/2697/))
 
 ### Fixed
 

engine/apps/api/tests/test_shift_swaps.py

@@ -10,7 +10,7 @@
 from rest_framework.test import APIClient
 
 from apps.api.permissions import LegacyAccessControlRole
-from apps.schedules.models import OnCallScheduleWeb, ShiftSwapRequest
+from apps.schedules.models import CustomOnCallShift, OnCallScheduleWeb, ShiftSwapRequest
 from common.api_helpers.utils import serialize_datetime_as_utc_timestamp
 from common.insight_log import EntityEvent
 
@@ -466,6 +466,53 @@ def test_partial_update_time_related_fields(ssr_setup, make_user_auth_headers):
     assert response.json() == expected_response
 
 
+@pytest.mark.django_db
+def test_related_shifts(ssr_setup, make_on_call_shift, make_user_auth_headers):
+    ssr, beneficiary, token, _ = ssr_setup()
+
+    schedule = ssr.schedule
+    organization = schedule.organization
+    user = beneficiary
+
+    today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    start = today + timezone.timedelta(days=2)
+    duration = timezone.timedelta(hours=8)
+    data = {
+        "start": start,
+        "rotation_start": start,
+        "duration": duration,
+        "priority_level": 1,
+        "frequency": CustomOnCallShift.FREQUENCY_DAILY,
+        "schedule": schedule,
+    }
+    on_call_shift = make_on_call_shift(
+        organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
+    )
+    on_call_shift.add_rolling_users([[user]])
+
+    client = APIClient()
+    url = reverse("api-internal:shift_swap-shifts", kwargs={"pk": ssr.public_primary_key})
+    auth_headers = make_user_auth_headers(beneficiary, token)
+    response = client.get(url, **auth_headers)
+
+    assert response.status_code == status.HTTP_200_OK
+    response_json = response.json()
+    expected = [
+        # start, end, user, swap request ID
+        (
+            start.strftime("%Y-%m-%dT%H:%M:%SZ"),
+            (start + duration).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            user.public_primary_key,
+            ssr.public_primary_key,
+        ),
+    ]
+    returned_events = [
+        (e["start"], e["end"], e["users"][0]["pk"], e["users"][0]["swap_request"]["pk"])
+        for e in response_json["events"]
+    ]
+    assert returned_events == expected
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize(
     "role,expected_status",
@@ -714,3 +761,28 @@ def test_take_permissions(
 
     response = client.post(url, format="json", **make_user_auth_headers(benefactor, token))
     assert response.status_code == expected_status
+
+
+@patch("apps.api.views.shift_swap.ShiftSwapViewSet.shifts", return_value=mock_success_response)
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "role,expected_status",
+    [
+        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
+        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
+        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
+    ],
+)
+def test_list_shifts_permissions(
+    mock_endpoint_handler,
+    ssr_setup,
+    make_user_auth_headers,
+    role,
+    expected_status,
+):
+    ssr, beneficiary, token, _ = ssr_setup(beneficiary_role=role)
+    client = APIClient()
+    url = reverse("api-internal:shift_swap-shifts", kwargs={"pk": ssr.public_primary_key})
+
+    response = client.get(url, format="json", **make_user_auth_headers(beneficiary, token))
+    assert response.status_code == expected_status

engine/apps/api/views/shift_swap.py

@@ -36,6 +36,7 @@ class ShiftSwapViewSet(PublicPrimaryKeyMixin, ModelViewSet):
         "partial_update": [RBACPermission.Permissions.SCHEDULES_WRITE],
         "destroy": [RBACPermission.Permissions.SCHEDULES_WRITE],
         "take": [RBACPermission.Permissions.SCHEDULES_WRITE],
+        "shifts": [RBACPermission.Permissions.SCHEDULES_READ],
     }
 
     is_beneficiary = IsOwner(ownership_field="beneficiary")
@@ -87,6 +88,13 @@ def perform_update(self, serializer) -> None:
 
         update_shift_swap_request_message.apply_async((shift_swap_request.pk,))
 
+    @action(methods=["get"], detail=True)
+    def shifts(self, request, pk) -> Response:
+        shift_swap = self.get_object()
+        result = {"events": shift_swap.shifts()}
+
+        return Response(result, status=status.HTTP_200_OK)
+
     @action(methods=["post"], detail=True)
     def take(self, request, pk) -> Response:
         shift_swap = self.get_object()

engine/apps/schedules/models/shift_swap_request.py

@@ -165,6 +165,17 @@ def hard_delete(self):
         # make sure final schedule ical representation is updated
         refresh_ical_final_schedule.apply_async((self.schedule.pk,))
 
+    def shifts(self):
+        """Return shifts affected by this swap request."""
+        schedule = self.schedule.get_real_instance()
+        events = schedule.final_events(self.swap_start, self.swap_end)
+        related_shifts = [
+            e
+            for e in events
+            if self.public_primary_key in set(u["swap_request"]["pk"] for u in e["users"] if u.get("swap_request"))
+        ]
+        return related_shifts
+
     def take(self, benefactor: "User") -> None:
         if benefactor == self.beneficiary:
             raise exceptions.BeneficiaryCannotTakeOwnShiftSwapRequest()

engine/apps/schedules/tests/test_shift_swap_request.py

@@ -2,9 +2,10 @@
 from unittest.mock import patch
 
 import pytest
+from django.utils import timezone
 
 from apps.schedules import exceptions
-from apps.schedules.models import ShiftSwapRequest
+from apps.schedules.models import CustomOnCallShift, ShiftSwapRequest
 
 
 @pytest.mark.django_db
@@ -116,3 +117,37 @@ def test_take_own_ssr(shift_swap_request_setup) -> None:
     ssr, beneficiary, _ = shift_swap_request_setup()
     with pytest.raises(exceptions.BeneficiaryCannotTakeOwnShiftSwapRequest):
         ssr.take(beneficiary)
+
+
+@pytest.mark.django_db
+def test_related_shifts(shift_swap_request_setup, make_on_call_shift) -> None:
+    ssr, beneficiary, _ = shift_swap_request_setup()
+
+    schedule = ssr.schedule
+    organization = schedule.organization
+    user = beneficiary
+
+    today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    start = today + timezone.timedelta(days=2)
+    duration = timezone.timedelta(hours=8)
+    data = {
+        "start": start,
+        "rotation_start": start,
+        "duration": duration,
+        "priority_level": 1,
+        "frequency": CustomOnCallShift.FREQUENCY_DAILY,
+        "schedule": schedule,
+    }
+    on_call_shift = make_on_call_shift(
+        organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
+    )
+    on_call_shift.add_rolling_users([[user]])
+
+    events = ssr.shifts()
+
+    expected = [
+        # start, end, user, swap request ID
+        (start, start + duration, user.public_primary_key, ssr.public_primary_key),
+    ]
+    returned_events = [(e["start"], e["end"], e["users"][0]["pk"], e["users"][0]["swap_request"]["pk"]) for e in events]
+    assert returned_events == expected