6f3f4e3f14
# What this PR does Enables the API to perform updates on the advanced webhooks created via the UI ## Which issue(s) this PR closes Closes #3958 ## Checklist - [x] Unit, integration, and e2e (if applicable) tests updated - [x] Documentation added (or `pr:no public docs` PR label added if not required) - [x] Added the relevant release notes label (see labels prefixed w/ `release:`). These labels dictate how your PR will show up in the autogenerated release notes.
1079 lines
35 KiB
Python
1079 lines
35 KiB
Python
import datetime
|
|
import json
|
|
import os
|
|
import sys
|
|
import typing
|
|
import uuid
|
|
from importlib import import_module, reload
|
|
|
|
import pytest
|
|
from celery import Task
|
|
from django.db.models.signals import post_save
|
|
from django.urls import clear_url_caches
|
|
from django.utils import timezone
|
|
from pytest_factoryboy import register
|
|
from telegram import Bot
|
|
|
|
from apps.alerts.models import (
|
|
Alert,
|
|
AlertGroupLogRecord,
|
|
AlertReceiveChannel,
|
|
MaintainableObject,
|
|
ResolutionNote,
|
|
listen_for_alertgrouplogrecord,
|
|
listen_for_alertreceivechannel_model_save,
|
|
)
|
|
from apps.alerts.signals import user_notification_action_triggered_signal
|
|
from apps.alerts.tests.factories import (
|
|
AlertFactory,
|
|
AlertGroupFactory,
|
|
AlertGroupLogRecordFactory,
|
|
AlertReceiveChannelConnectionFactory,
|
|
AlertReceiveChannelFactory,
|
|
ChannelFilterFactory,
|
|
CustomActionFactory,
|
|
EscalationChainFactory,
|
|
EscalationPolicyFactory,
|
|
InvitationFactory,
|
|
ResolutionNoteFactory,
|
|
ResolutionNoteSlackMessageFactory,
|
|
)
|
|
from apps.api.permissions import (
|
|
ACTION_PREFIX,
|
|
GrafanaAPIPermission,
|
|
LegacyAccessControlCompatiblePermission,
|
|
LegacyAccessControlRole,
|
|
RBACPermission,
|
|
)
|
|
from apps.auth_token.models import (
|
|
ApiAuthToken,
|
|
GoogleOAuth2Token,
|
|
IntegrationBacksyncAuthToken,
|
|
PluginAuthToken,
|
|
SlackAuthToken,
|
|
)
|
|
from apps.base.models.user_notification_policy_log_record import (
|
|
UserNotificationPolicyLogRecord,
|
|
listen_for_usernotificationpolicylogrecord_model_save,
|
|
)
|
|
from apps.base.tests.factories import (
|
|
LiveSettingFactory,
|
|
UserNotificationPolicyFactory,
|
|
UserNotificationPolicyLogRecordFactory,
|
|
)
|
|
from apps.email.tests.factories import EmailMessageFactory
|
|
from apps.google.tests.factories import GoogleOAuth2UserFactory
|
|
from apps.heartbeat.tests.factories import IntegrationHeartBeatFactory
|
|
from apps.labels.tests.factories import (
|
|
AlertGroupAssociatedLabelFactory,
|
|
AlertReceiveChannelAssociatedLabelFactory,
|
|
LabelKeyFactory,
|
|
LabelValueFactory,
|
|
WebhookAssociatedLabelFactory,
|
|
)
|
|
from apps.mobile_app.models import MobileAppAuthToken, MobileAppVerificationToken
|
|
from apps.phone_notifications.phone_backend import PhoneBackend
|
|
from apps.phone_notifications.tests.factories import PhoneCallRecordFactory, SMSRecordFactory
|
|
from apps.phone_notifications.tests.mock_phone_provider import MockPhoneProvider
|
|
from apps.schedules.ical_utils import memoized_users_in_ical
|
|
from apps.schedules.models import OnCallScheduleWeb
|
|
from apps.schedules.tests.factories import (
|
|
CustomOnCallShiftFactory,
|
|
OnCallScheduleCalendarFactory,
|
|
OnCallScheduleFactory,
|
|
OnCallScheduleICalFactory,
|
|
ShiftSwapRequestFactory,
|
|
)
|
|
from apps.slack.client import SlackClient
|
|
from apps.slack.tests.factories import (
|
|
SlackChannelFactory,
|
|
SlackMessageFactory,
|
|
SlackTeamIdentityFactory,
|
|
SlackUserGroupFactory,
|
|
SlackUserIdentityFactory,
|
|
)
|
|
from apps.telegram.tests.factories import (
|
|
TelegramChannelFactory,
|
|
TelegramChannelVerificationCodeFactory,
|
|
TelegramMessageFactory,
|
|
TelegramToUserConnectorFactory,
|
|
TelegramVerificationCodeFactory,
|
|
)
|
|
from apps.user_management.models.user import User, listen_for_user_model_save
|
|
from apps.user_management.tests.factories import OrganizationFactory, RegionFactory, TeamFactory, UserFactory
|
|
from apps.webhooks.presets.preset_options import WebhookPresetOptions
|
|
from apps.webhooks.tests.factories import CustomWebhookFactory, WebhookResponseFactory
|
|
from apps.webhooks.tests.test_webhook_presets import (
|
|
ADVANCED_WEBHOOK_PRESET_ID,
|
|
TEST_WEBHOOK_PRESET_ID,
|
|
TestAdvancedWebhookPreset,
|
|
TestWebhookPreset,
|
|
)
|
|
|
|
register(OrganizationFactory)
|
|
register(UserFactory)
|
|
register(TeamFactory)
|
|
register(AlertReceiveChannelFactory)
|
|
register(AlertReceiveChannelConnectionFactory)
|
|
register(ChannelFilterFactory)
|
|
register(EscalationPolicyFactory)
|
|
register(OnCallScheduleICalFactory)
|
|
register(OnCallScheduleCalendarFactory)
|
|
register(CustomOnCallShiftFactory)
|
|
register(ShiftSwapRequestFactory)
|
|
register(AlertFactory)
|
|
register(AlertGroupFactory)
|
|
register(AlertGroupLogRecordFactory)
|
|
register(InvitationFactory)
|
|
register(CustomActionFactory)
|
|
register(SlackUserGroupFactory)
|
|
register(SlackUserIdentityFactory)
|
|
register(SlackTeamIdentityFactory)
|
|
register(SlackMessageFactory)
|
|
register(TelegramToUserConnectorFactory)
|
|
register(TelegramChannelFactory)
|
|
register(TelegramVerificationCodeFactory)
|
|
register(TelegramChannelVerificationCodeFactory)
|
|
register(TelegramMessageFactory)
|
|
register(ResolutionNoteSlackMessageFactory)
|
|
register(PhoneCallRecordFactory)
|
|
register(SMSRecordFactory)
|
|
register(EmailMessageFactory)
|
|
register(IntegrationHeartBeatFactory)
|
|
register(LiveSettingFactory)
|
|
register(LabelKeyFactory)
|
|
register(LabelValueFactory)
|
|
register(AlertReceiveChannelAssociatedLabelFactory)
|
|
register(GoogleOAuth2UserFactory)
|
|
|
|
IS_RBAC_ENABLED = os.getenv("ONCALL_TESTING_RBAC_ENABLED", "True") == "True"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated_cache(settings):
|
|
"""
|
|
https://github.com/pytest-dev/pytest-django/issues/527#issuecomment-1115887487
|
|
"""
|
|
cache_version = uuid.uuid4().hex
|
|
|
|
for name in settings.CACHES.keys():
|
|
settings.CACHES[name]["VERSION"] = cache_version
|
|
|
|
from django.test.signals import clear_cache_handlers
|
|
|
|
clear_cache_handlers(setting="CACHES")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_slack_api_call(monkeypatch):
|
|
def mock_api_call(*args, **kwargs):
|
|
return {
|
|
"ts": timezone.now().isoformat(),
|
|
"status": 200,
|
|
"usergroups": [],
|
|
"channel": {"id": "TEST_CHANNEL_ID"},
|
|
"user": {
|
|
"name": "TEST_SLACK_LOGIN",
|
|
"real_name": "TEST_SLACK_NAME",
|
|
"profile": {"image_512": "TEST_SLACK_IMAGE"},
|
|
},
|
|
"team": {"name": "TEST_TEAM"},
|
|
}
|
|
|
|
monkeypatch.setattr(SlackClient, "api_call", mock_api_call)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_telegram_bot_username(monkeypatch):
|
|
def mock_username(*args, **kwargs):
|
|
return "oncall_bot"
|
|
|
|
monkeypatch.setattr(Bot, "username", mock_username)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_phone_provider(monkeypatch):
|
|
def mock_get_provider(*args, **kwargs):
|
|
return MockPhoneProvider()
|
|
|
|
monkeypatch.setattr(PhoneBackend, "_get_phone_provider", mock_get_provider)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_apply_async(monkeypatch):
|
|
def mock_apply_async(*args, **kwargs):
|
|
return uuid.uuid4()
|
|
|
|
monkeypatch.setattr(Task, "apply_async", mock_apply_async)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_ical_users_cache():
|
|
# clear users pks <-> organization cache (persisting between tests)
|
|
memoized_users_in_ical.cache_clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_is_labels_feature_enabled(settings):
|
|
settings.FEATURE_LABELS_ENABLED_FOR_ALL = True
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_is_labels_feature_enabled_for_org(settings):
|
|
def _mock_is_labels_feature_enabled_for_org(org_id):
|
|
settings.FEATURE_LABELS_ENABLED_FOR_ALL = False
|
|
settings.FEATURE_LABELS_ENABLED_PER_ORG = [org_id]
|
|
|
|
return _mock_is_labels_feature_enabled_for_org
|
|
|
|
|
|
@pytest.fixture
|
|
def make_organization():
|
|
def _make_organization(**kwargs):
|
|
return OrganizationFactory(
|
|
**kwargs, is_rbac_permissions_enabled=IS_RBAC_ENABLED, is_grafana_labels_enabled=True
|
|
)
|
|
|
|
return _make_organization
|
|
|
|
|
|
@pytest.fixture
|
|
def make_user_for_organization(make_user):
|
|
def _make_user_for_organization(organization, role: typing.Optional[LegacyAccessControlRole] = None, **kwargs):
|
|
post_save.disconnect(listen_for_user_model_save, sender=User)
|
|
user = make_user(organization=organization, role=role, **kwargs)
|
|
post_save.disconnect(listen_for_user_model_save, sender=User)
|
|
return user
|
|
|
|
return _make_user_for_organization
|
|
|
|
|
|
@pytest.fixture
|
|
def make_token_for_organization():
|
|
def _make_token_for_organization(organization):
|
|
return PluginAuthToken.create_auth_token(organization)
|
|
|
|
return _make_token_for_organization
|
|
|
|
|
|
@pytest.fixture
|
|
def make_token_for_integration():
|
|
def _make_token_for_integration(alert_receive_channel, organization):
|
|
return IntegrationBacksyncAuthToken.create_auth_token(alert_receive_channel, organization)
|
|
|
|
return _make_token_for_integration
|
|
|
|
|
|
@pytest.fixture
|
|
def make_mobile_app_verification_token_for_user():
|
|
def _make_mobile_app_verification_token_for_user(user, organization):
|
|
return MobileAppVerificationToken.create_auth_token(user, organization)
|
|
|
|
return _make_mobile_app_verification_token_for_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_mobile_app_auth_token_for_user():
|
|
def _make_mobile_app_auth_token_for_user(user, organization):
|
|
return MobileAppAuthToken.create_auth_token(user, organization)
|
|
|
|
return _make_mobile_app_auth_token_for_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_slack_token_for_user():
|
|
def _make_slack_token_for_user(user):
|
|
return SlackAuthToken.create_auth_token(organization=user.organization, user=user)
|
|
|
|
return _make_slack_token_for_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_google_oauth2_token_for_user():
|
|
def _make_google_oauth2_token_for_user(user):
|
|
return GoogleOAuth2Token.create_auth_token(organization=user.organization, user=user)
|
|
|
|
return _make_google_oauth2_token_for_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_public_api_token():
|
|
def _make_public_api_token(user, organization, name="test_api_token"):
|
|
return ApiAuthToken.create_auth_token(user, organization, name)
|
|
|
|
return _make_public_api_token
|
|
|
|
|
|
@pytest.fixture
|
|
def make_user_auth_headers():
|
|
def _make_user_auth_headers(
|
|
user,
|
|
token,
|
|
grafana_token: typing.Optional[str] = None,
|
|
grafana_context_data: typing.Optional[typing.Dict] = None,
|
|
):
|
|
instance_context_headers = {"stack_id": user.organization.stack_id, "org_id": user.organization.org_id}
|
|
grafana_context_headers = {"UserId": user.user_id}
|
|
if grafana_token is not None:
|
|
instance_context_headers["grafana_token"] = grafana_token
|
|
if grafana_context_data is not None:
|
|
grafana_context_headers.update(grafana_context_data)
|
|
|
|
return {
|
|
"HTTP_X-Instance-Context": json.dumps(instance_context_headers),
|
|
"HTTP_X-Grafana-Context": json.dumps(grafana_context_headers),
|
|
"HTTP_AUTHORIZATION": f"{token}",
|
|
}
|
|
|
|
return _make_user_auth_headers
|
|
|
|
|
|
RoleMapping = typing.Dict[LegacyAccessControlRole, typing.List[LegacyAccessControlCompatiblePermission]]
|
|
|
|
|
|
def get_user_permission_role_mapping_from_frontend_plugin_json() -> RoleMapping:
|
|
"""
|
|
This is used to take the RBAC permission -> basic role grants on the frontend
|
|
and test that the RBAC grants work the same way against the backend in terms of authorization
|
|
"""
|
|
|
|
class PluginJSONRoleDefinition(typing.TypedDict):
|
|
permissions: typing.List[GrafanaAPIPermission]
|
|
|
|
class PluginJSONRole(typing.TypedDict):
|
|
role: PluginJSONRoleDefinition
|
|
grants: typing.List[str]
|
|
|
|
class PluginJSON(typing.TypedDict):
|
|
roles: typing.List[PluginJSONRole]
|
|
|
|
with open("../grafana-plugin/src/plugin.json") as fp:
|
|
plugin_json: PluginJSON = json.load(fp)
|
|
|
|
role_mapping: RoleMapping = {
|
|
LegacyAccessControlRole.NONE: [],
|
|
LegacyAccessControlRole.VIEWER: [],
|
|
LegacyAccessControlRole.EDITOR: [],
|
|
LegacyAccessControlRole.ADMIN: [],
|
|
}
|
|
|
|
all_permission_classes: typing.Dict[str, LegacyAccessControlCompatiblePermission] = {
|
|
getattr(RBACPermission.Permissions, attr).value: getattr(RBACPermission.Permissions, attr)
|
|
for attr in dir(RBACPermission.Permissions)
|
|
if not attr.startswith("_")
|
|
}
|
|
|
|
# we just care about getting the basic role grants, everything else can be ignored
|
|
for role in plugin_json["roles"]:
|
|
if grants := role["grants"]:
|
|
for permission in role["role"]["permissions"]:
|
|
# only concerned with grafana-oncall-app specific grants
|
|
# ignore things like plugins.app:access actions
|
|
action = permission["action"]
|
|
permission_class = None
|
|
|
|
if action.startswith(ACTION_PREFIX):
|
|
permission_class = all_permission_classes[action]
|
|
|
|
if permission_class:
|
|
for grant in grants:
|
|
try:
|
|
role = LegacyAccessControlRole[grant.upper()]
|
|
if role not in role_mapping[role]:
|
|
role_mapping[role].append(permission_class)
|
|
except KeyError:
|
|
# may come across grants like "Grafana Admin"
|
|
# which we can ignore
|
|
continue
|
|
|
|
return role_mapping
|
|
|
|
|
|
ROLE_PERMISSION_MAPPING = get_user_permission_role_mapping_from_frontend_plugin_json()
|
|
|
|
|
|
@pytest.fixture
|
|
def make_user():
|
|
def _make_user(role: typing.Optional[LegacyAccessControlRole] = None, **kwargs):
|
|
role = LegacyAccessControlRole.ADMIN if role is None else role
|
|
permissions = kwargs.pop("permissions", None)
|
|
if permissions is None:
|
|
permissions_to_grant = ROLE_PERMISSION_MAPPING[role] if IS_RBAC_ENABLED else []
|
|
permissions = [GrafanaAPIPermission(action=perm.value) for perm in permissions_to_grant]
|
|
return UserFactory(role=role, permissions=permissions, **kwargs)
|
|
|
|
return _make_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_organization_and_user(make_organization, make_user_for_organization):
|
|
def _make_organization_and_user(role: typing.Optional[LegacyAccessControlRole] = None):
|
|
organization = make_organization()
|
|
user = make_user_for_organization(organization=organization, role=role)
|
|
return organization, user
|
|
|
|
return _make_organization_and_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_organization_and_user_with_slack_identities(
|
|
make_organization_with_slack_team_identity, make_user_with_slack_user_identity
|
|
):
|
|
def _make_organization_and_user_with_slack_identities(role: typing.Optional[LegacyAccessControlRole] = None):
|
|
organization, slack_team_identity = make_organization_with_slack_team_identity()
|
|
user, slack_user_identity = make_user_with_slack_user_identity(slack_team_identity, organization, role=role)
|
|
return organization, user, slack_team_identity, slack_user_identity
|
|
|
|
return _make_organization_and_user_with_slack_identities
|
|
|
|
|
|
@pytest.fixture
|
|
def make_user_with_slack_user_identity(make_user):
|
|
def _make_slack_user_identity_with_user(
|
|
slack_team_identity, organization, role: typing.Optional[LegacyAccessControlRole] = None, **kwargs
|
|
):
|
|
slack_user_identity = SlackUserIdentityFactory(slack_team_identity=slack_team_identity, **kwargs)
|
|
user = make_user(slack_user_identity=slack_user_identity, organization=organization, role=role)
|
|
return user, slack_user_identity
|
|
|
|
return _make_slack_user_identity_with_user
|
|
|
|
|
|
@pytest.fixture
|
|
def make_organization_with_slack_team_identity(make_slack_team_identity, make_organization):
|
|
def _make_slack_team_identity_with_organization(**kwargs):
|
|
slack_team_identity = make_slack_team_identity(**kwargs)
|
|
organization = make_organization(slack_team_identity=slack_team_identity)
|
|
return organization, slack_team_identity
|
|
|
|
return _make_slack_team_identity_with_organization
|
|
|
|
|
|
@pytest.fixture
|
|
def make_slack_team_identity():
|
|
def _make_slack_team_identity(**kwargs):
|
|
slack_team_identity = SlackTeamIdentityFactory(**kwargs)
|
|
return slack_team_identity
|
|
|
|
return _make_slack_team_identity
|
|
|
|
|
|
@pytest.fixture
|
|
def make_slack_user_identity():
|
|
def _make_slack_user_identity(**kwargs):
|
|
slack_user_identity = SlackUserIdentityFactory(**kwargs)
|
|
return slack_user_identity
|
|
|
|
return _make_slack_user_identity
|
|
|
|
|
|
@pytest.fixture
|
|
def make_slack_message():
|
|
def _make_slack_message(alert_group=None, organization=None, **kwargs):
|
|
organization = organization or alert_group.channel.organization
|
|
slack_message = SlackMessageFactory(
|
|
alert_group=alert_group,
|
|
organization=organization,
|
|
_slack_team_identity=organization.slack_team_identity,
|
|
**kwargs,
|
|
)
|
|
return slack_message
|
|
|
|
return _make_slack_message
|
|
|
|
|
|
@pytest.fixture
|
|
def make_team():
|
|
def _make_team(organization, **kwargs):
|
|
team = TeamFactory(organization=organization, **kwargs)
|
|
return team
|
|
|
|
return _make_team
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_receive_channel():
|
|
def _make_alert_receive_channel(organization, **kwargs):
|
|
if "integration" not in kwargs:
|
|
kwargs["integration"] = AlertReceiveChannel.INTEGRATION_GRAFANA
|
|
post_save.disconnect(listen_for_alertreceivechannel_model_save, sender=AlertReceiveChannel)
|
|
alert_receive_channel = AlertReceiveChannelFactory(organization=organization, **kwargs)
|
|
post_save.connect(listen_for_alertreceivechannel_model_save, sender=AlertReceiveChannel)
|
|
return alert_receive_channel
|
|
|
|
return _make_alert_receive_channel
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_receive_channel_connection():
|
|
def _make_alert_receive_channel_connection(source_alert_receive_channel, connected_alert_receive_channel, **kwargs):
|
|
alert_receive_channel_connection = AlertReceiveChannelConnectionFactory(
|
|
source_alert_receive_channel=source_alert_receive_channel,
|
|
connected_alert_receive_channel=connected_alert_receive_channel,
|
|
**kwargs,
|
|
)
|
|
return alert_receive_channel_connection
|
|
|
|
return _make_alert_receive_channel_connection
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_receive_channel_with_post_save_signal():
|
|
def _make_alert_receive_channel(organization, **kwargs):
|
|
if "integration" not in kwargs:
|
|
kwargs["integration"] = AlertReceiveChannel.INTEGRATION_GRAFANA
|
|
alert_receive_channel = AlertReceiveChannelFactory(organization=organization, **kwargs)
|
|
return alert_receive_channel
|
|
|
|
return _make_alert_receive_channel
|
|
|
|
|
|
@pytest.fixture
|
|
def make_channel_filter():
|
|
def _make_channel_filter(alert_receive_channel, filtering_term=None, **kwargs):
|
|
channel_filter = ChannelFilterFactory(
|
|
filtering_term=filtering_term,
|
|
alert_receive_channel=alert_receive_channel,
|
|
**kwargs,
|
|
)
|
|
return channel_filter
|
|
|
|
return _make_channel_filter
|
|
|
|
|
|
@pytest.fixture
|
|
def make_channel_filter_with_post_save():
|
|
def _make_channel_filter(alert_receive_channel, filtering_term=None, **kwargs):
|
|
channel_filter = ChannelFilterFactory(
|
|
filtering_term=filtering_term,
|
|
alert_receive_channel=alert_receive_channel,
|
|
**kwargs,
|
|
)
|
|
return channel_filter
|
|
|
|
return _make_channel_filter
|
|
|
|
|
|
@pytest.fixture
|
|
def make_escalation_chain():
|
|
def _make_escalation_chain(organization, **kwargs):
|
|
escalation_chain = EscalationChainFactory(organization=organization, **kwargs)
|
|
return escalation_chain
|
|
|
|
return _make_escalation_chain
|
|
|
|
|
|
@pytest.fixture
|
|
def make_escalation_policy():
|
|
def _make_escalation_policy(escalation_chain, escalation_policy_step, **kwargs):
|
|
escalation_policy = EscalationPolicyFactory(
|
|
escalation_chain=escalation_chain, step=escalation_policy_step, **kwargs
|
|
)
|
|
return escalation_policy
|
|
|
|
return _make_escalation_policy
|
|
|
|
|
|
@pytest.fixture
|
|
def make_user_notification_policy():
|
|
def _make_user_notification_policy(user, step, **kwargs):
|
|
user_notification_policy = UserNotificationPolicyFactory(user=user, step=step, **kwargs)
|
|
return user_notification_policy
|
|
|
|
return _make_user_notification_policy
|
|
|
|
|
|
@pytest.fixture
|
|
def make_user_notification_policy_log_record():
|
|
def _make_user_notification_policy_log_record(**kwargs):
|
|
post_save.disconnect(
|
|
listen_for_usernotificationpolicylogrecord_model_save, sender=UserNotificationPolicyLogRecord
|
|
)
|
|
user_notification_policy_log_record = UserNotificationPolicyLogRecordFactory(**kwargs)
|
|
post_save.connect(listen_for_usernotificationpolicylogrecord_model_save, sender=UserNotificationPolicyLogRecord)
|
|
|
|
return user_notification_policy_log_record
|
|
|
|
return _make_user_notification_policy_log_record
|
|
|
|
|
|
@pytest.fixture
|
|
def make_integration_escalation_chain_route_escalation_policy(
|
|
make_alert_receive_channel,
|
|
make_escalation_chain,
|
|
make_channel_filter,
|
|
make_escalation_policy,
|
|
):
|
|
def _make_integration_escalation_chain_route_escalation_policy(organization, escalation_policy_step):
|
|
alert_receive_channel = make_alert_receive_channel(organization)
|
|
escalation_chain = make_escalation_chain(organization)
|
|
default_channel_filter = make_channel_filter(
|
|
alert_receive_channel, escalation_chain=escalation_chain, is_default=True
|
|
)
|
|
escalation_policy = make_escalation_policy(escalation_chain, escalation_policy_step)
|
|
|
|
return alert_receive_channel, escalation_chain, default_channel_filter, escalation_policy
|
|
|
|
return _make_integration_escalation_chain_route_escalation_policy
|
|
|
|
|
|
@pytest.fixture
|
|
def make_invitation():
|
|
def _make_invitation(alert_group, author, invitee, **kwargs):
|
|
invitation = InvitationFactory(alert_group=alert_group, author=author, invitee=invitee, **kwargs)
|
|
return invitation
|
|
|
|
return _make_invitation
|
|
|
|
|
|
@pytest.fixture
|
|
def make_schedule():
|
|
def _make_schedule(organization, schedule_class, **kwargs):
|
|
factory = OnCallScheduleFactory.get_factory_for_class(schedule_class)
|
|
schedule = factory(organization=organization, **kwargs)
|
|
return schedule
|
|
|
|
return _make_schedule
|
|
|
|
|
|
@pytest.fixture
|
|
def make_on_call_shift():
|
|
def _make_on_call_shift(organization, shift_type, **kwargs):
|
|
on_call_shift = CustomOnCallShiftFactory(organization=organization, type=shift_type, **kwargs)
|
|
return on_call_shift
|
|
|
|
return _make_on_call_shift
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_group():
|
|
def _make_alert_group(alert_receive_channel, **kwargs):
|
|
alert_group = AlertGroupFactory(channel=alert_receive_channel, **kwargs)
|
|
return alert_group
|
|
|
|
return _make_alert_group
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_group_log_record():
|
|
def _make_alert_group_log_record(alert_group, type, author, **kwargs):
|
|
post_save.disconnect(listen_for_alertgrouplogrecord, sender=AlertGroupLogRecord)
|
|
log_record = AlertGroupLogRecordFactory(alert_group=alert_group, type=type, author=author, **kwargs)
|
|
post_save.connect(listen_for_alertgrouplogrecord, sender=AlertGroupLogRecord)
|
|
return log_record
|
|
|
|
return _make_alert_group_log_record
|
|
|
|
|
|
@pytest.fixture
|
|
def make_resolution_note():
|
|
def _make_resolution_note(alert_group, source=ResolutionNote.Source.WEB, author=None, **kwargs):
|
|
resolution_note = ResolutionNoteFactory(alert_group=alert_group, source=source, author=author, **kwargs)
|
|
return resolution_note
|
|
|
|
return _make_resolution_note
|
|
|
|
|
|
@pytest.fixture
|
|
def make_resolution_note_slack_message():
|
|
def _make_resolution_note_slack_message(alert_group, user, added_by_user, **kwargs):
|
|
return ResolutionNoteSlackMessageFactory(
|
|
alert_group=alert_group, user=user, added_by_user=added_by_user, **kwargs
|
|
)
|
|
|
|
return _make_resolution_note_slack_message
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert():
|
|
def _make_alert(alert_group, raw_request_data, **kwargs):
|
|
alert = AlertFactory(group=alert_group, raw_request_data=raw_request_data, **kwargs)
|
|
return alert
|
|
|
|
return _make_alert
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_with_custom_create_method():
|
|
def _make_alert_with_custom_create_method(
|
|
title,
|
|
message,
|
|
image_url,
|
|
link_to_upstream_details,
|
|
alert_receive_channel,
|
|
integration_unique_data,
|
|
raw_request_data,
|
|
**kwargs,
|
|
):
|
|
alert = Alert.create(
|
|
title,
|
|
message,
|
|
image_url,
|
|
link_to_upstream_details,
|
|
alert_receive_channel,
|
|
integration_unique_data,
|
|
raw_request_data,
|
|
**kwargs,
|
|
)
|
|
return alert
|
|
|
|
return _make_alert_with_custom_create_method
|
|
|
|
|
|
@pytest.fixture
|
|
def make_custom_action():
|
|
def _make_custom_action(organization, **kwargs):
|
|
custom_action = CustomActionFactory(organization=organization, **kwargs)
|
|
return custom_action
|
|
|
|
return _make_custom_action
|
|
|
|
|
|
@pytest.fixture
|
|
def make_custom_webhook():
|
|
def _make_custom_webhook(organization, **kwargs):
|
|
custom_webhook = CustomWebhookFactory(organization=organization, **kwargs)
|
|
return custom_webhook
|
|
|
|
return _make_custom_webhook
|
|
|
|
|
|
@pytest.fixture
|
|
def make_webhook_response():
|
|
def _make_webhook_response(**kwargs):
|
|
webhook_response = WebhookResponseFactory(**kwargs)
|
|
return webhook_response
|
|
|
|
return _make_webhook_response
|
|
|
|
|
|
@pytest.fixture
|
|
def make_slack_user_group():
|
|
def _make_slack_user_group(slack_team_identity, **kwargs):
|
|
slack_user_group = SlackUserGroupFactory(slack_team_identity=slack_team_identity, **kwargs)
|
|
return slack_user_group
|
|
|
|
return _make_slack_user_group
|
|
|
|
|
|
@pytest.fixture
|
|
def make_slack_channel():
|
|
def _make_slack_channel(slack_team_identity, **kwargs):
|
|
schedule = SlackChannelFactory(slack_team_identity=slack_team_identity, **kwargs)
|
|
return schedule
|
|
|
|
return _make_slack_channel
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_start_disable_maintenance_task(monkeypatch):
|
|
def mocked_start_disable_maintenance_task(*args, **kwargs):
|
|
return uuid.uuid4()
|
|
|
|
monkeypatch.setattr(MaintainableObject, "start_disable_maintenance_task", mocked_start_disable_maintenance_task)
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_organization_and_user_with_plugin_token(make_organization_and_user, make_token_for_organization):
|
|
def _make_organization_and_user_with_plugin_token(role: typing.Optional[LegacyAccessControlRole] = None):
|
|
organization, user = make_organization_and_user(role)
|
|
_, token = make_token_for_organization(organization)
|
|
return organization, user, token
|
|
|
|
return _make_organization_and_user_with_plugin_token
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_organization_and_user_with_mobile_app_verification_token(
|
|
make_organization_and_user, make_mobile_app_verification_token_for_user
|
|
):
|
|
def _make_organization_and_user_with_mobile_app_verification_token(
|
|
role: typing.Optional[LegacyAccessControlRole] = None,
|
|
):
|
|
organization, user = make_organization_and_user(role)
|
|
_, token = make_mobile_app_verification_token_for_user(user, organization)
|
|
return organization, user, token
|
|
|
|
return _make_organization_and_user_with_mobile_app_verification_token
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_organization_and_user_with_mobile_app_auth_token(
|
|
make_organization_and_user, make_mobile_app_auth_token_for_user
|
|
):
|
|
def _make_organization_and_user_with_mobile_app_auth_token(
|
|
role: typing.Optional[LegacyAccessControlRole] = None,
|
|
):
|
|
organization, user = make_organization_and_user(role)
|
|
_, token = make_mobile_app_auth_token_for_user(user, organization)
|
|
return organization, user, token
|
|
|
|
return _make_organization_and_user_with_mobile_app_auth_token
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_send_user_notification_signal(monkeypatch):
|
|
def mocked_send_signal(*args, **kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(user_notification_action_triggered_signal, "send", mocked_send_signal)
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_telegram_user_connector():
|
|
def _make_telegram_user_connector(user, **kwargs):
|
|
return TelegramToUserConnectorFactory(user=user, **kwargs)
|
|
|
|
return _make_telegram_user_connector
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_telegram_channel():
|
|
def _make_telegram_channel(organization, is_default_channel=False):
|
|
return TelegramChannelFactory(organization=organization, is_default_channel=is_default_channel)
|
|
|
|
return _make_telegram_channel
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_telegram_verification_code():
|
|
def _make_telegram_verification_code(user, **kwargs):
|
|
return TelegramVerificationCodeFactory(user=user, **kwargs)
|
|
|
|
return _make_telegram_verification_code
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_telegram_channel_verification_code():
|
|
def _make_telegram_channel_verification_code(organization, author, **kwargs):
|
|
return TelegramChannelVerificationCodeFactory(organization=organization, author=author, **kwargs)
|
|
|
|
return _make_telegram_channel_verification_code
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_telegram_message():
|
|
def _make_telegram_message(alert_group, message_type, **kwargs):
|
|
return TelegramMessageFactory(alert_group=alert_group, message_type=message_type, **kwargs)
|
|
|
|
return _make_telegram_message
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_phone_call_record():
|
|
def _make_phone_call_record(receiver, **kwargs):
|
|
return PhoneCallRecordFactory(receiver=receiver, **kwargs)
|
|
|
|
return _make_phone_call_record
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_sms_record():
|
|
def _make_sms_record(receiver, **kwargs):
|
|
return SMSRecordFactory(receiver=receiver, **kwargs)
|
|
|
|
return _make_sms_record
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_email_message():
|
|
def _make_email_message(receiver, **kwargs):
|
|
return EmailMessageFactory(receiver=receiver, **kwargs)
|
|
|
|
return _make_email_message
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_live_setting():
|
|
def _make_live_setting(name, **kwargs):
|
|
return LiveSettingFactory(name=name, **kwargs)
|
|
|
|
return _make_live_setting
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_integration_heartbeat():
|
|
def _make_integration_heartbeat(alert_receive_channel, timeout_seconds=60, last_heartbeat_time=None, **kwargs):
|
|
return IntegrationHeartBeatFactory(
|
|
alert_receive_channel=alert_receive_channel,
|
|
timeout_seconds=timeout_seconds,
|
|
last_heartbeat_time=last_heartbeat_time,
|
|
**kwargs,
|
|
)
|
|
|
|
return _make_integration_heartbeat
|
|
|
|
|
|
@pytest.fixture
|
|
def reload_urls(settings):
|
|
"""
|
|
Reloads Django URLs, especially useful when testing conditionally registered URLs
|
|
"""
|
|
|
|
def _reload_urls(app_url_file_to_reload: str = None):
|
|
clear_url_caches()
|
|
|
|
# this can be useful when testing conditionally registered URLs
|
|
# for example when a django app's urls.py file has conditional logic that is being
|
|
# overriden/tested, you will need to reload that urls.py file before relaoding the ROOT_URLCONF file
|
|
if app_url_file_to_reload is not None:
|
|
reload(import_module(app_url_file_to_reload))
|
|
|
|
urlconf = settings.ROOT_URLCONF
|
|
if urlconf in sys.modules:
|
|
reload(sys.modules[urlconf])
|
|
import_module(urlconf)
|
|
|
|
return _reload_urls
|
|
|
|
|
|
@pytest.fixture()
|
|
def load_slack_urls(settings, reload_urls):
|
|
settings.FEATURE_SLACK_INTEGRATION_ENABLED = True
|
|
reload_urls()
|
|
|
|
|
|
@pytest.fixture
|
|
def make_region():
|
|
def _make_region(**kwargs):
|
|
region = RegionFactory(**kwargs)
|
|
return region
|
|
|
|
return _make_region
|
|
|
|
|
|
@pytest.fixture
|
|
def make_organization_and_region(make_organization, make_region):
|
|
def _make_organization_and_region():
|
|
organization = make_organization()
|
|
region = make_region()
|
|
organization.migration_destination = region
|
|
return organization, region
|
|
|
|
return _make_organization_and_region
|
|
|
|
|
|
@pytest.fixture()
|
|
def make_organization_and_user_with_token(make_organization_and_user, make_public_api_token):
|
|
def _make_organization_and_user_with_token():
|
|
organization, user = make_organization_and_user()
|
|
_, token = make_public_api_token(user, organization)
|
|
return organization, user, token
|
|
|
|
return _make_organization_and_user_with_token
|
|
|
|
|
|
@pytest.fixture
|
|
def make_shift_swap_request():
|
|
def _make_shift_swap_request(schedule, beneficiary, **kwargs):
|
|
return ShiftSwapRequestFactory(schedule=schedule, beneficiary=beneficiary, **kwargs)
|
|
|
|
return _make_shift_swap_request
|
|
|
|
|
|
@pytest.fixture
|
|
def shift_swap_request_setup(
|
|
make_schedule, make_organization_and_user, make_user_for_organization, make_shift_swap_request
|
|
):
|
|
def _shift_swap_request_setup(**kwargs):
|
|
organization, beneficiary = make_organization_and_user()
|
|
benefactor = make_user_for_organization(organization)
|
|
|
|
schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb)
|
|
tomorrow = timezone.now() + datetime.timedelta(days=1)
|
|
two_days_from_now = tomorrow + datetime.timedelta(days=1)
|
|
|
|
ssr = make_shift_swap_request(schedule, beneficiary, swap_start=tomorrow, swap_end=two_days_from_now, **kwargs)
|
|
|
|
return ssr, beneficiary, benefactor
|
|
|
|
return _shift_swap_request_setup
|
|
|
|
|
|
@pytest.fixture()
|
|
def webhook_preset_api_setup():
|
|
WebhookPresetOptions.WEBHOOK_PRESETS = {
|
|
TEST_WEBHOOK_PRESET_ID: TestWebhookPreset(),
|
|
ADVANCED_WEBHOOK_PRESET_ID: TestAdvancedWebhookPreset(),
|
|
}
|
|
WebhookPresetOptions.WEBHOOK_PRESET_CHOICES = [
|
|
preset.metadata for preset in WebhookPresetOptions.WEBHOOK_PRESETS.values()
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def make_label_key():
|
|
def _make_label_key(organization, key_id=None, key_name=None, **kwargs):
|
|
if key_id is not None:
|
|
kwargs["id"] = key_id
|
|
|
|
if key_name is not None:
|
|
kwargs["name"] = key_name
|
|
|
|
return LabelKeyFactory(organization=organization, **kwargs)
|
|
|
|
return _make_label_key
|
|
|
|
|
|
@pytest.fixture
|
|
def make_label_value():
|
|
def _make_label_value(key, value_id=None, value_name=None, **kwargs):
|
|
if value_id is not None:
|
|
kwargs["id"] = value_id
|
|
|
|
if value_name is not None:
|
|
kwargs["name"] = value_name
|
|
|
|
return LabelValueFactory(key=key, **kwargs)
|
|
|
|
return _make_label_value
|
|
|
|
|
|
@pytest.fixture
|
|
def make_label_key_and_value(make_label_key, make_label_value):
|
|
def _make_label_key_and_value(organization, key_id=None, key_name=None, value_id=None, value_name=None):
|
|
key = make_label_key(organization=organization, key_id=key_id, key_name=key_name)
|
|
value = make_label_value(key=key, value_id=value_id, value_name=value_name)
|
|
return key, value
|
|
|
|
return _make_label_key_and_value
|
|
|
|
|
|
@pytest.fixture
|
|
def make_integration_label_association(make_label_key_and_value):
|
|
def _make_integration_label_association(
|
|
organization, alert_receive_channel, key_id=None, key_name=None, value_id=None, value_name=None, **kwargs
|
|
):
|
|
key, value = make_label_key_and_value(
|
|
organization, key_id=key_id, key_name=key_name, value_id=value_id, value_name=value_name
|
|
)
|
|
return AlertReceiveChannelAssociatedLabelFactory(
|
|
alert_receive_channel=alert_receive_channel, organization=organization, key=key, value=value, **kwargs
|
|
)
|
|
|
|
return _make_integration_label_association
|
|
|
|
|
|
@pytest.fixture
|
|
def make_alert_group_label_association():
|
|
def _make_alert_group_label_association(organization, alert_group, **kwargs):
|
|
return AlertGroupAssociatedLabelFactory(alert_group=alert_group, organization=organization, **kwargs)
|
|
|
|
return _make_alert_group_label_association
|
|
|
|
|
|
@pytest.fixture
|
|
def make_webhook_label_association(make_label_key_and_value):
|
|
def _make_integration_label_association(organization, webhook, **kwargs):
|
|
key, value = make_label_key_and_value(organization)
|
|
return WebhookAssociatedLabelFactory(webhook=webhook, organization=organization, key=key, value=value, **kwargs)
|
|
|
|
return _make_integration_label_association
|
|
|
|
|
|
@pytest.fixture
|
|
def make_google_oauth2_user_for_user():
|
|
def _make_google_oauth2_user_for_user(user):
|
|
return GoogleOAuth2UserFactory(user=user)
|
|
|
|
return _make_google_oauth2_user_for_user
|