centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oncall-engine/engine/apps/schedules/tests/test_custom_on_call_shift.py
Michael Derynck c65a3c9cea
Fix for user appearing in shift rotation when they should not be (#5064) 
# What this PR does
Fix calculation for interval when building ical events for schedules
with end dates and masked days. Add a test which makes it easier to test
different cases that can occur.

## Which issue(s) this PR closes

Related to https://github.com/grafana/support-escalations/issues/12388

<!--
*Note*: If you want the issue to be auto-closed once the PR is merged,
change "Related to" to "Closes" in the line above.
If you have more than one GitHub issue that this PR closes, be sure to
preface
each issue link with a [closing
keyword](https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/using-keywords-in-issues-and-pull-requests#linking-a-pull-request-to-an-issue).
This ensures that the issue(s) are auto-closed once the PR has been
merged.
-->

## 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] Added the relevant release notes label (see labels prefixed w/
`release:`). These labels dictate how your PR will
    show up in the autogenerated release notes.

---------

Co-authored-by: Matias Bordese <mbordese@gmail.com>
2024-10-03 20:04:55 +00:00

1902 lines
71 KiB
Python
Raw Permalink Blame History

import datetime
from calendar import monthrange
from collections import defaultdict
from unittest.mock import patch
from zoneinfo import ZoneInfo
import icalendar
import pytest
from django.utils import timezone
from recurring_ical_events import UnfoldableCalendar
from apps.schedules.ical_utils import list_users_to_notify_from_ical
from apps.schedules.models import CustomOnCallShift, OnCallSchedule, OnCallScheduleCalendar, OnCallScheduleWeb
from apps.schedules.tests.custom_shift_test_cases import CUSTOM_SHIFT_TEST_CASES
@pytest.mark.django_db
def test_get_on_call_users_from_single_event(make_organization_and_user, make_on_call_shift, make_schedule):
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
date = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_SINGLE_EVENT, **data
)
on_call_shift.users.add(user)
schedule.custom_on_call_shifts.add(on_call_shift)
# user is on-call
date = date + timezone.timedelta(minutes=5)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
@pytest.mark.django_db
def test_get_on_call_users_from_web_schedule_override(make_organization_and_user, make_on_call_shift, make_schedule):
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
date = timezone.now().replace(microsecond=0)
data = {
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"schedule": schedule,
}
on_call_shift = make_on_call_shift(organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **data)
on_call_shift.add_rolling_users([[user]])
# user is on-call
date = date + timezone.timedelta(minutes=5)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
@pytest.mark.django_db
def test_get_on_call_users_from_web_schedule_override_until(
make_organization_and_user, make_on_call_shift, make_schedule
):
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
date = timezone.now().replace(microsecond=0)
data = {
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"schedule": schedule,
"until": date + timezone.timedelta(seconds=3600),
}
on_call_shift = make_on_call_shift(organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **data)
on_call_shift.add_rolling_users([[user]])
# user is on-call
date = date + timezone.timedelta(minutes=5)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
# and the until is enforced
date = date + timezone.timedelta(hours=2)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_get_on_call_users_from_recurrent_event(make_organization_and_user, make_on_call_shift, make_schedule):
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
date = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 2,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_RECURRENT_EVENT, **data
)
on_call_shift.users.add(user)
schedule.custom_on_call_shifts.add(on_call_shift)
# user is on-call
date = date + timezone.timedelta(minutes=5)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
# user is not on-call according to event recurrence rules (interval = 2)
date = date + timezone.timedelta(days=1)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 0
# user is on-call again
date = date + timezone.timedelta(days=1)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
@pytest.mark.django_db
def test_get_on_call_users_from_web_schedule_recurrent_event(
make_organization_and_user, make_on_call_shift, make_schedule
):
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
date = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 2,
"schedule": schedule,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_RECURRENT_EVENT, **data
)
on_call_shift.users.add(user)
# user is on-call
date = date + timezone.timedelta(minutes=5)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
# user is not on-call according to event recurrence rules (interval = 2)
date = date + timezone.timedelta(days=1)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 0
# user is on-call again
date = date + timezone.timedelta(days=1)
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user in users_on_call
@pytest.mark.django_db
def test_get_on_call_users_from_rolling_users_event(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 2,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
schedule.custom_on_call_shifts.add(on_call_shift)
date = now + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date, date + timezone.timedelta(days=4)]
user_2_on_call_dates = [date + timezone.timedelta(days=2), date + timezone.timedelta(days=6)]
nobody_on_call_dates = [
date + timezone.timedelta(days=1),
date + timezone.timedelta(days=3),
date + timezone.timedelta(days=5),
]
for date in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for date in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for date in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_with_interval_hourly(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(hours=1),
"duration": timezone.timedelta(seconds=600),
"frequency": CustomOnCallShift.FREQUENCY_HOURLY,
"interval": 2,
"schedule": schedule,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date + timezone.timedelta(hours=4)]
user_2_on_call_dates = [date + timezone.timedelta(hours=2), date + timezone.timedelta(hours=6)]
nobody_on_call_dates = [
date,
date + timezone.timedelta(hours=1),
date + timezone.timedelta(hours=3),
date + timezone.timedelta(hours=5),
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for date in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, date)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_with_interval_daily(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 2,
"schedule": schedule,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date, date + timezone.timedelta(days=4)]
user_2_on_call_dates = [date + timezone.timedelta(days=2), date + timezone.timedelta(days=6)]
nobody_on_call_dates = [
date + timezone.timedelta(days=1),
date + timezone.timedelta(days=3),
date + timezone.timedelta(days=5),
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_daily_by_day(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
today_weekday = now.weekday()
delta_days = (0 - today_weekday) % 7 + (7 if today_weekday == 0 else 0)
next_week_monday = now + timezone.timedelta(days=delta_days)
# MO, WE, FR
weekdays = [0, 2, 4]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": next_week_monday,
"rotation_start": next_week_monday,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 1,
"by_day": by_day,
"schedule": schedule,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = next_week_monday + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date, date + timezone.timedelta(days=4), date + timezone.timedelta(days=9)]
user_2_on_call_dates = [date + timezone.timedelta(days=2), date + timezone.timedelta(days=7)]
nobody_on_call_dates = [
date + timezone.timedelta(days=1), # TU
date + timezone.timedelta(days=3), # TH
date + timezone.timedelta(days=5), # SAT
date + timezone.timedelta(days=6), # SUN
date + timezone.timedelta(days=8), # TU
date + timezone.timedelta(days=10), # TH
date + timezone.timedelta(days=12), # SAT
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_daily_by_day_off_start(make_organization_and_user, make_on_call_shift, make_schedule):
organization, user_1 = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
current_week_monday = now - timezone.timedelta(days=now.weekday())
# WE, FR
weekdays = [2, 4]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": current_week_monday,
"rotation_start": current_week_monday,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 1,
"by_day": by_day,
"schedule": schedule,
}
rolling_users = [[user_1]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = current_week_monday + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date + timezone.timedelta(days=2), date + timezone.timedelta(days=4)]
nobody_on_call_dates = [
date, # MO
date + timezone.timedelta(days=1), # TU
date + timezone.timedelta(days=3), # TH
date + timezone.timedelta(days=5), # SA
date + timezone.timedelta(days=6), # SU
date + timezone.timedelta(days=7), # MO
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_with_interval_daily_by_day(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
today_weekday = now.weekday()
delta_days = (0 - today_weekday) % 7 + (7 if today_weekday == 0 else 0)
next_week_monday = now + timezone.timedelta(days=delta_days)
# MO, WE, FR
weekdays = [0, 2, 4]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": next_week_monday,
"rotation_start": next_week_monday,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 2,
"by_day": by_day,
"schedule": schedule,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = next_week_monday + timezone.timedelta(minutes=5)
user_1_on_call_dates = [
date, # MO
date + timezone.timedelta(days=2), # WE
date + timezone.timedelta(days=9), # WE
date + timezone.timedelta(days=11), # FR
date + timezone.timedelta(days=18), # FR
date + timezone.timedelta(days=21), # MO
date + timezone.timedelta(days=28), # MO
date + timezone.timedelta(days=30), # WE
]
user_2_on_call_dates = [
date + timezone.timedelta(days=4), # FR
date + timezone.timedelta(days=7), # MO
date + timezone.timedelta(days=14), # MO
date + timezone.timedelta(days=16), # WE
date + timezone.timedelta(days=23), # WE
date + timezone.timedelta(days=25), # FR
date + timezone.timedelta(days=32), # FR
date + timezone.timedelta(days=35), # MO
]
nobody_on_call_dates = [
date + timezone.timedelta(days=1), # TU
date + timezone.timedelta(days=3), # TH
date + timezone.timedelta(days=5), # SAT
date + timezone.timedelta(days=6), # SUN
date + timezone.timedelta(days=8), # TU
date + timezone.timedelta(days=10), # TH
date + timezone.timedelta(days=12), # SAT
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_with_interval_weekly(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(hours=1),
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"interval": 2,
"week_start": now.weekday(),
"schedule": schedule,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
schedule.custom_on_call_shifts.add(on_call_shift)
date = now + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date + timezone.timedelta(days=28)]
user_2_on_call_dates = [date + timezone.timedelta(days=14), date + timezone.timedelta(days=42)]
nobody_on_call_dates = [
date,
date + timezone.timedelta(days=7),
date + timezone.timedelta(days=21),
date + timezone.timedelta(days=35),
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_event_with_interval_monthly(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
start_date = timezone.datetime(year=2022, month=10, day=1, hour=10, minute=30, tzinfo=datetime.timezone.utc)
days_for_next_month_1 = monthrange(2022, 10)[1]
days_for_next_month_2 = monthrange(2022, 11)[1] + days_for_next_month_1
days_for_next_month_3 = monthrange(2022, 12)[1] + days_for_next_month_2
days_for_next_month_4 = monthrange(2023, 1)[1] + days_for_next_month_3
data = {
"priority_level": 1,
"start": start_date,
"rotation_start": start_date + timezone.timedelta(hours=1),
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_MONTHLY,
"interval": 2,
"week_start": start_date.weekday(),
"schedule": schedule,
}
rolling_users = [[user_1], [user_2]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
schedule.custom_on_call_shifts.add(on_call_shift)
date = start_date + timezone.timedelta(minutes=5)
user_1_on_call_dates = [date + timezone.timedelta(days=days_for_next_month_4)]
user_2_on_call_dates = [date + timezone.timedelta(days=days_for_next_month_2)]
nobody_on_call_dates = [
date,
date + timezone.timedelta(days=days_for_next_month_1),
date + timezone.timedelta(days=days_for_next_month_3),
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_hourly(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(hours=2),
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_HOURLY,
"schedule": schedule,
"until": now + timezone.timedelta(hours=6, minutes=59),
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
# rotation starts from user_3, because user_1 and user_2 started earlier than rotation start date
user_1_on_call_dates = [date + timezone.timedelta(hours=3), date + timezone.timedelta(hours=6)]
user_2_on_call_dates = [date + timezone.timedelta(hours=4)]
user_3_on_call_dates = [date + timezone.timedelta(hours=2), date + timezone.timedelta(hours=5)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(hours=1), # less than rotation start
date + timezone.timedelta(hours=7), # higher than until
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_daily(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(days=1, hours=1),
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"schedule": schedule,
"until": now + timezone.timedelta(days=6, minutes=10),
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
# rotation starts from user_3, because user_1 and user_2 started earlier than rotation start date
user_1_on_call_dates = [date + timezone.timedelta(days=3), date + timezone.timedelta(days=6)]
user_2_on_call_dates = [date + timezone.timedelta(days=4)]
user_3_on_call_dates = [date + timezone.timedelta(days=2), date + timezone.timedelta(days=5)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(days=1), # less than rotation start
date + timezone.timedelta(days=7), # higher than until
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_weekly(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"week_start": now.weekday(),
"rotation_start": now + timezone.timedelta(days=7, hours=1),
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
"until": now + timezone.timedelta(days=42, minutes=10),
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
# rotation starts from user_3, because user_1 and user_2 started earlier than rotation start date
user_1_on_call_dates = [date + timezone.timedelta(days=21), date + timezone.timedelta(days=42)]
user_2_on_call_dates = [date + timezone.timedelta(days=28)]
user_3_on_call_dates = [date + timezone.timedelta(days=14), date + timezone.timedelta(days=35)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(days=7), # less than rotation start
date + timezone.timedelta(days=43), # higher than until
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_weekly_by_day_weekend(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
today_weekday = now.weekday()
delta_days = (0 - today_weekday) % 7 + (7 if today_weekday == 0 else 0)
next_week_monday = now + timezone.timedelta(days=delta_days)
# SAT, SUN
weekdays = [5, 6]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": now,
"week_start": 0,
"rotation_start": next_week_monday,
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
"until": next_week_monday + timezone.timedelta(days=30, minutes=1),
"by_day": by_day,
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
first_sat = next_week_monday + timezone.timedelta(days=5) + timezone.timedelta(minutes=5)
user_1_on_call_dates = [first_sat + timezone.timedelta(days=15)]
user_2_on_call_dates = [first_sat, first_sat + timezone.timedelta(days=22)]
user_3_on_call_dates = [first_sat + timezone.timedelta(days=7), first_sat + timezone.timedelta(days=8)]
nobody_on_call_dates = [
now, # less than rotation start
first_sat - timezone.timedelta(days=7), # before rotation start
first_sat + timezone.timedelta(days=9), # weekday value not in by_day
first_sat + timezone.timedelta(days=30), # higher than until
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_weekly_by_day(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
today_weekday = now.weekday()
weekdays = [(today_weekday + 1) % 7, (today_weekday + 3) % 7]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": now,
"week_start": today_weekday,
"rotation_start": now + timezone.timedelta(days=8, hours=1),
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
"until": now + timezone.timedelta(days=23, minutes=1),
"by_day": by_day,
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
# week 1: weekdays[0] - no (+1 day from start) ; weekdays[1] - no (+3 days from start) user_1
# week 2: weekdays[0] - no (+8 days from start) ; weekdays[1] - yes (+10 days from start) user_2
# week 3: weekdays[0] - yes (+15 days from start) ; weekdays[1] - yes (+17 days from start) user_3
# week 4: weekdays[0] - yes (+22 days from start) ; weekdays[1] - no (+24 days from start) user_1
user_1_on_call_dates = [date + timezone.timedelta(days=22)]
user_2_on_call_dates = [date + timezone.timedelta(days=10)]
user_3_on_call_dates = [date + timezone.timedelta(days=15), date + timezone.timedelta(days=17)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(days=1), # less than rotation start
date + timezone.timedelta(days=3), # less than rotation start
date + timezone.timedelta(days=8), # less than rotation start
date + timezone.timedelta(days=9), # weekday value not in by_day
date + timezone.timedelta(days=24), # higher than until
]
for dt in user_1_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_1 in users_on_call
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_monthly(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
start_date = timezone.datetime(year=2022, month=12, day=1, hour=10, minute=30, tzinfo=datetime.timezone.utc)
days_in_curr_month = monthrange(2022, 12)[1]
days_in_next_month = monthrange(2023, 1)[1]
data = {
"priority_level": 1,
"start": start_date,
"week_start": start_date.weekday(),
"rotation_start": start_date + timezone.timedelta(days=days_in_curr_month - 1, hours=1),
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_MONTHLY,
"schedule": schedule,
"until": start_date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 10, minutes=1),
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = start_date + timezone.timedelta(minutes=5)
# rotation starts from user_2, because user_1 started earlier than rotation start date
user_2_on_call_dates = [date + timezone.timedelta(days=days_in_curr_month)]
user_3_on_call_dates = [date + timezone.timedelta(days=days_in_curr_month + days_in_next_month)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(days=days_in_curr_month - 1), # less than rotation start
date + timezone.timedelta(days=days_in_curr_month + 1), # higher than event end
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 2), # higher than event end
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 11), # higher than until
]
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_monthly_by_monthday(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
start_date = timezone.datetime(year=2022, month=12, day=1, hour=10, minute=30, tzinfo=datetime.timezone.utc)
days_in_curr_month = monthrange(2022, 12)[1]
days_in_next_month = monthrange(2023, 1)[1]
data = {
"priority_level": 1,
"start": start_date,
"week_start": start_date.weekday(),
"rotation_start": start_date + timezone.timedelta(days=days_in_curr_month - 1, hours=1),
"duration": timezone.timedelta(seconds=1800),
"frequency": CustomOnCallShift.FREQUENCY_MONTHLY,
"schedule": schedule,
"until": start_date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 10, minutes=1),
"by_monthday": [i for i in range(1, 5)],
}
rolling_users = [[user_1], [user_2], [user_3]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = start_date + timezone.timedelta(minutes=5)
# rotation starts from user_2, because user_1 started earlier than rotation start date
user_2_on_call_dates = [
date + timezone.timedelta(days=days_in_curr_month),
date + timezone.timedelta(days=days_in_curr_month + 1),
date + timezone.timedelta(days=days_in_curr_month + 2),
date + timezone.timedelta(days=days_in_curr_month + 3),
]
user_3_on_call_dates = [
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month),
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 1),
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 2),
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 3),
]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(days=3), # less than rotation start
date + timezone.timedelta(days=days_in_curr_month + 4), # out of by_monthday range
date + timezone.timedelta(days=days_in_curr_month + 6), # out of by_monthday range
date + timezone.timedelta(days=days_in_curr_month + 10), # out of by_monthday range
date + timezone.timedelta(days=days_in_curr_month + days_in_next_month + 11), # higher than until
]
for dt in user_2_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_2 in users_on_call
for dt in user_3_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user_3 in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_get_oncall_users_with_respect_to_rotation_start_and_until_dates_hourly(
make_organization_and_user,
make_on_call_shift,
make_schedule,
):
"""Test calculation start and end event dates for one event with respect to rotation start and until"""
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(minutes=10),
"duration": timezone.timedelta(hours=1),
"frequency": CustomOnCallShift.FREQUENCY_HOURLY,
"schedule": schedule,
"until": now + timezone.timedelta(minutes=40),
"source": CustomOnCallShift.SOURCE_WEB,
}
rolling_users = [[user]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=2)
user_on_call_dates = [date + timezone.timedelta(minutes=10), date + timezone.timedelta(minutes=35)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(minutes=5), # less than rotation start
date + timezone.timedelta(minutes=40), # higher than until
]
for dt in user_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_get_oncall_users_with_respect_to_rotation_start_and_until_dates_daily(
make_organization_and_user,
make_on_call_shift,
make_schedule,
):
"""Test calculation start and end event dates for one event with respect to rotation start and until"""
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(hours=5),
"duration": timezone.timedelta(days=1),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"schedule": schedule,
"until": now + timezone.timedelta(hours=15),
"source": CustomOnCallShift.SOURCE_WEB,
}
rolling_users = [[user]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
user_on_call_dates = [date + timezone.timedelta(hours=5), date + timezone.timedelta(hours=10)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(hours=4), # less than rotation start
date + timezone.timedelta(hours=15), # higher than until
]
for dt in user_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_get_oncall_users_with_respect_to_rotation_start_and_until_dates_weekly(
make_organization_and_user,
make_on_call_shift,
make_schedule,
):
"""Test calculation start and end event dates for one event with respect to rotation start and until"""
organization, user = make_organization_and_user()
# simple weekly event
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(days=1),
"duration": timezone.timedelta(days=7),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
"until": now + timezone.timedelta(days=6),
"week_start": now.weekday(),
"source": CustomOnCallShift.SOURCE_WEB,
}
rolling_users = [[user]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
user_on_call_dates = [date + timezone.timedelta(days=1), date + timezone.timedelta(days=5)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(hours=23), # less than rotation start
date + timezone.timedelta(days=6), # higher than until
]
for dt in user_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
# weekly event with by_day
schedule_2 = make_schedule(organization, schedule_class=OnCallScheduleWeb)
today_weekday = now.weekday()
weekdays = [today_weekday, (today_weekday + 1) % 7, (today_weekday + 2) % 7, (today_weekday + 5) % 7]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(days=1),
"duration": timezone.timedelta(hours=12),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule_2,
"until": now + timezone.timedelta(days=4, hours=23),
"week_start": today_weekday,
"by_day": by_day,
"source": CustomOnCallShift.SOURCE_WEB,
}
on_call_shift_2 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift_2.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
user_on_call_dates = [date + timezone.timedelta(days=1), date + timezone.timedelta(days=2)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(hours=23), # less than rotation start
date + timezone.timedelta(days=3), # out of by_day
date + timezone.timedelta(days=4), # out of by_day
date + timezone.timedelta(days=5), # higher than until
]
for dt in user_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule_2, dt)
assert len(users_on_call) == 1
assert user in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule_2, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_get_oncall_users_with_respect_to_rotation_start_and_until_dates_monthly(
make_organization_and_user,
make_on_call_shift,
make_schedule,
):
"""Test calculation start and end event dates for one event with respect to rotation start and until"""
organization, user = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(microsecond=0)
data = {
"priority_level": 1,
"start": now,
"rotation_start": now + timezone.timedelta(days=5),
"duration": timezone.timedelta(days=30),
"frequency": CustomOnCallShift.FREQUENCY_MONTHLY,
"schedule": schedule,
"until": now + timezone.timedelta(days=15),
"source": CustomOnCallShift.SOURCE_WEB,
}
rolling_users = [[user]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
date = now + timezone.timedelta(minutes=5)
user_on_call_dates = [date + timezone.timedelta(days=5), date + timezone.timedelta(days=10)]
nobody_on_call_dates = [
date, # less than rotation start
date + timezone.timedelta(days=4), # less than rotation start
date + timezone.timedelta(days=15), # higher than until
]
for dt in user_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 1
assert user in users_on_call
for dt in nobody_on_call_dates:
users_on_call = list_users_to_notify_from_ical(schedule, dt)
assert len(users_on_call) == 0
@pytest.mark.django_db
def test_get_oncall_users_for_empty_schedule(
make_organization,
make_schedule,
):
organization = make_organization()
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
schedules = OnCallSchedule.objects.filter(pk=schedule.pk)
assert list(schedules.get_oncall_users()[schedule]) == []
@pytest.mark.django_db
def test_get_oncall_users_for_multiple_schedules(
make_organization,
make_user_for_organization,
make_on_call_shift,
make_schedule,
):
organization = make_organization()
user_1 = make_user_for_organization(organization)
user_2 = make_user_for_organization(organization)
user_3 = make_user_for_organization(organization)
schedule_1 = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
schedule_2 = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
now = timezone.now().replace(microsecond=0)
on_call_shift_1 = make_on_call_shift(
organization=organization,
shift_type=CustomOnCallShift.TYPE_SINGLE_EVENT,
priority_level=1,
start=now,
rotation_start=now,
duration=timezone.timedelta(minutes=30),
)
on_call_shift_2 = make_on_call_shift(
organization=organization,
shift_type=CustomOnCallShift.TYPE_SINGLE_EVENT,
priority_level=1,
start=now,
rotation_start=now,
duration=timezone.timedelta(minutes=10),
)
on_call_shift_3 = make_on_call_shift(
organization=organization,
shift_type=CustomOnCallShift.TYPE_SINGLE_EVENT,
priority_level=1,
start=now + timezone.timedelta(minutes=10),
rotation_start=now + timezone.timedelta(minutes=10),
duration=timezone.timedelta(minutes=30),
)
on_call_shift_1.users.add(user_1)
on_call_shift_1.users.add(user_2)
on_call_shift_2.users.add(user_2)
on_call_shift_3.users.add(user_3)
schedule_1.custom_on_call_shifts.add(on_call_shift_1)
schedule_2.custom_on_call_shifts.add(on_call_shift_2)
schedule_2.custom_on_call_shifts.add(on_call_shift_3)
schedules = OnCallSchedule.objects.filter(pk__in=[schedule_1.pk, schedule_2.pk])
def _extract_oncall_users_from_schedules(schedules):
return set(user for schedule in schedules.values() for user in schedule)
expected = _extract_oncall_users_from_schedules(
schedules.get_oncall_users(events_datetime=now + timezone.timedelta(seconds=1))
)
assert expected == {user_1, user_2}
expected = _extract_oncall_users_from_schedules(
schedules.get_oncall_users(events_datetime=now + timezone.timedelta(minutes=10, seconds=1))
)
assert expected == {user_1, user_2, user_3}
assert _extract_oncall_users_from_schedules(
schedules.get_oncall_users(events_datetime=now + timezone.timedelta(minutes=30, seconds=1))
) == {user_3}
assert (
_extract_oncall_users_from_schedules(
schedules.get_oncall_users(events_datetime=now + timezone.timedelta(minutes=40, seconds=1))
)
== set()
)
@pytest.mark.django_db
def test_get_oncall_users_for_multiple_schedules_emails_case_insensitive(
get_ical,
make_organization,
make_user_for_organization,
make_on_call_shift,
make_schedule,
):
"""
Test that emails are case insensitive when matching users to on-call shifts.
https://github.com/grafana/oncall/issues/1296
"""
organization = make_organization()
# user's email case is the opposite of the one in the ICal file below (Test@TEST.test)
user = make_user_for_organization(organization, email="tEST@test.TEST")
schedule = make_schedule(organization, schedule_class=OnCallScheduleCalendar)
# Load ICal file with an event for user with email Test@TEST.test for 6 February 2023, 11:00 UTC - 12:00 UTC
calendar = get_ical("override_email_case_sensitivity.ics")
schedule.cached_ical_file_overrides = calendar.to_ical().decode()
schedule.save(update_fields=["cached_ical_file_overrides"])
# Get on-call users for 6 February 2023 11:30 UTC
events_datetime = timezone.datetime(2023, 2, 6, 11, 30, tzinfo=datetime.timezone.utc)
schedules = OnCallSchedule.objects.filter(pk=schedule.pk)
oncall_users = schedules.get_oncall_users(events_datetime=events_datetime)
assert len(oncall_users) == 1
assert list(oncall_users[schedule]) == [user]
@pytest.mark.django_db
def test_shift_convert_to_ical(make_organization_and_user, make_on_call_shift):
organization, user = make_organization_and_user()
date = timezone.now().replace(microsecond=0)
until = date + timezone.timedelta(days=30)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_HOURLY,
"interval": 1,
"until": until,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_RECURRENT_EVENT, **data
)
on_call_shift.users.add(user)
ical_data = on_call_shift.convert_to_ical()
ical_rrule_until = on_call_shift.until.strftime("%Y%m%dT%H%M%S")
expected_rrule = f"RRULE:FREQ=HOURLY;UNTIL={ical_rrule_until}Z;INTERVAL=1;WKST=SU"
assert expected_rrule in ical_data
@pytest.mark.django_db
def test_rolling_users_shift_convert_to_ical(
make_organization_and_user,
make_user_for_organization,
make_on_call_shift,
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
date = timezone.now().replace(microsecond=0)
until = date + timezone.timedelta(days=30)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_HOURLY,
"interval": 2,
"until": until,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
rolling_users = [[user_1], [user_2]]
on_call_shift.add_rolling_users(rolling_users)
ical_data = on_call_shift.convert_to_ical()
ical_rrule_until = on_call_shift.until.strftime("%Y%m%dT%H%M%S")
expected_rrule = f"RRULE:FREQ=HOURLY;UNTIL={ical_rrule_until}Z;INTERVAL=4;WKST=SU"
assert on_call_shift.event_interval == len(rolling_users) * data["interval"]
assert expected_rrule in ical_data
@pytest.mark.django_db
def test_rolling_users_event_daily_by_day_start_none_convert_to_ical(
make_organization_and_user, make_user_for_organization, make_on_call_shift, make_schedule
):
organization, user_1 = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
today_weekday = now.weekday()
delta_days = (0 - today_weekday) % 7 + (7 if today_weekday == 0 else 0)
next_week_monday = now + timezone.timedelta(days=delta_days)
# MO
weekdays = [0]
by_day = [CustomOnCallShift.ICAL_WEEKDAY_MAP[day] for day in weekdays]
data = {
"priority_level": 1,
"start": now + timezone.timedelta(hours=12),
"rotation_start": next_week_monday,
"duration": timezone.timedelta(seconds=3600),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"interval": 1,
"by_day": by_day,
"schedule": schedule,
"until": now,
}
rolling_users = [[user_1]]
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users(rolling_users)
ical_data = on_call_shift.convert_to_ical()
# empty result since there is no event in the defined time range
assert ical_data == ""
@pytest.mark.django_db
def test_etc_utc_timezone_convert_to_ical(
make_organization_and_user,
make_user_for_organization,
make_on_call_shift,
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
date = timezone.now().replace(microsecond=0)
until = date + timezone.timedelta(days=30)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_HOURLY,
"interval": 2,
"until": until,
"time_zone": "Etc/UTC",
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
rolling_users = [[user_1], [user_2]]
on_call_shift.add_rolling_users(rolling_users)
ical_data = on_call_shift.convert_to_ical()
ical_rrule_until = on_call_shift.until.strftime("%Y%m%dT%H%M%S")
expected_rrule = f"RRULE:FREQ=HOURLY;UNTIL={ical_rrule_until}Z;INTERVAL=4;WKST=SU"
assert on_call_shift.event_interval == len(rolling_users) * data["interval"]
assert expected_rrule in ical_data
@pytest.mark.django_db
@pytest.mark.parametrize(
"starting_day,force,deleted",
[
(-1, False, False),
(-1, True, True),
(1, False, True),
],
)
def test_delete_shift(make_organization_and_user, make_schedule, make_on_call_shift, starting_day, force, deleted):
organization, user_1 = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
start_date = (timezone.now() + timezone.timedelta(days=starting_day)).replace(microsecond=0)
data = {
"priority_level": 1,
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(seconds=10800),
"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.delete(force=force)
if deleted:
with pytest.raises(CustomOnCallShift.DoesNotExist):
on_call_shift.refresh_from_db()
else:
on_call_shift.refresh_from_db()
assert on_call_shift.until is not None
@pytest.mark.django_db
def test_delete_shift_updates_linked_shift(
make_organization_and_user, make_user_for_organization, make_schedule, make_on_call_shift
):
organization, user_1 = make_organization_and_user()
other_user = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
start_date = (timezone.now() - timezone.timedelta(days=7)).replace(microsecond=0)
updated_shifts = (
(start_date, 3600, user_1),
(start_date, 3600 * 2, user_1),
(start_date, 3600, other_user),
)
shifts = []
previous_shift = None
for start_date, duration, user in reversed(updated_shifts):
data = {
"priority_level": 1,
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(seconds=duration),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"schedule": schedule,
"updated_shift": previous_shift,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user]])
previous_shift = on_call_shift
shifts.append(on_call_shift)
last_shift, intermediate_shift, first_shift = shifts
intermediate_shift.delete(force=True)
# deleted shift does not exist
with pytest.raises(CustomOnCallShift.DoesNotExist):
intermediate_shift.refresh_from_db()
# first shift now is linked to the following one
first_shift.refresh_from_db()
assert first_shift.updated_shift == last_shift
@pytest.mark.django_db
@pytest.mark.parametrize(
"starting_day,duration,deleted",
[
(-1, 2, False),
(-2, 1, False),
(1, 1, True),
],
)
def test_delete_override(
make_organization_and_user, make_schedule, make_on_call_shift, starting_day, duration, deleted
):
organization, _ = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
start_date = (timezone.now() + timezone.timedelta(days=starting_day)).replace(microsecond=0)
data = {
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(days=duration),
"schedule": schedule,
}
on_call_shift = make_on_call_shift(organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **data)
original_duration = on_call_shift.duration
on_call_shift.delete()
if deleted:
with pytest.raises(CustomOnCallShift.DoesNotExist):
on_call_shift.refresh_from_db()
else:
on_call_shift.refresh_from_db()
assert on_call_shift.until is not None
assert (
on_call_shift.duration == original_duration
if (starting_day + duration) < 0
else on_call_shift.duration < original_duration
)
@pytest.mark.django_db
def test_until_rrule_must_be_utc(
make_organization_and_user,
make_user_for_organization,
make_schedule,
make_on_call_shift,
):
organization, user_1 = make_organization_and_user()
user_2 = make_user_for_organization(organization)
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb, time_zone="Europe/Warsaw")
date = timezone.now().replace(microsecond=0) - timezone.timedelta(days=7)
data = {
"priority_level": 1,
"start": date,
"rotation_start": date,
"duration": timezone.timedelta(seconds=10800),
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"interval": 2,
"time_zone": "Europe/Warsaw",
"schedule": schedule,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
rolling_users = [[user_1], [user_2]]
on_call_shift.add_rolling_users(rolling_users)
# finish the rotation, will set until value
on_call_shift.delete()
on_call_shift.refresh_from_db()
assert on_call_shift.until.tzname() == "UTC"
ical_data = on_call_shift.convert_to_ical()
ical_rrule_until = on_call_shift.until.strftime("%Y%m%dT%H%M%S")
expected_rrule = f"RRULE:FREQ=WEEKLY;UNTIL={ical_rrule_until}Z;INTERVAL=4;WKST=SU"
assert expected_rrule in ical_data
@pytest.mark.django_db
def test_week_start_changed_daily_shift(
make_organization_and_user,
make_schedule,
make_on_call_shift,
):
organization, user_1 = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb, time_zone="Europe/Warsaw")
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
today_weekday = now.weekday()
last_sunday = now - timezone.timedelta(days=7 + (today_weekday + 1) % 7)
last_saturday = last_sunday - timezone.timedelta(days=1)
# set week start to Sunday, so first event should be on last_sunday itself
data = {
"priority_level": 1,
"start": last_saturday,
"rotation_start": last_sunday,
"duration": timezone.timedelta(seconds=3600),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"by_day": ["MO", "SU"],
"week_start": 5, # SU
"interval": 1,
"schedule": schedule,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
rolling_users = [[user_1]]
on_call_shift.add_rolling_users(rolling_users)
ical_data = on_call_shift.convert_to_ical()
expected_start = "DTSTART:{}T000000Z".format(last_sunday.strftime("%Y%m%d"))
assert expected_start in ical_data
@pytest.mark.django_db
def test_week_start_changed_daily_shift_until(
make_organization_and_user,
make_schedule,
make_on_call_shift,
):
organization, user_1 = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb, time_zone="Europe/Warsaw")
now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
today_weekday = now.weekday()
last_sunday = now - timezone.timedelta(days=7 + (today_weekday + 1) % 7)
last_saturday = last_sunday - timezone.timedelta(days=1)
thursday = last_sunday + timezone.timedelta(days=4)
data = {
"priority_level": 1,
"start": last_saturday,
"rotation_start": last_sunday,
"duration": timezone.timedelta(seconds=3600),
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"by_day": ["MO", "SU"],
"week_start": 5, # SU
"interval": 1,
"until": thursday,
"schedule": schedule,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
rolling_users = [[user_1]]
on_call_shift.add_rolling_users(rolling_users)
ical_data = on_call_shift.convert_to_ical()
# setting UNTIL to Thursday was generating extra events for current week Wednesday and Thursday
unexpected_by_days = ("BYDAY=WE", "BYDAY=TH")
for unexpected in unexpected_by_days:
assert unexpected not in ical_data
@pytest.mark.django_db
@pytest.mark.parametrize(
"shift_type",
[
CustomOnCallShift.TYPE_OVERRIDE,
CustomOnCallShift.TYPE_ROLLING_USERS_EVENT,
],
)
def test_refresh_schedule(make_organization_and_user, make_schedule, make_on_call_shift, shift_type):
organization, _ = make_organization_and_user()
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
start_date = timezone.now()
frequency = CustomOnCallShift.FREQUENCY_DAILY if shift_type == CustomOnCallShift.TYPE_ROLLING_USERS_EVENT else None
data = {
"priority_level": 1,
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(seconds=10800),
"frequency": frequency,
"schedule": schedule,
}
on_call_shift = make_on_call_shift(organization=organization, shift_type=shift_type, **data)
assert schedule.cached_ical_file_primary is None
assert schedule.cached_ical_file_overrides is None
with patch("apps.schedules.models.custom_on_call_shift.refresh_ical_final_schedule") as mock_refresh_final:
on_call_shift.refresh_schedule()
assert mock_refresh_final.apply_async.called
assert schedule.cached_ical_file_primary is not None
assert schedule.cached_ical_file_overrides is not None
@pytest.mark.parametrize(
"users_per_group, shift_start, day_mask, total_days, frequency, interval, expected_result", CUSTOM_SHIFT_TEST_CASES
)
@pytest.mark.django_db
def test_ical_shift_generation(
make_organization,
make_user_for_organization,
make_schedule,
make_on_call_shift,
users_per_group,
shift_start,
day_mask,
total_days,
frequency,
interval,
expected_result,
):
organization = make_organization()
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_web_schedule",
)
total_users = sum(users_per_group)
users = [make_user_for_organization(organization, username=chr(i + 64)) for i in range(1, total_users + 1)]
start = datetime.datetime.strptime(shift_start, "%Y-%m-%d").replace(tzinfo=ZoneInfo("UTC"))
data = {
"start": start,
"rotation_start": start,
"until": start + timezone.timedelta(days=total_days),
"duration": timezone.timedelta(hours=12),
"frequency": frequency,
"by_day": day_mask,
"schedule": schedule,
"interval": interval,
"priority_level": 1,
"week_start": CustomOnCallShift.MONDAY,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
rolling_users = []
start_user = 0
for count in users_per_group:
end = start_user + count
rolling_users.append(users[start_user:end])
start_user = end
on_call_shift.add_rolling_users(rolling_users)
query_start = start
query_end = data["until"]
calendar = icalendar.Calendar.from_ical(schedule._ical_file_primary)
events = UnfoldableCalendar(calendar).between(query_start, query_end)
day_events = defaultdict(str)
for event in events:
event_start = event["DTSTART"].dt
event_summary = event["SUMMARY"].strip()[-1]
event_date = event_start.date().strftime("%Y-%m-%d")
day_events[event_date] += event_summary
for k, v in day_events.items():
day_events[k] = "".join(sorted(v))
assert day_events == expected_result
Reference in a new issue View git blame Copy permalink