centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #368 from grafana/matiasb/preview-schedule-shift

Browse source 
Add shift preview endpoint for web schedule
This commit is contained in:
Matias Bordese 2022-08-16 11:08:19 -03:00 • committed by GitHub
commit 5f9640366b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 473 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

173
engine/apps/api/tests/test_oncall_shift.py
View file

@ -7,7 +7,7 @@ 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 apps.schedules.models import CustomOnCallShift, OnCallSchedule, OnCallScheduleWeb
from common.constants.role import Role
@ -1140,3 +1140,174 @@ def test_on_call_shift_days_options_permissions(
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_403_FORBIDDEN),
(Role.VIEWER, status.HTTP_403_FORBIDDEN),
],
)
def test_on_call_shift_preview_permissions(
make_organization_and_user_with_plugin_token,
make_schedule,
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()
client = APIClient()
shift_start = (start_date + timezone.timedelta(hours=12)).strftime("%Y-%m-%dT%H:%M:%SZ")
shift_end = (start_date + timezone.timedelta(hours=13)).strftime("%Y-%m-%dT%H:%M:%SZ")
shift_data = {
"schedule": schedule.public_primary_key,
"type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT,
"rotation_start": shift_start,
"shift_start": shift_start,
"shift_end": shift_end,
"rolling_users": [[user.public_primary_key]],
"priority_level": 2,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
url = reverse("api-internal:oncall_shifts-preview")
response = client.post(url, shift_data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == expected_status
@pytest.mark.django_db
def test_on_call_shift_preview_missing_data(
make_organization_and_user_with_plugin_token,
make_schedule,
make_user_auth_headers,
):
organization, user, token = make_organization_and_user_with_plugin_token()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
client = APIClient()
shift_data = {
"schedule": schedule.public_primary_key,
"type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT,
"rolling_users": [[user.public_primary_key]],
"priority_level": 2,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
url = reverse("api-internal:oncall_shifts-preview")
response = client.post(url, shift_data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_on_call_shift_preview(
make_organization_and_user_with_plugin_token,
make_user_for_organization,
make_user_auth_headers,
make_schedule,
make_on_call_shift,
):
organization, user, token = make_organization_and_user_with_plugin_token()
client = APIClient()
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_web_schedule",
)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
start_date = now - timezone.timedelta(days=7)
request_date = start_date
user = make_user_for_organization(organization)
other_user = make_user_for_organization(organization)
data = {
"start": start_date + timezone.timedelta(hours=9),
"rotation_start": start_date + timezone.timedelta(hours=9),
"duration": timezone.timedelta(hours=9),
"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]])
url = "{}?date={}&days={}".format(
reverse("api-internal:oncall_shifts-preview"), request_date.strftime("%Y-%m-%d"), 1
)
shift_start = (start_date + timezone.timedelta(hours=12)).strftime("%Y-%m-%dT%H:%M:%SZ")
shift_end = (start_date + timezone.timedelta(hours=13)).strftime("%Y-%m-%dT%H:%M:%SZ")
shift_data = {
"schedule": schedule.public_primary_key,
"type": CustomOnCallShift.TYPE_ROLLING_USERS_EVENT,
"rotation_start": shift_start,
"shift_start": shift_start,
"shift_end": shift_end,
"rolling_users": [[other_user.public_primary_key]],
"priority_level": 2,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
response = client.post(url, shift_data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
# check rotation events
rotation_events = response.json()["rotation"]
expected_rotation_events = [
{
"calendar_type": OnCallSchedule.TYPE_ICAL_PRIMARY,
"start": shift_start,
"end": shift_end,
"all_day": False,
"is_override": False,
"is_empty": False,
"is_gap": False,
"priority_level": 2,
"missing_users": [],
"users": [{"display_name": other_user.username, "pk": other_user.public_primary_key}],
"source": "web",
}
]
# there isn't a saved shift, we don't care/know the temp pk
_ = [r.pop("shift") for r in rotation_events]
assert rotation_events == expected_rotation_events
# check final schedule events
final_events = response.json()["final"]
expected = (
# start (h), duration (H), user, priority
(9, 3, user.username, 1), # 9-12 user
(12, 1, other_user.username, 2), # 12-13 other_user
(13, 5, user.username, 1), # 13-18 C
)
expected_events = [
{
"end": (start_date + timezone.timedelta(hours=start + duration)).strftime("%Y-%m-%dT%H:%M:%SZ"),
"priority_level": priority,
"start": (start_date + timezone.timedelta(hours=start, milliseconds=1 if start == 0 else 0)).strftime(
"%Y-%m-%dT%H:%M:%SZ"
),
"user": user,
}
for start, duration, user, priority in expected
]
returned_events = [
{
"end": e["end"],
"priority_level": e["priority_level"],
"start": e["start"],
"user": e["users"][0]["display_name"] if e["users"] else None,
}
for e in final_events
if not e["is_override"] and not e["is_gap"]
]
assert returned_events == expected_events

24
engine/apps/api/views/on_call_shifts.py
View file

@ -1,5 +1,6 @@
from django.db.models import Q
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
@ -12,6 +13,7 @@ from apps.schedules.models import CustomOnCallShift
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.mixins import PublicPrimaryKeyMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
from common.api_helpers.utils import get_date_range_from_request
class OnCallShiftView(PublicPrimaryKeyMixin, UpdateSerializerMixin, ModelViewSet):
@ -19,7 +21,7 @@ class OnCallShiftView(PublicPrimaryKeyMixin, UpdateSerializerMixin, ModelViewSet
permission_classes = (IsAuthenticated, ActionPermission)
action_permissions = {
IsAdmin: MODIFY_ACTIONS,
IsAdmin: (*MODIFY_ACTIONS, "preview"),
AnyRole: (*READ_ACTIONS, "details", "frequency_options", "days_options"),
}
@ -77,6 +79,26 @@ class OnCallShiftView(PublicPrimaryKeyMixin, UpdateSerializerMixin, ModelViewSet
create_organization_log(organization, user, OrganizationLogType.TYPE_ON_CALL_SHIFT_DELETED, description)
instance.delete()
@action(detail=False, methods=["post"])
def preview(self, request):
user_tz, starting_date, days = get_date_range_from_request(self.request)
serializer = self.get_serializer(data=request.data)
if not serializer.is_valid():
return Response(data=serializer.errors, status=status.HTTP_400_BAD_REQUEST)
validated_data = serializer._correct_validated_data(
serializer.validated_data["type"], serializer.validated_data
)
shift = CustomOnCallShift(**validated_data)
schedule = shift.schedule
shift_events, final_events = schedule.preview_shift(shift, user_tz, starting_date, days)
data = {
"rotation": shift_events,
"final": final_events,
}
return Response(data=data, status=status.HTTP_200_OK)
@action(detail=False, methods=["get"])
def frequency_options(self, request):
return Response(

20
engine/apps/api/views/schedule.py
View file

@ -1,5 +1,3 @@
import datetime
import pytz
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import OuterRef, Subquery
@ -35,7 +33,7 @@ from common.api_helpers.mixins import (
ShortSerializerMixin,
UpdateSerializerMixin,
)
from common.api_helpers.utils import create_engine_url
from common.api_helpers.utils import create_engine_url, get_date_range_from_request
EVENTS_FILTER_BY_ROTATION = "rotation"
EVENTS_FILTER_BY_OVERRIDE = "override"
@ -224,27 +222,17 @@ class ScheduleView(
@action(detail=True, methods=["get"])
def filter_events(self, request, pk):
user_tz, date = self.get_request_timezone()
filter_by = self.request.query_params.get("type")
user_tz, starting_date, days = get_date_range_from_request(self.request)
filter_by = self.request.query_params.get("type")
valid_filters = (EVENTS_FILTER_BY_ROTATION, EVENTS_FILTER_BY_OVERRIDE, EVENTS_FILTER_BY_FINAL)
if filter_by is not None and filter_by not in valid_filters:
raise BadRequest(detail="Invalid type value")
resolve_schedule = filter_by is None or filter_by == EVENTS_FILTER_BY_FINAL
starting_date = date if self.request.query_params.get("date") else None
if starting_date is None:
# default to current week start
starting_date = date - datetime.timedelta(days=date.weekday())
try:
days = int(self.request.query_params.get("days", 7)) # fallback to a week
except ValueError:
raise BadRequest(detail="Invalid days format")
schedule = self.original_get_object()
if filter_by is not None:
if filter_by is not None and filter_by != EVENTS_FILTER_BY_FINAL:
filter_by = OnCallSchedule.PRIMARY if filter_by == EVENTS_FILTER_BY_ROTATION else OnCallSchedule.OVERRIDES
events = schedule.filter_events(
user_tz, starting_date, days=days, with_empty=True, with_gap=resolve_schedule, filter_by=filter_by

45
engine/apps/schedules/models/on_call_schedule.py
View file

@ -1,4 +1,5 @@
import datetime
import itertools
import icalendar
from django.apps import apps
@ -509,10 +510,12 @@ class OnCallScheduleCalendar(OnCallSchedule):
class OnCallScheduleWeb(OnCallSchedule):
time_zone = models.CharField(max_length=100, default="UTC")
def _generate_ical_file_from_shifts(self, qs):
def _generate_ical_file_from_shifts(self, qs, extra_shifts=None):
"""Generate iCal events file from custom on-call shifts."""
ical = None
if qs.exists():
if qs.exists() or extra_shifts is not None:
if extra_shifts is None:
extra_shifts = []
end_line = "END:VCALENDAR"
calendar = Calendar()
calendar.add("prodid", "-//web schedule//oncall//")
@ -521,7 +524,7 @@ class OnCallScheduleWeb(OnCallSchedule):
ical_file = calendar.to_ical().decode()
ical = ical_file.replace(end_line, "").strip()
ical = f"{ical}\r\n"
for event in qs.all():
for event in itertools.chain(qs.all(), extra_shifts):
ical += event.convert_to_ical(self.time_zone)
ical += f"{end_line}\r\n"
return ical
@ -559,3 +562,39 @@ class OnCallScheduleWeb(OnCallSchedule):
self.prev_ical_file_overrides = self.cached_ical_file_overrides
self.cached_ical_file_overrides = self._generate_ical_file_overrides()
self.save(update_fields=["cached_ical_file_overrides", "prev_ical_file_overrides"])
def preview_shift(self, custom_shift, user_tz, starting_date, days):
"""Return unsaved rotation and final schedule preview events."""
if custom_shift.type == CustomOnCallShift.TYPE_OVERRIDE:
qs = self.custom_shifts.filter(type=CustomOnCallShift.TYPE_OVERRIDE)
ical_attr = "cached_ical_file_overrides"
ical_property = "_ical_file_overrides"
elif custom_shift.type == CustomOnCallShift.TYPE_ROLLING_USERS_EVENT:
qs = self.custom_shifts.exclude(type=CustomOnCallShift.TYPE_OVERRIDE)
ical_attr = "cached_ical_file_primary"
ical_property = "_ical_file_primary"
else:
raise ValueError("Invalid shift type")
def _invalidate_cache(schedule, prop_name):
"""Invalidate cached property cache"""
try:
delattr(schedule, prop_name)
except AttributeError:
pass
ical_file = self._generate_ical_file_from_shifts(qs, extra_shifts=[custom_shift])
original_value = getattr(self, ical_attr)
_invalidate_cache(self, ical_property)
setattr(self, ical_attr, ical_file)
# filter events using a temporal overriden calendar including the not-yet-saved shift
events = self.filter_events(user_tz, starting_date, days=days, with_empty=True, with_gap=True)
shift_events = [e for e in events if e["shift"]["pk"] == custom_shift.public_primary_key]
final_events = self._resolve_schedule(events)
_invalidate_cache(self, ical_property)
setattr(self, ical_attr, original_value)
return shift_events, final_events

194
engine/apps/schedules/tests/test_on_call_schedule.py
View file

@ -117,7 +117,7 @@ def test_filter_events_include_gaps(make_organization, make_user_for_organizatio
data = {
"start": start_date + timezone.timedelta(hours=10),
"rotation_start": start_date + timezone.timedelta(days=1, hours=10),
"rotation_start": start_date + timezone.timedelta(hours=10),
"duration": timezone.timedelta(hours=8),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
@ -192,7 +192,7 @@ def test_filter_events_include_empty(make_organization, make_user_for_organizati
data = {
"start": start_date + timezone.timedelta(hours=10),
"rotation_start": start_date + timezone.timedelta(days=1, hours=10),
"rotation_start": start_date + timezone.timedelta(hours=10),
"duration": timezone.timedelta(hours=8),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
@ -320,3 +320,193 @@ def test_final_schedule_events(make_organization, make_user_for_organization, ma
for e in returned_events
]
assert returned_events == expected_events
@pytest.mark.django_db
def test_preview_shift(make_organization, make_user_for_organization, make_schedule, make_on_call_shift):
organization = make_organization()
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_web_schedule",
)
user = make_user_for_organization(organization)
other_user = make_user_for_organization(organization)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
start_date = now - timezone.timedelta(days=7)
data = {
"start": start_date + timezone.timedelta(hours=9),
"rotation_start": start_date + timezone.timedelta(hours=9),
"duration": timezone.timedelta(hours=9),
"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]])
schedule_primary_ical = schedule._ical_file_primary
# proposed shift
new_shift = CustomOnCallShift(
type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT,
organization=organization,
schedule=schedule,
name="testing",
start=start_date + timezone.timedelta(hours=12),
rotation_start=start_date + timezone.timedelta(hours=12),
duration=timezone.timedelta(seconds=3600),
frequency=CustomOnCallShift.FREQUENCY_DAILY,
priority_level=2,
rolling_users=[{other_user.pk: other_user.public_primary_key}],
)
rotation_events, final_events = schedule.preview_shift(new_shift, "UTC", start_date, days=1)
# check rotation events
expected_rotation_events = [
{
"calendar_type": OnCallSchedule.TYPE_ICAL_PRIMARY,
"start": new_shift.start,
"end": new_shift.start + new_shift.duration,
"all_day": False,
"is_override": False,
"is_empty": False,
"is_gap": False,
"priority_level": new_shift.priority_level,
"missing_users": [],
"users": [{"display_name": other_user.username, "pk": other_user.public_primary_key}],
"shift": {"pk": new_shift.public_primary_key},
"source": "api",
}
]
assert rotation_events == expected_rotation_events
# check final schedule events
expected = (
# start (h), duration (H), user, priority
(9, 3, user.username, 1), # 9-12 user
(12, 1, other_user.username, 2), # 12-13 other_user
(13, 5, user.username, 1), # 13-18 C
)
expected_events = [
{
"end": start_date + timezone.timedelta(hours=start + duration),
"priority_level": priority,
"start": start_date + timezone.timedelta(hours=start, milliseconds=1 if start == 0 else 0),
"user": user,
}
for start, duration, user, priority in expected
]
returned_events = [
{
"end": e["end"],
"priority_level": e["priority_level"],
"start": e["start"],
"user": e["users"][0]["display_name"] if e["users"] else None,
}
for e in final_events
if not e["is_override"] and not e["is_gap"]
]
assert returned_events == expected_events
# final ical schedule didn't change
assert schedule._ical_file_primary == schedule_primary_ical
@pytest.mark.django_db
def test_preview_override_shift(make_organization, make_user_for_organization, make_schedule, make_on_call_shift):
organization = make_organization()
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_web_schedule",
)
user = make_user_for_organization(organization)
other_user = make_user_for_organization(organization)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
start_date = now - timezone.timedelta(days=7)
data = {
"start": start_date + timezone.timedelta(hours=9),
"rotation_start": start_date + timezone.timedelta(hours=9),
"duration": timezone.timedelta(hours=9),
"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]])
schedule_overrides_ical = schedule._ical_file_overrides
# proposed override
new_shift = CustomOnCallShift(
type=CustomOnCallShift.TYPE_OVERRIDE,
organization=organization,
schedule=schedule,
name="testing",
start=start_date + timezone.timedelta(hours=12),
rotation_start=start_date + timezone.timedelta(hours=12),
duration=timezone.timedelta(seconds=3600),
rolling_users=[{other_user.pk: other_user.public_primary_key}],
)
rotation_events, final_events = schedule.preview_shift(new_shift, "UTC", start_date, days=1)
# check rotation events
expected_rotation_events = [
{
"calendar_type": OnCallSchedule.TYPE_ICAL_OVERRIDES,
"start": new_shift.start,
"end": new_shift.start + new_shift.duration,
"all_day": False,
"is_override": True,
"is_empty": False,
"is_gap": False,
"priority_level": None,
"missing_users": [],
"users": [{"display_name": other_user.username, "pk": other_user.public_primary_key}],
"shift": {"pk": new_shift.public_primary_key},
"source": "api",
}
]
assert rotation_events == expected_rotation_events
# check final schedule events
expected = (
# start (h), duration (H), user, priority, is_override
(9, 3, user.username, 1, False), # 9-12 user
(12, 1, other_user.username, None, True), # 12-13 other_user
(13, 5, user.username, 1, False), # 13-18 C
)
expected_events = [
{
"end": start_date + timezone.timedelta(hours=start + duration),
"priority_level": priority,
"start": start_date + timezone.timedelta(hours=start, milliseconds=1 if start == 0 else 0),
"user": user,
"is_override": is_override,
}
for start, duration, user, priority, is_override in expected
]
returned_events = [
{
"end": e["end"],
"priority_level": e["priority_level"],
"start": e["start"],
"user": e["users"][0]["display_name"] if e["users"] else None,
"is_override": e["is_override"],
}
for e in final_events
if not e["is_gap"]
]
assert returned_events == expected_events
# final ical schedule didn't change
assert schedule._ical_file_overrides == schedule_overrides_ical

40
engine/common/api_helpers/utils.py
View file

@ -1,10 +1,15 @@
import datetime
from urllib.parse import urljoin
import pytz
import requests
from django.conf import settings
from django.utils import dateparse, timezone
from icalendar import Calendar
from rest_framework import serializers
from common.api_helpers.exceptions import BadRequest
class CurrentOrganizationDefault:
"""
@ -71,3 +76,38 @@ def create_engine_url(path, override_base=None):
base += "/"
trimmed_path = path.lstrip("/")
return urljoin(base, trimmed_path)
def get_date_range_from_request(request):
"""Extract timezone, starting date and number of days params from request.
Used mainly for schedules and shifts API.
"""
user_tz = request.query_params.get("user_tz", "UTC")
try:
pytz.timezone(user_tz)
except pytz.exceptions.UnknownTimeZoneError:
raise BadRequest(detail="Invalid tz format")
date = timezone.now().date()
date_param = request.query_params.get("date")
if date_param is not None:
try:
date = dateparse.parse_date(date_param)
except ValueError:
raise BadRequest(detail="Invalid date format")
else:
if date is None:
raise BadRequest(detail="Invalid date format")
starting_date = date if request.query_params.get("date") else None
if starting_date is None:
# default to current week start
starting_date = date - datetime.timedelta(days=date.weekday())
try:
days = int(request.query_params.get("days", 7)) # fallback to a week
except ValueError:
raise BadRequest(detail="Invalid days format")
return user_tz, starting_date, days