oncall-engine/engine/apps/slack/tests/test_reset_slack.py

from unittest.mock import patch

import pytest
from django.conf import settings
from django.urls import reverse
from rest_framework import status
from rest_framework.response import Response
from rest_framework.test import APIClient

from apps.api.permissions import LegacyAccessControlRole
from apps.schedules.models import OnCallScheduleWeb
from apps.slack.tasks import (
    clean_slack_channel_leftovers,
    clean_slack_integration_leftovers,
    unpopulate_slack_user_identities,
)
from apps.user_management.models import User


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_reset_slack_integration_permissions(
    make_organization_and_user_with_plugin_token, load_slack_urls, make_user_auth_headers, role, expected_status
):
    settings.FEATURE_SLACK_INTEGRATION_ENABLED = True

    _, user, token = make_organization_and_user_with_plugin_token(role=role)
    client = APIClient()

    url = reverse("reset-slack")
    with patch("apps.slack.views.ResetSlackView.post", return_value=Response(status=status.HTTP_200_OK)):
        response = client.post(url, format="json", **make_user_auth_headers(user, token))

    assert response.status_code == expected_status


@pytest.mark.django_db
def test_clean_slack_channel_leftovers(
    make_slack_team_identity,
    make_slack_channel,
    make_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_slack_user_group,
    make_schedule,
):
    slack_team_identity = make_slack_team_identity()
    slack_channel = make_slack_channel(slack_team_identity)
    organization = make_organization(slack_team_identity=slack_team_identity, default_slack_channel=slack_channel)

    # create channel filter with Slack channel
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, slack_channel=slack_channel)

    # create schedule with Slack channel and user group
    user_group = make_slack_user_group(slack_team_identity)
    schedule = make_schedule(
        organization,
        schedule_class=OnCallScheduleWeb,
        slack_channel=slack_channel,
        user_group=user_group,
    )

    assert channel_filter.slack_channel == slack_channel
    assert schedule.slack_channel == slack_channel
    assert schedule.user_group == user_group
    assert organization.default_slack_channel_slack_id == slack_channel.slack_id

    # clean Slack channel leftovers
    clean_slack_channel_leftovers(slack_team_identity.pk, slack_channel.slack_id)
    channel_filter.refresh_from_db()
    schedule.refresh_from_db()
    organization.refresh_from_db()

    # check that references to Slack objects are removed
    assert channel_filter.slack_channel is None
    assert organization.default_slack_channel is None

    # NOTE: user groups shouldn't be updated for schedules, only the channel
    assert schedule.slack_channel is None
    assert schedule.user_group == user_group


@pytest.mark.django_db
def test_clean_slack_integration_leftovers(
    make_slack_team_identity,
    make_slack_channel,
    make_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_slack_user_group,
    make_schedule,
):
    slack_team_identity = make_slack_team_identity()
    slack_channel = make_slack_channel(slack_team_identity)
    organization = make_organization(slack_team_identity=slack_team_identity, default_slack_channel=slack_channel)

    # create channel filter with Slack channel
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, slack_channel=slack_channel)

    # create schedule with Slack channel and user group
    user_group = make_slack_user_group(slack_team_identity)
    schedule = make_schedule(
        organization,
        schedule_class=OnCallScheduleWeb,
        slack_channel=slack_channel,
        user_group=user_group,
    )

    assert channel_filter.slack_channel == slack_channel
    assert schedule.slack_channel == slack_channel
    assert schedule.user_group == user_group

    # clean Slack integration leftovers
    clean_slack_integration_leftovers(organization.pk)
    channel_filter.refresh_from_db()
    schedule.refresh_from_db()

    # check that references to Slack objects are removed
    assert channel_filter.slack_channel is None
    assert schedule.slack_channel is None
    assert schedule.user_group is None


@pytest.mark.django_db
def test_unpopulate_slack_user_identities(
    make_slack_team_identity,
    make_slack_channel,
    make_organization,
    make_user_for_organization,
    make_user_with_slack_user_identity,
):
    # create organization and user with Slack connected
    slack_team_identity = make_slack_team_identity()
    slack_channel = make_slack_channel(slack_team_identity)
    organization = make_organization(slack_team_identity=slack_team_identity, default_slack_channel=slack_channel)
    user = make_user_for_organization(organization)

    assert organization.default_slack_channel_slack_id is not None

    # create & delete user with Slack connected
    deleted_user, _ = make_user_with_slack_user_identity(slack_team_identity, organization)
    User.objects.filter(pk=deleted_user.pk).delete()

    # unpopulate Slack user identities
    unpopulate_slack_user_identities(organization.pk, force=True)
    user.refresh_from_db()
    deleted_user.refresh_from_db()
    organization.refresh_from_db()

    # check that references to Slack user identities are removed
    assert user.slack_user_identity is None
    assert deleted_user.slack_user_identity is None

    # check that Slack specific info is reset for organization
    assert organization.slack_team_identity is None
    assert organization.default_slack_channel_slack_id is None


@pytest.mark.django_db
def test_delete_slack_channel_and_cascade_deletes(
    make_slack_team_identity,
    make_slack_channel,
    make_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_schedule,
):
    slack_team_identity = make_slack_team_identity()
    slack_channel = make_slack_channel(slack_team_identity)
    organization = make_organization(slack_team_identity=slack_team_identity, default_slack_channel=slack_channel)

    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, slack_channel=slack_channel)
    schedule = make_schedule(organization, schedule_class=OnCallScheduleWeb, slack_channel=slack_channel)

    assert channel_filter.slack_channel == slack_channel
    assert schedule.slack_channel == slack_channel

    slack_channel.delete()
    channel_filter.refresh_from_db()
    schedule.refresh_from_db()

    assert channel_filter.slack_channel is None
    assert schedule.slack_channel is None