oncall-engine/engine/conftest.py

import binascii
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,
    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,
    RelatedIncidentFactory,
    ResolutionNoteFactory,
    ResolutionNoteSlackMessageFactory,
    UserNotificationBundleFactory,
)
from apps.api.permissions import (
    GrafanaAPIPermission,
    LegacyAccessControlCompatiblePermission,
    LegacyAccessControlRole,
    RBACPermission,
)
from apps.auth_token import constants as auth_token_constants
from apps.auth_token.crypto import hash_token_string
from apps.auth_token.models import (
    ApiAuthToken,
    GoogleOAuth2Token,
    IntegrationBacksyncAuthToken,
    PluginAuthToken,
    ServiceAccountToken,
    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 import constants as google_constants
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,
    ServiceAccountFactory,
    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,
)
from common.constants.plugin_ids import PluginID

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)
register(UserNotificationBundleFactory)

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):
        if "is_rbac_permissions_enabled" not in kwargs:
            kwargs["is_rbac_permissions_enabled"] = IS_RBAC_ENABLED
        return OrganizationFactory(**kwargs, 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_service_account_for_organization(make_user):
    def _make_service_account_for_organization(organization, **kwargs):
        return ServiceAccountFactory(organization=organization, **kwargs)

    return _make_service_account_for_organization


@pytest.fixture
def make_token_for_service_account():
    def _make_token_for_service_account(service_account, token_string):
        prefix_length = len(ServiceAccountToken.GRAFANA_SA_PREFIX)
        token_key = token_string[prefix_length : prefix_length + auth_token_constants.TOKEN_KEY_LENGTH]
        hashable_token = binascii.hexlify(token_string.encode()).decode()
        digest = hash_token_string(hashable_token)
        return ServiceAccountToken.objects.create(
            service_account=service_account,
            token_key=token_key,
            digest=digest,
        )

    return _make_token_for_service_account


@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,
        organization=None,
    ):
        org = organization or user.organization
        instance_context_headers = {"stack_id": org.stack_id, "org_id": org.org_id}
        grafana_context_headers = {"UserId": user.user_id if user else None}
        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)

    # NOTE: we need to manually add grafana-labels-app permissions here since these
    # are granted to basic roles via the grafana-labels-app itself, and not
    # ../grafana-plugin/src/plugin.json
    #
    # However, we do sync these permissions into our backend. See
    # https://github.com/grafana/irm/pull/200 for more details
    #
    # We don't currently add the label delete permission here because we don't currently
    # use this in OnCall
    role_mapping: RoleMapping = {
        LegacyAccessControlRole.NONE: [],
        LegacyAccessControlRole.VIEWER: [
            RBACPermission.Permissions.LABEL_READ,
        ],
        LegacyAccessControlRole.EDITOR: [
            RBACPermission.Permissions.LABEL_READ,
            RBACPermission.Permissions.LABEL_WRITE,
            RBACPermission.Permissions.LABEL_CREATE,
        ],
        LegacyAccessControlRole.ADMIN: [
            RBACPermission.Permissions.LABEL_READ,
            RBACPermission.Permissions.LABEL_WRITE,
            RBACPermission.Permissions.LABEL_CREATE,
        ],
    }

    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(PluginID.ONCALL):
                    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(channel, alert_group=None, organization=None, **kwargs):
        # TODO: once _channel_id has been fully migrated to channel, remove _channel_id
        # see https://raintank-corp.slack.com/archives/C06K1MQ07GS/p1732555465144099
        return SlackMessageFactory(
            alert_group=alert_group,
            organization=organization or alert_group.channel.organization,
            _channel_id=channel.slack_id,
            channel=channel,
            **kwargs,
        )

    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 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, **kwargs):
        oauth_scope = kwargs.pop("oauth_scope", " ".join(google_constants.REQUIRED_OAUTH_SCOPES))
        return GoogleOAuth2UserFactory(user=user, oauth_scope=oauth_scope, **kwargs)

    return _make_google_oauth2_user_for_user


@pytest.fixture
def make_user_notification_bundle():
    def _make_user_notification_bundle(user, notification_channel, important=False, **kwargs):
        return UserNotificationBundleFactory(
            user=user, notification_channel=notification_channel, important=important, **kwargs
        )

    return _make_user_notification_bundle


@pytest.fixture
def make_related_incident():
    def _make_related_incident(incident_id, organization, channel_filter):
        return RelatedIncidentFactory(incident_id=incident_id, organization=organization, channel_filter=channel_filter)

    return _make_related_incident