from unittest import mock from unittest.mock import Mock, PropertyMock, patch import pytest from django.core.cache import cache from django.core.exceptions import ObjectDoesNotExist from django.test import override_settings 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.api import permissions from apps.api.serializers.user import UserHiddenFieldsSerializer from apps.api.views.user import UPCOMING_SHIFTS_DEFAULT_DAYS from apps.base.models import UserNotificationPolicy from apps.phone_notifications.exceptions import FailedToFinishVerification from apps.schedules.models import CustomOnCallShift, OnCallScheduleWeb from apps.user_management.models.user import default_working_hours @pytest.fixture(autouse=True) def clear_cache(): # Ratelimit keys are stored in cache, clean to prevent ratelimits cache.clear() @pytest.mark.django_db def test_current_user( 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() url = reverse("api-internal:api-user") expected_response = { "pk": user.public_primary_key, "organization": {"pk": organization.public_primary_key, "name": organization.org_title}, "current_team": None, "email": user.email, "hide_phone_number": False, "username": user.username, "name": user.name, "role": user.role, "rbac_permissions": user.permissions, "timezone": None, "working_hours": default_working_hours(), "is_currently_oncall": False, "unverified_phone_number": None, "verified_phone_number": None, "telegram_configuration": None, "messaging_backends": { "TESTONLY": { "user": user.username, } }, "cloud_connection_status": mock.ANY, "notification_chain_verbal": {"default": "", "important": ""}, "slack_user_identity": None, "avatar": user.avatar_url, "avatar_full": user.avatar_full_url(organization), "has_google_oauth2_connected": False, "google_calendar_settings": None, "google_oauth2_token_is_missing_scopes": False, } response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json() == expected_response # current user is on-call today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb) on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, start=today, rotation_start=today, duration=timezone.timedelta(seconds=24 * 60 * 60), priority_level=1, frequency=CustomOnCallShift.FREQUENCY_DAILY, schedule=schedule, ) 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(user, token)) assert response.status_code == status.HTTP_200_OK expected_response["is_currently_oncall"] = True assert response.json() == expected_response data_to_update = {"hide_phone_number": True} response = client.put(url, data=data_to_update, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json() == expected_response | data_to_update @pytest.mark.django_db def test_update_user( make_organization, make_team, make_user_for_organization, make_token_for_organization, make_user_auth_headers, ): organization = make_organization() admin = make_user_for_organization(organization) team = make_team(organization) team.users.add(admin) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": admin.public_primary_key}) data = { "unverified_phone_number": "+79123456789", "current_team": team.public_primary_key, } response = client.put(url, data, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["unverified_phone_number"] == data["unverified_phone_number"] assert response.json()["current_team"] == data["current_team"] @pytest.mark.parametrize("oncall_schedules_to_consider_for_shift_swaps", [True, False]) @pytest.mark.django_db def test_update_user_google_calendar_settings( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_schedule, oncall_schedules_to_consider_for_shift_swaps, ): organization = make_organization() admin = make_user_for_organization(organization) _, token = make_token_for_organization(organization) schedule1 = make_schedule(organization, schedule_class=OnCallScheduleWeb) schedule2 = make_schedule(organization, schedule_class=OnCallScheduleWeb) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": admin.public_primary_key}) schedule_public_primary_keys = ( [schedule1.public_primary_key, schedule2.public_primary_key] if oncall_schedules_to_consider_for_shift_swaps else [] ) data = { "google_calendar_settings": { "oncall_schedules_to_consider_for_shift_swaps": schedule_public_primary_keys, }, } response = client.put(url, data, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["google_calendar_settings"] == { "oncall_schedules_to_consider_for_shift_swaps": schedule_public_primary_keys, } @override_settings(GRAFANA_CLOUD_NOTIFICATIONS_ENABLED=False) @pytest.mark.django_db def test_update_user_cant_change_email_and_username( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, ): organization = make_organization() admin = make_user_for_organization(organization) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": admin.public_primary_key}) phone_number = "+79123456789" data = { "unverified_phone_number": phone_number, "email": "test@amixr.io", "username": "bob", } expected_response = { "pk": admin.public_primary_key, "organization": {"pk": organization.public_primary_key, "name": organization.org_title}, "current_team": None, "email": admin.email, "hide_phone_number": False, "username": admin.username, "name": admin.name, "role": admin.role, "timezone": None, "working_hours": default_working_hours(), "is_currently_oncall": False, "unverified_phone_number": phone_number, "verified_phone_number": None, "telegram_configuration": None, "messaging_backends": { "TESTONLY": { "user": admin.username, } }, "cloud_connection_status": None, "notification_chain_verbal": {"default": "", "important": ""}, "slack_user_identity": None, "avatar": admin.avatar_url, "avatar_full": admin.avatar_full_url(organization), "has_google_oauth2_connected": False, "google_calendar_settings": None, } response = client.put(url, data, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert response.json() == expected_response @override_settings(GRAFANA_CLOUD_NOTIFICATIONS_ENABLED=False) @pytest.mark.django_db def test_list_users( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, ): organization = make_organization() admin = make_user_for_organization(organization, _verified_phone_number="1234567890") editor = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-list") expected_payload = { "count": 2, "next": None, "previous": None, "results": [ { "pk": admin.public_primary_key, "organization": {"pk": organization.public_primary_key, "name": organization.org_title}, "current_team": None, "email": admin.email, "hide_phone_number": False, "username": admin.username, "name": admin.name, "role": admin.role, "timezone": None, "working_hours": default_working_hours(), "unverified_phone_number": None, "verified_phone_number": admin.verified_phone_number, "telegram_configuration": None, "messaging_backends": { "TESTONLY": { "user": admin.username, } }, "notification_chain_verbal": {"default": "", "important": ""}, "slack_user_identity": None, "avatar": admin.avatar_url, "avatar_full": admin.avatar_full_url(organization), "cloud_connection_status": None, "has_google_oauth2_connected": False, }, { "pk": editor.public_primary_key, "organization": {"pk": organization.public_primary_key, "name": organization.org_title}, "current_team": None, "email": editor.email, "hide_phone_number": False, "username": editor.username, "name": editor.name, "role": editor.role, "timezone": None, "working_hours": default_working_hours(), "unverified_phone_number": None, "verified_phone_number": None, "telegram_configuration": None, "messaging_backends": { "TESTONLY": { "user": editor.username, } }, "notification_chain_verbal": {"default": "", "important": ""}, "slack_user_identity": None, "avatar": editor.avatar_url, "avatar_full": editor.avatar_full_url(organization), "cloud_connection_status": None, "has_google_oauth2_connected": False, }, ], "current_page_number": 1, "page_size": 100, "total_pages": 1, } # as admin response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert response.json() == expected_payload # as editor response = client.get(url, format="json", **make_user_auth_headers(editor, token)) assert response.status_code == status.HTTP_200_OK admin_user_data = expected_payload["results"][0] for f_name in admin_user_data: if f_name not in UserHiddenFieldsSerializer.fields_available_for_all_users: admin_user_data[f_name] = "******" admin_user_data["hidden_fields"] = True assert response.json() == expected_payload @pytest.mark.parametrize("is_grafana_irm_enabled", [False, True]) @pytest.mark.django_db def test_list_users_filtered_by_granted_permission( is_grafana_irm_enabled, make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, ): permission = permissions.RBACPermission.Permissions.NOTIFICATIONS_READ admin_perm_required_to_call_endpoint = permissions.RBACPermission.Permissions.USER_SETTINGS_READ perm_to_filter_on = ( permissions.convert_oncall_permission_to_irm(permission) if is_grafana_irm_enabled else permission.value ) perms_to_grant = permissions.GrafanaAPIPermissions.construct_permissions([perm_to_filter_on]) organization = make_organization(is_grafana_irm_enabled=is_grafana_irm_enabled, is_rbac_permissions_enabled=True) admin_user = make_user_for_organization( organization, # NOTE: need to explicitly grant this permission here because otherwise the permissions granted by the # make_user_for_organization fixture will only grant the oncall flavour of the permission permissions=permissions.GrafanaAPIPermissions.construct_permissions( [ permissions.convert_oncall_permission_to_irm(admin_perm_required_to_call_endpoint) if is_grafana_irm_enabled else admin_perm_required_to_call_endpoint.value ] ), ) user1 = make_user_for_organization(organization, permissions=perms_to_grant) user2 = make_user_for_organization(organization, permissions=perms_to_grant) user3 = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-list") response = client.get( f"{url}?permission={perm_to_filter_on}", format="json", **make_user_auth_headers(admin_user, token) ) assert response.status_code == status.HTTP_200_OK returned_user_pks = [u["pk"] for u in response.json()["results"]] assert len(returned_user_pks) == 2 assert user1.public_primary_key in returned_user_pks assert user2.public_primary_key in returned_user_pks assert user3.public_primary_key not in returned_user_pks @pytest.mark.django_db def test_list_users_filtered_by_public_primary_key( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, ): organization = make_organization() admin_user = make_user_for_organization(organization) user1 = make_user_for_organization(organization) make_user_for_organization(organization) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-list") response = client.get( f"{url}?search={user1.public_primary_key}", format="json", **make_user_auth_headers(admin_user, token) ) assert response.status_code == status.HTTP_200_OK returned_user_pks = [u["pk"] for u in response.json()["results"]] assert returned_user_pks == [user1.public_primary_key] @pytest.mark.django_db def test_list_users_filtered_by_team( make_organization, make_team, make_user_for_organization, make_token_for_organization, make_user_auth_headers, ): organization = make_organization() user1 = make_user_for_organization(organization) user2 = make_user_for_organization(organization) user3 = make_user_for_organization(organization) team1 = make_team(organization) team2 = make_team(organization) team3 = make_team(organization) user1.teams.add(team1) user1.teams.add(team2) user2.teams.add(team2) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-list") def _get_user_pks(teams): response = client.get( url, data={"team": [team.public_primary_key if team else "null" for team in teams]}, # these are query params **make_user_auth_headers(user1, token), ) assert response.status_code == status.HTTP_200_OK return [u["pk"] for u in response.json()["results"]] assert _get_user_pks([team1]) == [user1.public_primary_key] assert _get_user_pks([team1, team2]) == [user1.public_primary_key, user2.public_primary_key] assert _get_user_pks([team3]) == [] assert _get_user_pks([team1, None]) == [user1.public_primary_key, user3.public_primary_key] assert _get_user_pks([None]) == [user3.public_primary_key] # check non-existent team returns bad request response = client.get(f"{url}?team=non-existing", **make_user_auth_headers(user1, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @pytest.mark.django_db def test_notification_chain_verbal( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_user_notification_policy, ): organization = make_organization() admin = make_user_for_organization(organization) _, token = make_token_for_organization(organization) data_for_creation = [ {"step": UserNotificationPolicy.Step.NOTIFY, "notify_by": UserNotificationPolicy.NotificationChannel.SLACK}, {"step": UserNotificationPolicy.Step.WAIT, "wait_delay": timezone.timedelta(minutes=5)}, { "step": UserNotificationPolicy.Step.NOTIFY, "notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL, }, {"step": UserNotificationPolicy.Step.WAIT, "wait_delay": None}, {"step": UserNotificationPolicy.Step.NOTIFY, "notify_by": UserNotificationPolicy.NotificationChannel.TELEGRAM}, {"step": None}, { "step": UserNotificationPolicy.Step.NOTIFY, "notify_by": UserNotificationPolicy.NotificationChannel.SLACK, "important": True, }, {"step": UserNotificationPolicy.Step.WAIT, "wait_delay": timezone.timedelta(minutes=5), "important": True}, { "step": UserNotificationPolicy.Step.NOTIFY, "notify_by": UserNotificationPolicy.NotificationChannel.PHONE_CALL, "important": True, }, {"step": UserNotificationPolicy.Step.WAIT, "wait_delay": None, "important": True}, { "step": UserNotificationPolicy.Step.NOTIFY, "notify_by": UserNotificationPolicy.NotificationChannel.TELEGRAM, "important": True, }, ] for data in data_for_creation: make_user_notification_policy(admin, **data) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": admin.public_primary_key}) expected_notification_chain = { "default": "Slack - 5 min - \U0000260E - Telegram", "important": "Slack - 5 min - \U0000260E - Telegram", } response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert response.json()["notification_chain_verbal"] == expected_notification_chain @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_update_self_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": tester.public_primary_key}) with patch( "apps.api.views.user.UserView.update", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.put(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_update_other_permissions( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, role, expected_status, ): organization, tester, token = make_organization_and_user_with_plugin_token(role) admin = make_user_for_organization(organization) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": admin.public_primary_key}) data = {"unverified_phone_number": "+79123456789"} response = client.put(url, data, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_list_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:user-list") with patch( "apps.api.views.user.UserView.list", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_detail_self_permissions( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": tester.public_primary_key}) with patch( "apps.api.views.user.UserView.list", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_detail_other_permissions( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, role, expected_status, ): organization, tester, token = make_organization_and_user_with_plugin_token(role) admin = make_user_for_organization(organization) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": admin.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status # hidden information for editor/viewer available_fields = UserHiddenFieldsSerializer.fields_available_for_all_users + ["hidden_fields"] if role in (permissions.LegacyAccessControlRole.EDITOR, permissions.LegacyAccessControlRole.VIEWER): user_details = response.json() for f_name in user_details: if f_name not in available_fields: assert user_details[f_name] == "******" @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_get_own_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": tester.public_primary_key}) with patch( "apps.api.views.user.UserView.get_verification_code", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_get_other_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, role, expected_status, ): organization, tester, token = make_organization_and_user_with_plugin_token(role) admin = make_user_for_organization(organization) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": admin.public_primary_key}) with patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()): response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db def test_validation_of_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) with patch( "apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True ) as verify_phone_number: url_with_token = f"{url}?token=some_token" r = client.put(url_with_token, format="json", **make_user_auth_headers(user, token)) assert r.status_code == 200 assert verify_phone_number.call_count == 1 url_without_token = f"{url}" r = client.put(url_without_token, format="json", **make_user_auth_headers(user, token)) assert r.status_code == 400 assert verify_phone_number.call_count == 1 url_with_empty_token = f"{url}?token=" r = client.put(url_with_empty_token, format="json", **make_user_auth_headers(user, token)) assert r.status_code == 400 assert verify_phone_number.call_count == 1 @pytest.mark.django_db def test_verification_code_provider_exception( make_organization_and_user_with_plugin_token, make_user_auth_headers, ): organization, user, token = make_organization_and_user_with_plugin_token() client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) with patch( "apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", side_effect=FailedToFinishVerification, ) as verify_phone_number: url_with_token = f"{url}?token=some_token" r = client.put(url_with_token, format="json", **make_user_auth_headers(user, token)) assert r.status_code == 503 assert verify_phone_number.call_count == 1 @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_verify_own_phone( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": tester.public_primary_key}) with patch( "apps.api.views.user.UserView.verify_number", return_value=Response( status=status.HTTP_200_OK, ), ): response = client.put(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_400_BAD_REQUEST), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_400_BAD_REQUEST), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_get_own_telegram_verification_code_with_telegram_connected( make_organization_and_user_with_plugin_token, make_telegram_user_connector, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() make_telegram_user_connector(tester) url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": tester.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status """ Tests below are outdated """ @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_verify_another_phone( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, role, expected_status, ): organization, tester, token = make_organization_and_user_with_plugin_token(role) other_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": other_user.public_primary_key}) with patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True): response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_get_own_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers, role, expected_status, ): _, tester, token = make_organization_and_user_with_plugin_token(role) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": tester.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN), (permissions.LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN), ], ) def test_user_get_another_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, role, expected_status, ): organization, tester, token = make_organization_and_user_with_plugin_token(role) other_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": other_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(tester, token)) assert response.status_code == expected_status @pytest.mark.django_db def test_admin_can_update_user( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() data = { "email": "test@amixr.io", "username": "updated_test_username", "unverified_phone_number": "+1234567890", "slack_login": "", } url = reverse("api-internal:user-detail", kwargs={"pk": first_user.public_primary_key}) response = client.put(url, format="json", data=data, **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_admin_can_update_himself(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() data = { "email": "test@amixr.io", "username": "updated_test_username", "unverified_phone_number": "+1234567890", "slack_login": "", } url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) response = client.put(url, format="json", data=data, **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_admin_can_list_users(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-list") 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_admin_can_detail_users( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": first_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.django_db def test_admin_can_get_own_verification_code( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_auth_headers, ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.django_db def test_admin_can_get_another_user_verification_code( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": first_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_admin_can_verify_own_phone( mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_auth_headers, ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_admin_can_verify_another_user_phone( mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": first_user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_admin_can_get_own_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": user.public_primary_key}) 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_admin_can_get_another_user_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": first_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_admin_can_get_another_user_backend_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = ( reverse("api-internal:user-get-backend-verification-code", kwargs={"pk": first_user.public_primary_key}) + "?backend=TESTONLY" ) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_admin_can_unlink_another_user_backend_account( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.ADMIN) client = APIClient() url = ( reverse("api-internal:user-unlink-backend", kwargs={"pk": first_user.public_primary_key}) + "?backend=TESTONLY" ) response = client.post(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_admin_can_unlink_another_user_slack_account( make_organization_with_slack_team_identity, make_user_for_organization, make_user_with_slack_user_identity, make_token_for_organization, make_user_auth_headers, ): organization, slack_team_identity = make_organization_with_slack_team_identity() _, token = make_token_for_organization(organization) user, _ = make_user_with_slack_user_identity( slack_team_identity, organization, slack_id="user_2", role=permissions.LegacyAccessControlRole.ADMIN ) other_user = make_user_for_organization(organization) client = APIClient() url = reverse("api-internal:user-unlink-slack", kwargs={"pk": other_user.public_primary_key}) response = client.post(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK other_user.refresh_from_db() assert other_user.slack_user_identity is None """Test user permissions""" @pytest.mark.django_db def test_user_cant_update_user( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() data = { "email": "test@amixr.io", "username": "updated_test_username", "unverified_phone_number": "+1234567890", "slack_login": "", } url = reverse("api-internal:user-detail", kwargs={"pk": first_user.public_primary_key}) response = client.put(url, format="json", data=data, **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_user_can_update_themself(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() data = { "email": "test@amixr.io", "username": "updated_test_username", "unverified_phone_number": "+1234567890", "slack_login": "", } url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) response = client.put(url, format="json", data=data, **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_user_can_list_users(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-list") 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_user_can_detail_users( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": first_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_200_OK # hidden information though user_details = response.json() available_fields = UserHiddenFieldsSerializer.fields_available_for_all_users + ["hidden_fields"] for f_name in user_details: if f_name not in available_fields: assert user_details[f_name] == "******" @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.django_db def test_user_can_get_own_verification_code( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": user.public_primary_key}) response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.django_db def test_user_cant_get_another_user_verification_code( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": first_user.public_primary_key}) response = client.get(f"{url}", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_user_can_verify_own_phone( mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_user_cant_verify_another_user_phone( mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": first_user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_user_can_get_own_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": user.public_primary_key}) response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_user_cant_get_another_user_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": first_user.public_primary_key}) response = client.get(f"{url}", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_user_can_get_own_backend_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = ( reverse("api-internal:user-get-backend-verification-code", kwargs={"pk": user.public_primary_key}) + "?backend=TESTONLY" ) with patch( "apps.base.tests.messaging_backend.TestOnlyBackend.generate_user_verification_code", return_value="the-code", ) as mock_generate_code: response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert response.json() == "the-code" mock_generate_code.assert_called_once_with(user) @pytest.mark.django_db def test_user_cant_get_another_user_backend_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = ( reverse("api-internal:user-get-backend-verification-code", kwargs={"pk": first_user.public_primary_key}) + "?backend=TESTONLY" ) response = client.get(f"{url}", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_user_can_unlink_own_slack_account( make_organization_with_slack_team_identity, make_user_with_slack_user_identity, make_token_for_organization, make_user_auth_headers, ): organization, slack_team_identity = make_organization_with_slack_team_identity() user, _ = make_user_with_slack_user_identity( slack_team_identity, organization, slack_id="user_2", role=permissions.LegacyAccessControlRole.EDITOR ) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-unlink-slack", kwargs={"pk": user.public_primary_key}) response = client.post(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK user.refresh_from_db() assert user.slack_user_identity is None @pytest.mark.django_db def test_user_can_unlink_backend_own_account(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-unlink-backend", kwargs={"pk": user.public_primary_key}) + "?backend=TESTONLY" response = client.post(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @pytest.mark.django_db def test_user_unlink_backend_invalid_backend_id(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-unlink-backend", kwargs={"pk": user.public_primary_key}) + "?backend=INVALID" response = client.post(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @pytest.mark.django_db def test_user_unlink_backend_backend_account_not_found( make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-unlink-backend", kwargs={"pk": user.public_primary_key}) + "?backend=TESTONLY" with patch("apps.base.tests.messaging_backend.TestOnlyBackend.unlink_user", side_effect=ObjectDoesNotExist): response = client.post(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @pytest.mark.django_db def test_user_cant_unlink_slack_another_user( make_organization_with_slack_team_identity, make_user_with_slack_user_identity, make_token_for_organization, make_user_auth_headers, ): organization, slack_team_identity = make_organization_with_slack_team_identity() first_user, _ = make_user_with_slack_user_identity( slack_team_identity, organization, slack_id="user_1", role=permissions.LegacyAccessControlRole.EDITOR ) second_user, _ = make_user_with_slack_user_identity( slack_team_identity, organization, slack_id="user_2", role=permissions.LegacyAccessControlRole.EDITOR ) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-unlink-slack", kwargs={"pk": first_user.public_primary_key}) response = client.post(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN first_user.refresh_from_db() assert first_user.slack_user_identity is not None @pytest.mark.django_db def test_user_cant_unlink_backend_another_user( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = ( reverse("api-internal:user-unlink-backend", kwargs={"pk": first_user.public_primary_key}) + "?backend=TESTONLY" ) response = client.post(f"{url}", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN """Test stakeholder permissions""" @pytest.mark.django_db def test_viewer_cant_update_user( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.VIEWER ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) data = { "email": "test@amixr.io", "role": permissions.LegacyAccessControlRole.EDITOR, "username": "updated_test_username", "unverified_phone_number": "+1234567890", "slack_login": "", } client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": first_user.public_primary_key}) response = client.put(url, format="json", data=data, **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_viewer_cant_update_himself(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.VIEWER) data = { "email": "test@amixr.io", "role": permissions.LegacyAccessControlRole.VIEWER, "username": "updated_test_username", "unverified_phone_number": "+1234567890", "slack_login": "", } client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) response = client.put(url, format="json", data=data, **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_viewer_can_list_users(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-list") response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.django_db def test_viewer_cant_get_own_verification_code( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.django_db def test_viewer_cant_get_another_user_verification_code( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": first_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_viewer_cant_verify_own_phone( mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_viewer_cant_verify_another_user_phone( mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-verify-number", kwargs={"pk": first_user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_viewer_cant_get_own_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_auth_headers ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": user.public_primary_key}) response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_viewer_cant_get_another_user_telegram_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": first_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status,initial_unverified_number,initial_verified_number", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK, "+1234567890", None), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK, "+1234567890", None), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN, "+1234567890", None), (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK, None, "+1234567890"), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_200_OK, None, "+1234567890"), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN, None, "+1234567890"), ], ) def test_forget_own_number( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers, role, expected_status, initial_unverified_number, initial_verified_number, ): organization, admin, token = make_organization_and_user_with_plugin_token() user = make_user_for_organization( organization, role=role, unverified_phone_number=initial_unverified_number, _verified_phone_number=initial_verified_number, ) client = APIClient() url = reverse("api-internal:user-forget-number", kwargs={"pk": user.public_primary_key}) with patch("apps.phone_notifications.phone_backend.PhoneBackend._notify_disconnected_number", return_value=None): response = client.put(url, None, format="json", **make_user_auth_headers(user, token)) assert response.status_code == expected_status user_detail_url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) response = client.get(user_detail_url, None, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK if expected_status == status.HTTP_200_OK: assert not response.json()["unverified_phone_number"] assert not response.json()["verified_phone_number"] else: assert response.json()["unverified_phone_number"] == initial_unverified_number assert response.json()["verified_phone_number"] == initial_verified_number @pytest.mark.django_db @pytest.mark.parametrize( "role,expected_status,initial_unverified_number,initial_verified_number", [ (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK, "+1234567890", None), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN, "+1234567890", None), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN, "+1234567890", None), (permissions.LegacyAccessControlRole.ADMIN, status.HTTP_200_OK, None, "+1234567890"), (permissions.LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN, None, "+1234567890"), (permissions.LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN, None, "+1234567890"), ], ) def test_forget_other_number( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, role, expected_status, initial_unverified_number, initial_verified_number, ): organization = make_organization() _, token = make_token_for_organization(organization) admin = make_user_for_organization( organization, unverified_phone_number=initial_unverified_number, _verified_phone_number=initial_verified_number ) other_user = make_user_for_organization(organization, role=role) admin_primary_key = admin.public_primary_key client = APIClient() url = reverse("api-internal:user-forget-number", kwargs={"pk": admin_primary_key}) with patch("apps.phone_notifications.phone_backend.PhoneBackend._notify_disconnected_number", return_value=None): response = client.put(url, None, format="json", **make_user_auth_headers(other_user, token)) assert response.status_code == expected_status user_detail_url = reverse("api-internal:user-detail", kwargs={"pk": admin_primary_key}) response = client.get(user_detail_url, None, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK if expected_status == status.HTTP_200_OK: assert not response.json()["unverified_phone_number"] assert not response.json()["verified_phone_number"] else: assert response.json()["unverified_phone_number"] == initial_unverified_number assert response.json()["verified_phone_number"] == initial_verified_number @pytest.mark.django_db def test_viewer_cant_get_another_user_backend_verification_code( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = ( reverse("api-internal:user-get-backend-verification-code", kwargs={"pk": first_user.public_primary_key}) + "?backend=TESTONLY" ) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_viewer_cant_unlink_backend_another_user( make_organization_and_user_with_plugin_token, make_user_for_organization, make_user_auth_headers ): organization, first_user, token = make_organization_and_user_with_plugin_token( role=permissions.LegacyAccessControlRole.EDITOR ) second_user = make_user_for_organization(organization, role=permissions.LegacyAccessControlRole.VIEWER) client = APIClient() url = ( reverse("api-internal:user-unlink-backend", kwargs={"pk": first_user.public_primary_key}) + "?backend=TESTONLY" ) response = client.post(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_403_FORBIDDEN @pytest.mark.django_db def test_change_timezone(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) data = {"timezone": "Europe/London"} response = client.put(f"{url}", data, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert "timezone" in response.json() assert response.json()["timezone"] == "Europe/London" @pytest.mark.django_db @pytest.mark.parametrize("timezone", ["", 1, "NotATimezone"]) def test_invalid_timezone(make_organization_and_user_with_plugin_token, make_user_auth_headers, timezone): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) data = {"timezone": timezone} response = client.put(f"{url}", data, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @pytest.mark.django_db def test_change_working_hours(make_organization_and_user_with_plugin_token, make_user_auth_headers): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) periods = [{"start": "05:00:00", "end": "23:00:00"}] working_hours = { day: periods for day in ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] } data = {"working_hours": working_hours} response = client.put(f"{url}", data, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK assert "working_hours" in response.json() assert response.json()["working_hours"] == working_hours @pytest.mark.django_db @pytest.mark.parametrize( "working_hours_extra", [ {}, {"sunday": 1}, {"sunday": ""}, {"sunday": {"start": "18:00:00"}}, {"sunday": {"start": "", "end": ""}}, {"sunday": {"start": "18:00:00", "end": None}}, {"sunday": {"start": "18:00:00", "end": "18:00:00"}}, {"sunday": {"start": "18:00:00", "end": "9:00:00"}}, {"sunday": {"start": "18:00:00", "end": "9:00:00", "extra": 1}}, ], ) def test_invalid_working_hours( make_organization_and_user_with_plugin_token, make_user_auth_headers, working_hours_extra ): _, user, token = make_organization_and_user_with_plugin_token(role=permissions.LegacyAccessControlRole.EDITOR) client = APIClient() url = reverse("api-internal:user-detail", kwargs={"pk": user.public_primary_key}) periods = [{"start": "05:00:00", "end": "23:00:00"}] working_hours = {day: periods for day in ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday"]} working_hours.update(working_hours_extra) data = {"working_hours": working_hours} response = client.put(f"{url}", data, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_phone_number_verification_flow_ratelimit_per_user( mock_verification_start, mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_auth_headers, ): _, user, token = make_organization_and_user_with_plugin_token() client = APIClient() url = reverse("api-internal:user-get-verification-code", kwargs={"pk": user.public_primary_key}) with patch( "apps.api.throttlers.GetPhoneVerificationCodeThrottlerPerUser.rate", new_callable=PropertyMock, ) as mocked_rate: mocked_rate.return_value = "1/10m" # first get_verification_code request is succesfull response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK # second get_verification_code request is ratelimited response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) with patch( "apps.api.throttlers.VerifyPhoneNumberThrottlerPerUser.rate", new_callable=PropertyMock, ) as mocked_rate: mocked_rate.return_value = "1/10m" # first verify_number request is succesfull, because it uses different ratelimit scope response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) # second verify_number request is succesfull, because it ratelimited response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @patch("apps.phone_notifications.phone_backend.PhoneBackend.verify_phone_number", return_value=True) @pytest.mark.django_db def test_phone_number_verification_flow_ratelimit_per_org( mock_verification_start, mocked_verification_check, make_organization_and_user_with_plugin_token, make_user_auth_headers, make_user_for_organization, ): """ This test is checks per-org ratelimits for phone verification flow. It makes two get_verification_code and two verify_number requests from different users and expect that second call will be ratelimited. """ org, user, token = make_organization_and_user_with_plugin_token() second_user = make_user_for_organization(org) client = APIClient() with patch( "apps.api.throttlers.GetPhoneVerificationCodeThrottlerPerOrg.rate", new_callable=PropertyMock, ) as mocked_rate: mocked_rate.return_value = "1/10m" url = reverse("api-internal:user-get-verification-code", kwargs={"pk": user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK url = reverse("api-internal:user-get-verification-code", kwargs={"pk": second_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS with patch( "apps.api.throttlers.VerifyPhoneNumberThrottlerPerOrg.rate", new_callable=PropertyMock, ) as mocked_rate: mocked_rate.return_value = "1/10m" url = reverse("api-internal:user-verify-number", kwargs={"pk": user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(user, token)) assert response.status_code == status.HTTP_200_OK url = reverse("api-internal:user-verify-number", kwargs={"pk": second_user.public_primary_key}) response = client.put(f"{url}?token=12345", format="json", **make_user_auth_headers(second_user, token)) assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS @patch("apps.phone_notifications.phone_backend.PhoneBackend.send_verification_sms", return_value=Mock()) @pytest.mark.parametrize( "recaptcha_testing_pass,expected_status", [ (True, status.HTTP_200_OK), (False, status.HTTP_400_BAD_REQUEST), ], ) @pytest.mark.django_db @override_settings(RECAPTCHA_V3_ENABLED=True) def test_phone_number_verification_recaptcha( mock_verification_start, make_organization_and_user_with_plugin_token, make_user_auth_headers, recaptcha_testing_pass, expected_status, ): _, user, token = make_organization_and_user_with_plugin_token() recaptcha_token = "asdasdfasdf" client = APIClient() request_headers = {"HTTP_X-OnCall-Recaptcha": recaptcha_token, **make_user_auth_headers(user, token)} url = reverse("api-internal:user-get-verification-code", kwargs={"pk": user.public_primary_key}) with patch("apps.api.views.user.check_recaptcha_internal_api", return_value=recaptcha_testing_pass): response = client.get(url, format="json", **request_headers) assert response.status_code == expected_status if expected_status == status.HTTP_200_OK: mock_verification_start.assert_called_once_with(user) else: mock_verification_start.assert_not_called() @pytest.mark.django_db @pytest.mark.parametrize( "days", ["invalid", 75, -2, 0], ) def test_upcoming_shifts_invalid_days( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, days ): organization = make_organization() admin = make_user_for_organization(organization) _, token = make_token_for_organization(organization) client = APIClient() url = reverse("api-internal:user-upcoming-shifts", kwargs={"pk": admin.public_primary_key}) + "?days={}".format( days ) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_400_BAD_REQUEST @pytest.mark.django_db def test_upcoming_shifts_oncall( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization = make_organization() admin = make_user_for_organization(organization) other_user = make_user_for_organization(organization) _, token = make_token_for_organization(organization) schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, ) shifts = ( # user, priority, start time (h), duration (seconds) (admin, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 ) today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) for 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() client = APIClient() url = reverse("api-internal:user-upcoming-shifts", kwargs={"pk": admin.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK returned_data = response.data[0] assert returned_data["schedule_id"] == schedule.public_primary_key assert returned_data["schedule_name"] == schedule.name assert returned_data["is_oncall"] assert returned_data["current_shift"]["start"] == on_call_shift.start next_shift_start = on_call_shift.start + timezone.timedelta(days=1) assert returned_data["next_shift"]["start"] == next_shift_start assert len(returned_data["upcoming_shifts"]) == UPCOMING_SHIFTS_DEFAULT_DAYS for i, shift in enumerate(returned_data["upcoming_shifts"], start=1): assert shift["start"] == on_call_shift.start + timezone.timedelta(days=i) # empty response for other user url = reverse("api-internal:user-upcoming-shifts", kwargs={"pk": other_user.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK assert response.data == [] @pytest.mark.django_db def test_upcoming_shifts_override( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization = make_organization() admin = make_user_for_organization(organization) _, token = make_token_for_organization(organization) schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, ) tomorrow = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) + timezone.timedelta(days=1) override_data = { "start": tomorrow + timezone.timedelta(hours=22), "rotation_start": tomorrow + 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([[admin]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() client = APIClient() url = reverse("api-internal:user-upcoming-shifts", kwargs={"pk": admin.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK returned_data = response.data[0] assert returned_data["schedule_id"] == schedule.public_primary_key assert returned_data["schedule_name"] == schedule.name assert returned_data["is_oncall"] is False assert returned_data["current_shift"] is None assert returned_data["next_shift"]["start"] == override.start assert returned_data["upcoming_shifts"] == [returned_data["next_shift"]] @pytest.mark.django_db def test_upcoming_shifts_multiple_schedules( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization = make_organization() admin = make_user_for_organization(organization) _, token = make_token_for_organization(organization) schedules = [] # create schedules in a reversed order to check the output is sorted later for i in range(2, -1, -1): schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, ) shifts = ( # user, priority, start time (h), duration (seconds) (admin, 1, 0, (24 * 60 * 60) - 1), # r1-1: 0-23:59:59 ) today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) for user, priority, start_h, duration in shifts: data = { "start": today + timezone.timedelta(hours=start_h) + timezone.timedelta(days=i), "rotation_start": today + timezone.timedelta(hours=start_h) + timezone.timedelta(days=i), "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() schedules.append(schedule) client = APIClient() url = reverse("api-internal:user-upcoming-shifts", kwargs={"pk": admin.public_primary_key}) num_days = 2 response = client.get(url + f"?days={num_days}", format="json", **make_user_auth_headers(admin, token)) assert response.status_code == status.HTTP_200_OK returned_data = response.data for i, schedule in enumerate(reversed(schedules)): assert returned_data[i]["schedule_name"] == schedule.name expected_start = today + timezone.timedelta(hours=start_h) + timezone.timedelta(days=i) if i == 0: assert returned_data[i]["is_oncall"] assert returned_data[i]["current_shift"]["start"] == expected_start assert returned_data[i]["next_shift"]["start"] == expected_start + timezone.timedelta(days=1) assert len(returned_data[i]["upcoming_shifts"]) == 2 for delta, shift in enumerate(returned_data[i]["upcoming_shifts"], start=1): assert shift["start"] == expected_start + timezone.timedelta(days=delta) else: assert returned_data[i]["is_oncall"] is False assert returned_data[i]["current_shift"] is None assert returned_data[i]["next_shift"]["start"] == expected_start assert len(returned_data[i]["upcoming_shifts"]) == num_days - i + 1 for delta, shift in enumerate(returned_data[i]["upcoming_shifts"], start=0): assert shift["start"] == expected_start + timezone.timedelta(days=delta) @pytest.mark.django_db def test_users_is_currently_oncall_attribute_works_properly( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization = make_organization() user1 = make_user_for_organization(organization) user2 = make_user_for_organization(organization) _, token = make_token_for_organization(organization) schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, ) today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, start=today, rotation_start=today, duration=timezone.timedelta(seconds=24 * 60 * 60), priority_level=1, frequency=CustomOnCallShift.FREQUENCY_DAILY, schedule=schedule, ) on_call_shift.add_rolling_users([[user1]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() client = APIClient() url = f"{reverse('api-internal:user-list')}?is_currently_oncall=all" response = client.get(url, format="json", **make_user_auth_headers(user1, token)) oncall_statuses = { user1.public_primary_key: True, user2.public_primary_key: False, } for user in response.json(): assert user["teams"] == [] assert user["is_currently_oncall"] == oncall_statuses[user["pk"]] # getting specific user details include currently on-call info url = reverse("api-internal:user-detail", kwargs={"pk": user1.public_primary_key}) response = client.get(url, format="json", **make_user_auth_headers(user1, token)) assert response.json()["is_currently_oncall"] @pytest.mark.django_db def test_list_users_filtered_by_is_currently_oncall( make_organization, make_user_for_organization, make_token_for_organization, make_user_auth_headers, make_schedule, make_on_call_shift, ): organization = make_organization() user1 = make_user_for_organization(organization) user2 = make_user_for_organization(organization) _, token = make_token_for_organization(organization) schedule = make_schedule( organization, schedule_class=OnCallScheduleWeb, ) today = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) on_call_shift = make_on_call_shift( organization=organization, shift_type=CustomOnCallShift.TYPE_ROLLING_USERS_EVENT, start=today, rotation_start=today, duration=timezone.timedelta(seconds=24 * 60 * 60), priority_level=1, frequency=CustomOnCallShift.FREQUENCY_DAILY, schedule=schedule, ) on_call_shift.add_rolling_users([[user1]]) schedule.refresh_ical_file() schedule.refresh_ical_final_schedule() client = APIClient() url = reverse("api-internal:user-list") response = client.get(f"{url}?is_currently_oncall=true", format="json", **make_user_auth_headers(user1, token)) response = response.json()["results"] assert len(response) == 1 assert response[0]["pk"] == user1.public_primary_key response = client.get(f"{url}?is_currently_oncall=false", format="json", **make_user_auth_headers(user1, token)) response = response.json()["results"] user = response[0] assert len(response) == 1 assert user["pk"] == user2.public_primary_key assert user["teams"] == []