import json import textwrap from unittest.mock import PropertyMock, patch import pytest from django.urls import reverse from django.utils import timezone from rest_framework import status from rest_framework.response import Response from rest_framework.test import APIClient from apps.alerts.models import EscalationPolicy from apps.api.permissions import LegacyAccessControlRole from apps.api.serializers.schedule_base import ScheduleBaseSerializer from apps.api.serializers.user import ScheduleUserSerializer from apps.schedules.models import ( CustomOnCallShift, OnCallSchedule, OnCallScheduleCalendar, OnCallScheduleICal, OnCallScheduleWeb, ) from apps.slack.models import SlackUserGroup from common.api_helpers.utils import create_engine_url, serialize_datetime_as_utc_timestamp ICAL_URL = "https://example.com" @pytest.fixture(autouse=True) def patch_fetch_ical_file(): """ NOTE: we patch this method for all tests in this file to avoid making actual HTTP requests.. we don't really need to test the actual fetching of the ical file in these tests, so just simply mock out the response as an empty string Alternatively, if we really needed to, we could save .ical files locally here, and read/return those as the return_value """ with patch("apps.schedules.ical_utils.fetch_ical_file", return_value=""): yield @pytest.fixture() def schedule_internal_api_setup( make_organization_and_user_with_plugin_token, make_slack_channel, make_schedule, ): organization, user, token = make_organization_and_user_with_plugin_token() slack_channel = make_slack_channel( organization.slack_team_identity, ) calendar_schedule = make_schedule( organization, schedule_class=OnCallScheduleCalendar, name="test_calendar_schedule", ) ical_schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) web_schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) return user, token, calendar_schedule, ical_schedule, web_schedule, slack_channel @pytest.mark.django_db def test_get_list_schedules( schedule_internal_api_setup, make_escalation_chain, make_escalation_policy, make_user_auth_headers ): user, token, calendar_schedule, ical_schedule, web_schedule, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-list") # setup escalation chain linked to web schedule escalation_chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=escalation_chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=web_schedule, ) expected_payload = { "count": 3, "next": None, "previous": None, "results": [ { "id": calendar_schedule.public_primary_key, "type": 0, "team": None, "name": "test_calendar_schedule", "time_zone": "UTC", "slack_channel": None, "user_group": None, "warnings": [], "ical_url_overrides": None, "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, }, { "id": ical_schedule.public_primary_key, "type": 1, "team": None, "name": "test_ical_schedule", "ical_url_primary": ICAL_URL, "ical_url_overrides": None, "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, }, { "id": web_schedule.public_primary_key, "type": 2, "time_zone": "UTC", "team": None, "name": "test_web_schedule", "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 1, "enable_web_overrides": True, }, ], "current_page_number": 1, "page_size": 15, "total_pages": 1, } response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json() == expected_payload @pytest.mark.django_db def test_get_list_schedules_pagination( schedule_internal_api_setup, make_escalation_chain, make_escalation_policy, make_user_auth_headers ): user, token, calendar_schedule, ical_schedule, web_schedule, _ = schedule_internal_api_setup # setup escalation chain linked to web schedule escalation_chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=escalation_chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=web_schedule, ) available_schedules = [ { "id": calendar_schedule.public_primary_key, "type": 0, "team": None, "name": "test_calendar_schedule", "time_zone": "UTC", "slack_channel": None, "user_group": None, "warnings": [], "ical_url_overrides": None, "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, }, { "id": ical_schedule.public_primary_key, "type": 1, "team": None, "name": "test_ical_schedule", "ical_url_primary": ICAL_URL, "ical_url_overrides": None, "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, }, { "id": web_schedule.public_primary_key, "type": 2, "time_zone": "UTC", "team": None, "name": "test_web_schedule", "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 1, "enable_web_overrides": True, }, ] client = APIClient() schedule_list_url = reverse("api-internal:schedule-list") absolute_url = create_engine_url(schedule_list_url) for p, schedule in enumerate(available_schedules, start=1): # patch oncall_users to check a paginated queryset is used def mock_oncall_now(qs, events_datetime): # only one schedule is passed here assert qs.count() == 1 return {} url = "{}?page={}&perpage=1".format(schedule_list_url, p) with patch( "apps.schedules.models.on_call_schedule.get_oncall_users_for_multiple_schedules", side_effect=mock_oncall_now, ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK previous_url = None next_url = "{}?page={}&perpage=1".format(absolute_url, p + 1) if p == 2: previous_url = "{}?perpage=1".format(absolute_url) elif p > 2: previous_url = "{}?page={}&perpage=1".format(absolute_url, p - 1) next_url = None expected_payload = { "count": 3, "next": next_url, "previous": previous_url, "results": [schedule], "current_page_number": p, "page_size": 1, "total_pages": 3, } assert response.json() == expected_payload @pytest.mark.django_db def test_get_list_schedules_by_type( schedule_internal_api_setup, make_escalation_chain, make_escalation_policy, make_user_auth_headers ): user, token, calendar_schedule, ical_schedule, web_schedule, slack_channel = schedule_internal_api_setup client = APIClient() # setup escalation chain linked to web schedule escalation_chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=escalation_chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=web_schedule, ) available_schedules = [ { "id": calendar_schedule.public_primary_key, "type": 0, "team": None, "name": "test_calendar_schedule", "time_zone": "UTC", "slack_channel": None, "user_group": None, "warnings": [], "ical_url_overrides": None, "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, }, { "id": ical_schedule.public_primary_key, "type": 1, "team": None, "name": "test_ical_schedule", "ical_url_primary": ICAL_URL, "ical_url_overrides": None, "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, }, { "id": web_schedule.public_primary_key, "type": 2, "time_zone": "UTC", "team": None, "name": "test_web_schedule", "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 1, "enable_web_overrides": True, }, ] for schedule_type in range(3): url = reverse("api-internal:schedule-list") + "?type={}".format(schedule_type) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK expected_payload = { "count": 1, "next": None, "previous": None, "results": [available_schedules[schedule_type]], "current_page_number": 1, "page_size": 15, "total_pages": 1, } assert response.json() == expected_payload # request multiple types url = reverse("api-internal:schedule-list") + "?type=0&type=1" response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK expected_payload = { "count": 2, "next": None, "previous": None, "results": [available_schedules[0], available_schedules[1]], "current_page_number": 1, "page_size": 15, "total_pages": 1, } assert response.json() == expected_payload @pytest.mark.django_db @pytest.mark.parametrize( "query_param, expected_schedule_names", [ ("?used=true", ["test_web_schedule"]), ("?used=false", ["test_calendar_schedule", "test_ical_schedule"]), ("?used=null", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]), ("", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]), ], ) def test_get_list_schedules_by_used( schedule_internal_api_setup, make_escalation_chain, make_escalation_policy, make_user_auth_headers, query_param, expected_schedule_names, ): user, token, calendar_schedule, ical_schedule, web_schedule, slack_channel = schedule_internal_api_setup client = APIClient() # setup escalation chain linked to web schedule escalation_chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=escalation_chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=web_schedule, ) url = reverse("api-internal:schedule-list") + query_param response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["count"] == len(expected_schedule_names) schedule_names = [schedule["name"] for schedule in response.json()["results"]] assert set(schedule_names) == set(expected_schedule_names) @pytest.mark.django_db @pytest.mark.parametrize( "query_param, expected_schedule_names", [ ("?mine=true", ["test_web_schedule"]), ("?mine=false", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]), ("?mine=null", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]), ("", ["test_calendar_schedule", "test_ical_schedule", "test_web_schedule"]), ], ) def test_get_list_schedules_by_mine( schedule_internal_api_setup, make_user_auth_headers, make_on_call_shift, query_param, expected_schedule_names, ): user, token, calendar_schedule, ical_schedule, web_schedule, slack_channel = schedule_internal_api_setup client = APIClient() today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) # setup user shift in web schedule override_data = { "start": today + timezone.timedelta(hours=22), "rotation_start": today + timezone.timedelta(hours=22), "duration": timezone.timedelta(hours=1), "schedule": web_schedule, } override = make_on_call_shift( organization=user.organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) override.add_rolling_users([[user]]) web_schedule.refresh_ical_file() web_schedule.refresh_ical_final_schedule() url = reverse("api-internal:schedule-list") + query_param response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["count"] == len(expected_schedule_names) schedule_names = [schedule["name"] for schedule in response.json()["results"]] assert set(schedule_names) == set(expected_schedule_names) @pytest.mark.django_db def test_get_list_schedules_pagination_respects_search( schedule_internal_api_setup, make_escalation_chain, make_escalation_policy, make_user_auth_headers, ): user, token, _, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-list") + "?search=ical" response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["count"] == 1 assert len(response.json()["results"]) == 1 @pytest.mark.django_db def test_get_detail_calendar_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, calendar_schedule, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": calendar_schedule.public_primary_key}) expected_payload = { "id": calendar_schedule.public_primary_key, "type": 0, "team": None, "name": "test_calendar_schedule", "time_zone": "UTC", "slack_channel": None, "user_group": None, "warnings": [], "ical_url_overrides": None, "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, } response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.data == expected_payload @pytest.mark.django_db def test_get_detail_ical_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, _, ical_schedule, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": ical_schedule.public_primary_key}) expected_payload = { "id": ical_schedule.public_primary_key, "team": None, "ical_url_primary": ICAL_URL, "ical_url_overrides": None, "name": "test_ical_schedule", "type": 1, "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": False, } response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.data == expected_payload @pytest.mark.django_db def test_get_detail_web_schedule( schedule_internal_api_setup, make_escalation_chain, make_escalation_policy, make_user_auth_headers ): user, token, _, _, web_schedule, _ = schedule_internal_api_setup # setup escalation chain linked to web schedule escalation_chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=escalation_chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=web_schedule, ) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": web_schedule.public_primary_key}) expected_payload = { "id": web_schedule.public_primary_key, "team": None, "name": "test_web_schedule", "type": 2, "time_zone": "UTC", "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 1, "enable_web_overrides": True, } response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.data == expected_payload @pytest.mark.django_db def test_get_detail_schedule_oncall_now_multipage_objects( make_organization_and_user_with_plugin_token, make_schedule, make_on_call_shift, make_user_auth_headers ): organization, user, token = make_organization_and_user_with_plugin_token() # make sure our schedule would be in the second page of the listing page for i in range(16): make_schedule(organization, schedule_class=OnCallScheduleWeb, name=f"schedule {i}") schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) start_date = now - timezone.timedelta(days=7) data = { "start": start_date, "rotation_start": start_date, "duration": timezone.timedelta(seconds=86400), "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([[user]]) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) expected_payload = { "id": schedule.public_primary_key, "team": None, "name": "test_web_schedule", "type": 2, "time_zone": "UTC", "slack_channel": None, "user_group": None, "warnings": [], "on_call_now": [ { "pk": user.public_primary_key, "username": user.username, "avatar": user.avatar_url, "avatar_full": user.avatar_full_url(organization), } ], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "number_of_escalation_chains": 0, "enable_web_overrides": True, } response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.data == expected_payload @pytest.mark.django_db def test_create_calendar_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, _, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-list") data = { "ical_url_overrides": None, "type": 0, "name": "created_calendar_schedule", "time_zone": "UTC", "user_group": None, "team": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, "enable_web_overrides": True, } response = client.post(url, data, format="json", **make_user_auth_headers(user, token)) # modify initial data by adding id and None for optional fields schedule = OnCallSchedule.objects.get(public_primary_key=response.data["id"]) data["id"] = schedule.public_primary_key data["number_of_escalation_chains"] = 0 assert response.status_code == status.HTTP_201_CREATED assert response.data == data @pytest.mark.django_db def test_create_ical_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, _, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-list") data = { "ical_url_primary": ICAL_URL, "ical_url_overrides": None, "name": "created_ical_schedule", "type": 1, "user_group": None, "team": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, } with patch( "apps.api.serializers.schedule_ical.ScheduleICalSerializer.validate_ical_url_primary", return_value=ICAL_URL ), patch("apps.schedules.tasks.refresh_ical_final_schedule.apply_async") as mock_refresh_final: response = client.post(url, data, format="json", **make_user_auth_headers(user, token)) # modify initial data by adding id and None for optional fields schedule = OnCallSchedule.objects.get(public_primary_key=response.data["id"]) data["id"] = schedule.public_primary_key data["number_of_escalation_chains"] = 0 data["enable_web_overrides"] = False assert response.status_code == status.HTTP_201_CREATED assert response.data == data # check final schedule refresh triggered mock_refresh_final.assert_called_once_with((schedule.pk,)) @pytest.mark.django_db def test_create_web_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, _, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-list") data = { "name": "created_web_schedule", "type": 2, "time_zone": "UTC", "user_group": None, "team": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, } response = client.post(url, data, format="json", **make_user_auth_headers(user, token)) # modify initial data by adding id and None for optional fields schedule = OnCallSchedule.objects.get(public_primary_key=response.data["id"]) data["id"] = schedule.public_primary_key data["number_of_escalation_chains"] = 0 data["enable_web_overrides"] = True assert response.status_code == status.HTTP_201_CREATED assert response.data == data @pytest.mark.django_db @pytest.mark.parametrize("calendar_type", [0, 2]) def test_create_schedule_invalid_time_zone(schedule_internal_api_setup, make_user_auth_headers, calendar_type): user, token, _, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-list") data = { "name": "created_web_schedule", "type": calendar_type, "time_zone": "asdfasdfasdf", "slack_channel_id": None, "user_group": None, "team": None, "warnings": [], "on_call_now": [], "has_gaps": False, "mention_oncall_next": False, "mention_oncall_start": True, "notify_empty_oncall": 0, "notify_oncall_shift_freq": 1, } response = client.post(url, data, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST assert response.json() == {"time_zone": ["Invalid timezone"]} @pytest.mark.django_db def test_update_calendar_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, calendar_schedule, _, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": calendar_schedule.public_primary_key}) data = { "name": "updated_calendar_schedule", "type": 0, "team": None, } response = client.put( url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) ) updated_instance = OnCallSchedule.objects.get(public_primary_key=calendar_schedule.public_primary_key) assert response.status_code == status.HTTP_200_OK assert updated_instance.name == "updated_calendar_schedule" @pytest.mark.django_db def test_update_ical_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, _, ical_schedule, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": ical_schedule.public_primary_key}) data = { "name": "updated_ical_schedule", "type": 1, "team": None, } with patch("apps.schedules.tasks.refresh_ical_final_schedule.apply_async") as mock_refresh_final: response = client.put( url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) ) updated_instance = OnCallSchedule.objects.get(public_primary_key=ical_schedule.public_primary_key) assert response.status_code == status.HTTP_200_OK assert updated_instance.name == "updated_ical_schedule" # check refresh final is not triggered (url unchanged) assert not mock_refresh_final.called @pytest.mark.django_db def test_update_ical_schedule_url(schedule_internal_api_setup, make_user_auth_headers): user, token, _, ical_schedule, _, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": ical_schedule.public_primary_key}) updated_url = "another-url" data = { "name": ical_schedule.name, "type": 1, "ical_url_primary": updated_url, } with patch( "apps.api.serializers.schedule_ical.ScheduleICalSerializer.validate_ical_url_primary", return_value=updated_url ), patch("apps.schedules.tasks.refresh_ical_final_schedule.apply_async") as mock_refresh_final: response = client.put( url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) ) updated_instance = OnCallSchedule.objects.get(public_primary_key=ical_schedule.public_primary_key) assert response.status_code == status.HTTP_200_OK # check refresh final triggered (changing url) mock_refresh_final.assert_called_once_with((updated_instance.pk,)) @pytest.mark.django_db def test_update_web_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, _, _, web_schedule, _ = schedule_internal_api_setup client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": web_schedule.public_primary_key}) data = { "name": "updated_web_schedule", "type": 2, "team": None, } response = client.put( url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) ) updated_instance = OnCallSchedule.objects.get(public_primary_key=web_schedule.public_primary_key) assert response.status_code == status.HTTP_200_OK assert updated_instance.name == "updated_web_schedule" @pytest.mark.django_db @pytest.mark.parametrize("calendar_type", [0, 2]) def test_update_schedule_invalid_time_zone(schedule_internal_api_setup, make_user_auth_headers, calendar_type): user, token, *calendars, _ = schedule_internal_api_setup schedule = calendars[calendar_type] client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) data = {"name": "updated_web_schedule", "type": calendar_type, "time_zone": "asdfasdfasdf"} response = client.put( url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) ) assert response.status_code == status.HTTP_400_BAD_REQUEST assert response.json() == {"time_zone": ["Invalid timezone"]} @pytest.mark.django_db def test_delete_schedule(schedule_internal_api_setup, make_user_auth_headers): user, token, calendar_schedule, ical_schedule, _, _ = schedule_internal_api_setup client = APIClient() for calendar in (calendar_schedule, ical_schedule): url = reverse("api-internal:schedule-detail", kwargs={"pk": calendar.public_primary_key}) response = client.delete(url, **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_204_NO_CONTENT @pytest.mark.django_db def test_events_calendar( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleCalendar, name="test_calendar_schedule", ) start_date = timezone.now().replace(microsecond=0) data = { "start": start_date, "rotation_start": start_date, "duration": timezone.timedelta(seconds=7200), "priority_level": 2, } 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) url = reverse("api-internal:schedule-events", kwargs={"pk": schedule.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(user, token)) expected_result = { "id": schedule.public_primary_key, "name": "test_calendar_schedule", "type": 0, "slack_channel": None, "events": [ { "all_day": False, "start": on_call_shift.start, "end": on_call_shift.start + on_call_shift.duration, "users": [ { "display_name": user.username, "pk": user.public_primary_key, "email": user.email, "avatar_full": user.avatar_full_url(organization), }, ], "missing_users": [], "priority_level": on_call_shift.priority_level, "source": "api", "calendar_type": OnCallSchedule.PRIMARY, "is_empty": False, "is_gap": False, "is_override": False, "shift": { "pk": on_call_shift.public_primary_key, }, } ], } assert response.status_code == status.HTTP_200_OK assert response.data == expected_result @pytest.mark.django_db def test_filter_events_calendar( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) 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=7200), "priority_level": 1, "frequency": CustomOnCallShift.FREQUENCY_WEEKLY, "by_day": ["MO", "FR"], "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_RECURRENT_EVENT, **data ) on_call_shift.users.add(user) url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?type=rotation" response = client.get(url, format="json", **make_user_auth_headers(user, token)) # current week events are expected mon_start = now - timezone.timedelta(days=start_date.weekday()) fri_start = mon_start + timezone.timedelta(days=4) expected_result = { "id": schedule.public_primary_key, "name": "test_web_schedule", "type": 2, "events": [ { "all_day": False, "start": mon_start, "end": mon_start + on_call_shift.duration, "users": [ { "display_name": user.username, "pk": user.public_primary_key, "email": user.email, "avatar_full": user.avatar_full_url(organization), }, ], "missing_users": [], "priority_level": on_call_shift.priority_level, "source": "api", "calendar_type": OnCallSchedule.PRIMARY, "is_empty": False, "is_gap": False, "is_override": False, "shift": { "pk": on_call_shift.public_primary_key, "name": on_call_shift.name, "type": on_call_shift.type, }, }, { "all_day": False, "start": fri_start, "end": fri_start + on_call_shift.duration, "users": [ { "display_name": user.username, "pk": user.public_primary_key, "email": user.email, "avatar_full": user.avatar_full_url(organization), } ], "missing_users": [], "priority_level": on_call_shift.priority_level, "source": "api", "calendar_type": OnCallSchedule.PRIMARY, "is_empty": False, "is_gap": False, "is_override": False, "shift": { "pk": on_call_shift.public_primary_key, "name": on_call_shift.name, "type": on_call_shift.type, }, }, ], } assert response.status_code == status.HTTP_200_OK assert response.data == expected_result @pytest.mark.django_db def test_filter_events_range_calendar( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) now = timezone.now().replace(microsecond=0) start_date = now - timezone.timedelta(days=7) mon_start = now - timezone.timedelta(days=start_date.weekday()) request_date = mon_start + timezone.timedelta(days=2) data = { "start": start_date, "rotation_start": start_date, "duration": timezone.timedelta(seconds=7200), "priority_level": 1, "frequency": CustomOnCallShift.FREQUENCY_WEEKLY, "by_day": ["MO", "FR"], "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) # add override shift override_start = request_date + timezone.timedelta(seconds=3600) override_data = { "start": override_start, "rotation_start": override_start, "duration": timezone.timedelta(seconds=3600), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) other_user = make_user_for_organization(organization) override.users.add(other_user) url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?date={}&days=3&type=rotation".format(request_date.strftime("%Y-%m-%d")) response = client.get(url, format="json", **make_user_auth_headers(user, token)) # only friday occurrence is expected fri_start = mon_start + timezone.timedelta(days=4) expected_result = { "id": schedule.public_primary_key, "name": "test_web_schedule", "type": 2, "events": [ { "all_day": False, "start": fri_start, "end": fri_start + on_call_shift.duration, "users": [ { "display_name": user.username, "pk": user.public_primary_key, "email": user.email, "avatar_full": user.avatar_full_url(organization), }, ], "missing_users": [], "priority_level": on_call_shift.priority_level, "source": "api", "calendar_type": OnCallSchedule.PRIMARY, "is_empty": False, "is_gap": False, "is_override": False, "shift": { "pk": on_call_shift.public_primary_key, "name": on_call_shift.name, "type": on_call_shift.type, }, } ], } assert response.status_code == status.HTTP_200_OK assert response.data == expected_result @pytest.mark.django_db def test_filter_events_overrides( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) now = timezone.now().replace(microsecond=0) start_date = now - timezone.timedelta(days=7) mon_start = now - timezone.timedelta(days=start_date.weekday()) request_date = mon_start + timezone.timedelta(days=2) data = { "start": start_date, "rotation_start": start_date, "duration": timezone.timedelta(seconds=7200), "priority_level": 1, "frequency": CustomOnCallShift.FREQUENCY_WEEKLY, "by_day": ["MO", "FR"], "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) # add override shift override_start = request_date + timezone.timedelta(seconds=3600) override_data = { "start": override_start, "rotation_start": override_start, "duration": timezone.timedelta(seconds=3600), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) other_user = make_user_for_organization(organization) override.add_rolling_users([[other_user]]) url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?date={}&days=3&type=override".format(request_date.strftime("%Y-%m-%d")) response = client.get(url, format="json", **make_user_auth_headers(user, token)) # only override occurrence is expected expected_result = { "id": schedule.public_primary_key, "name": "test_web_schedule", "type": 2, "events": [ { "all_day": False, "start": override_start, "end": override_start + override.duration, "users": [ { "display_name": other_user.username, "pk": other_user.public_primary_key, "email": other_user.email, "avatar_full": other_user.avatar_full_url(organization), } ], "missing_users": [], "priority_level": 0, "source": "api", "calendar_type": OnCallSchedule.OVERRIDES, "is_empty": False, "is_gap": False, "is_override": True, "shift": { "pk": override.public_primary_key, "name": override.name, "type": override.type, }, } ], } assert response.status_code == status.HTTP_200_OK assert response.data == expected_result @pytest.mark.django_db def test_filter_events_final_schedule( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) start_date = now - timezone.timedelta(days=7) request_date = start_date user_a, user_b, user_c, user_d, user_e = (make_user_for_organization(organization, username=i) for i in "ABCDE") shifts = ( # user, priority, start time (h), duration (hs) (user_a, 1, 10, 5), # r1-1: 10-15 / A (user_b, 1, 11, 2), # r1-2: 11-13 / B (user_a, 1, 16, 3), # r1-3: 16-19 / A (user_a, 1, 21, 1), # r1-4: 21-22 / A (user_b, 1, 22, 2), # r1-5: 22-00 / B (user_c, 2, 12, 2), # r2-1: 12-14 / C (user_d, 2, 14, 1), # r2-2: 14-15 / D (user_d, 2, 17, 1), # r2-3: 17-18 / D (user_d, 2, 20, 3), # r2-4: 20-23 / D ) for user, priority, start_h, duration in shifts: data = { "start": start_date + timezone.timedelta(hours=start_h), "rotation_start": start_date, "duration": timezone.timedelta(hours=duration), "priority_level": priority, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) # override: 22-23 / E override_data = { "start": start_date + timezone.timedelta(hours=22), "rotation_start": start_date + timezone.timedelta(hours=22), "duration": timezone.timedelta(hours=1), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) override.add_rolling_users([[user_e]]) url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?date={}&days=1".format(request_date.strftime("%Y-%m-%d")) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK expected = ( # start (h), duration (H), user, priority, is_gap, is_override (0, 10, None, None, True, False), # 0-10 gap (10, 2, "A", 1, False, False), # 10-12 A (11, 1, "B", 1, False, False), # 11-12 B (12, 2, "C", 2, False, False), # 12-14 C (14, 1, "D", 2, False, False), # 14-15 D (15, 1, None, None, True, False), # 15-16 gap (16, 1, "A", 1, False, False), # 16-17 A (17, 1, "D", 2, False, False), # 17-18 D (18, 1, "A", 1, False, False), # 18-19 A (19, 1, None, None, True, False), # 19-20 gap (20, 2, "D", 2, False, False), # 20-22 D (22, 1, "E", None, False, True), # 22-23 E (override) (23, 1, "B", 1, False, False), # 23-00 B ) expected_events = [ { "calendar_type": 1 if is_override else None if is_gap else 0, "end": start_date + timezone.timedelta(hours=start + duration), "is_gap": is_gap, "is_override": is_override, "priority_level": priority or 0, "start": start_date + timezone.timedelta(hours=start), "user": user, } for start, duration, user, priority, is_gap, is_override in expected ] returned_events = [ { "calendar_type": e["calendar_type"], "end": e["end"], "is_gap": e["is_gap"], "is_override": e["is_override"], "priority_level": e["priority_level"], "start": e["start"], "user": e["users"][0]["display_name"] if e["users"] else None, } for e in response.data["events"] ] assert returned_events == expected_events @pytest.mark.django_db def test_filter_swap_requests( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_shift_swap_request, ): organization, admin, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) other_schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="other_web_schedule", ) user_a, user_b, user_c = (make_user_for_organization(organization, username=i) for i in "ABC") today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) start_date = today - timezone.timedelta(days=7) request_date = start_date # swap for other schedule make_shift_swap_request( other_schedule, user_a, swap_start=start_date + timezone.timedelta(days=1), swap_end=start_date + timezone.timedelta(days=3), ) # swap out of range make_shift_swap_request( schedule, user_a, swap_start=start_date + timezone.timedelta(days=10), swap_end=start_date + timezone.timedelta(days=13), ) # expected swaps swap_a = make_shift_swap_request( schedule, user_a, swap_start=start_date + timezone.timedelta(days=1), swap_end=start_date + timezone.timedelta(days=3), ) swap_b = make_shift_swap_request( schedule, user_b, swap_start=start_date, swap_end=start_date + timezone.timedelta(days=1), benefactor=user_c, ) url = reverse("api-internal:schedule-filter-shift-swaps", kwargs={"pk": schedule.public_primary_key}) url += "?date={}&days=1".format(request_date.strftime("%Y-%m-%d")) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK def _serialized_user(u): if u: return { "display_name": u.username, "email": u.email, "pk": u.public_primary_key, "avatar_full": u.avatar_full_url(organization), } expected = [ { "pk": swap.public_primary_key, "swap_start": serialize_datetime_as_utc_timestamp(swap.swap_start), "swap_end": serialize_datetime_as_utc_timestamp(swap.swap_end), "beneficiary": _serialized_user(swap.beneficiary), "benefactor": _serialized_user(swap.benefactor), } for swap in (swap_a, swap_b) ] returned = [ { "pk": s["id"], "swap_start": s["swap_start"], "swap_end": s["swap_end"], "beneficiary": s["beneficiary"], "benefactor": s["benefactor"], } for s in response.data["shift_swaps"] ] assert returned == expected @pytest.mark.django_db def test_next_shifts_per_user( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, admin, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) tomorrow = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) + timezone.timedelta(days=1) users = ( ("A", "Europe/London"), ("B", "UTC"), ("C", None), ("D", "America/Montevideo"), ("E", None), ) user_a, user_b, user_c, user_d, user_e = ( make_user_for_organization(organization, username=i, _timezone=tz) for i, tz in users ) shifts = ( # user, priority, start time (h), duration (hs) (user_a, 1, 8, 2), # r1-1: 8-10 / A (user_a, 1, 15, 2), # r1-2: 15-17 / A (user_b, 2, 7, 5), # r2-1: 7-12 / B (user_b, 2, 16, 2), # r2-2: 16-18 / B (user_c, 2, 18, 2), # r2-3: 18-20 / C ) for user, priority, start_h, duration in shifts: data = { "start": tomorrow + timezone.timedelta(hours=start_h), "rotation_start": tomorrow + timezone.timedelta(hours=start_h), "duration": timezone.timedelta(hours=duration), "priority_level": priority, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) # override in the past, won't be listed: 17-18 / D override_data = { "start": tomorrow - timezone.timedelta(days=3), "rotation_start": tomorrow - timezone.timedelta(days=3), "duration": timezone.timedelta(hours=1), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) override.add_rolling_users([[user_d]]) # override: 17-18 / C # this is before C's shift, so it will be listed as upcoming override_data = { "start": tomorrow + timezone.timedelta(hours=17), "rotation_start": tomorrow + timezone.timedelta(hours=17), "duration": timezone.timedelta(hours=1), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) override.add_rolling_users([[user_c]]) # override: 17-18 / E fifteend_days_later = tomorrow + timezone.timedelta(days=15) override_data = { "start": fifteend_days_later + timezone.timedelta(hours=17), "rotation_start": fifteend_days_later + timezone.timedelta(hours=17), "duration": timezone.timedelta(hours=1), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) override.add_rolling_users([[user_e]]) # final schedule: 7-12: B, 15-16: A, 16-17: B, 17-18: C (override), 18-20: C schedule.refresh_ical_final_schedule() url = reverse("api-internal:schedule-next-shifts-per-user", kwargs={"pk": schedule.public_primary_key}) # check for users with shifts in the next week response = client.get(url + "?days=7", format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK expected = { user_a.public_primary_key: ( tomorrow + timezone.timedelta(hours=15), tomorrow + timezone.timedelta(hours=16), user_a.timezone, ), user_b.public_primary_key: ( tomorrow + timezone.timedelta(hours=7), tomorrow + timezone.timedelta(hours=12), user_b.timezone, ), user_c.public_primary_key: ( tomorrow + timezone.timedelta(hours=17), tomorrow + timezone.timedelta(hours=18), user_c.timezone, ), } returned_data = { u: (ev.get("start"), ev.get("end"), ev.get("user_timezone")) for u, ev in response.data["users"].items() } assert returned_data == expected # by default it will check for shifts in the next 45 days response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK # include user E with the override expected[user_e.public_primary_key] = ( fifteend_days_later + timezone.timedelta(hours=17), fifteend_days_later + timezone.timedelta(hours=18), user_e.timezone, ) returned_data = { u: (ev.get("start"), ev.get("end"), ev.get("user_timezone")) for u, ev in response.data["users"].items() } assert returned_data == expected @pytest.mark.django_db def test_next_shifts_per_user_ical_schedule_using_emails( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule ): organization, admin, token = make_organization_and_user_with_plugin_token() client = APIClient() user = make_user_for_organization(organization, username="testing", email="testing@testing.com") # ical file using emails as reference cached_ical_primary_schedule = textwrap.dedent( """ BEGIN:VCALENDAR VERSION:2.0 PRODID:testing CALSCALE:GREGORIAN BEGIN:VEVENT CREATED:20220316T121102Z LAST-MODIFIED:20230127T151619Z DTSTAMP:20230127T151619Z UID:something SUMMARY:testing@testing.com RRULE:FREQ=WEEKLY DTSTART;TZID=Europe/Madrid:20220309T130000 DTEND;TZID=Europe/Madrid:20220309T133000 END:VEVENT BEGIN:VEVENT CREATED:20220316T121102Z LAST-MODIFIED:20230127T151619Z DTSTAMP:20230127T151619Z UID:something-else SUMMARY:testing_unknown@testing.com RRULE:FREQ=WEEKLY DTSTART;TZID=Europe/Madrid:20220309T150000 DTEND;TZID=Europe/Madrid:20220309T153000 END:VEVENT END:VCALENDAR """ ) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, cached_ical_file_primary=cached_ical_primary_schedule, ) schedule.refresh_ical_final_schedule() url = reverse("api-internal:schedule-next-shifts-per-user", kwargs={"pk": schedule.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert set(response.data["users"].keys()) == {user.public_primary_key} @pytest.mark.django_db def test_related_users( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, admin, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) tomorrow = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) + timezone.timedelta(days=1) user_a, user_b, user_c, _ = (make_user_for_organization(organization, username=i) for i in "ABCD") shifts = ( # user, priority, start time (h), duration (hs) (user_a, 1, 8, 2), # r1-1: 8-10 / A (user_b, 2, 16, 2), # r2-2: 16-18 / B ) for user, priority, start_h, duration in shifts: data = { "start": tomorrow + timezone.timedelta(hours=start_h), "rotation_start": tomorrow + timezone.timedelta(hours=start_h), "duration": timezone.timedelta(hours=duration), "priority_level": priority, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) # override: 17-18 / C override_data = { "start": tomorrow + timezone.timedelta(hours=17), "rotation_start": tomorrow + timezone.timedelta(hours=17), "duration": timezone.timedelta(hours=1), "schedule": schedule, } override = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_OVERRIDE, **override_data ) override.add_rolling_users([[user_c]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() url = reverse("api-internal:schedule-related-users", kwargs={"pk": schedule.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK expected = [ScheduleUserSerializer(u).data for u in (user_a, user_b, user_c)] assert sorted(response.data["users"], key=lambda u: u["username"]) == sorted(expected, key=lambda u: u["username"]) @pytest.mark.django_db def test_related_escalation_chains( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, make_escalation_chain, make_escalation_policy, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) # setup escalation chains linked to web schedule escalation_chains = [] for _ in range(3): chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=schedule, ) escalation_chains.append(chain) # setup other unrelated schedule other_schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) other_chain = make_escalation_chain(user.organization) make_escalation_policy( escalation_chain=other_chain, escalation_policy_step=EscalationPolicy.STEP_NOTIFY_SCHEDULE, notify_schedule=other_schedule, ) url = reverse("api-internal:schedule-related-escalation-chains", kwargs={"pk": schedule.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK expected = [{"name": chain.name, "pk": chain.public_primary_key} for chain in escalation_chains] assert sorted(response.data, key=lambda e: e["name"]) == sorted(expected, key=lambda e: e["name"]) @pytest.mark.django_db def test_merging_same_shift_events( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) now = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) start_date = now - timezone.timedelta(days=7) # tomorrow request_date = now + timezone.timedelta(days=1) user_a = make_user_for_organization(organization) user_b = make_user_for_organization(organization) user_c = make_user_for_organization(organization, role=LegacyAccessControlRole.VIEWER) data = { "start": start_date + timezone.timedelta(hours=10), "rotation_start": start_date + timezone.timedelta(hours=10), "duration": timezone.timedelta(hours=2), "priority_level": 1, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user_a, user_c, user_b]]) expected_users = { "users": sorted([user_a.username, user_b.username]), "missing_users": [user_c.username], } # final schedule url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?date={}&days=3".format(request_date.strftime("%Y-%m-%d")) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK returned_users = [ { "users": sorted([u["display_name"] for u in e["users"]]) if e["users"] else None, "missing_users": e["missing_users"], } for e in response.data["events"] if not e["is_gap"] ] for users in returned_users: assert users == expected_users # rotations url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?date={}&days=3&type=rotation".format(request_date.strftime("%Y-%m-%d")) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK returned_users = [ { "users": sorted([u["display_name"] for u in e["users"]]) if e["users"] else None, "missing_users": e["missing_users"], } for e in response.data["events"] if not e["is_gap"] ] for users in returned_users: assert users == expected_users @pytest.mark.django_db def test_filter_events_invalid_type( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) url = reverse("api-internal:schedule-filter-events", kwargs={"pk": schedule.public_primary_key}) url += "?type=invalid" response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_create_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-list") with patch( "apps.api.views.schedule.ScheduleView.create", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.post(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_update_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) with patch( "apps.api.views.schedule.ScheduleView.update", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.put(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status response = client.patch(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_list_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-list") with patch( "apps.api.views.schedule.ScheduleView.list", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_retrieve_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) with patch( "apps.api.views.schedule.ScheduleView.retrieve", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_204_NO_CONTENT), (LegacyAccessControlRole.EDITOR, status.HTTP_204_NO_CONTENT), (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_delete_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) with patch( "apps.api.views.schedule.ScheduleView.destroy", return_value=Response( status=status.HTTP_204_NO_CONTENT, ), ): response = client.delete(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_events_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-events", kwargs={"pk": schedule.public_primary_key}) with patch( "apps.api.views.schedule.ScheduleView.events", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_filter_shift_swaps_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-filter-shift-swaps", kwargs={"pk": schedule.public_primary_key}) with patch( "apps.api.views.schedule.ScheduleView.filter_shift_swaps", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_reload_ical_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) schedule = make_schedule( organization, schedule_class=OnCallScheduleICal, name="test_ical_schedule", ical_url_primary=ICAL_URL, ) client = APIClient() url = reverse("api-internal:schedule-reload-ical", kwargs={"pk": schedule.public_primary_key}) with patch( "apps.api.views.schedule.ScheduleView.reload_ical", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.post(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_notify_oncall_shift_freq_options_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): _, user, token = make_organization_and_user_with_plugin_token(role) url = reverse("api-internal:schedule-notify-oncall-shift-freq-options") client = APIClient() response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_notify_empty_oncall_options_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): _, user, token = make_organization_and_user_with_plugin_token(role) url = reverse("api-internal:schedule-notify-empty-oncall-options") client = APIClient() response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_schedule_mention_options_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, role, expected_status, ): _, user, token = make_organization_and_user_with_plugin_token(role) url = reverse("api-internal:schedule-mention-options") client = APIClient() response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_current_user_events_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): organization, user, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:schedule-current-user-events") with patch( "apps.api.views.schedule.ScheduleView.current_user_events", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status @pytest.mark.django_db def test_get_schedule_from_other_team_with_flag( make_organization_and_user_with_plugin_token, make_team, make_user_auth_headers, make_schedule, ): organization, user, token = make_organization_and_user_with_plugin_token() team = make_team(organization) calendar_schedule = make_schedule( organization, schedule_class=OnCallScheduleCalendar, name="test_calendar_schedule", team=team, ) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": calendar_schedule.public_primary_key}) url = f"{url}?from_organization=true" response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_get_schedule_on_call_now( make_organization, make_user_for_organization, make_token_for_organization, make_schedule, make_user_auth_headers ): organization = make_organization(grafana_url="https://example.com") user = make_user_for_organization(organization, username="test", avatar_url="/avatar/test123") _, token = make_token_for_organization(organization) schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, name="test_web_schedule", ) client = APIClient() url = reverse("api-internal:schedule-list") with patch( "apps.api.views.schedule.get_oncall_users_for_multiple_schedules", return_value={schedule: [user]}, ): response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["results"][0]["on_call_now"] == [ { "pk": user.public_primary_key, "username": "test", "avatar": "/avatar/test123", "avatar_full": "https://example.com/avatar/test123", } ] @pytest.mark.django_db def test_current_user_events( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_user_for_organization, make_schedule, make_on_call_shift, ): organization, current_user, token = make_organization_and_user_with_plugin_token() other_user = make_user_for_organization(organization) client = APIClient() url = reverse("api-internal:schedule-current-user-events") schedule_with_current_user = make_schedule(organization, schedule_class=OnCallScheduleWeb) other_schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) shifts = ( # schedule, user, priority, start time (h), duration (seconds) (other_schedule, other_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 (schedule_with_current_user, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 ) now = timezone.now() today = now.replace(hour=0, minute=0, second=0, microsecond=0) for schedule, user, priority, start_h, duration in shifts: data = { "start": today + timezone.timedelta(hours=start_h), "rotation_start": today + timezone.timedelta(hours=start_h), "duration": timezone.timedelta(seconds=duration), "priority_level": priority, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() response = client.get(url, format="json", **make_user_auth_headers(current_user, token)) result = response.json() assert response.status_code == status.HTTP_200_OK assert result["is_oncall"] is True assert len(result["schedules"]) == 1 assert result["schedules"][0]["id"] == schedule_with_current_user.public_primary_key assert result["schedules"][0]["name"] == schedule_with_current_user.name assert len(result["schedules"][0]["events"]) > 0 for event in result["schedules"][0]["events"]: # check the current user shift is populated assert event["shift"] == { "pk": on_call_shift.public_primary_key, "name": on_call_shift.name, "type": on_call_shift.type, } @pytest.mark.django_db def test_current_user_events_out_of_range( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, current_user, token = make_organization_and_user_with_plugin_token() client = APIClient() schedule_with_current_user = make_schedule(organization, schedule_class=OnCallScheduleWeb) shifts = ( # schedule, user, priority, start time (h), duration (seconds) (schedule_with_current_user, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 ) now = timezone.now() today = now.replace(hour=0, minute=0, second=0, microsecond=0) days = 3 start_date = today + timezone.timedelta(days=days) for schedule, user, priority, start_h, duration in shifts: data = { "start": start_date + timezone.timedelta(hours=start_h), "rotation_start": start_date + timezone.timedelta(hours=start_h), "duration": timezone.timedelta(seconds=duration), "priority_level": priority, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() url = reverse("api-internal:schedule-current-user-events") + f"?days={days}" response = client.get(url, format="json", **make_user_auth_headers(current_user, token)) result = response.json() assert response.status_code == status.HTTP_200_OK assert result["is_oncall"] is False assert len(result["schedules"]) == 0 @pytest.mark.django_db def test_current_user_events_no_schedules( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, current_user, token = make_organization_and_user_with_plugin_token() client = APIClient() url = reverse("api-internal:schedule-current-user-events") response = client.get(url, format="json", **make_user_auth_headers(current_user, token)) result = response.json() assert response.status_code == status.HTTP_200_OK assert result["is_oncall"] is False assert len(result["schedules"]) == 0 @pytest.mark.django_db def test_current_user_events_multiple_schedules( make_organization_and_user_with_plugin_token, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization, current_user, token = make_organization_and_user_with_plugin_token() client = APIClient() url = reverse("api-internal:schedule-current-user-events") schedule_1 = make_schedule(organization, schedule_class=OnCallScheduleWeb) schedule_2 = make_schedule(organization, schedule_class=OnCallScheduleWeb) shifts = ( # schedule, user, priority, start time (h), duration (seconds) (schedule_1, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 (schedule_2, current_user, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 ) now = timezone.now() today = now.replace(hour=0, minute=0, second=0, microsecond=0) for schedule, user, priority, start_h, duration in shifts: data = { "start": today + timezone.timedelta(hours=start_h), "rotation_start": today + timezone.timedelta(hours=start_h), "duration": timezone.timedelta(seconds=duration), "priority_level": priority, "frequency": CustomOnCallShift.FREQUENCY_DAILY, "schedule": schedule, } on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, **data ) on_call_shift.add_rolling_users([[user]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() response = client.get(url, format="json", **make_user_auth_headers(current_user, token)) result = response.json() assert response.status_code == status.HTTP_200_OK assert result["is_oncall"] is True assert len(result["schedules"]) == 2 assert result["schedules"][0]["id"] != result["schedules"][1]["id"] assert result["schedules"][0]["id"] in (schedule_1.public_primary_key, schedule_2.public_primary_key) assert result["schedules"][0]["name"] in (schedule_1.name, schedule_2.name) assert result["schedules"][1]["id"] in (schedule_1.public_primary_key, schedule_2.public_primary_key) assert result["schedules"][1]["name"] in (schedule_1.name, schedule_2.name) assert len(result["schedules"][0]["events"]) > 0 assert len(result["schedules"][1]["events"]) > 0 @pytest.mark.django_db def test_team_not_updated_if_not_in_data( make_organization_and_user_with_plugin_token, make_team, make_schedule, make_user_auth_headers, ): organization, user, token = make_organization_and_user_with_plugin_token() team = make_team(organization) schedule = make_schedule(organization, team=team, schedule_class=OnCallScheduleWeb) assert schedule.team == team client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) data = {"name": "renamed", "type": 2} response = client.put(url, data=data, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["team"] == schedule.team.public_primary_key schedule.refresh_from_db() assert schedule.team == team # we don't need to validate the ical URL when creating an ical schedule.. so just patch that functionality @patch("apps.api.serializers.schedule_ical.ScheduleICalSerializer.validate_ical_url_primary", return_value=ICAL_URL) @pytest.mark.parametrize( "schedule_type,other_create_data", [ (0, {}), (1, {"ical_url_primary": ICAL_URL}), (2, {}), ], ) @pytest.mark.django_db def test_can_update_slack_channel( _mock_validate_ical_url_primary, make_organization_and_user_with_plugin_token, make_slack_team_identity, make_slack_channel, make_user_auth_headers, schedule_type, other_create_data, ): organization, user, token = make_organization_and_user_with_plugin_token() auth_headers = make_user_auth_headers(user, token) slack_team_identity = make_slack_team_identity() organization.slack_team_identity = slack_team_identity organization.save() slack_channel1 = make_slack_channel(slack_team_identity) slack_channel2 = make_slack_channel(slack_team_identity) client = APIClient() # we can set it when creating response = client.post( reverse("api-internal:schedule-list"), { "name": "created_schedule", "type": schedule_type, "slack_channel_id": slack_channel1.public_primary_key, **other_create_data, }, format="json", **auth_headers, ) assert response.status_code == status.HTTP_201_CREATED response_data = response.json() schedule_id = response_data["id"] url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule_id}) # NOTE: the response returned by the POST/PUT endpoint currently doesn't include slack_channel_id # as it's not used by the UI.. additionally, there was already a bug in it that despite specifying it, it # would return null.. the proper way to refactor this is to change the name of slack_channel_id used in the # request (as this clashes with the slack_channel_id db column) def _assert_slack_channel_updated(new_slack_channel): response = client.get(url, **auth_headers) assert response.status_code == status.HTTP_200_OK assert response.json()["slack_channel"] == new_slack_channel # we can update it response = client.patch( url, data={ "slack_channel_id": slack_channel2.public_primary_key, }, format="json", **auth_headers, ) assert response.status_code == status.HTTP_200_OK _assert_slack_channel_updated( { "id": slack_channel2.public_primary_key, "display_name": slack_channel2.name, "slack_id": slack_channel2.slack_id, } ) # we can unset it response = client.patch(url, data={"slack_channel_id": None}, format="json", **auth_headers) assert response.status_code == status.HTTP_200_OK _assert_slack_channel_updated(None) @patch.object(SlackUserGroup, "can_be_updated", new_callable=PropertyMock) @pytest.mark.django_db def test_can_update_user_groups( mock_user_group_can_be_updated, make_organization_and_user_with_plugin_token, make_slack_team_identity, make_schedule, make_slack_user_group, make_user_auth_headers, ): mock_user_group_can_be_updated.return_value = True organization, user, token = make_organization_and_user_with_plugin_token() slack_team_identity = make_slack_team_identity() organization.slack_team_identity = slack_team_identity organization.save() inactive_user_group = make_slack_user_group(slack_team_identity, is_active=False) schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb, user_group=inactive_user_group) client = APIClient() url = reverse("api-internal:schedule-detail", kwargs={"pk": schedule.public_primary_key}) response = client.get(url, **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["warnings"] == [ScheduleBaseSerializer.CANT_UPDATE_USER_GROUP_WARNING] mock_user_group_can_be_updated.assert_not_called() # should not be called for inactive user group (is_active=False) active_user_group = make_slack_user_group(slack_team_identity, is_active=True) schedule.user_group = active_user_group schedule.save() response = client.get(url, **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["warnings"] == [] mock_user_group_can_be_updated.assert_called_once() # should be called for active user group (is_active=True)