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