centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oncall-engine/engine/apps/alerts/tests/test_notify_ical_schedule_shift.py
Joey Orlando 4a5c4263e0
feat: convert schedule.channel (char field) to schedule.slack_channel (foreign key) (#5199) 
# What this PR does

`OnCallSchedule` equivalent of
https://github.com/grafana/oncall/pull/5191.

**NOTE**: merge after https://github.com/grafana/oncall/pull/5224 (so
that I can use some of the new serializer fields defined in there)

### Migration
```bash
Running migrations:                                                                                                                                                                                                │
│ source=engine:app google_trace_id=none logger=apps.schedules.migrations.0019_auto_20241021_1735 Starting migration to populate slack_channel field.                                                                │
│ source=engine:app google_trace_id=none logger=apps.schedules.migrations.0019_auto_20241021_1735 Total schedules to process: 1                                                                                      │
│ source=engine:app google_trace_id=none logger=apps.schedules.migrations.0019_auto_20241021_1735 Schedule 26 updated with SlackChannel 2 (slack_id: C043LL6RTS7).                                                   │
│ source=engine:app google_trace_id=none logger=apps.schedules.migrations.0019_auto_20241021_1735 Bulk updated 1 OnCallSchedules with their Slack channel.                                                           │
│ source=engine:app google_trace_id=none logger=apps.schedules.migrations.0019_auto_20241021_1735 Finished migration. Total schedules processed: 1. Schedules updated: 1. Missing SlackChannels: 0.                  │
│   Applying schedules.0019_auto_20241021_1735... OK
```

### Tested Public API
```txt
POST {{oncall_host}}/api/v1/schedules/
Authorization: {{oncall_api_key}}
Content-Type: application/json

{
    "name": "Demo testy testy2",
    "type": "web",
    "time_zone": "America/Los_Angeles",
    "slack": {
        "channel_id": "C05PPLYN1U1"
    }
}

HTTP/1.1 201 Created
Content-Type: application/json
Vary: Accept, Origin
Allow: GET, POST, HEAD, OPTIONS
X-Frame-Options: DENY
Content-Length: 198
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{
  "id": "SBBN73UTUTVCE",
  "team_id": null,
  "name": "Demo testy testy2",
  "time_zone": "America/Los_Angeles",
  "on_call_now": [],
  "shifts": [],
  "slack": {
    "channel_id": "C05PPLYN1U1",
    "user_group_id": null
  },
  "type": "web"
}
```

### Tested via UI (eg; internal API)

https://www.loom.com/share/e66bf3468b144dd782da5eb6e0bfd0af

## 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.
2024-11-04 14:27:21 -05:00

971 lines
31 KiB
Python
Raw Permalink Blame History

import datetime
import json
import textwrap
from unittest.mock import Mock, patch
import pytest
from django.utils import timezone
from apps.alerts.tasks.notify_ical_schedule_shift import (
MIN_DAYS_TO_LOOKUP_FOR_THE_END_OF_EVENT,
notify_ical_schedule_shift,
)
from apps.schedules.models import CustomOnCallShift, OnCallScheduleCalendar, OnCallScheduleICal, OnCallScheduleWeb
ICAL_DATA = """
BEGIN:VCALENDAR
PRODID:-//Google Inc//Google Calendar 70.9054//EN
VERSION:2.0
CALSCALE:GREGORIAN
METHOD:PUBLISH
BEGIN:VEVENT
DTSTART;VALUE=DATE:20211005
DTEND;VALUE=DATE:20211012
RRULE:FREQ=WEEKLY;WKST=SU;INTERVAL=7;BYDAY=WE
DTSTAMP:20210930T125523Z
UID:id1@google.com
CREATED:20210928T202349Z
DESCRIPTION:
LAST-MODIFIED:20210929T204751Z
LOCATION:
SEQUENCE:1
STATUS:CONFIRMED
SUMMARY:user1
TRANSP:TRANSPARENT
END:VEVENT
BEGIN:VEVENT
DTSTART;VALUE=DATE:20210928
DTEND;VALUE=DATE:20211005
RRULE:FREQ=WEEKLY;WKST=SU;INTERVAL=7;BYDAY=WE
DTSTAMP:20210930T125523Z
UID:id2@google.com
CREATED:20210928T202331Z
DESCRIPTION:
LAST-MODIFIED:20210929T204744Z
LOCATION:
SEQUENCE:2
STATUS:CONFIRMED
SUMMARY:user2
TRANSP:TRANSPARENT
END:VEVENT
END:VCALENDAR
"""
@pytest.mark.django_db
def test_current_overrides_ical_schedule_is_none(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_schedule,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
ical_schedule = make_schedule(
organization,
schedule_class=OnCallScheduleICal,
name="test_ical_schedule",
slack_channel=slack_channel,
ical_url_primary="url",
prev_ical_file_primary=ICAL_DATA,
cached_ical_file_primary=ICAL_DATA,
prev_ical_file_overrides=ICAL_DATA,
cached_ical_file_overrides=None,
)
# this should not raise
notify_ical_schedule_shift(ical_schedule.pk)
@pytest.mark.django_db
def test_next_shift_notification_long_shifts(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_schedule,
make_user,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
make_user(organization=organization, username="user1")
make_user(organization=organization, username="user2")
ical_schedule = make_schedule(
organization,
schedule_class=OnCallScheduleICal,
name="test_ical_schedule",
slack_channel=slack_channel,
ical_url_primary="url",
prev_ical_file_primary=ICAL_DATA,
cached_ical_file_primary=ICAL_DATA,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
with patch("apps.alerts.tasks.notify_ical_schedule_shift.datetime", Mock(wraps=datetime)) as mock_datetime:
mock_datetime.datetime.now.return_value = datetime.datetime(2021, 9, 29, 12, 0, tzinfo=datetime.timezone.utc)
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(ical_schedule.pk)
slack_blocks = mock_slack_api_call.call_args_list[0][1]["blocks"]
notification = slack_blocks[1]["text"]["text"]
assert "*New on-call shift*\nuser2" in notification
notification = slack_blocks[2]["text"]["text"]
assert "*Next on-call shift*\nuser1" in notification
@pytest.mark.django_db
def test_overrides_changes_no_current_no_triggering_notification(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
ical_before = textwrap.dedent(
"""
BEGIN:VCALENDAR
PRODID:-//Google Inc//Google Calendar 70.9054//EN
VERSION:2.0
CALSCALE:GREGORIAN
METHOD:PUBLISH
BEGIN:VEVENT
DTSTART:20230101T020000
DTEND:20230101T170000
DTSTAMP:20230101T000000
UID:id1@google.com
CREATED:20230101T000000
DESCRIPTION:
LAST-MODIFIED:20230101T000000
LOCATION:
SEQUENCE:1
STATUS:CONFIRMED
SUMMARY:user1
TRANSP:TRANSPARENT
END:VEVENT
END:VCALENDAR"""
)
# event outside current time is changed
ical_after = textwrap.dedent(
"""
BEGIN:VCALENDAR
PRODID:-//Google Inc//Google Calendar 70.9054//EN
VERSION:2.0
CALSCALE:GREGORIAN
METHOD:PUBLISH
BEGIN:VEVENT
DTSTART:20230101T020000
DTEND:20230101T210000
DTSTAMP:20230101T000000
UID:id1@google.com
CREATED:20230101T000000
DESCRIPTION:
LAST-MODIFIED:20230101T000000
LOCATION:
SEQUENCE:2
STATUS:CONFIRMED
SUMMARY:user1
TRANSP:TRANSPARENT
END:VEVENT
END:VCALENDAR"""
)
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=ical_before,
)
now = timezone.now().replace(microsecond=0)
start_date = now - timezone.timedelta(days=7, minutes=1)
data = {
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
on_call_shift.schedules.add(schedule)
# setup current shifts before checking/triggering for notifications
current_shifts = schedule.final_events(now, now, False, False)
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
schedule.cached_ical_file_overrides = ical_after
schedule.prev_ical_file_overrides = ical_before
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert not mock_slack_api_call.called
@pytest.mark.django_db
def test_no_changes_no_triggering_notification(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date = now - timezone.timedelta(days=7, minutes=1)
data = {
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
on_call_shift.schedules.add(schedule)
# setup current shifts before checking/triggering for notifications
current_shifts = schedule.final_events(now, now, False, False)
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert not mock_slack_api_call.called
@pytest.mark.django_db
def test_current_shift_changes_trigger_notification(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date = now - datetime.timedelta(days=7)
data = {
"start": start_date,
"rotation_start": start_date,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
on_call_shift.schedules.add(schedule)
schedule.refresh_ical_file()
# setup empty current shifts before checking/triggering for notifications
schedule.current_shifts = json.dumps({}, default=str)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert mock_slack_api_call.called
@pytest.mark.django_db
@pytest.mark.parametrize("swap_taken", [False, True])
def test_current_shift_changes_swap_split(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
make_shift_swap_request,
swap_taken,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
user2 = make_user(organization=organization, username="user2")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
duration = timezone.timedelta(hours=23, minutes=59, seconds=59)
data = {
"start": today,
"rotation_start": today,
"duration": duration,
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
"schedule": schedule,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
# setup in progress swap request
swap_request = make_shift_swap_request(
schedule,
user1,
swap_start=today,
swap_end=today + timezone.timedelta(days=2),
)
if swap_taken:
swap_request.benefactor = user2
swap_request.save()
schedule.refresh_ical_file()
# setup empty current shifts before checking/triggering for notifications
schedule.current_shifts = json.dumps({}, default=str)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
text_block = mock_slack_api_call.call_args_list[0][1]["blocks"][1]["text"]["text"]
assert "user2" in text_block if swap_taken else "user1" in text_block
@pytest.mark.django_db
def test_current_shift_changes_end_affected_by_swap(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
make_shift_swap_request,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
user2 = make_user(organization=organization, username="user2")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
duration = timezone.timedelta(days=3)
data = {
"start": today,
"rotation_start": today,
"duration": duration,
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
# setup in progress swap request
swap_start = today + timezone.timedelta(days=1)
swap_end = today + timezone.timedelta(days=2)
make_shift_swap_request(
schedule,
user1,
swap_start=swap_start,
swap_end=swap_end,
benefactor=user2,
)
schedule.refresh_ical_file()
# setup empty current shifts before checking/triggering for notifications
schedule.current_shifts = json.dumps({}, default=str)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
current_text_block = mock_slack_api_call.call_args_list[0][1]["blocks"][1]["text"]["text"]
assert "user1" in current_text_block
assert today.strftime("%Y-%m-%d") in current_text_block
assert swap_start.strftime("%Y-%m-%d") in current_text_block
next_text_block = mock_slack_api_call.call_args_list[0][1]["blocks"][2]["text"]["text"]
assert "user2" in next_text_block
assert swap_start.strftime("%Y-%m-%d") in next_text_block
assert swap_end.strftime("%Y-%m-%d") in next_text_block
@pytest.mark.django_db
def test_next_shift_changes_no_triggering_notification(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
user2 = make_user(organization=organization, username="user2")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date_1 = now - datetime.timedelta(days=7, minutes=1)
data_1 = {
"start": start_date_1,
"rotation_start": start_date_1,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift_1 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data_1
)
on_call_shift_1.add_rolling_users([[user1]])
on_call_shift_1.schedules.add(schedule)
start_date_2 = now + datetime.timedelta(minutes=10)
data_2 = {
"start": start_date_2,
"rotation_start": start_date_2,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 2,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift_2 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data_2
)
on_call_shift_2.add_rolling_users([[user1]])
on_call_shift_2.schedules.add(schedule)
schedule.refresh_ical_file()
# setup current shifts before checking/triggering for notifications
next_shifts = schedule.final_events(
now, now + datetime.timedelta(days=MIN_DAYS_TO_LOOKUP_FOR_THE_END_OF_EVENT), False, False
)
current_shifts = [e for e in next_shifts if now > e["start"]]
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
schedule.save()
on_call_shift_2.add_rolling_users([[user2]])
schedule.refresh_ical_file()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert not mock_slack_api_call.called
@pytest.mark.django_db
def test_current_shifts_using_microseconds(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
schedule.refresh_ical_file()
schedule.current_shifts = json.dumps(
{
"test_shift_uid": {
"users": [user1.pk],
"start": timezone.now().replace(microsecond=123456),
"end": timezone.now().replace(microsecond=654321) + timezone.timedelta(days=1),
"all_day": False,
"priority": 1,
"priority_increased_by": 0,
}
},
default=str,
)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert mock_slack_api_call.called
@pytest.mark.django_db
def test_lower_priority_changes_no_triggering_notification(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
user2 = make_user(organization=organization, username="user2")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date = now - datetime.timedelta(days=7, minutes=1)
data_1 = {
"start": start_date,
"rotation_start": start_date,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 2,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift_1 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data_1
)
on_call_shift_1.add_rolling_users([[user1]])
on_call_shift_1.schedules.add(schedule)
data_2 = {
"start": start_date,
"rotation_start": start_date,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift_2 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data_2
)
on_call_shift_2.add_rolling_users([[user1]])
on_call_shift_2.schedules.add(schedule)
schedule.refresh_ical_file()
# setup empty current shifts before checking/triggering for notifications
current_shifts = schedule.final_events(now, now, False, False)
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
schedule.save()
on_call_shift_2.add_rolling_users([[user2]])
schedule.refresh_ical_file()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert not mock_slack_api_call.called
@pytest.mark.django_db
def test_vtimezone_changes_no_triggering_notification(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
make_user(organization=organization, username="user1")
ical_before = textwrap.dedent(
"""
BEGIN:VCALENDAR
PRODID:-//Google Inc//Google Calendar 70.9054//EN
VERSION:2.0
CALSCALE:GREGORIAN
X-WR-TIMEZONE:Europe/London
METHOD:PUBLISH
BEGIN:VTIMEZONE
TZID:Europe/Rome
BEGIN:STANDARD
TZOFFSETFROM:0200
TZOFFSETTO:0100
TZNAME:CET
DTSTART:19701025T030000
END:STANDARD
END:VTIMEZONE
BEGIN:VTIMEZONE
TZID:America/Argentina/Buenos_Aires
X-LIC-LOCATION:America/Argentina/Buenos_Aires
BEGIN:STANDARD
TZOFFSETFROM:-0300
TZOFFSETTO:-0300
TZNAME:-03
DTSTART:19700101T000000
END:STANDARD
END:VTIMEZONE
BEGIN:VEVENT
DTSTART;VALUE=DATE:20230101
DTEND;VALUE=DATE:20230102
RRULE:FREQ=DAILY
DTSTAMP:20230101T000000
UID:id1@google.com
CREATED:20230101T000000
DESCRIPTION:
LAST-MODIFIED:20230101T000000
LOCATION:
SEQUENCE:1
STATUS:CONFIRMED
SUMMARY:user1
TRANSP:TRANSPARENT
END:VEVENT
END:VCALENDAR"""
)
# same data, timezones in different order (eg. google usually randomly reorders them)
ical_after = textwrap.dedent(
"""
BEGIN:VCALENDAR
PRODID:-//Google Inc//Google Calendar 70.9054//EN
VERSION:2.0
CALSCALE:GREGORIAN
METHOD:PUBLISH
X-WR-TIMEZONE:Europe/London
BEGIN:VTIMEZONE
TZID:America/Argentina/Buenos_Aires
X-LIC-LOCATION:America/Argentina/Buenos_Aires
BEGIN:STANDARD
TZOFFSETFROM:-0300
TZOFFSETTO:-0300
TZNAME:-03
DTSTART:19700101T000000
END:STANDARD
END:VTIMEZONE
BEGIN:VTIMEZONE
TZID:Europe/Rome
BEGIN:STANDARD
TZOFFSETFROM:0200
TZOFFSETTO:0100
TZNAME:CET
DTSTART:19701025T030000
END:STANDARD
END:VTIMEZONE
BEGIN:VEVENT
DTSTART;VALUE=DATE:20230101
DTEND;VALUE=DATE:20230102
RRULE:FREQ=DAILY
DTSTAMP:20230101T000000
UID:id1@google.com
CREATED:20230101T000000
DESCRIPTION:
LAST-MODIFIED:20230101T000000
LOCATION:
SEQUENCE:1
STATUS:CONFIRMED
SUMMARY:user1
TRANSP:TRANSPARENT
END:VEVENT
END:VCALENDAR"""
)
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleICal,
name="test_ical_schedule",
slack_channel=slack_channel,
ical_url_primary="url",
prev_ical_file_primary=None,
cached_ical_file_primary=ical_before,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
# setup current shifts before checking/triggering for notifications
now = datetime.datetime.now(datetime.timezone.utc)
current_shifts = schedule.final_events(now, now, False, False)
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
# update schedule cached ical to ical_after
schedule.prev_ical_file_primary = ical_before
schedule.cached_ical_file_primary = ical_after
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert not mock_slack_api_call.called
@pytest.mark.django_db
def test_no_changes_no_triggering_notification_from_old_to_new_task_version(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date = now - timezone.timedelta(days=7)
data = {
"start": start_date,
"rotation_start": start_date,
"duration": timezone.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
on_call_shift.schedules.add(schedule)
# setup current shifts with old version of shifts structure before checking/triggering for notifications
current_shifts = {
"test_shift_uid": {
"users": [user1.pk],
"start": start_date,
"end": start_date + data["duration"],
"all_day": False,
"priority": data["priority_level"],
"priority_increased_by": 0,
}
}
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert not mock_slack_api_call.called
@pytest.mark.django_db
def test_current_shift_changes_trigger_notification_from_old_to_new_task_version(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
user2 = make_user(organization=organization, username="user2")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleCalendar,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date = now - datetime.timedelta(days=7)
data = {
"start": start_date,
"rotation_start": start_date,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_DAILY,
}
on_call_shift = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data
)
on_call_shift.add_rolling_users([[user1]])
on_call_shift.schedules.add(schedule)
schedule.refresh_ical_file()
# setup current shifts with old version of shifts structure before checking/triggering for notifications
current_shifts = {
"test_shift_uid": {
"users": [user1.pk],
"start": start_date,
"end": start_date + data["duration"],
"all_day": False,
"priority": data["priority_level"],
"priority_increased_by": 0,
}
}
schedule.current_shifts = json.dumps(current_shifts, default=str)
schedule.empty_oncall = False
schedule.save()
on_call_shift.add_rolling_users([[user2]])
schedule.refresh_ical_file()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert mock_slack_api_call.called
@pytest.mark.django_db
def test_next_shift_notification_long_and_short_shifts(
make_slack_team_identity,
make_slack_channel,
make_organization,
make_user,
make_schedule,
make_on_call_shift,
):
slack_team_identity = make_slack_team_identity()
slack_channel = make_slack_channel(slack_team_identity)
organization = make_organization(slack_team_identity=slack_team_identity)
user1 = make_user(organization=organization, username="user1")
user2 = make_user(organization=organization, username="user2")
user3 = make_user(organization=organization, username="user3")
schedule = make_schedule(
organization,
schedule_class=OnCallScheduleWeb,
name="test_schedule",
slack_channel=slack_channel,
prev_ical_file_overrides=None,
cached_ical_file_overrides=None,
)
now = timezone.now().replace(microsecond=0)
start_date_1 = now - datetime.timedelta(days=1)
data_1 = {
"start": start_date_1,
"rotation_start": start_date_1,
"duration": datetime.timedelta(seconds=3600 * 24 * 7), # one week duration
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
}
on_call_shift_1 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data_1
)
on_call_shift_1.add_rolling_users([[user1], [user2]])
start_date_2 = now - datetime.timedelta(hours=1)
data_2 = {
"start": start_date_2,
"rotation_start": start_date_2,
"duration": datetime.timedelta(seconds=3600 * 24),
"priority_level": 1,
"frequency": CustomOnCallShift.FREQUENCY_WEEKLY,
"schedule": schedule,
}
on_call_shift_2 = make_on_call_shift(
organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data_2
)
on_call_shift_2.add_rolling_users([[user3]])
schedule.refresh_ical_file()
# setup empty current shifts before checking/triggering for notifications
schedule.current_shifts = json.dumps({}, default=str)
schedule.empty_oncall = False
schedule.save()
with patch("apps.slack.client.SlackClient.chat_postMessage") as mock_slack_api_call:
notify_ical_schedule_shift(schedule.pk)
assert mock_slack_api_call.called
new_shift_notification = mock_slack_api_call.call_args[1]["blocks"][1]["text"]["text"]
next_shift_notification = mock_slack_api_call.call_args[1]["blocks"][2]["text"]["text"]
assert "*New on-call shift*\n[L1] user1" in new_shift_notification
assert "[L1] user3" in new_shift_notification
assert "*Next on-call shift*\n[L1] user2" in next_shift_notification
Reference in a new issue View git blame Copy permalink