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

Merge pull request #432 from grafana/fix-getting-rotation-start-date

Browse source 
Fix events start date calculation
This commit is contained in:
Yulya Artyukhina 2022-09-02 17:13:38 +03:00 • committed by GitHub
commit 4ada990ba4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 101 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
engine/apps/schedules/models/custom_on_call_shift.py
View file

@ -381,6 +381,23 @@ class CustomOnCallShift(models.Model):
days_for_next_event += next_month_days
next_event_start = current_event_start + timezone.timedelta(days=days_for_next_event)
end_date = None
# get the period for calculating the current rotation end date for long events with frequency weekly and monthly
if self.frequency == CustomOnCallShift.FREQUENCY_WEEKLY:
DAYS_IN_A_WEEK = 7
days_diff = 0
# get the last day of the week with respect to the week_start
if next_event_start.weekday() != self.week_start:
days_diff = DAYS_IN_A_WEEK + next_event_start.weekday() - self.week_start
days_diff %= DAYS_IN_A_WEEK
end_date = next_event_start + timezone.timedelta(days=DAYS_IN_A_WEEK - days_diff - ONE_DAY)
elif self.frequency == CustomOnCallShift.FREQUENCY_MONTHLY:
# get the last day of the month
current_day_number = next_event_start.day
number_of_days = monthrange(next_event_start.year, next_event_start.month)[1]
days_diff = number_of_days - current_day_number
end_date = next_event_start + timezone.timedelta(days=days_diff)
next_event = None
# repetitions generate the next event shift according with the recurrence rules
repetitions = UnfoldableCalendar(current_event).RepeatedEvent(
@ -388,12 +405,21 @@ class CustomOnCallShift(models.Model):
)
ical_iter = repetitions.__iter__()
for event in ical_iter:
if event.start >= next_event_start:
next_event = event
break
next_event_dt = next_event.start if next_event is not None else None
if end_date: # end_date exists for long events with frequency weekly and monthly
if end_date >= event.start >= next_event_start:
if event.start >= self.rotation_start:
next_event = event
break
elif end_date < event.start:
break
else:
if event.start >= next_event_start:
next_event = event
break
if self.until and next_event_dt and next_event_dt > self.until:
next_event_dt = next_event.start if next_event is not None else next_event_start
if self.until and next_event_dt > self.until:
return
return next_event_dt

71
engine/apps/schedules/tests/test_custom_on_call_shift.py
View file

@ -575,7 +575,7 @@ def test_rolling_users_with_diff_start_and_rotation_start_weekly(
@pytest.mark.django_db
def test_rolling_users_with_diff_start_and_rotation_start_weekly_by_day(
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()
@ -640,6 +640,75 @@ def test_rolling_users_with_diff_start_and_rotation_start_weekly_by_day(
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