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

Add endpoint for getting schedules events for current user (#2928)

Browse source 
# What this PR does

## Which issue(s) this PR fixes
https://github.com/grafana/oncall/issues/2915

## Checklist

- [x] Unit, integration, and e2e (if applicable) tests updated
- [x] Documentation added (or `pr:no public docs` PR label added if not
required)
- [x] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not
required)
This commit is contained in:
Yulya Artyukhina 2023-09-05 09:22:08 +02:00 • committed by GitHub
parent a2851d3f81
commit ecb4ba0057
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 422 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
CHANGELOG.md
View file

@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## Unreleased
### Added
- Add internal API endpoint for getting schedules shifts for current user by @Ferril([#2928](https://github.com/grafana/oncall/pull/2928))
### Changed
- Make Slack integration not post an alert group message if it's already deleted + refactor AlertGroup and

199
engine/apps/api/tests/test_schedules.py
View file

@ -2008,6 +2008,36 @@ def test_schedule_mention_options_permissions(
assert response.status_code == expected_status
@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_current_user_events_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:schedule-current-user-events")
with patch(
"apps.api.views.schedule.ScheduleView.current_user_events",
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
def test_get_schedule_from_other_team_with_flag(
make_organization_and_user_with_plugin_token,
@ -2065,3 +2095,172 @@ def test_get_schedule_on_call_now(
"avatar_full": "https://example.com/avatar/test123",
}
]
@pytest.mark.django_db
def test_current_user_events(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization, current_user, token = make_organization_and_user_with_plugin_token()
other_user = make_user_for_organization(organization)
client = APIClient()
url = reverse("api-internal:schedule-current-user-events")
schedule_with_current_user = make_schedule(organization, schedule_class=OnCallScheduleWeb)
other_schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
shifts = (
# schedule, user, priority, start time (h), duration (seconds)
(schedule_with_current_user, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
(other_schedule, other_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
)
now = timezone.now()
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
for schedule, user, priority, start_h, duration in shifts:
data = {
"start": today + timezone.timedelta(hours=start_h),
"rotation_start": today + timezone.timedelta(hours=start_h),
"duration": timezone.timedelta(seconds=duration),
"priority_level": priority,
"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.refresh_ical_file()
schedule.refresh_ical_final_schedule()
response = client.get(url, format="json", **make_user_auth_headers(current_user, token))
result = response.json()
assert response.status_code == status.HTTP_200_OK
assert result["is_oncall"] is True
assert len(result["schedules"]) == 1
assert result["schedules"][0]["id"] == schedule_with_current_user.public_primary_key
assert result["schedules"][0]["name"] == schedule_with_current_user.name
assert len(result["schedules"][0]["events"]) > 0
@pytest.mark.django_db
def test_current_user_events_out_of_range(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_schedule,
make_on_call_shift,
):
organization, current_user, token = make_organization_and_user_with_plugin_token()
client = APIClient()
schedule_with_current_user = make_schedule(organization, schedule_class=OnCallScheduleWeb)
shifts = (
# schedule, user, priority, start time (h), duration (seconds)
(schedule_with_current_user, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
)
now = timezone.now()
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
days = 3
start_date = today + timezone.timedelta(days=days)
for schedule, user, priority, start_h, duration in shifts:
data = {
"start": start_date + timezone.timedelta(hours=start_h),
"rotation_start": start_date + timezone.timedelta(hours=start_h),
"duration": timezone.timedelta(seconds=duration),
"priority_level": priority,
"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.refresh_ical_file()
schedule.refresh_ical_final_schedule()
url = reverse("api-internal:schedule-current-user-events") + f"?days={days}"
response = client.get(url, format="json", **make_user_auth_headers(current_user, token))
result = response.json()
assert response.status_code == status.HTTP_200_OK
assert result["is_oncall"] is False
assert len(result["schedules"]) == 0
@pytest.mark.django_db
def test_current_user_events_no_schedules(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_schedule,
make_on_call_shift,
):
organization, current_user, token = make_organization_and_user_with_plugin_token()
client = APIClient()
url = reverse("api-internal:schedule-current-user-events")
response = client.get(url, format="json", **make_user_auth_headers(current_user, token))
result = response.json()
assert response.status_code == status.HTTP_200_OK
assert result["is_oncall"] is False
assert len(result["schedules"]) == 0
@pytest.mark.django_db
def test_current_user_events_multiple_schedules(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
make_schedule,
make_on_call_shift,
):
organization, current_user, token = make_organization_and_user_with_plugin_token()
client = APIClient()
url = reverse("api-internal:schedule-current-user-events")
schedule_1 = make_schedule(organization, schedule_class=OnCallScheduleWeb)
schedule_2 = make_schedule(organization, schedule_class=OnCallScheduleWeb)
shifts = (
# schedule, user, priority, start time (h), duration (seconds)
(schedule_1, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
(schedule_2, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
)
now = timezone.now()
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
for schedule, user, priority, start_h, duration in shifts:
data = {
"start": today + timezone.timedelta(hours=start_h),
"rotation_start": today + timezone.timedelta(hours=start_h),
"duration": timezone.timedelta(seconds=duration),
"priority_level": priority,
"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.refresh_ical_file()
schedule.refresh_ical_final_schedule()
response = client.get(url, format="json", **make_user_auth_headers(current_user, token))
result = response.json()
assert response.status_code == status.HTTP_200_OK
assert result["is_oncall"] is True
assert len(result["schedules"]) == 2
assert result["schedules"][0]["id"] != result["schedules"][1]["id"]
assert result["schedules"][0]["id"] in (schedule_1.public_primary_key, schedule_2.public_primary_key)
assert result["schedules"][0]["name"] in (schedule_1.name, schedule_2.name)
assert result["schedules"][1]["id"] in (schedule_1.public_primary_key, schedule_2.public_primary_key)
assert result["schedules"][1]["name"] in (schedule_1.name, schedule_2.name)
assert len(result["schedules"][0]["events"]) > 0
assert len(result["schedules"][1]["events"]) > 0

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

@ -92,6 +92,7 @@ class ScheduleView(
"notify_oncall_shift_freq_options": [RBACPermission.Permissions.SCHEDULES_READ],
"mention_options": [RBACPermission.Permissions.SCHEDULES_READ],
"related_escalation_chains": [RBACPermission.Permissions.SCHEDULES_READ],
"current_user_events": [RBACPermission.Permissions.SCHEDULES_READ],
"create": [RBACPermission.Permissions.SCHEDULES_WRITE],
"update": [RBACPermission.Permissions.SCHEDULES_WRITE],
"partial_update": [RBACPermission.Permissions.SCHEDULES_WRITE],
@ -406,6 +407,29 @@ class ScheduleView(
return Response(schedule.quality_report(datetime_start, days))
@action(detail=False, methods=["get"])
def current_user_events(self, request):
user_tz, starting_date, days = get_date_range_from_request(self.request)
pytz_tz = pytz.timezone(user_tz)
datetime_start = datetime.datetime.combine(starting_date, datetime.time.min, tzinfo=pytz_tz)
schedules = OnCallSchedule.objects.related_to_user(self.request.user)
schedules_events = []
is_oncall = False
for schedule in schedules:
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(
user=self.request.user, datetime_start=datetime_start, days=days
)
all_shifts = passed_shifts + current_shifts + upcoming_shifts
if all_shifts:
schedules_events.append(
{"id": schedule.public_primary_key, "name": schedule.name, "events": all_shifts}
)
if current_shifts and not is_oncall:
is_oncall = True
result = {"schedules": schedules_events, "is_oncall": is_oncall}
return Response(result, status=status.HTTP_200_OK)
@action(detail=False, methods=["get"])
def type_options(self, request):
# TODO: check if it needed

12
engine/apps/api/views/user.py
View file

@ -5,6 +5,7 @@ from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from django.db.utils import IntegrityError
from django.urls import reverse
from django.utils import timezone
from django_filters import rest_framework as filters
from rest_framework import mixins, status, viewsets
from rest_framework.decorators import action
@ -577,21 +578,22 @@ class UserView(
if days <= 0 or days > UPCOMING_SHIFTS_MAX_DAYS:
return Response(status=status.HTTP_400_BAD_REQUEST)
now = timezone.now()
# filter user-related schedules
schedules = OnCallSchedule.objects.related_to_user(user)
# check upcoming shifts
upcoming = []
for schedule in schedules:
current_shift, upcoming_shift = schedule.upcoming_shift_for_user(user, days=days)
if current_shift or upcoming_shift:
_, current_shifts, upcoming_shifts = schedule.shifts_for_user(user, datetime_start=now, days=days)
if current_shifts or upcoming_shifts:
upcoming.append(
{
"schedule_id": schedule.public_primary_key,
"schedule_name": schedule.name,
"is_oncall": current_shift is not None,
"current_shift": current_shift,
"next_shift": upcoming_shift,
"is_oncall": len(current_shifts) > 0,
"current_shift": current_shifts[0] if current_shifts else None,
"next_shift": upcoming_shifts[0] if upcoming_shifts else None,
}
)

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

@ -505,33 +505,32 @@ class OnCallSchedule(PolymorphicModel):
self.cached_ical_final_schedule = ical_data
self.save(update_fields=["cached_ical_final_schedule"])
def upcoming_shift_for_user(self, user, days=7):
def shifts_for_user(
self, user: User, datetime_start: datetime.datetime, days: int = 7
) -> typing.Tuple[ScheduleEvents, ScheduleEvents, ScheduleEvents]:
now = timezone.now()
# consider an extra day before to include events from UTC yesterday
datetime_start = now - datetime.timedelta(days=1)
datetime_end = datetime_start + datetime.timedelta(days=days)
current_shift = upcoming_shift = None
passed_shifts: ScheduleEvents = []
current_shifts: ScheduleEvents = []
upcoming_shifts: ScheduleEvents = []
if self.cached_ical_final_schedule is None:
# no final schedule info available
return None, None
return passed_shifts, current_shifts, upcoming_shifts
events = self.filter_events(datetime_start, datetime_end, all_day_datetime=True, from_cached_final=True)
for e in events:
if e["end"] < now:
# shift is finished, ignore
continue
users = {u["pk"] for u in e["users"]}
events.sort(key=lambda e: e["start"])
for event in events:
users = {u["pk"] for u in event["users"]}
if user.public_primary_key in users:
if e["start"] < now and e["end"] > now:
# shift is in progress
current_shift = e
continue
upcoming_shift = e
break
if event["end"] <= now:
passed_shifts.append(event)
elif event["start"] <= now < event["end"]:
current_shifts.append(event)
else:
upcoming_shifts.append(event)
return current_shift, upcoming_shift
return passed_shifts, current_shifts, upcoming_shifts
def quality_report(self, date: typing.Optional[datetime.datetime], days: typing.Optional[int]) -> QualityReport:
"""

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

@ -1570,49 +1570,6 @@ def test_user_related_schedules_only_username(
assert set(schedules) == {schedule1, schedule2}
@pytest.mark.django_db
def test_upcoming_shift_for_user(
make_organization,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization = make_organization()
admin = make_user_for_organization(organization)
other_user = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
shifts = (
# user, priority, start time (h), duration (seconds)
(admin, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
)
today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
for user, priority, start_h, duration in shifts:
data = {
"start": today + timezone.timedelta(hours=start_h),
"rotation_start": today + timezone.timedelta(hours=start_h),
"duration": timezone.timedelta(seconds=duration),
"priority_level": priority,
"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.refresh_ical_file()
schedule.refresh_ical_final_schedule()
current_shift, upcoming_shift = schedule.upcoming_shift_for_user(admin)
assert current_shift is not None and current_shift["start"] == on_call_shift.start
next_shift_start = on_call_shift.start + timezone.timedelta(days=1)
assert upcoming_shift is not None and upcoming_shift["start"] == next_shift_start
current_shift, upcoming_shift = schedule.upcoming_shift_for_user(other_user)
assert current_shift is None
assert upcoming_shift is None
@pytest.mark.django_db
def test_refresh_ical_final_schedule_ok(
make_organization,
@ -2552,3 +2509,174 @@ def test_filter_events_ical_duplicated_uid(make_organization, make_user_for_orga
assert len(events) == 2
assert events[0]["shift"]["pk"] == "eventuid@google.com_1"
assert events[1]["shift"]["pk"] == "eventuid@google.com_2_1970-01-01T01:00:00+01:00"
@pytest.mark.django_db
def test_shifts_for_user(
make_organization,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization = make_organization()
admin = make_user_for_organization(organization)
other_user = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
shifts = (
# user, priority, start time (h), duration (seconds)
(admin, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59
)
now = timezone.now()
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
for user, priority, start_h, duration in shifts:
data = {
"start": today + timezone.timedelta(hours=start_h),
"rotation_start": today + timezone.timedelta(hours=start_h),
"duration": timezone.timedelta(seconds=duration),
"priority_level": priority,
"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.refresh_ical_file()
schedule.refresh_ical_final_schedule()
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(admin, now)
assert len(passed_shifts) == 0
assert len(current_shifts) == 1
assert len(upcoming_shifts) == 7
current_shift = current_shifts[0]
assert current_shift is not None and current_shift["start"] == on_call_shift.start
next_shift_start = on_call_shift.start + timezone.timedelta(days=1)
upcoming_shift = upcoming_shifts[0]
assert upcoming_shift is not None and upcoming_shift["start"] == next_shift_start
for shifts in (passed_shifts, current_shifts, upcoming_shifts):
for shift in shifts:
users = {u["pk"] for u in shift["users"]}
assert admin.public_primary_key in users
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(other_user, now)
assert len(passed_shifts) == 0
assert len(current_shifts) == 0
assert len(upcoming_shifts) == 0
@pytest.mark.django_db
def test_shifts_for_user_only_two_users_with_shifts(
make_organization,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization = make_organization()
current_user = make_user_for_organization(organization)
user2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now()
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
start_date = today - timezone.timedelta(days=2)
days = 7
data = {
"start": now + timezone.timedelta(hours=1),
"rotation_start": now + timezone.timedelta(hours=1),
"duration": timezone.timedelta(hours=2),
"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([[current_user]])
# shift with another user
data = {
"start": start_date + timezone.timedelta(hours=10),
"rotation_start": start_date + timezone.timedelta(hours=10),
"duration": timezone.timedelta(hours=24),
"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([[user2]])
schedule.refresh_ical_final_schedule()
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(current_user, start_date, days)
assert len(passed_shifts) == 0
assert len(current_shifts) == 0
assert len(upcoming_shifts) == 5
for shift in upcoming_shifts:
users = {u["pk"] for u in shift["users"]}
assert current_user.public_primary_key in users
assert shift["start"] > now
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(user2, start_date, days)
assert len(passed_shifts) > 0
assert len(current_shifts) > 0
assert len(upcoming_shifts) > 0
for shift in passed_shifts:
users = {u["pk"] for u in shift["users"]}
assert user2.public_primary_key in users
assert shift["end"] < now
for shift in current_shifts:
users = {u["pk"] for u in shift["users"]}
assert user2.public_primary_key in users
assert shift["start"] <= now < shift["end"]
for shift in upcoming_shifts:
users = {u["pk"] for u in shift["users"]}
assert user2.public_primary_key in users
assert shift["start"] > now
@pytest.mark.django_db
def test_shifts_for_user_no_events(
make_organization,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization = make_organization()
current_user = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now()
today = now.replace(hour=0, minute=0, second=0, microsecond=0)
start_date = today - timezone.timedelta(days=2)
days = 7
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(current_user, start_date, days)
assert len(passed_shifts) == 0
assert len(current_shifts) == 0
assert len(upcoming_shifts) == 0
@pytest.mark.django_db
def test_shifts_for_user_without_final_ical(
make_organization,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization = make_organization()
user = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
start_date = today - timezone.timedelta(days=2)
days = 7
passed_shifts, current_shifts, upcoming_shifts = schedule.shifts_for_user(user, start_date, days)
assert len(passed_shifts) == 0
assert len(current_shifts) == 0
assert len(upcoming_shifts) == 0