diff --git a/engine/apps/api/serializers/on_call_shifts.py b/engine/apps/api/serializers/on_call_shifts.py index 794c466f..9cc9e66c 100644 --- a/engine/apps/api/serializers/on_call_shifts.py +++ b/engine/apps/api/serializers/on_call_shifts.py @@ -30,13 +30,14 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): queryset=User.objects, required=False, allow_null=True ), # todo: filter by team? ) + updated_shift = serializers.CharField(read_only=True, allow_null=True, source="updated_shift.public_primary_key") class Meta: model = CustomOnCallShift fields = [ "id", "organization", - "name", + "title", "type", "schedule", "priority_level", @@ -49,20 +50,20 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): "by_day", "source", "rolling_users", + "updated_shift", ] extra_kwargs = { "interval": {"required": False, "allow_null": True}, "source": {"required": False, "write_only": True}, } - SELECT_RELATED = ["schedule"] + SELECT_RELATED = ["schedule", "updated_shift"] def get_shift_end(self, obj): return obj.start + obj.duration def to_internal_value(self, data): data["source"] = CustomOnCallShift.SOURCE_WEB - data["week_start"] = CustomOnCallShift.MONDAY if not data.get("shift_end"): raise serializers.ValidationError({"shift_end": ["This field is required."]}) @@ -73,19 +74,6 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): result = super().to_representation(instance) return result - def validate_name(self, name): - organization = self.context["request"].auth.organization - if name is None: - return name - try: - obj = CustomOnCallShift.objects.get(organization=organization, name=name) - except CustomOnCallShift.DoesNotExist: - return name - if self.instance and obj.id == self.instance.id: - return name - else: - raise serializers.ValidationError(["On-call shift with this name already exists"]) - def validate_by_day(self, by_day): if by_day: for day in by_day: @@ -105,7 +93,7 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): for users in rolling_users: users_dict = dict() for user in users: - users_dict[user.pk] = user.public_primary_key + users_dict[str(user.pk)] = user.public_primary_key result.append(users_dict) return result @@ -113,7 +101,7 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): if end <= start: raise serializers.ValidationError({"shift_end": ["Incorrect shift end date"]}) - def _validate_frequency(self, frequency, event_type, rolling_users, interval, by_day): + def _validate_frequency(self, frequency, event_type, rolling_users, interval, by_day, until): if frequency is None: if rolling_users and len(rolling_users) > 1: raise serializers.ValidationError( @@ -123,6 +111,8 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): raise serializers.ValidationError({"interval": ["Cannot set interval for non-recurrent shifts"]}) if by_day: raise serializers.ValidationError({"by_day": ["Cannot set days value for non-recurrent shifts"]}) + if until: + raise serializers.ValidationError({"until": ["Cannot set 'until' for non-recurrent shifts"]}) else: if event_type == CustomOnCallShift.TYPE_OVERRIDE: raise serializers.ValidationError( @@ -142,10 +132,6 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): def _correct_validated_data(self, event_type, validated_data): fields_to_update_for_overrides = [ "priority_level", - "frequency", - "interval", - "by_day", - "until", "rotation_start", ] if event_type == CustomOnCallShift.TYPE_OVERRIDE: @@ -163,6 +149,7 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): validated_data.get("rolling_users"), validated_data.get("interval"), validated_data.get("by_day"), + validated_data.get("until"), ) self._validate_rotation_start(validated_data["start"], validated_data["rotation_start"]) self._validate_until(validated_data["rotation_start"], validated_data.get("until")) @@ -176,11 +163,15 @@ class OnCallShiftSerializer(EagerLoadingMixin, serializers.ModelSerializer): if validated_data.get("schedule"): validated_data["team"] = validated_data["schedule"].team + validated_data["week_start"] = CustomOnCallShift.MONDAY + return validated_data def create(self, validated_data): validated_data = self._correct_validated_data(validated_data["type"], validated_data) - + validated_data["name"] = CustomOnCallShift.generate_name( + validated_data["schedule"], validated_data["priority_level"], validated_data["type"] + ) instance = super().create(validated_data) instance.start_drop_ical_and_check_schedule_tasks(instance.schedule) @@ -196,8 +187,26 @@ class OnCallShiftUpdateSerializer(OnCallShiftSerializer): def update(self, instance, validated_data): validated_data = self._correct_validated_data(instance.type, validated_data) + change_only_title = True + create_or_update_last_shift = False - result = super().update(instance, validated_data) + for field in validated_data: + if field != "title" and validated_data[field] != getattr(instance, field): + change_only_title = False + break + + if not change_only_title: + if instance.type != CustomOnCallShift.TYPE_OVERRIDE: + if instance.event_is_started: + create_or_update_last_shift = True + + elif instance.event_is_finished: + raise serializers.ValidationError(["This event cannot be updated"]) + + if create_or_update_last_shift: + result = instance.create_or_update_last_shift(validated_data) + else: + result = super().update(instance, validated_data) instance.start_drop_ical_and_check_schedule_tasks(instance.schedule) return result diff --git a/engine/apps/api/tests/test_oncall_shift.py b/engine/apps/api/tests/test_oncall_shift.py new file mode 100644 index 00000000..a40fbd46 --- /dev/null +++ b/engine/apps/api/tests/test_oncall_shift.py @@ -0,0 +1,1142 @@ +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.response import Response +from rest_framework.test import APIClient + +from apps.schedules.models import CustomOnCallShift, OnCallScheduleWeb +from common.constants.role import Role + + +@pytest.fixture() +def on_call_shift_internal_api_setup( + make_organization_and_user_with_plugin_token, + make_schedule, + make_user_for_organization, +): + organization, first_user, token = make_organization_and_user_with_plugin_token() + second_user = make_user_for_organization(organization) + + schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) + return token, first_user, second_user, organization, schedule + + +@pytest.mark.django_db +def test_create_on_call_shift_rotation(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + data = { + "title": "Test Shift", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 1, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": 1, + "interval": None, + "by_day": [ + CustomOnCallShift.ICAL_WEEKDAY_MAP[CustomOnCallShift.MONDAY], + CustomOnCallShift.ICAL_WEEKDAY_MAP[CustomOnCallShift.FRIDAY], + ], + "rolling_users": [[user1.public_primary_key], [user2.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + expected_payload = data | {"id": response.data["id"], "updated_shift": None} + + assert response.status_code == status.HTTP_201_CREATED + assert response.json() == expected_payload + + +@pytest.mark.django_db +def test_create_on_call_shift_override(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + data = { + "title": "Test Shift Override", + "type": CustomOnCallShift.TYPE_OVERRIDE, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key, user2.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + expected_payload = data | {"id": response.data["id"], "updated_shift": None} + + assert response.status_code == status.HTTP_201_CREATED + assert response.json() == expected_payload + + +@pytest.mark.django_db +def test_get_on_call_shift( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = timezone.now().replace(microsecond=0) + + title = "Test Shift Rotation" + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}, {user2.pk: user2.public_primary_key}], + ) + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + response = client.get(url, format="json", **make_user_auth_headers(user1, token)) + expected_payload = { + "id": response.data["id"], + "title": title, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key], [user2.public_primary_key]], + "updated_shift": None, + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + +@pytest.mark.django_db +def test_list_on_call_shift( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = timezone.now().replace(microsecond=0) + title = "Test Shift Rotation" + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}, {user2.pk: user2.public_primary_key}], + ) + url = reverse("api-internal:oncall_shifts-list") + + response = client.get(url, format="json", **make_user_auth_headers(user1, token)) + expected_payload = { + "count": 1, + "next": None, + "previous": None, + "results": [ + { + "id": on_call_shift.public_primary_key, + "title": title, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key], [user2.public_primary_key]], + "updated_shift": None, + } + ], + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + +@pytest.mark.django_db +def test_list_on_call_shift_filter_schedule_id( + on_call_shift_internal_api_setup, + make_schedule, + make_on_call_shift, + make_user_auth_headers, +): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + schedule_without_shifts = make_schedule(organization, schedule_class=OnCallScheduleWeb) + + client = APIClient() + + start_date = timezone.now().replace(microsecond=0) + title = "Test Shift Rotation" + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}, {user2.pk: user2.public_primary_key}], + ) + url = reverse("api-internal:oncall_shifts-list") + + response = client.get( + url + f"?schedule_id={schedule.public_primary_key}", format="json", **make_user_auth_headers(user1, token) + ) + expected_payload = { + "count": 1, + "next": None, + "previous": None, + "results": [ + { + "id": on_call_shift.public_primary_key, + "title": title, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key], [user2.public_primary_key]], + "updated_shift": None, + } + ], + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + expected_payload = { + "count": 0, + "next": None, + "previous": None, + "results": [], + } + + response = client.get( + url + f"?schedule_id={schedule_without_shifts.public_primary_key}", + format="json", + **make_user_auth_headers(user1, token), + ) + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + +@pytest.mark.django_db +def test_update_future_on_call_shift( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + """Test updating the shift that has not started (rotation_start > now)""" + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = (timezone.now() + timezone.timedelta(days=1)).replace(microsecond=0) + + title = "Test Shift Rotation" + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + ) + data_to_update = { + "title": title, + "priority_level": 2, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + assert on_call_shift.priority_level != data_to_update["priority_level"] + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + response = client.put(url, data=data_to_update, format="json", **make_user_auth_headers(user1, token)) + + expected_payload = { + "id": on_call_shift.public_primary_key, + "title": title, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 2, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + "updated_shift": None, + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + on_call_shift.refresh_from_db() + assert on_call_shift.priority_level == data_to_update["priority_level"] + + +@pytest.mark.django_db +def test_update_started_on_call_shift( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + """Test updating the shift that has started (rotation_start < now)""" + + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = (timezone.now() - timezone.timedelta(hours=1)).replace(microsecond=0) + + title = "Test Shift Rotation" + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=3), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + ) + data_to_update = { + "title": title, + "priority_level": 2, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + assert on_call_shift.priority_level != data_to_update["priority_level"] + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + response = client.put(url, data=data_to_update, format="json", **make_user_auth_headers(user1, token)) + + expected_payload = { + "id": response.data["id"], + "title": title, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 2, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": response.data["rotation_start"], + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + "updated_shift": None, + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + # check that another shift was created + assert response.data["id"] != on_call_shift.public_primary_key + on_call_shift.refresh_from_db() + assert on_call_shift.priority_level != data_to_update["priority_level"] + assert on_call_shift.updated_shift.public_primary_key == response.data["id"] + # check if until date was changed + assert on_call_shift.until is not None + assert on_call_shift.until == on_call_shift.updated_shift.rotation_start + + +@pytest.mark.django_db +def test_update_old_on_call_shift_with_future_version( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + """Test updating the shift that has the newer version (updated_shift is not None)""" + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = (timezone.now() - timezone.timedelta(days=3)).replace(microsecond=0) + next_rotation_start_date = start_date + timezone.timedelta(days=5) + updated_duration = timezone.timedelta(hours=4) + + title = "Test Shift Rotation" + new_on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=3), + rotation_start=next_rotation_start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + ) + old_on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=3), + rotation_start=start_date, + until=next_rotation_start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + updated_shift=new_on_call_shift, + ) + # update shift_end and priority_level + data_to_update = { + "title": title, + "priority_level": 2, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + updated_duration).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": next_rotation_start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + assert old_on_call_shift.duration != updated_duration + assert old_on_call_shift.priority_level != data_to_update["priority_level"] + assert new_on_call_shift.duration != updated_duration + assert new_on_call_shift.priority_level != data_to_update["priority_level"] + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": old_on_call_shift.public_primary_key}) + + response = client.put(url, data=data_to_update, format="json", **make_user_auth_headers(user1, token)) + + expected_payload = data_to_update | { + "id": new_on_call_shift.public_primary_key, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "updated_shift": None, + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + new_on_call_shift.refresh_from_db() + # check if the newest version of shift was changed + assert old_on_call_shift.duration != updated_duration + assert old_on_call_shift.priority_level != data_to_update["priority_level"] + assert new_on_call_shift.duration == updated_duration + assert new_on_call_shift.priority_level == data_to_update["priority_level"] + + +@pytest.mark.django_db +def test_update_started_on_call_shift_title( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + """Test updating the title for the shift that has started (rotation_start < now)""" + + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = (timezone.now() - timezone.timedelta(hours=1)).replace(microsecond=0) + + title = "Test Shift Rotation" + new_title = "Test Shift Rotation RENAMED" + + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + source=CustomOnCallShift.SOURCE_WEB, + week_start=CustomOnCallShift.MONDAY, + ) + # update only title + data_to_update = { + "title": new_title, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + assert on_call_shift.title != new_title + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + response = client.put(url, data=data_to_update, format="json", **make_user_auth_headers(user1, token)) + + expected_payload = data_to_update | { + "id": on_call_shift.public_primary_key, + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "updated_shift": None, + } + + assert response.status_code == status.HTTP_200_OK + assert response.json() == expected_payload + + on_call_shift.refresh_from_db() + assert on_call_shift.title == new_title + + +@pytest.mark.django_db +def test_delete_started_on_call_shift( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + """Test deleting the shift that has started (rotation_start < now)""" + + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = (timezone.now() - timezone.timedelta(hours=1)).replace(microsecond=0) + + title = "Test Shift Rotation" + + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + ) + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + assert on_call_shift.until is None + + response = client.delete(url, **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_204_NO_CONTENT + + on_call_shift.refresh_from_db() + assert on_call_shift.until is not None + + +@pytest.mark.django_db +def test_delete_future_on_call_shift( + on_call_shift_internal_api_setup, + make_on_call_shift, + make_user_auth_headers, +): + """Test deleting the shift that has not started (rotation_start > now)""" + + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + + client = APIClient() + start_date = (timezone.now() + timezone.timedelta(days=1)).replace(microsecond=0) + + title = "Test Shift Rotation" + + on_call_shift = make_on_call_shift( + schedule.organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + title=title, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + rolling_users=[{user1.pk: user1.public_primary_key}], + ) + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + response = client.delete(url, **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_204_NO_CONTENT + + with pytest.raises(CustomOnCallShift.DoesNotExist): + on_call_shift.refresh_from_db() + + +@pytest.mark.django_db +def test_create_on_call_shift_invalid_data_rotation_start( + on_call_shift_internal_api_setup, + make_user_auth_headers, +): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + # rotation_start < shift_start + data = { + "title": "Test Shift 1", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": (start_date - timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["rotation_start"][0] == "Incorrect rotation start date" + + +@pytest.mark.django_db +def test_create_on_call_shift_invalid_data_until(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + # until < rotation_start + data = { + "title": "Test Shift", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 1, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": (start_date - timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "frequency": 1, + "interval": None, + "by_day": [ + CustomOnCallShift.ICAL_WEEKDAY_MAP[CustomOnCallShift.MONDAY], + CustomOnCallShift.ICAL_WEEKDAY_MAP[CustomOnCallShift.FRIDAY], + ], + "rolling_users": [[user1.public_primary_key], [user2.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["until"][0] == "Incorrect rotation end date" + + # until with non-recurrent shift + data = { + "title": "Test Shift 2", + "type": CustomOnCallShift.TYPE_OVERRIDE, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["until"][0] == "Cannot set 'until' for non-recurrent shifts" + + +@pytest.mark.django_db +def test_create_on_call_shift_invalid_data_by_day(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + # by_day with non-recurrent shift + data = { + "title": "Test Shift 1", + "type": CustomOnCallShift.TYPE_OVERRIDE, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": [CustomOnCallShift.ICAL_WEEKDAY_MAP[CustomOnCallShift.MONDAY]], + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["by_day"][0] == "Cannot set days value for non-recurrent shifts" + + # by_day with non-weekly frequency + data = { + "title": "Test Shift 2", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": CustomOnCallShift.FREQUENCY_DAILY, + "interval": None, + "by_day": [CustomOnCallShift.ICAL_WEEKDAY_MAP[CustomOnCallShift.MONDAY]], + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["by_day"][0] == "Cannot set days value for this frequency type" + + +@pytest.mark.django_db +def test_create_on_call_shift_invalid_data_interval(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + # interval with non-recurrent shift + data = { + "title": "Test Shift 2", + "type": CustomOnCallShift.TYPE_OVERRIDE, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": 2, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["interval"][0] == "Cannot set interval for non-recurrent shifts" + + +@pytest.mark.django_db +def test_create_on_call_shift_invalid_data_shift_end(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + # shift_end is None + data = { + "title": "Test Shift 1", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": None, + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": 1, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["shift_end"][0] == "This field is required." + + # shift_end < shift_start + data = { + "title": "Test Shift 2", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date - timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["shift_end"][0] == "Incorrect shift end date" + + +@pytest.mark.django_db +def test_create_on_call_shift_invalid_data_rolling_users( + on_call_shift_internal_api_setup, + make_user_auth_headers, +): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + data = { + "title": "Test Shift 1", + "type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": None, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key], [user2.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["rolling_users"][0] == "Cannot set multiple user groups for non-recurrent shifts" + + +@pytest.mark.django_db +def test_create_on_call_shift_override_invalid_data(on_call_shift_internal_api_setup, make_user_auth_headers): + token, user1, user2, organization, schedule = on_call_shift_internal_api_setup + client = APIClient() + url = reverse("api-internal:oncall_shifts-list") + start_date = timezone.now().replace(microsecond=0, tzinfo=None) + + # override shift with frequency + data = { + "title": "Test Shift Override", + "type": CustomOnCallShift.TYPE_OVERRIDE, + "schedule": schedule.public_primary_key, + "priority_level": 0, + "shift_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "shift_end": (start_date + timezone.timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "rotation_start": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"), + "until": None, + "frequency": 1, + "interval": None, + "by_day": None, + "rolling_users": [[user1.public_primary_key]], + } + + response = client.post(url, data, format="json", **make_user_auth_headers(user1, token)) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["frequency"][0] == "Cannot set 'frequency' for shifts with type 'override'" + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_201_CREATED), + (Role.EDITOR, status.HTTP_403_FORBIDDEN), + (Role.VIEWER, status.HTTP_403_FORBIDDEN), + ], +) +def test_on_call_shift_create_permissions( + make_organization_and_user_with_plugin_token, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + + client = APIClient() + + url = reverse("api-internal:oncall_shifts-list") + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.create", + return_value=Response( + status=status.HTTP_201_CREATED, + ), + ): + response = client.post(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_200_OK), + (Role.EDITOR, status.HTTP_403_FORBIDDEN), + (Role.VIEWER, status.HTTP_403_FORBIDDEN), + ], +) +def test_on_call_shift_update_permissions( + make_organization_and_user_with_plugin_token, + make_schedule, + make_on_call_shift, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + + client = APIClient() + schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) + start_date = timezone.now() + on_call_shift = make_on_call_shift( + organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + ) + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.update", + return_value=Response( + status=status.HTTP_200_OK, + ), + ): + + response = client.put(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + response = client.patch(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_200_OK), + (Role.EDITOR, status.HTTP_200_OK), + (Role.VIEWER, status.HTTP_200_OK), + ], +) +def test_on_call_shift_list_permissions( + make_organization_and_user_with_plugin_token, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + client = APIClient() + + url = reverse("api-internal:oncall_shifts-list") + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.list", + return_value=Response( + status=status.HTTP_200_OK, + ), + ): + response = client.get(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_200_OK), + (Role.EDITOR, status.HTTP_200_OK), + (Role.VIEWER, status.HTTP_200_OK), + ], +) +def test_on_call_shift_retrieve_permissions( + make_organization_and_user_with_plugin_token, + make_schedule, + make_on_call_shift, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) + start_date = timezone.now() + on_call_shift = make_on_call_shift( + organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + ) + client = APIClient() + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.retrieve", + return_value=Response( + status=status.HTTP_200_OK, + ), + ): + response = client.get(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_204_NO_CONTENT), + (Role.EDITOR, status.HTTP_403_FORBIDDEN), + (Role.VIEWER, status.HTTP_403_FORBIDDEN), + ], +) +def test_on_call_shift_delete_permissions( + make_organization_and_user_with_plugin_token, + make_schedule, + make_on_call_shift, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) + start_date = timezone.now() + on_call_shift = make_on_call_shift( + organization, + shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, + schedule=schedule, + start=start_date, + duration=timezone.timedelta(hours=1), + rotation_start=start_date, + ) + client = APIClient() + + url = reverse("api-internal:oncall_shifts-detail", kwargs={"pk": on_call_shift.public_primary_key}) + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.destroy", + return_value=Response( + status=status.HTTP_204_NO_CONTENT, + ), + ): + response = client.delete(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_200_OK), + (Role.EDITOR, status.HTTP_200_OK), + (Role.VIEWER, status.HTTP_200_OK), + ], +) +def test_on_call_shift_frequency_options_permissions( + make_organization_and_user_with_plugin_token, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + client = APIClient() + + url = reverse("api-internal:oncall_shifts-frequency-options") + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.frequency_options", + return_value=Response( + status=status.HTTP_200_OK, + ), + ): + response = client.get(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status + + +@pytest.mark.django_db +@pytest.mark.parametrize( + "role,expected_status", + [ + (Role.ADMIN, status.HTTP_200_OK), + (Role.EDITOR, status.HTTP_200_OK), + (Role.VIEWER, status.HTTP_200_OK), + ], +) +def test_on_call_shift_days_options_permissions( + make_organization_and_user_with_plugin_token, + make_user_auth_headers, + role, + expected_status, +): + organization, user, token = make_organization_and_user_with_plugin_token(role) + client = APIClient() + + url = reverse("api-internal:oncall_shifts-days-options") + + with patch( + "apps.api.views.on_call_shifts.OnCallShiftView.days_options", + return_value=Response( + status=status.HTTP_200_OK, + ), + ): + response = client.get(url, format="json", **make_user_auth_headers(user, token)) + + assert response.status_code == expected_status diff --git a/engine/apps/schedules/migrations/0007_customoncallshift_updated_shift.py b/engine/apps/schedules/migrations/0007_customoncallshift_updated_shift.py new file mode 100644 index 00000000..c034f53a --- /dev/null +++ b/engine/apps/schedules/migrations/0007_customoncallshift_updated_shift.py @@ -0,0 +1,24 @@ +# Generated by Django 3.2.13 on 2022-07-22 11:54 + +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + dependencies = [ + ('schedules', '0006_customoncallshift_rotation_start'), + ] + + operations = [ + migrations.AddField( + model_name='customoncallshift', + name='updated_shift', + field=models.OneToOneField(default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='parent_shift', to='schedules.customoncallshift'), + ), + migrations.AddField( + model_name='customoncallshift', + name='title', + field=models.CharField(default=None, max_length=200, null=True), + ), + ] diff --git a/engine/apps/schedules/models/custom_on_call_shift.py b/engine/apps/schedules/models/custom_on_call_shift.py index c19d2250..94782050 100644 --- a/engine/apps/schedules/models/custom_on_call_shift.py +++ b/engine/apps/schedules/models/custom_on_call_shift.py @@ -1,4 +1,6 @@ import logging +import random +import string from calendar import monthrange from uuid import uuid4 @@ -6,8 +8,9 @@ import pytz from django.apps import apps from django.conf import settings from django.core.validators import MinLengthValidator -from django.db import models +from django.db import models, transaction from django.db.models import JSONField +from django.forms.models import model_to_dict from django.utils import timezone from django.utils.functional import cached_property from icalendar.cal import Event @@ -165,6 +168,7 @@ class CustomOnCallShift(models.Model): default=None, ) name = models.CharField(max_length=200) + title = models.CharField(max_length=200, null=True, default=None) time_zone = models.CharField(max_length=100, null=True, default=None) source = models.IntegerField(choices=SOURCE_CHOICES, default=SOURCE_API) users = models.ManyToManyField("user_management.User") # users in single and recurrent events @@ -196,16 +200,44 @@ class CustomOnCallShift(models.Model): by_month = JSONField(default=None, null=True) # [] BYMONTH - what months (1, 2, 3, ...) - ical format by_monthday = JSONField(default=None, null=True) # [] BYMONTHDAY - what days of month (1, 2, -3) - ical format + updated_shift = models.OneToOneField( + "schedules.CustomOnCallShift", + on_delete=models.SET_NULL, + default=None, + null=True, + related_name="parent_shift", + ) + class Meta: unique_together = ("name", "organization") def delete(self, *args, **kwargs): - for schedule in self.schedules.all(): - self.start_drop_ical_and_check_schedule_tasks(schedule) + schedules_to_update = list(self.schedules.all()) if self.schedule: - self.start_drop_ical_and_check_schedule_tasks(self.schedule) - # todo: add soft delete - super().delete(*args, **kwargs) + schedules_to_update.append(self.schedule) + + # do soft delete for started shifts that were created for web schedule + if self.schedule and self.event_is_started: + self.until = timezone.now().replace(microsecond=0) + self.save(update_fields=["until"]) + else: + super().delete(*args, **kwargs) + + for schedule in schedules_to_update: + self.start_drop_ical_and_check_schedule_tasks(schedule) + + @property + def event_is_started(self): + return bool(self.rotation_start <= timezone.now()) + + @property + def event_is_finished(self): + if self.frequency is not None: + is_finished = bool(self.until <= timezone.now()) if self.until else False + else: + is_finished = bool(self.start + self.duration <= timezone.now()) + + return is_finished @property def repr_settings_for_client_side_logging(self) -> str: @@ -328,7 +360,7 @@ class CustomOnCallShift(models.Model): if event.start.date() >= next_event_start.date(): next_event = event break - next_event_dt = next_event.start + next_event_dt = next_event.start if next_event is not None else None return next_event_dt @cached_property @@ -399,3 +431,51 @@ class CustomOnCallShift(models.Model): drop_cached_ical_task.apply_async((schedule.pk,)) schedule_notify_about_empty_shifts_in_schedule.apply_async((schedule.pk,)) schedule_notify_about_gaps_in_schedule.apply_async((schedule.pk,)) + + @cached_property + def last_updated_shift(self): + last_shift = self.updated_shift + if last_shift is not None: + while last_shift.updated_shift is not None: + last_shift = last_shift.updated_shift + return last_shift + + def create_or_update_last_shift(self, data): + # rotation start date cannot be earlier than now + data["rotation_start"] = max(data["rotation_start"], timezone.now().replace(microsecond=0)) + # prepare dict with params of existing instance with last updates and remove unique and m2m fields from it + shift_to_update = self.last_updated_shift or self + instance_data = model_to_dict(shift_to_update) + fields_to_remove = ["id", "public_primary_key", "uuid", "users", "updated_shift", "name"] + for field in fields_to_remove: + instance_data.pop(field) + + instance_data.update(data) + instance_data["schedule"] = self.schedule + instance_data["team"] = self.team + + if self.last_updated_shift is None or self.last_updated_shift.event_is_started: + # create new shift + instance_data["name"] = CustomOnCallShift.generate_name( + self.schedule, instance_data["priority_level"], instance_data["type"] + ) + with transaction.atomic(): + shift = CustomOnCallShift(**instance_data) + shift.save() + shift_to_update.until = data["rotation_start"] + shift_to_update.updated_shift = shift + shift_to_update.save(update_fields=["until", "updated_shift"]) + else: + shift = self.last_updated_shift + for key in instance_data: + setattr(shift, key, instance_data[key]) + shift.save(update_fields=list(instance_data)) + + return shift + + @staticmethod + def generate_name(schedule, priority_level, shift_type): + shift_type_name = "override" if shift_type == CustomOnCallShift.TYPE_OVERRIDE else "rotation" + name = f"{schedule.name}-{shift_type_name}-{priority_level}-" + name += "".join(random.choice(string.ascii_lowercase) for _ in range(5)) + return name