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

Remove demo token related code/logic

Browse source
This commit is contained in:
Matias Bordese 2022-06-06 16:07:28 -03:00
parent 3fae0c9ff2
commit 0e26568857
41 changed files with 40 additions and 2156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
engine/apps/auth_token/auth.py
View file

@ -9,7 +9,6 @@ from rest_framework.authentication import BaseAuthentication, get_authorization_
from rest_framework.request import Request
from apps.grafana_plugin.helpers.gcom import check_token
from apps.public_api import constants as public_api_constants
from apps.user_management.models import User
from apps.user_management.models.organization import Organization
from common.constants.role import Role
@ -29,12 +28,6 @@ class ApiTokenAuthentication(BaseAuthentication):
def authenticate(self, request):
auth = get_authorization_header(request).decode("utf-8")
if auth == public_api_constants.DEMO_AUTH_TOKEN:
user = User.objects.get(public_primary_key=public_api_constants.DEMO_USER_ID)
auth_token = user.auth_tokens.first()
return user, auth_token
user, auth_token = self.authenticate_credentials(auth)
if user.role != Role.ADMIN:

5
engine/apps/grafana_plugin/tasks/sync.py
View file

@ -6,7 +6,6 @@ from django.utils import timezone
from apps.grafana_plugin.helpers import GcomAPIClient
from apps.grafana_plugin.helpers.gcom import get_active_instance_ids
from apps.public_api.constants import DEMO_ORGANIZATION_ID
from apps.user_management.models import Organization
from apps.user_management.sync import sync_organization
from common.custom_celery_tasks import shared_dedicated_queue_retry_task
@ -23,9 +22,7 @@ SYNC_PERIOD = timezone.timedelta(minutes=25)
def start_sync_organizations():
sync_threshold = timezone.now() - SYNC_PERIOD
organization_qs = Organization.objects.exclude(public_primary_key=DEMO_ORGANIZATION_ID).filter(
last_time_synced__lte=sync_threshold
)
organization_qs = Organization.objects.filter(last_time_synced__lte=sync_threshold)
active_instance_ids, is_cloud_configured = get_active_instance_ids()
if is_cloud_configured:

8
engine/apps/oss_installation/utils.py
View file

@ -1,10 +1,8 @@
import logging
from contextlib import suppress
from django.apps import apps
from django.utils import timezone
from apps.public_api.constants import DEMO_USER_ID
from apps.schedules.ical_utils import list_users_to_notify_from_ical_for_period
logger = logging.getLogger(__name__)
@ -18,7 +16,6 @@ def active_oss_users_count():
AlertGroupLogRecord = apps.get_model("alerts", "AlertGroupLogRecord")
EscalationPolicy = apps.get_model("alerts", "EscalationPolicy")
UserNotificationPolicyLogRecord = apps.get_model("base", "UserNotificationPolicyLogRecord")
User = apps.get_model("user_management", "User")
# Take logs for previous 24 hours
start = timezone.now() - timezone.timedelta(hours=24)
@ -67,9 +64,4 @@ def active_oss_users_count():
for user in users_from_schedule:
unique_active_users.add(user.pk)
# Remove demo user from active users
with suppress(User.DoesNotExist):
demo_user = User.objects.get(public_primary_key=DEMO_USER_ID)
with suppress(KeyError):
unique_active_users.remove(demo_user.pk)
return len(unique_active_users)

66
engine/apps/public_api/constants.py
View file

@ -1,69 +1,3 @@
from django.utils import dateparse
DEMO_USER_ID = "U4DNY931HHJS5"
DEMO_ORGANIZATION_ID = "TCNPY4A1BWUMP"
DEMO_SLACK_USER_ID = "UALEXSLACKDJPK"
DEMO_SLACK_TEAM_ID = "TALEXSLACKDJPK"
DEMO_AUTH_TOKEN = "meowmeowmeow"
DEMO_USER_USERNAME = "Alex"
DEMO_USER_EMAIL = "public-api-demo-user-1@amixr.io"
DEMO_INTEGRATION_ID = "CFRPV98RPR1U8"
DEMO_INTEGRATION_LINK_TOKEN = "mReAoNwDm0eMwKo1mTeTwYo"
DEMO_INTEGRATION_NAME = "Grafana :blush:"
DEMO_ROUTE_ID_1 = "RIYGUJXCPFHXY"
DEMO_ROUTE_ID_2 = "RVBE4RKQSCGJ2"
DEMO_SLACK_CHANNEL_FOR_ROUTE_ID = "CH23212D"
DEMO_ESCALATION_CHAIN_ID = "F5JU6KJET33FE"
DEMO_ESCALATION_POLICY_ID_1 = "E3GA6SJETWWJS"
DEMO_ESCALATION_POLICY_ID_2 = "E5JJTU52M5YM4"
DEMO_SCHEDULE_ID_ICAL = "SBM7DV7BKFUYU"
DEMO_SCHEDULE_ID_CALENDAR = "S3Z477AHDXTMF"
DEMO_SCHEDULE_NAME_ICAL = "Demo schedule iCal"
DEMO_SCHEDULE_NAME_CALENDAR = "Demo schedule Calendar"
DEMO_SCHEDULE_ICAL_URL_PRIMARY = "https://example.com/meow_calendar.ics"
DEMO_SCHEDULE_ICAL_URL_OVERRIDES = "https://example.com/meow_calendar_overrides.ics"
DEMO_INCIDENT_ID = "I68T24C13IFW1"
DEMO_INCIDENT_CREATED_AT = "2020-05-19T12:37:01.430444Z"
DEMO_INCIDENT_RESOLVED_AT = "2020-05-19T13:37:01.429805Z"
DEMO_ALERT_IDS = [
("AA74DN7T4JQB6", "2020-05-11T20:07:43Z"),
("AR9SSYFKE2PV7", "2020-05-11T20:07:54Z"),
("AWJQSGEYYUFGH", "2020-05-11T20:07:58Z"),
]
DEMO_ALERT_PAYLOAD = {
"evalMatches": [
{"value": 100, "metric": "High value", "tags": None},
{"value": 200, "metric": "Higher Value", "tags": None},
],
"message": "Someone is testing the alert notification within grafana.",
"ruleId": 0,
"ruleName": "Test notification",
"ruleUrl": "https://amixr.io/",
"state": "alerting",
"title": "[Alerting] Test notification",
}
VALID_DATE_FOR_DELETE_INCIDENT = dateparse.parse_date("2020-07-04")
DEMO_SLACK_CHANNEL_NAME = "meow_channel"
DEMO_SLACK_CHANNEL_SLACK_ID = "MEOW_SLACK_ID"
DEMO_PERSONAL_NOTIFICATION_ID_1 = "NT79GA9I7E4DJ"
DEMO_PERSONAL_NOTIFICATION_ID_2 = "ND9EHN5LN1DUU"
DEMO_PERSONAL_NOTIFICATION_ID_3 = "NEF49YQ1HNPDD"
DEMO_PERSONAL_NOTIFICATION_ID_4 = "NWAL6WFJNWDD8"
DEMO_RESOLUTION_NOTE_ID = "M4BTQUS3PRHYQ"
DEMO_RESOLUTION_NOTE_TEXT = "Demo resolution note"
DEMO_RESOLUTION_NOTE_CREATED_AT = "2020-06-19T12:40:01.429805Z"
DEMO_RESOLUTION_NOTE_SOURCE = "web"
DEMO_CUSTOM_ACTION_ID = "KGEFG74LU1D8L"
DEMO_CUSTOM_ACTION_NAME = "Publish Incident To Jira"
DEMO_SLACK_USER_GROUP_ID = "GPFAPH7J7BKJB"
DEMO_SLACK_USER_GROUP_SLACK_ID = "MEOW_SLACK_ID"
DEMO_SLACK_USER_GROUP_NAME = "Meow Group"
DEMO_SLACK_USER_GROUP_HANDLE = "meow_group"
DEMO_ON_CALL_SHIFT_ID_1 = "OH3V5FYQEYJ6M"
DEMO_ON_CALL_SHIFT_ID_2 = "O9WTH7CKM3KZW"
DEMO_ON_CALL_SHIFT_NAME_1 = "Demo single event"
DEMO_ON_CALL_SHIFT_NAME_2 = "Demo recurrent event"
DEMO_ON_CALL_SHIFT_START_1 = "2020-09-10T08:00:00"
DEMO_ON_CALL_SHIFT_START_2 = "2020-09-10T16:00:00"
DEMO_ON_CALL_SHIFT_DURATION = 10800
DEMO_ON_CALL_SHIFT_BY_DAY = ["MO", "WE", "FR"]

8
engine/apps/public_api/helpers.py
View file

@ -1,14 +1,8 @@
from apps.public_api.constants import DEMO_AUTH_TOKEN, VALID_DATE_FOR_DELETE_INCIDENT
from apps.public_api.constants import VALID_DATE_FOR_DELETE_INCIDENT
from apps.slack.slack_client import SlackClientWithErrorHandling
from apps.slack.slack_client.exceptions import SlackAPITokenException
def is_demo_token_request(request):
if DEMO_AUTH_TOKEN == request.headers.get("Authorization"):
return True
return False
def team_has_slack_token_for_deleting(alert_group):
if alert_group.slack_message and alert_group.slack_message.slack_team_identity:
sc = SlackClientWithErrorHandling(alert_group.slack_message.slack_team_identity.bot_access_token)

8
engine/apps/public_api/serializers/integrations.py
View file

@ -4,8 +4,6 @@ from rest_framework import fields, serializers
from apps.alerts.grafana_alerting_sync_manager.grafana_alerting_sync import GrafanaAlertingSyncManager
from apps.alerts.models import AlertReceiveChannel
from apps.public_api.constants import DEMO_INTEGRATION_LINK_TOKEN
from apps.public_api.helpers import is_demo_token_request
from common.api_helpers.custom_fields import TeamPrimaryKeyRelatedField
from common.api_helpers.exceptions import BadRequest
from common.api_helpers.mixins import EagerLoadingMixin
@ -62,12 +60,6 @@ class IntegrationSerializer(EagerLoadingMixin, serializers.ModelSerializer, Main
default_route = self._get_default_route_iterative(instance)
serializer = DefaultChannelFilterSerializer(default_route, context=self.context)
result["default_route"] = serializer.data
if is_demo_token_request(self.context["request"]):
# Replace integration token to not receive alerts on demo integration
link = result["link"]
real_token = instance.token
link = link.replace(real_token, DEMO_INTEGRATION_LINK_TOKEN)
result["link"] = link
return result

13
engine/apps/public_api/serializers/schedules_base.py
View file

@ -2,8 +2,6 @@ from django.apps import apps
from django.utils import timezone
from rest_framework import serializers
from apps.public_api import constants as public_api_constants
from apps.public_api.helpers import is_demo_token_request
from apps.schedules.ical_utils import list_users_to_notify_from_ical
from apps.schedules.models import OnCallSchedule
from apps.slack.models import SlackUserGroup
@ -36,14 +34,11 @@ class ScheduleBaseSerializer(serializers.ModelSerializer):
raise BadRequest(detail="Schedule with this name already exists")
def get_on_call_now(self, obj):
if not is_demo_token_request(self.context["request"]):
users_on_call = list_users_to_notify_from_ical(obj, timezone.datetime.now(timezone.utc))
if users_on_call is not None:
return [user.public_primary_key for user in users_on_call]
else:
return []
users_on_call = list_users_to_notify_from_ical(obj, timezone.datetime.now(timezone.utc))
if users_on_call is not None:
return [user.public_primary_key for user in users_on_call]
else:
return [public_api_constants.DEMO_USER_ID]
return []
def _correct_validated_data(self, validated_data):
slack_field = validated_data.pop("slack", {})

226
engine/apps/public_api/tests/conftest.py
View file

@ -1,14 +1,7 @@
import pytest
from django.utils import dateparse, timezone
from pytest_factoryboy import register
from apps.alerts.models import EscalationPolicy, ResolutionNote
from apps.auth_token.models import ApiAuthToken
from apps.base.models import UserNotificationPolicy
from apps.public_api import constants as public_api_constants
from apps.schedules.models import CustomOnCallShift, OnCallScheduleCalendar, OnCallScheduleICal
from apps.user_management.tests.factories import OrganizationFactory, UserFactory
from common.constants.role import Role
register(UserFactory)
register(OrganizationFactory)
@ -22,222 +15,3 @@ def make_organization_and_user_with_token(make_organization_and_user, make_publi
return organization, user, token
return _make_organization_and_user_with_token
@pytest.fixture()
def make_organization_and_user_with_slack_identities_for_demo_token(
make_slack_team_identity,
make_organization,
make_slack_user_identity,
make_user,
):
def _make_organization_and_user_with_slack_identities_for_demo_token():
slack_team_identity = make_slack_team_identity(slack_id=public_api_constants.DEMO_SLACK_TEAM_ID)
organization = make_organization(
slack_team_identity=slack_team_identity, public_primary_key=public_api_constants.DEMO_ORGANIZATION_ID
)
slack_user_identity = make_slack_user_identity(
slack_id=public_api_constants.DEMO_SLACK_USER_ID,
slack_team_identity=slack_team_identity,
)
user = make_user(
organization=organization,
public_primary_key=public_api_constants.DEMO_USER_ID,
email=public_api_constants.DEMO_USER_EMAIL,
username=public_api_constants.DEMO_USER_USERNAME,
role=Role.ADMIN,
slack_user_identity=slack_user_identity,
)
ApiAuthToken.create_auth_token(user, organization, public_api_constants.DEMO_AUTH_TOKEN)
token = public_api_constants.DEMO_AUTH_TOKEN
return organization, user, token
return _make_organization_and_user_with_slack_identities_for_demo_token
@pytest.fixture()
def make_data_for_demo_token(
make_alert_receive_channel,
make_channel_filter,
make_escalation_chain,
make_escalation_policy,
make_alert_group,
make_alert,
make_resolution_note,
make_custom_action,
make_slack_user_group,
make_schedule,
make_on_call_shift,
make_slack_channel,
make_user_notification_policy,
):
def _make_data_for_demo_token(organization, user):
alert_receive_channel = make_alert_receive_channel(
organization,
public_primary_key=public_api_constants.DEMO_INTEGRATION_ID,
verbal_name=public_api_constants.DEMO_INTEGRATION_NAME,
)
route_1 = make_channel_filter(
public_primary_key=public_api_constants.DEMO_ROUTE_ID_1,
alert_receive_channel=alert_receive_channel,
slack_channel_id=public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID,
filtering_term="us-(east|west)",
order=0,
)
make_channel_filter(
public_primary_key=public_api_constants.DEMO_ROUTE_ID_2,
alert_receive_channel=alert_receive_channel,
slack_channel_id=public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID,
filtering_term=".*",
order=1,
is_default=True,
)
escalation_chain = make_escalation_chain(
organization, public_primary_key=public_api_constants.DEMO_ESCALATION_CHAIN_ID
)
make_escalation_policy(
escalation_chain,
public_primary_key=public_api_constants.DEMO_ESCALATION_POLICY_ID_1,
escalation_policy_step=EscalationPolicy.STEP_WAIT,
order=0,
wait_delay=EscalationPolicy.ONE_MINUTE,
)
escalation_policy_2 = make_escalation_policy(
escalation_chain,
public_primary_key=public_api_constants.DEMO_ESCALATION_POLICY_ID_2,
escalation_policy_step=EscalationPolicy.STEP_NOTIFY_USERS_QUEUE,
order=1,
)
escalation_policy_2.notify_to_users_queue.add(user)
alert_group = make_alert_group(
alert_receive_channel,
public_primary_key=public_api_constants.DEMO_INCIDENT_ID,
resolved=True,
channel_filter=route_1,
)
alert_group.started_at = dateparse.parse_datetime(public_api_constants.DEMO_INCIDENT_CREATED_AT)
alert_group.resolved_at = dateparse.parse_datetime(public_api_constants.DEMO_INCIDENT_RESOLVED_AT)
alert_group.save(update_fields=["started_at", "resolved_at"])
for alert_id, created_at in public_api_constants.DEMO_ALERT_IDS:
alert = make_alert(
public_primary_key=alert_id,
alert_group=alert_group,
raw_request_data=public_api_constants.DEMO_ALERT_PAYLOAD,
)
alert.created_at = dateparse.parse_datetime(created_at)
alert.save(update_fields=["created_at"])
resolution_note = make_resolution_note(
alert_group=alert_group,
source=ResolutionNote.Source.WEB,
author=user,
public_primary_key=public_api_constants.DEMO_RESOLUTION_NOTE_ID,
message_text=public_api_constants.DEMO_RESOLUTION_NOTE_TEXT,
)
resolution_note.created_at = dateparse.parse_datetime(public_api_constants.DEMO_RESOLUTION_NOTE_CREATED_AT)
resolution_note.save(update_fields=["created_at"])
make_custom_action(
public_primary_key=public_api_constants.DEMO_CUSTOM_ACTION_ID,
organization=organization,
name=public_api_constants.DEMO_CUSTOM_ACTION_NAME,
)
user_group = make_slack_user_group(
public_primary_key=public_api_constants.DEMO_SLACK_USER_GROUP_ID,
name=public_api_constants.DEMO_SLACK_USER_GROUP_NAME,
handle=public_api_constants.DEMO_SLACK_USER_GROUP_HANDLE,
slack_id=public_api_constants.DEMO_SLACK_USER_GROUP_SLACK_ID,
slack_team_identity=organization.slack_team_identity,
)
# ical schedule
make_schedule(
organization=organization,
schedule_class=OnCallScheduleICal,
public_primary_key=public_api_constants.DEMO_SCHEDULE_ID_ICAL,
ical_url_primary=public_api_constants.DEMO_SCHEDULE_ICAL_URL_PRIMARY,
ical_url_overrides=public_api_constants.DEMO_SCHEDULE_ICAL_URL_OVERRIDES,
name=public_api_constants.DEMO_SCHEDULE_NAME_ICAL,
channel=public_api_constants.DEMO_SLACK_CHANNEL_SLACK_ID,
user_group=user_group,
)
# calendar schedule
schedule_calendar = make_schedule(
organization=organization,
schedule_class=OnCallScheduleCalendar,
public_primary_key=public_api_constants.DEMO_SCHEDULE_ID_CALENDAR,
name=public_api_constants.DEMO_SCHEDULE_NAME_CALENDAR,
channel=public_api_constants.DEMO_SLACK_CHANNEL_SLACK_ID,
user_group=user_group,
time_zone="America/New_york",
)
on_call_shift_1 = make_on_call_shift(
shift_type=CustomOnCallShift.TYPE_SINGLE_EVENT,
organization=organization,
public_primary_key=public_api_constants.DEMO_ON_CALL_SHIFT_ID_1,
name=public_api_constants.DEMO_ON_CALL_SHIFT_NAME_1,
start=dateparse.parse_datetime(public_api_constants.DEMO_ON_CALL_SHIFT_START_1),
duration=timezone.timedelta(seconds=public_api_constants.DEMO_ON_CALL_SHIFT_DURATION),
)
on_call_shift_1.users.add(user)
on_call_shift_2 = make_on_call_shift(
shift_type=CustomOnCallShift.TYPE_RECURRENT_EVENT,
organization=organization,
public_primary_key=public_api_constants.DEMO_ON_CALL_SHIFT_ID_2,
name=public_api_constants.DEMO_ON_CALL_SHIFT_NAME_2,
start=dateparse.parse_datetime(public_api_constants.DEMO_ON_CALL_SHIFT_START_2),
duration=timezone.timedelta(seconds=public_api_constants.DEMO_ON_CALL_SHIFT_DURATION),
frequency=CustomOnCallShift.FREQUENCY_WEEKLY,
interval=2,
by_day=public_api_constants.DEMO_ON_CALL_SHIFT_BY_DAY,
source=CustomOnCallShift.SOURCE_TERRAFORM,
)
on_call_shift_2.users.add(user)
schedule_calendar.custom_on_call_shifts.add(on_call_shift_1)
schedule_calendar.custom_on_call_shifts.add(on_call_shift_2)
make_slack_channel(
organization.slack_team_identity,
slack_id=public_api_constants.DEMO_SLACK_CHANNEL_SLACK_ID,
name=public_api_constants.DEMO_SLACK_CHANNEL_NAME,
)
make_user_notification_policy(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_1,
important=False,
user=user,
notify_by=UserNotificationPolicy.NotificationChannel.SMS,
step=UserNotificationPolicy.Step.NOTIFY,
order=0,
)
make_user_notification_policy(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_2,
important=False,
user=user,
step=UserNotificationPolicy.Step.WAIT,
wait_delay=UserNotificationPolicy.FIVE_MINUTES,
order=1,
)
make_user_notification_policy(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_3,
important=False,
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.PHONE_CALL,
order=2,
)
make_user_notification_policy(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_4,
important=True,
user=user,
step=UserNotificationPolicy.Step.NOTIFY,
notify_by=UserNotificationPolicy.NotificationChannel.PHONE_CALL,
order=0,
)
return
return _make_data_for_demo_token

0
engine/apps/public_api/tests/test_demo_token/__init__.py
View file

110
engine/apps/public_api/tests/test_demo_token/test_alerts.py
View file

@ -1,110 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
demo_alerts_results = []
for alert_id, created_at in public_api_constants.DEMO_ALERT_IDS:
demo_alerts_results.append(
{
"id": alert_id,
"alert_group_id": public_api_constants.DEMO_INCIDENT_ID,
"created_at": created_at,
"payload": {
"state": "alerting",
"title": "[Alerting] Test notification",
"ruleId": 0,
"message": "Someone is testing the alert notification within grafana.",
"ruleUrl": "https://amixr.io/",
"ruleName": "Test notification",
"evalMatches": [
{"tags": None, "value": 100, "metric": "High value"},
{"tags": None, "value": 200, "metric": "Higher Value"},
],
},
}
)
# https://api-docs.amixr.io/#list-alerts
demo_alerts_payload = {"count": 3, "next": None, "previous": None, "results": demo_alerts_results}
@pytest.mark.django_db
def test_get_alerts(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alerts-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_alerts_payload
@pytest.mark.django_db
def test_get_alerts_filter_by_incident(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alerts-list")
response = client.get(
url + f"?alert_group_id={public_api_constants.DEMO_INCIDENT_ID}", format="json", HTTP_AUTHORIZATION=token
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_alerts_payload
@pytest.mark.django_db
def test_get_alerts_filter_by_incident_no_results(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alerts-list")
response = client.get(url + "?alert_group_id=impossible_alert_group_id", format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.data["results"] == []
@pytest.mark.django_db
def test_get_alerts_search(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alerts-list")
response = client.get(url + "?search=evalMatches", format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_alerts_payload
@pytest.mark.django_db
def test_get_alerts_search_no_results(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alerts-list")
response = client.get(url + "?search=impossible_payload", format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.data["results"] == []

32
engine/apps/public_api/tests/test_demo_token/test_custom_actions.py
View file

@ -1,32 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
demo_custom_action_payload = {
"id": public_api_constants.DEMO_CUSTOM_ACTION_ID,
"name": public_api_constants.DEMO_CUSTOM_ACTION_NAME,
"team_id": None,
}
demo_custom_action_payload_list = {"count": 1, "next": None, "previous": None, "results": [demo_custom_action_payload]}
@pytest.mark.django_db
def test_demo_get_custom_actions_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:actions-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_custom_action_payload_list

169
engine/apps/public_api/tests/test_demo_token/test_escalation_policies.py
View file

@ -1,169 +0,0 @@
import pytest
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from rest_framework.test import APIClient
from apps.alerts.models import EscalationPolicy
from apps.public_api import constants as public_api_constants
# https://api-docs.amixr.io/#get-escalation-policy
demo_escalation_policy_payload = {
"id": public_api_constants.DEMO_ESCALATION_POLICY_ID_1,
"escalation_chain_id": public_api_constants.DEMO_ESCALATION_CHAIN_ID,
"position": 0,
"type": "wait",
"duration": timezone.timedelta(seconds=60).seconds,
}
# https://api-docs.amixr.io/#list-escalation-policies
demo_escalation_policies_payload = {
"count": 2,
"next": None,
"previous": None,
"results": [
{
"id": public_api_constants.DEMO_ESCALATION_POLICY_ID_1,
"escalation_chain_id": public_api_constants.DEMO_ESCALATION_CHAIN_ID,
"position": 0,
"type": "wait",
"duration": timezone.timedelta(seconds=60).seconds,
},
{
"id": public_api_constants.DEMO_ESCALATION_POLICY_ID_2,
"escalation_chain_id": public_api_constants.DEMO_ESCALATION_CHAIN_ID,
"position": 1,
"type": "notify_person_next_each_time",
"persons_to_notify_next_each_time": ["U4DNY931HHJS5"],
},
],
}
@pytest.mark.django_db
def test_get_escalation_policies(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:escalation_policies-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_escalation_policies_payload
@pytest.mark.django_db
def test_get_escalation_policies_filter_by_route(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:escalation_policies-list")
response = client.get(
url + f"?route_id={public_api_constants.DEMO_ROUTE_ID_1}", format="json", HTTP_AUTHORIZATION=token
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_escalation_policies_payload
@pytest.mark.django_db
def test_create_escalation_policy(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data_for_create = {
"escalation_chain_id": public_api_constants.DEMO_ESCALATION_CHAIN_ID,
"type": "notify_person_next_each_time",
"position": 0,
"persons_to_notify_next_each_time": [user.public_primary_key],
}
url = reverse("api-public:escalation_policies-list")
response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_201_CREATED
# check on nothing change
assert response.json() == demo_escalation_policy_payload
@pytest.mark.django_db
def test_invalid_step_type(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data_for_create = {
"escalation_chain_id": public_api_constants.DEMO_ESCALATION_CHAIN_ID,
"type": "this_is_invalid_step_type", # invalid step type
"position": 0,
"persons_to_notify_next_each_time": [user.public_primary_key],
}
url = reverse("api-public:escalation_policies-list")
response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_201_CREATED
# check on nothing change
assert response.json() == demo_escalation_policy_payload
@pytest.mark.django_db
def test_update_escalation_step(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data_for_update = {
"route_id": public_api_constants.DEMO_ROUTE_ID_1,
"type": "notify_person_next_each_time",
"position": 1,
"persons_to_notify_next_each_time": [user.public_primary_key],
}
url = reverse(
"api-public:escalation_policies-detail", kwargs={"pk": public_api_constants.DEMO_ESCALATION_POLICY_ID_1}
)
response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
# check on nothing change
assert response.json() == demo_escalation_policy_payload
@pytest.mark.django_db
def test_delete_escalation_policy(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
escalation_policy = EscalationPolicy.objects.get(
public_primary_key=public_api_constants.DEMO_ESCALATION_POLICY_ID_1
)
url = reverse("api-public:escalation_policies-detail", args=[escalation_policy.public_primary_key])
response = client.delete(url, format="json", HTTP_AUTHORIZATION=token)
escalation_policy.refresh_from_db()
assert response.status_code == status.HTTP_204_NO_CONTENT
# check on nothing change
escalation_policy.refresh_from_db()
assert escalation_policy is not None

82
engine/apps/public_api/tests/test_demo_token/test_incidents.py
View file

@ -1,82 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.alerts.models import AlertGroup
from apps.public_api import constants as public_api_constants
demo_incidents_payload = {
"count": 1,
"next": None,
"previous": None,
"results": [
{
"id": public_api_constants.DEMO_INCIDENT_ID,
"integration_id": public_api_constants.DEMO_INTEGRATION_ID,
"route_id": public_api_constants.DEMO_ROUTE_ID_1,
"alerts_count": 3,
"state": "resolved",
"created_at": public_api_constants.DEMO_INCIDENT_CREATED_AT,
"resolved_at": public_api_constants.DEMO_INCIDENT_RESOLVED_AT,
"acknowledged_at": None,
"title": None,
}
],
}
@pytest.mark.django_db
def test_create_incidents(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alert_groups-list")
response = client.post(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
@pytest.mark.django_db
def test_get_incidents(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alert_groups-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_incidents_payload
@pytest.mark.django_db
def test_delete_incidents(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:alert_groups-list")
incidents = AlertGroup.unarchived_objects.filter(public_primary_key=public_api_constants.DEMO_INCIDENT_ID)
total_count = incidents.count()
incident = incidents[0]
data = {
"mode": "delete",
}
response = client.delete(url + f"/{incident.public_primary_key}/", data, format="json", HTTP_AUTHORIZATION=token)
new_count = AlertGroup.unarchived_objects.filter(public_primary_key=public_api_constants.DEMO_INCIDENT_ID).count()
assert response.status_code == status.HTTP_204_NO_CONTENT
incident.refresh_from_db()
assert total_count == new_count
assert incident is not None

239
engine/apps/public_api/tests/test_demo_token/test_integrations.py
View file

@ -1,239 +0,0 @@
from urllib.parse import urljoin
import pytest
from django.conf import settings
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.alerts.models import AlertReceiveChannel
from apps.public_api import constants as public_api_constants
# https://api-docs.amixr.io/#post-integration
demo_integration_post_payload = {
"id": public_api_constants.DEMO_INTEGRATION_ID,
"team_id": None,
"name": "Grafana :blush:",
"link": urljoin(settings.BASE_URL, f"/integrations/v1/grafana/{public_api_constants.DEMO_INTEGRATION_LINK_TOKEN}/"),
"heartbeat": None,
"default_route": {
"escalation_chain_id": None,
"id": public_api_constants.DEMO_ROUTE_ID_2,
"slack": {"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID},
},
"type": "grafana",
"templates": {
"grouping_key": None,
"resolve_signal": None,
"acknowledge_signal": None,
"slack": {"title": None, "message": None, "image_url": None},
"web": {"title": None, "message": None, "image_url": None},
"sms": {
"title": None,
},
"phone_call": {
"title": None,
},
"email": {
"title": None,
"message": None,
},
"telegram": {
"title": None,
"message": None,
"image_url": None,
},
},
"maintenance_mode": None,
"maintenance_started_at": None,
"maintenance_end_at": None,
}
# https://api-docs.amixr.io/#get-integration
demo_integration_payload = {
"id": public_api_constants.DEMO_INTEGRATION_ID,
"team_id": None,
"name": "Grafana :blush:",
"link": urljoin(settings.BASE_URL, f"/integrations/v1/grafana/{public_api_constants.DEMO_INTEGRATION_LINK_TOKEN}/"),
"default_route": {
"escalation_chain_id": None,
"id": public_api_constants.DEMO_ROUTE_ID_2,
"slack": {"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID},
},
"type": "grafana",
"heartbeat": None,
"templates": {
"grouping_key": None,
"resolve_signal": None,
"acknowledge_signal": None,
"slack": {"title": None, "message": None, "image_url": None},
"web": {"title": None, "message": None, "image_url": None},
"sms": {
"title": None,
},
"phone_call": {
"title": None,
},
"email": {
"title": None,
"message": None,
},
"telegram": {
"title": None,
"message": None,
"image_url": None,
},
},
"maintenance_mode": None,
"maintenance_started_at": None,
"maintenance_end_at": None,
}
# https://api-docs.amixr.io/#list-integrations
demo_integrations_payload = {
"count": 1,
"next": None,
"previous": None,
"results": [
{
"id": public_api_constants.DEMO_INTEGRATION_ID,
"team_id": None,
"name": "Grafana :blush:",
"link": urljoin(
settings.BASE_URL, f"/integrations/v1/grafana/{public_api_constants.DEMO_INTEGRATION_LINK_TOKEN}/"
),
"default_route": {
"escalation_chain_id": None,
"id": public_api_constants.DEMO_ROUTE_ID_2,
"slack": {"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID},
},
"type": "grafana",
"heartbeat": None,
"templates": {
"grouping_key": None,
"resolve_signal": None,
"acknowledge_signal": None,
"slack": {
"title": None,
"message": None,
"image_url": None,
},
"web": {"title": None, "message": None, "image_url": None},
"sms": {
"title": None,
},
"phone_call": {
"title": None,
},
"email": {
"title": None,
"message": None,
},
"telegram": {
"title": None,
"message": None,
"image_url": None,
},
},
"maintenance_mode": None,
"maintenance_started_at": None,
"maintenance_end_at": None,
},
],
}
@pytest.mark.django_db
def test_get_integrations(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:integrations-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_integrations_payload
@pytest.mark.django_db
def test_create_integration(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data_for_create = {"type": "grafana"}
url = reverse("api-public:integrations-list")
response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_201_CREATED
# check on nothing change
assert response.json() == demo_integration_post_payload
@pytest.mark.django_db
def test_update_integration(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
integration = AlertReceiveChannel.objects.get(public_primary_key=public_api_constants.DEMO_INTEGRATION_ID)
data_for_update = {"name": "new_name"}
url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=token)
integration.refresh_from_db()
assert response.status_code == status.HTTP_200_OK
# check on nothing change
assert response.json() == demo_integration_payload
@pytest.mark.django_db
def test_invalid_integration_type(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data_for_create = {"type": "this_is_invalid_integration_type"}
url = reverse("api-public:integrations-list")
response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_201_CREATED
# check on nothing change
assert response.json() == demo_integration_post_payload
@pytest.mark.django_db
def test_delete_integration(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
integration = AlertReceiveChannel.objects.get(public_primary_key=public_api_constants.DEMO_INTEGRATION_ID)
url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
response = client.delete(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_204_NO_CONTENT
# check on nothing change
integration.refresh_from_db()
assert integration is not None

172
engine/apps/public_api/tests/test_demo_token/test_on_call_shift.py
View file

@ -1,172 +0,0 @@
import pytest
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
from apps.schedules.models import CustomOnCallShift
demo_on_call_shift_payload_1 = {
"id": public_api_constants.DEMO_ON_CALL_SHIFT_ID_1,
"team_id": None,
"name": public_api_constants.DEMO_ON_CALL_SHIFT_NAME_1,
"type": "single_event",
"time_zone": None,
"level": 0,
"start": public_api_constants.DEMO_ON_CALL_SHIFT_START_1,
"duration": public_api_constants.DEMO_ON_CALL_SHIFT_DURATION,
"users": [public_api_constants.DEMO_USER_ID],
}
demo_on_call_shift_payload_2 = {
"id": public_api_constants.DEMO_ON_CALL_SHIFT_ID_2,
"team_id": None,
"name": public_api_constants.DEMO_ON_CALL_SHIFT_NAME_2,
"type": "recurrent_event",
"time_zone": None,
"level": 0,
"start": public_api_constants.DEMO_ON_CALL_SHIFT_START_2,
"duration": public_api_constants.DEMO_ON_CALL_SHIFT_DURATION,
"frequency": "weekly",
"interval": 2,
"week_start": "SU",
"users": [public_api_constants.DEMO_USER_ID],
"by_day": public_api_constants.DEMO_ON_CALL_SHIFT_BY_DAY,
"by_month": None,
"by_monthday": None,
}
demo_on_call_shift_payload_list = {
"count": 2,
"next": None,
"previous": None,
"results": [demo_on_call_shift_payload_1, demo_on_call_shift_payload_2],
}
@pytest.mark.django_db
def test_demo_get_on_call_shift_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:on_call_shifts-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_on_call_shift_payload_list
@pytest.mark.django_db
@pytest.mark.parametrize(
"demo_on_call_shift_id,payload",
[
(public_api_constants.DEMO_ON_CALL_SHIFT_ID_1, demo_on_call_shift_payload_1),
(public_api_constants.DEMO_ON_CALL_SHIFT_ID_2, demo_on_call_shift_payload_2),
],
)
def test_demo_get_on_call_shift_1(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
demo_on_call_shift_id,
payload,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:on_call_shifts-detail", kwargs={"pk": demo_on_call_shift_id})
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == payload
@pytest.mark.django_db
def test_demo_post_on_call_shift(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:on_call_shifts-list")
data = {
"schedule_id": public_api_constants.DEMO_SCHEDULE_ID_CALENDAR,
"name": "New demo shift",
"type": CustomOnCallShift.TYPE_SINGLE_EVENT,
"start": timezone.now().replace(tzinfo=None, microsecond=0).isoformat(),
"duration": 3600,
}
response = client.post(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_201_CREATED
assert response.data == demo_on_call_shift_payload_1
@pytest.mark.django_db
@pytest.mark.parametrize(
"demo_on_call_shift_id,payload",
[
(public_api_constants.DEMO_ON_CALL_SHIFT_ID_1, demo_on_call_shift_payload_1),
(public_api_constants.DEMO_ON_CALL_SHIFT_ID_2, demo_on_call_shift_payload_2),
],
)
def test_demo_update_on_call_shift(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
demo_on_call_shift_id,
payload,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data = {"name": "Updated demo name"}
url = reverse("api-public:on_call_shifts-detail", kwargs={"pk": demo_on_call_shift_id})
response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == payload
@pytest.mark.django_db
@pytest.mark.parametrize(
"demo_on_call_shift_id",
[
public_api_constants.DEMO_ON_CALL_SHIFT_ID_1,
public_api_constants.DEMO_ON_CALL_SHIFT_ID_2,
],
)
def test_demo_delete_on_call_shift(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
demo_on_call_shift_id,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:on_call_shifts-detail", kwargs={"pk": demo_on_call_shift_id})
response = client.delete(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_204_NO_CONTENT
assert CustomOnCallShift.objects.filter(public_primary_key=demo_on_call_shift_id).exists()

225
engine/apps/public_api/tests/test_demo_token/test_personal_notification_rules.py
View file

@ -1,225 +0,0 @@
import pytest
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from rest_framework.test import APIClient
from apps.base.models import UserNotificationPolicy
from apps.base.models.user_notification_policy import NotificationChannelPublicAPIOptions
from apps.public_api import constants as public_api_constants
TYPE_WAIT = "wait"
demo_personal_notification_rule_payload_1 = {
"id": public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_1,
"user_id": public_api_constants.DEMO_USER_ID,
"position": 0,
"important": False,
"type": "notify_by_sms",
}
demo_personal_notification_rule_payload_2 = {
"id": public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_2,
"user_id": public_api_constants.DEMO_USER_ID,
"position": 1,
"duration": timezone.timedelta(seconds=300).seconds,
"important": False,
"type": "wait",
}
demo_personal_notification_rule_payload_3 = {
"id": public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_3,
"user_id": public_api_constants.DEMO_USER_ID,
"position": 2,
"important": False,
"type": "notify_by_phone_call",
}
demo_personal_notification_rule_payload_4 = {
"id": public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_4,
"user_id": public_api_constants.DEMO_USER_ID,
"position": 0,
"important": True,
"type": "notify_by_phone_call",
}
demo_personal_notification_rules_payload = {
"count": 4,
"next": None,
"previous": None,
"results": [
demo_personal_notification_rule_payload_1,
demo_personal_notification_rule_payload_2,
demo_personal_notification_rule_payload_3,
demo_personal_notification_rule_payload_4,
],
}
demo_personal_notification_rules_non_important_payload = {
"count": 3,
"next": None,
"previous": None,
"results": [
demo_personal_notification_rule_payload_1,
demo_personal_notification_rule_payload_2,
demo_personal_notification_rule_payload_3,
],
}
demo_personal_notification_rules_important_payload = {
"count": 1,
"next": None,
"previous": None,
"results": [
demo_personal_notification_rule_payload_4,
],
}
@pytest.mark.django_db
def test_get_personal_notification_rule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
demo_personal_notification_rule_1 = UserNotificationPolicy.objects.get(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_1
)
client = APIClient()
url = reverse(
"api-public:personal_notification_rules-detail",
kwargs={"pk": demo_personal_notification_rule_1.public_primary_key},
)
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_personal_notification_rule_payload_1
@pytest.mark.django_db
def test_get_personal_notification_rules_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
client = APIClient()
url = reverse("api-public:personal_notification_rules-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_personal_notification_rules_payload
@pytest.mark.django_db
def test_get_personal_notification_rules_list_important(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
client = APIClient()
url = reverse("api-public:personal_notification_rules-list")
response = client.get(url + "?important=true", format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_personal_notification_rules_important_payload
@pytest.mark.django_db
def test_get_personal_notification_rules_list_non_important(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
client = APIClient()
url = reverse("api-public:personal_notification_rules-list")
response = client.get(url + "?important=false", format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_personal_notification_rules_non_important_payload
@pytest.mark.django_db
def test_update_personal_notification_rule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
demo_personal_notification_rule_1 = UserNotificationPolicy.objects.get(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_1
)
client = APIClient()
url = reverse(
"api-public:personal_notification_rules-detail",
kwargs={"pk": demo_personal_notification_rule_1.public_primary_key},
)
data_to_update = {
"type": NotificationChannelPublicAPIOptions.LABELS[UserNotificationPolicy.NotificationChannel.SLACK]
}
response = client.put(url, format="json", HTTP_AUTHORIZATION=token, data=data_to_update)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_personal_notification_rule_payload_1
# check on nothing change
demo_personal_notification_rule_1.refresh_from_db()
assert demo_personal_notification_rule_1.notify_by != UserNotificationPolicy.NotificationChannel.SLACK
@pytest.mark.django_db
def test_create_personal_notification_rule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
client = APIClient()
url = reverse("api-public:personal_notification_rules-list")
data_for_create = {
"user_id": user.public_primary_key,
"type": TYPE_WAIT,
"position": 1,
"duration": timezone.timedelta(seconds=300).seconds,
}
response = client.post(url, format="json", HTTP_AUTHORIZATION=token, data=data_for_create)
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == demo_personal_notification_rule_payload_1
@pytest.mark.django_db
def test_delete_personal_notification_rule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
_ = make_data_for_demo_token(organization, user)
demo_personal_notification_rule_1 = UserNotificationPolicy.objects.get(
public_primary_key=public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_1
)
client = APIClient()
url = reverse(
"api-public:personal_notification_rules-detail",
kwargs={"pk": demo_personal_notification_rule_1.public_primary_key},
)
response = client.delete(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_204_NO_CONTENT
# check on nothing change
demo_personal_notification_rule_1.refresh_from_db()
assert demo_personal_notification_rule_1 is not None

117
engine/apps/public_api/tests/test_demo_token/test_resolution_notes.py
View file

@ -1,117 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.alerts.models import ResolutionNote
from apps.public_api import constants as public_api_constants
demo_resolution_note_payload = {
"id": public_api_constants.DEMO_RESOLUTION_NOTE_ID,
"alert_group_id": public_api_constants.DEMO_INCIDENT_ID,
"author": public_api_constants.DEMO_USER_ID,
"source": public_api_constants.DEMO_RESOLUTION_NOTE_SOURCE,
"created_at": public_api_constants.DEMO_RESOLUTION_NOTE_CREATED_AT,
"text": public_api_constants.DEMO_RESOLUTION_NOTE_TEXT,
}
demo_resolution_note_payload_list = {
"count": 1,
"next": None,
"previous": None,
"results": [demo_resolution_note_payload],
}
@pytest.mark.django_db
def test_demo_get_resolution_note_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:resolution_notes-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_resolution_note_payload_list
@pytest.mark.django_db
def test_demo_get_resolution_note(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:resolution_notes-detail", kwargs={"pk": public_api_constants.DEMO_RESOLUTION_NOTE_ID})
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_resolution_note_payload
@pytest.mark.django_db
def test_demo_post_resolution_note(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:resolution_notes-list")
data = {"alert_group_id": public_api_constants.DEMO_INCIDENT_ID, "text": "New demo text"}
response = client.post(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_201_CREATED
assert response.data == demo_resolution_note_payload
@pytest.mark.django_db
def test_demo_update_resolution_note(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
data = {"alert_group_id": public_api_constants.DEMO_INCIDENT_ID, "text": "Updated demo text"}
url = reverse("api-public:resolution_notes-detail", kwargs={"pk": public_api_constants.DEMO_RESOLUTION_NOTE_ID})
response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_resolution_note_payload
@pytest.mark.django_db
def test_demo_delete_resolution_note(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:resolution_notes-detail", kwargs={"pk": public_api_constants.DEMO_RESOLUTION_NOTE_ID})
response = client.delete(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_204_NO_CONTENT
assert ResolutionNote.objects.filter(public_primary_key=public_api_constants.DEMO_RESOLUTION_NOTE_ID).exists()

182
engine/apps/public_api/tests/test_demo_token/test_routes.py
View file

@ -1,182 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.alerts.models import ChannelFilter
from apps.public_api import constants as public_api_constants
# https://api-docs.amixr.io/#get-route
demo_route_payload = {
"id": public_api_constants.DEMO_ROUTE_ID_1,
"escalation_chain_id": None,
"integration_id": public_api_constants.DEMO_INTEGRATION_ID,
"routing_regex": "us-(east|west)",
"position": 0,
"is_the_last_route": False,
"slack": {"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID},
}
# https://api-docs.amixr.io/#list-routes
demo_routes_payload = {
"count": 2,
"next": None,
"previous": None,
"results": [
{
"id": public_api_constants.DEMO_ROUTE_ID_1,
"escalation_chain_id": None,
"integration_id": public_api_constants.DEMO_INTEGRATION_ID,
"routing_regex": "us-(east|west)",
"position": 0,
"is_the_last_route": False,
"slack": {"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID},
},
{
"id": public_api_constants.DEMO_ROUTE_ID_2,
"escalation_chain_id": None,
"integration_id": public_api_constants.DEMO_INTEGRATION_ID,
"routing_regex": ".*",
"position": 1,
"is_the_last_route": True,
"slack": {"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_FOR_ROUTE_ID},
},
],
}
@pytest.mark.django_db
def test_get_route(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
channel_filter = ChannelFilter.objects.get(public_primary_key=public_api_constants.DEMO_ROUTE_ID_1)
url = reverse("api-public:routes-detail", kwargs={"pk": channel_filter.public_primary_key})
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_route_payload
@pytest.mark.django_db
def test_get_routes_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:routes-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_routes_payload
@pytest.mark.django_db
def test_get_routes_filter_by_integration_id(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:routes-list")
response = client.get(
url + f"?integration_id={public_api_constants.DEMO_INTEGRATION_ID}", format="json", HTTP_AUTHORIZATION=token
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_routes_payload
@pytest.mark.django_db
def test_create_route(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:routes-list")
data_for_create = {
"integration_id": public_api_constants.DEMO_INTEGRATION_ID,
"routing_regex": "testreg",
}
response = client.post(url, format="json", HTTP_AUTHORIZATION=token, data=data_for_create)
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == demo_route_payload
@pytest.mark.django_db
def test_invalid_route_data(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:routes-list")
data_for_create = {
"integration_id": public_api_constants.DEMO_INTEGRATION_ID,
"routing_regex": None, # routing_regex cannot be null for non-default filters
}
response = client.post(url, format="json", HTTP_AUTHORIZATION=token, data=data_for_create)
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == demo_route_payload
@pytest.mark.django_db
def test_update_route(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
channel_filter = ChannelFilter.objects.get(public_primary_key=public_api_constants.DEMO_ROUTE_ID_1)
url = reverse("api-public:routes-detail", kwargs={"pk": channel_filter.public_primary_key})
data_to_update = {
"routing_regex": "testreg_updated",
}
assert channel_filter.filtering_term != data_to_update["routing_regex"]
response = client.put(url, format="json", HTTP_AUTHORIZATION=token, data=data_to_update)
assert response.status_code == status.HTTP_200_OK
# check on nothing change
channel_filter.refresh_from_db()
assert response.json() == demo_route_payload
assert channel_filter.filtering_term != data_to_update["routing_regex"]
@pytest.mark.django_db
def test_delete_route(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
channel_filter = ChannelFilter.objects.get(public_primary_key=public_api_constants.DEMO_ROUTE_ID_1)
url = reverse("api-public:routes-detail", kwargs={"pk": channel_filter.public_primary_key})
response = client.delete(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_204_NO_CONTENT
# check on nothing change
channel_filter.refresh_from_db()
assert channel_filter is not None

164
engine/apps/public_api/tests/test_demo_token/test_schedules.py
View file

@ -1,164 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
from apps.schedules.models import OnCallSchedule
demo_ical_schedule_payload = {
"id": public_api_constants.DEMO_SCHEDULE_ID_ICAL,
"team_id": None,
"name": public_api_constants.DEMO_SCHEDULE_NAME_ICAL,
"type": "ical",
"ical_url_primary": public_api_constants.DEMO_SCHEDULE_ICAL_URL_PRIMARY,
"ical_url_overrides": public_api_constants.DEMO_SCHEDULE_ICAL_URL_OVERRIDES,
"on_call_now": [public_api_constants.DEMO_USER_ID],
"slack": {
"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_SLACK_ID,
"user_group_id": public_api_constants.DEMO_SLACK_USER_GROUP_SLACK_ID,
},
}
demo_calendar_schedule_payload = {
"id": public_api_constants.DEMO_SCHEDULE_ID_CALENDAR,
"team_id": None,
"name": public_api_constants.DEMO_SCHEDULE_NAME_CALENDAR,
"type": "calendar",
"time_zone": "America/New_york",
"on_call_now": [public_api_constants.DEMO_USER_ID],
"shifts": [
public_api_constants.DEMO_ON_CALL_SHIFT_ID_1,
public_api_constants.DEMO_ON_CALL_SHIFT_ID_2,
],
"slack": {
"channel_id": public_api_constants.DEMO_SLACK_CHANNEL_SLACK_ID,
"user_group_id": public_api_constants.DEMO_SLACK_USER_GROUP_SLACK_ID,
},
"ical_url_overrides": None,
}
demo_schedules_payload = {
"count": 2,
"next": None,
"previous": None,
"results": [
demo_ical_schedule_payload,
demo_calendar_schedule_payload,
],
}
@pytest.mark.django_db
def test_get_schedule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
schedule = OnCallSchedule.objects.get(public_primary_key=public_api_constants.DEMO_SCHEDULE_ID_ICAL)
url = reverse("api-public:schedules-detail", kwargs={"pk": schedule.public_primary_key})
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_ical_schedule_payload
@pytest.mark.django_db
def test_create_schedule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:schedules-list")
data = {
"name": "schedule test name",
"type": "ical",
}
response = client.post(url, data=data, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_201_CREATED
# check that demo instance was returned
assert response.data == demo_ical_schedule_payload
@pytest.mark.django_db
def test_update_ical_schedule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
schedule = OnCallSchedule.objects.get(public_primary_key=public_api_constants.DEMO_SCHEDULE_ID_ICAL)
url = reverse("api-public:schedules-detail", kwargs={"pk": schedule.public_primary_key})
data = {
"name": "NEW NAME",
}
response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
# check on nothing change
schedule.refresh_from_db()
assert schedule.name != data["name"]
assert response.data == demo_ical_schedule_payload
@pytest.mark.django_db
def test_update_calendar_schedule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
schedule = OnCallSchedule.objects.get(public_primary_key=public_api_constants.DEMO_SCHEDULE_ID_CALENDAR)
url = reverse("api-public:schedules-detail", kwargs={"pk": schedule.public_primary_key})
data = {
"name": "NEW NAME",
}
response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
# check on nothing change
schedule.refresh_from_db()
assert schedule.name != data["name"]
assert response.data == demo_calendar_schedule_payload
@pytest.mark.django_db
def test_delete_schedule(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
schedule = OnCallSchedule.objects.get(public_primary_key=public_api_constants.DEMO_SCHEDULE_ID_ICAL)
url = reverse("api-public:schedules-detail", kwargs={"pk": schedule.public_primary_key})
response = client.delete(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_204_NO_CONTENT
# check on nothing change
schedule.refresh_from_db()
assert schedule is not None

34
engine/apps/public_api/tests/test_demo_token/test_slack_channels.py
View file

@ -1,34 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
demo_slack_channels_payload = {
"count": 1,
"next": None,
"previous": None,
"results": [
{
"name": public_api_constants.DEMO_SLACK_CHANNEL_NAME,
"slack_id": public_api_constants.DEMO_SLACK_CHANNEL_SLACK_ID,
}
],
}
@pytest.mark.django_db
def test_get_slack_channels_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:slack_channels-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_slack_channels_payload

36
engine/apps/public_api/tests/test_demo_token/test_user_groups.py
View file

@ -1,36 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
demo_user_group_payload = {
"id": public_api_constants.DEMO_SLACK_USER_GROUP_ID,
"type": "slack_based",
"slack": {
"id": public_api_constants.DEMO_SLACK_USER_GROUP_SLACK_ID,
"name": public_api_constants.DEMO_SLACK_USER_GROUP_NAME,
"handle": public_api_constants.DEMO_SLACK_USER_GROUP_HANDLE,
},
}
demo_user_group_payload_list = {"count": 1, "next": None, "previous": None, "results": [demo_user_group_payload]}
@pytest.mark.django_db
def test_demo_get_user_groups_list(
make_organization_and_user_with_slack_identities_for_demo_token,
make_data_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
_ = make_data_for_demo_token(organization, user)
url = reverse("api-public:user_groups-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
assert response.status_code == status.HTTP_200_OK
assert response.data == demo_user_group_payload_list

91
engine/apps/public_api/tests/test_demo_token/test_users.py
View file

@ -1,91 +0,0 @@
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.public_api import constants as public_api_constants
# NB can compare with https://api-docs.amixr.io/#get-user
demo_token_user_payload = {
"id": public_api_constants.DEMO_USER_ID,
"email": public_api_constants.DEMO_USER_EMAIL,
"slack": {"user_id": public_api_constants.DEMO_SLACK_USER_ID, "team_id": public_api_constants.DEMO_SLACK_TEAM_ID},
"username": public_api_constants.DEMO_USER_USERNAME,
"role": "admin",
"is_phone_number_verified": False,
}
# https://api-docs.amixr.io/#list-users
demo_token_users_payload = {
"count": 1,
"next": None,
"previous": None,
"results": [
{
"id": public_api_constants.DEMO_USER_ID,
"email": public_api_constants.DEMO_USER_EMAIL,
"slack": {
"user_id": public_api_constants.DEMO_SLACK_USER_ID,
"team_id": public_api_constants.DEMO_SLACK_TEAM_ID,
},
"username": public_api_constants.DEMO_USER_USERNAME,
"role": "admin",
"is_phone_number_verified": False,
}
],
}
@pytest.mark.django_db
def test_get_user(
make_organization_and_user_with_slack_identities_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
url = reverse("api-public:users-detail", args=[user.public_primary_key])
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_token_user_payload
# get current user
url = reverse("api-public:users-detail", args=["current"])
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_token_user_payload
@pytest.mark.django_db
def test_get_users(
make_organization_and_user_with_slack_identities_for_demo_token,
):
organization, user, token = make_organization_and_user_with_slack_identities_for_demo_token()
client = APIClient()
url = reverse("api-public:users-list")
response = client.get(url, format="json", HTTP_AUTHORIZATION=token)
assert response.status_code == status.HTTP_200_OK
assert response.json() == demo_token_users_payload
@pytest.mark.django_db
def test_forbidden_access(
make_organization_and_user_with_slack_identities_for_demo_token,
make_organization_and_user_with_token,
):
_, user, _ = make_organization_and_user_with_slack_identities_for_demo_token()
_, _, another_org_token = make_organization_and_user_with_token()
client = APIClient()
url = reverse("api-public:users-detail", args=[user.public_primary_key])
response = client.get(url, format="json", HTTP_AUTHORIZATION=another_org_token)
assert response.status_code == status.HTTP_404_NOT_FOUND

4
engine/apps/public_api/views/action.py
View file

@ -8,11 +8,11 @@ from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api.serializers.action import ActionSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from common.api_helpers.filters import ByTeamFilter
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin
from common.api_helpers.mixins import RateLimitHeadersMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class ActionView(RateLimitHeadersMixin, DemoTokenMixin, mixins.ListModelMixin, GenericViewSet):
class ActionView(RateLimitHeadersMixin, mixins.ListModelMixin, GenericViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
pagination_class = FiftyPageSizePaginator

7
engine/apps/public_api/views/alerts.py
View file

@ -6,14 +6,13 @@ from rest_framework.viewsets import GenericViewSet
from apps.alerts.models import Alert
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers.alerts import AlertSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin
from common.api_helpers.mixins import RateLimitHeadersMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class AlertView(RateLimitHeadersMixin, DemoTokenMixin, mixins.ListModelMixin, GenericViewSet):
class AlertView(RateLimitHeadersMixin, mixins.ListModelMixin, GenericViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -23,8 +22,6 @@ class AlertView(RateLimitHeadersMixin, DemoTokenMixin, mixins.ListModelMixin, Ge
serializer_class = AlertSerializer
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_ALERT_IDS[0]
def get_queryset(self):
alert_group_id = self.request.query_params.get("alert_group_id", None)
search = self.request.query_params.get("search", None)

7
engine/apps/public_api/views/escalation_policies.py
View file

@ -5,15 +5,14 @@ from rest_framework.viewsets import ModelViewSet
from apps.alerts.models import EscalationPolicy
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers import EscalationPolicySerializer, EscalationPolicyUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class EscalationPolicyView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerMixin, ModelViewSet):
class EscalationPolicyView(RateLimitHeadersMixin, UpdateSerializerMixin, ModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -25,8 +24,6 @@ class EscalationPolicyView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializ
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_ESCALATION_POLICY_ID_1
def get_queryset(self):
escalation_chain_id = self.request.query_params.get("escalation_chain_id", None)
queryset = EscalationPolicy.objects.filter(

9
engine/apps/public_api/views/incidents.py
View file

@ -8,14 +8,13 @@ from rest_framework.viewsets import GenericViewSet
from apps.alerts.models import AlertGroup
from apps.alerts.tasks import delete_alert_group, wipe
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.constants import VALID_DATE_FOR_DELETE_INCIDENT
from apps.public_api.helpers import is_valid_group_creation_date, team_has_slack_token_for_deleting
from apps.public_api.serializers import IncidentSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from common.api_helpers.exceptions import BadRequest
from common.api_helpers.filters import ByTeamModelFieldFilterMixin, get_team_queryset
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin
from common.api_helpers.mixins import RateLimitHeadersMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
@ -30,9 +29,7 @@ class IncidentByTeamFilter(ByTeamModelFieldFilterMixin, filters.FilterSet):
)
class IncidentView(
RateLimitHeadersMixin, DemoTokenMixin, mixins.ListModelMixin, mixins.DestroyModelMixin, GenericViewSet
):
class IncidentView(RateLimitHeadersMixin, mixins.ListModelMixin, mixins.DestroyModelMixin, GenericViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -42,8 +39,6 @@ class IncidentView(
serializer_class = IncidentSerializer
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_INCIDENT_ID
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = IncidentByTeamFilter

11
engine/apps/public_api/views/integrations.py
View file

@ -6,17 +6,11 @@ from rest_framework.viewsets import ModelViewSet
from apps.alerts.models import AlertReceiveChannel
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers import IntegrationSerializer, IntegrationUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.filters import ByTeamFilter
from common.api_helpers.mixins import (
DemoTokenMixin,
FilterSerializerMixin,
RateLimitHeadersMixin,
UpdateSerializerMixin,
)
from common.api_helpers.mixins import FilterSerializerMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
from .maintaiable_object_mixin import MaintainableObjectMixin
@ -24,7 +18,6 @@ from .maintaiable_object_mixin import MaintainableObjectMixin
class IntegrationView(
RateLimitHeadersMixin,
DemoTokenMixin,
FilterSerializerMixin,
UpdateSerializerMixin,
MaintainableObjectMixin,
@ -41,8 +34,6 @@ class IntegrationView(
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_INTEGRATION_ID
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = ByTeamFilter

7
engine/apps/public_api/views/on_call_shifts.py
View file

@ -4,17 +4,16 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.viewsets import ModelViewSet
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers import CustomOnCallShiftSerializer, CustomOnCallShiftUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.schedules.models import CustomOnCallShift
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.filters import ByTeamFilter
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class CustomOnCallShiftView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerMixin, ModelViewSet):
class CustomOnCallShiftView(RateLimitHeadersMixin, UpdateSerializerMixin, ModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -29,8 +28,6 @@ class CustomOnCallShiftView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSeriali
filter_backends = [DjangoFilterBackend]
filterset_class = ByTeamFilter
demo_default_id = public_api_constants.DEMO_ON_CALL_SHIFT_ID_1
def get_queryset(self):
name = self.request.query_params.get("name", None)
schedule_id = self.request.query_params.get("schedule_id", None)

6
engine/apps/public_api/views/organizations.py
View file

@ -3,17 +3,15 @@ from rest_framework.settings import api_settings
from rest_framework.viewsets import ReadOnlyModelViewSet
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers import OrganizationSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.user_management.models import Organization
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin
from common.api_helpers.mixins import RateLimitHeadersMixin
from common.api_helpers.paginators import TwentyFivePageSizePaginator
class OrganizationView(
RateLimitHeadersMixin,
DemoTokenMixin,
ReadOnlyModelViewSet,
):
authentication_classes = (ApiTokenAuthentication,)
@ -26,8 +24,6 @@ class OrganizationView(
pagination_class = TwentyFivePageSizePaginator
demo_default_id = public_api_constants.DEMO_ORGANIZATION_ID
def get_queryset(self):
# It's a dirty hack to get queryset from the object. Just in case we'll return multiple teams in the future.
return Organization.objects.filter(pk=self.request.auth.organization.pk)

7
engine/apps/public_api/views/personal_notifications.py
View file

@ -6,17 +6,16 @@ from rest_framework.viewsets import ModelViewSet
from apps.auth_token.auth import ApiTokenAuthentication
from apps.base.models import UserNotificationPolicy
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers import PersonalNotificationRuleSerializer, PersonalNotificationRuleUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.user_management.models import User
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.exceptions import BadRequest
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class PersonalNotificationView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerMixin, ModelViewSet):
class PersonalNotificationView(RateLimitHeadersMixin, UpdateSerializerMixin, ModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -28,8 +27,6 @@ class PersonalNotificationView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSeri
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_PERSONAL_NOTIFICATION_ID_1
def get_queryset(self):
user_id = self.request.query_params.get("user_id", None)
important = self.request.query_params.get("important", None)

7
engine/apps/public_api/views/resolution_notes.py
View file

@ -6,14 +6,13 @@ from rest_framework.viewsets import ModelViewSet
from apps.alerts.models import ResolutionNote
from apps.alerts.tasks import send_update_resolution_note_signal
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers.resolution_notes import ResolutionNoteSerializer, ResolutionNoteUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class ResolutionNoteView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerMixin, ModelViewSet):
class ResolutionNoteView(RateLimitHeadersMixin, UpdateSerializerMixin, ModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -28,8 +27,6 @@ class ResolutionNoteView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializer
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_RESOLUTION_NOTE_ID
def get_queryset(self):
alert_group_id = self.request.query_params.get("alert_group_id", None)
queryset = ResolutionNote.objects.filter(

7
engine/apps/public_api/views/routes.py
View file

@ -7,16 +7,15 @@ from rest_framework.viewsets import ModelViewSet
from apps.alerts.models import ChannelFilter
from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.serializers import ChannelFilterSerializer, ChannelFilterUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.exceptions import BadRequest
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import TwentyFivePageSizePaginator
class ChannelFilterView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerMixin, ModelViewSet):
class ChannelFilterView(RateLimitHeadersMixin, UpdateSerializerMixin, ModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -31,8 +30,6 @@ class ChannelFilterView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerM
filter_backends = [DjangoFilterBackend]
filterset_fields = ["alert_receive_channel"]
demo_default_id = public_api_constants.DEMO_ROUTE_ID_1
def get_queryset(self):
integration_id = self.request.query_params.get("integration_id", None)
routing_regex = self.request.query_params.get("routing_regex", None)

7
engine/apps/public_api/views/schedules.py
View file

@ -7,7 +7,6 @@ from rest_framework.views import Response
from rest_framework.viewsets import ModelViewSet
from apps.auth_token.auth import ApiTokenAuthentication, ScheduleExportAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.custom_renderers import CalendarRenderer
from apps.public_api.serializers import PolymorphicScheduleSerializer, PolymorphicScheduleUpdateSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
@ -16,11 +15,11 @@ from apps.schedules.models import OnCallSchedule
from apps.slack.tasks import update_slack_user_group_for_schedules
from apps.user_management.organization_log_creator import OrganizationLogType, create_organization_log
from common.api_helpers.filters import ByTeamFilter
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, UpdateSerializerMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class OnCallScheduleChannelView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSerializerMixin, ModelViewSet):
class OnCallScheduleChannelView(RateLimitHeadersMixin, UpdateSerializerMixin, ModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -32,8 +31,6 @@ class OnCallScheduleChannelView(RateLimitHeadersMixin, DemoTokenMixin, UpdateSer
pagination_class = FiftyPageSizePaginator
demo_default_id = public_api_constants.DEMO_SCHEDULE_ID_ICAL
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = ByTeamFilter

4
engine/apps/public_api/views/slack_channels.py
View file

@ -6,11 +6,11 @@ from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api.serializers.slack_channel import SlackChannelSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.slack.models import SlackChannel
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin
from common.api_helpers.mixins import RateLimitHeadersMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class SlackChannelView(RateLimitHeadersMixin, DemoTokenMixin, mixins.ListModelMixin, GenericViewSet):
class SlackChannelView(RateLimitHeadersMixin, mixins.ListModelMixin, GenericViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
pagination_class = FiftyPageSizePaginator

4
engine/apps/public_api/views/user_groups.py
View file

@ -6,11 +6,11 @@ from apps.auth_token.auth import ApiTokenAuthentication
from apps.public_api.serializers.user_groups import UserGroupSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.slack.models import SlackUserGroup
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin
from common.api_helpers.mixins import RateLimitHeadersMixin
from common.api_helpers.paginators import FiftyPageSizePaginator
class UserGroupView(RateLimitHeadersMixin, DemoTokenMixin, mixins.ListModelMixin, GenericViewSet):
class UserGroupView(RateLimitHeadersMixin, mixins.ListModelMixin, GenericViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
pagination_class = FiftyPageSizePaginator

7
engine/apps/public_api/views/users.py
View file

@ -5,19 +5,18 @@ from rest_framework.views import Response
from rest_framework.viewsets import ReadOnlyModelViewSet
from apps.auth_token.auth import ApiTokenAuthentication, UserScheduleExportAuthentication
from apps.public_api import constants as public_api_constants
from apps.public_api.custom_renderers import CalendarRenderer
from apps.public_api.serializers import FastUserSerializer, UserSerializer
from apps.public_api.throttlers.user_throttle import UserThrottle
from apps.schedules.ical_utils import user_ical_export
from apps.schedules.models import OnCallSchedule
from apps.user_management.models import User
from common.api_helpers.mixins import DemoTokenMixin, RateLimitHeadersMixin, ShortSerializerMixin
from common.api_helpers.mixins import RateLimitHeadersMixin, ShortSerializerMixin
from common.api_helpers.paginators import HundredPageSizePaginator
from common.constants.role import Role
class UserView(RateLimitHeadersMixin, ShortSerializerMixin, DemoTokenMixin, ReadOnlyModelViewSet):
class UserView(RateLimitHeadersMixin, ShortSerializerMixin, ReadOnlyModelViewSet):
authentication_classes = (ApiTokenAuthentication,)
permission_classes = (IsAuthenticated,)
@ -29,8 +28,6 @@ class UserView(RateLimitHeadersMixin, ShortSerializerMixin, DemoTokenMixin, Read
throttle_classes = [UserThrottle]
demo_default_id = public_api_constants.DEMO_USER_ID
def get_queryset(self):
username = self.request.query_params.get("username")
email = self.request.query_params.get("email")

7
engine/apps/schedules/tasks/notify_about_empty_shifts_in_schedule.py
View file

@ -4,7 +4,6 @@ from django.apps import apps
from django.core.cache import cache
from django.utils import timezone
from apps.public_api.constants import DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL
from apps.schedules.ical_utils import list_of_empty_shifts_in_schedule
from apps.slack.utils import format_datetime_to_slack, post_message_to_channel
from common.custom_celery_tasks import shared_dedicated_queue_retry_task
@ -19,9 +18,7 @@ def start_check_empty_shifts_in_schedule():
task_logger.info("Start start_notify_about_empty_shifts_in_schedule")
schedules = OnCallSchedule.objects.exclude(
public_primary_key__in=(DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL)
)
schedules = OnCallSchedule.objects.all()
for schedule in schedules:
check_empty_shifts_in_schedule.apply_async((schedule.pk,))
@ -58,7 +55,7 @@ def start_notify_about_empty_shifts_in_schedule():
schedules = OnCallSchedule.objects.filter(
empty_shifts_report_sent_at__lte=week_ago,
channel__isnull=False,
).exclude(public_primary_key__in=(DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL))
)
for schedule in schedules:
notify_about_empty_shifts_in_schedule.apply_async((schedule.pk,))

7
engine/apps/schedules/tasks/notify_about_gaps_in_schedule.py
View file

@ -4,7 +4,6 @@ from django.apps import apps
from django.core.cache import cache
from django.utils import timezone
from apps.public_api.constants import DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL
from apps.schedules.ical_utils import list_of_gaps_in_schedule
from apps.slack.utils import format_datetime_to_slack, post_message_to_channel
from common.custom_celery_tasks import shared_dedicated_queue_retry_task
@ -18,9 +17,7 @@ def start_check_gaps_in_schedule():
task_logger.info("Start start_check_gaps_in_schedule")
schedules = OnCallSchedule.objects.exclude(
public_primary_key__in=(DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL)
)
schedules = OnCallSchedule.objects.all()
for schedule in schedules:
check_gaps_in_schedule.apply_async((schedule.pk,))
@ -57,7 +54,7 @@ def start_notify_about_gaps_in_schedule():
schedules = OnCallSchedule.objects.filter(
gaps_report_sent_at__lte=week_ago,
channel__isnull=False,
).exclude(public_primary_key__in=(DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL))
)
for schedule in schedules:
notify_about_gaps_in_schedule.apply_async((schedule.pk,))

5
engine/apps/schedules/tasks/refresh_ical_files.py
View file

@ -2,7 +2,6 @@ from celery.utils.log import get_task_logger
from django.apps import apps
from apps.alerts.tasks import notify_ical_schedule_shift
from apps.public_api.constants import DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL
from apps.schedules.ical_utils import is_icals_equal
from apps.schedules.tasks import notify_about_empty_shifts_in_schedule, notify_about_gaps_in_schedule
from apps.slack.tasks import start_update_slack_user_group_for_schedules
@ -17,9 +16,7 @@ def start_refresh_ical_files():
task_logger.info("Start refresh ical files")
schedules = OnCallSchedule.objects.all().exclude(
public_primary_key__in=(DEMO_SCHEDULE_ID_CALENDAR, DEMO_SCHEDULE_ID_ICAL)
)
schedules = OnCallSchedule.objects.all()
for schedule in schedules:
refresh_ical_file.apply_async((schedule.pk,))

11
engine/apps/slack/tasks.py
View file

@ -9,8 +9,6 @@ from django.core.cache import cache
from django.utils import timezone
from apps.alerts.tasks.compare_escalations import compare_escalations
from apps.public_api import constants as public_constants
from apps.public_api.constants import DEMO_SLACK_USER_GROUP_ID
from apps.slack.constants import CACHE_UPDATE_INCIDENT_SLACK_MESSAGE_LIFETIME, SLACK_BOT_ID
from apps.slack.scenarios.escalation_delivery import EscalationDeliveryStep
from apps.slack.scenarios.scenario_step import ScenarioStep
@ -499,7 +497,7 @@ def populate_slack_usergroups():
slack_team_identities = SlackTeamIdentity.objects.filter(
detected_token_revoked__isnull=True,
).exclude(slack_id=public_constants.DEMO_SLACK_TEAM_ID)
)
delay = 0
counter = 0
@ -642,10 +640,7 @@ def start_update_slack_user_group_for_schedules():
SlackUserGroup = apps.get_model("slack", "SlackUserGroup")
user_group_pks = (
SlackUserGroup.objects.exclude(public_primary_key=DEMO_SLACK_USER_GROUP_ID)
.filter(oncall_schedules__isnull=False)
.distinct()
.values_list("pk", flat=True)
SlackUserGroup.objects.filter(oncall_schedules__isnull=False).distinct().values_list("pk", flat=True)
)
for user_group_pk in user_group_pks:
@ -673,7 +668,7 @@ def populate_slack_channels():
slack_team_identities = SlackTeamIdentity.objects.filter(
detected_token_revoked__isnull=True,
).exclude(slack_id=public_constants.DEMO_SLACK_TEAM_ID)
)
delay = 0
counter = 0

78
engine/common/api_helpers/mixins.py
View file

@ -19,7 +19,6 @@ from apps.alerts.incident_appearance.templaters import (
TemplateLoader,
)
from apps.base.messaging import get_messaging_backends
from apps.public_api.helpers import is_demo_token_request
from common.api_helpers.exceptions import BadRequest
from common.jinja_templater import apply_jinja_template
@ -125,83 +124,6 @@ class EagerLoadingMixin:
return queryset
class DemoTokenMixin:
"""
The view mixin for requests to public api with demo token authorization.
"""
def dispatch(self, request, *args, **kwargs):
"""
Overridden dispatch method of APIView
https://github.com/encode/django-rest-framework/blob/master/rest_framework/views.py#L485
"""
method = request.method.lower()
if is_demo_token_request(request) and method in ["post", "put", "delete"]:
self.args = args
self.kwargs = kwargs
request = self.initialize_request(request, *args, **kwargs)
self.request = request
# there is a strange comment about this
# https://github.com/encode/django-rest-framework/blob/master/rest_framework/views.py#L494
self.headers = self.default_response_headers
try:
self.initial(request, *args, **kwargs)
"""
check for allowed request methods
from APIView:
If `request.method` does not correspond to a handler method,
determine what kind of exception to raise.
def http_method_not_allowed(self, request, *args, **kwargs):
raise exceptions.MethodNotAllowed(request.method)
"""
if method in self.http_method_names:
handler = getattr(self, method, self.http_method_not_allowed)
else:
handler = self.http_method_not_allowed
# function comparison explanation
# https://stackoverflow.com/a/18217024
if handler == self.http_method_not_allowed:
response = handler(request, *args, **kwargs)
elif method == "post":
# It excludes a real instance creation.
# It returns the instance with public primary key
# is equal to demo_default_id
instance = self.model._default_manager.get(public_primary_key=self.demo_default_id)
serializer = self.get_serializer(instance)
headers = self.get_success_headers(serializer.data)
response = Response(data=serializer.data, status=status.HTTP_201_CREATED, headers=headers)
elif method == "put":
# It excludes a instance update.
# It returns the instance with public primary key
# is equal to demo_default_id
instance = self.get_object()
serializer = self.get_serializer(instance)
headers = self.get_success_headers(serializer.data)
response = Response(data=serializer.data, status=status.HTTP_200_OK, headers=headers)
elif method == "delete":
# In this case we return nothing just success response.
response = Response(status=status.HTTP_204_NO_CONTENT)
except Exception as exc:
response = self.handle_exception(exc)
self.response = self.finalize_response(request, response, *args, **kwargs)
return self.response
return super().dispatch(request, *args, **kwargs)
class RateLimitHeadersMixin:
# This mixin add RateLimit-Reset header to RateLimited response
def handle_exception(self, exc):