oncall-engine/engine/apps/api/tests/test_alert_group.py

from unittest.mock import Mock, patch

import pytest
from django.core.cache import cache
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from rest_framework.response import Response
from rest_framework.test import APIClient

from apps.alerts.constants import ActionSource
from apps.alerts.models import (
    AlertGroup,
    AlertGroupExternalID,
    AlertGroupLogRecord,
    AlertReceiveChannel,
    ResolutionNote,
)
from apps.alerts.paging import direct_paging
from apps.alerts.tasks import wipe
from apps.api.errors import AlertGroupAPIError
from apps.api.permissions import LegacyAccessControlRole
from apps.api.serializers.alert import AlertFieldsCacheSerializerMixin
from apps.api.serializers.alert_group import AlertGroupFieldsCacheSerializerMixin
from apps.base.models import UserNotificationPolicyLogRecord
from common.api_helpers.filters import DateRangeFilterMixin

alert_raw_request_data = {
    "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": "http://localhost:3000/",
    "state": "alerting",
    "title": "[Alerting] Test notification",
}


@pytest.fixture()
def alert_group_internal_api_setup(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_resolved_ack_new_silenced_alert_groups,
):
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)
    alert_groups = make_resolved_ack_new_silenced_alert_groups(
        alert_receive_channel, default_channel_filter, alert_raw_request_data
    )
    return user, token, alert_groups


@pytest.mark.django_db
def test_get_filter_by_integration(
    alert_group_internal_api_setup, make_alert_receive_channel, make_alert_group, make_user_auth_headers
):
    user, token, alert_groups = alert_group_internal_api_setup

    ag = alert_groups[0]
    # channel filter could be None, but the alert group still belongs to the original integration
    ag.channel_filter = None
    ag.save()

    # make an alert group in other integration
    alert_receive_channel = make_alert_receive_channel(user.organization)
    make_alert_group(alert_receive_channel)

    client = APIClient()
    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + f"?integration={ag.channel.public_primary_key}",
        format="json",
        **make_user_auth_headers(user, token),
    )

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 4


@pytest.mark.django_db
def test_get_alert_groups_from_deleted_integration(alert_group_internal_api_setup, make_user_auth_headers):
    user, token, alert_groups = alert_group_internal_api_setup

    alert_receive_channel = alert_groups[0].channel
    alert_receive_channel.delete()

    client = APIClient()

    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url,
        format="json",
        **make_user_auth_headers(user, token),
    )

    assert response.status_code == status.HTTP_200_OK
    # Alert groups from deleted integrations should be returned
    assert len(response.data["results"]) == 4


@pytest.mark.django_db
def test_get_filter_started_at(alert_group_internal_api_setup, make_user_auth_headers):
    user, token, _ = alert_group_internal_api_setup
    client = APIClient()

    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + "?started_at=1970-01-01T00:00:00_2099-01-01T23:59:59",
        format="json",
        **make_user_auth_headers(user, token),
    )

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 4


@pytest.mark.django_db
def test_get_filter_resolved_at_alertgroup_empty_result(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, _ = alert_group_internal_api_setup

    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + "?resolved_at=1970-01-01T00:00:00_1970-01-01T23:59:59",
        format="json",
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 0


@pytest.mark.django_db
def test_get_filter_resolved_at_alertgroup_invalid_format(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, _ = alert_group_internal_api_setup

    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + "?resolved_at=invalid_date_format", format="json", **make_user_auth_headers(user, token)
    )
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_get_filter_resolved_at(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup

    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + "?resolved_at=1970-01-01T00:00:00_2099-01-01T23:59:59",
        format="json",
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1


@pytest.mark.django_db
def test_status_new(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups

    url = reverse("api-internal:alertgroup-list")
    response = client.get(url + "?status=0", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1
    assert response.data["results"][0]["pk"] == new_alert_group.public_primary_key


@pytest.mark.django_db
def test_status_ack(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    _, ack_alert_group, _, _ = alert_groups

    url = reverse("api-internal:alertgroup-list")
    response = client.get(url + "?status=1", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1
    assert response.data["results"][0]["pk"] == ack_alert_group.public_primary_key


@pytest.mark.django_db
def test_status_resolved(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, _, _, _ = alert_groups

    url = reverse("api-internal:alertgroup-list")
    response = client.get(url + "?status=2", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1
    assert response.data["results"][0]["pk"] == resolved_alert_group.public_primary_key


@pytest.mark.django_db
def test_status_silenced(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    _, _, _, silenced_alert_group = alert_groups

    url = reverse("api-internal:alertgroup-list")
    response = client.get(url + "?status=3", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1
    assert response.data["results"][0]["pk"] == silenced_alert_group.public_primary_key


@pytest.mark.django_db
def test_all_statuses(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, _, _, _ = alert_groups

    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + "?status=0&status=1&&status=2&status=3", format="json", **make_user_auth_headers(user, token)
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 4


@pytest.mark.django_db
def test_get_filter_resolved_by(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolved_ack_new_silenced_alert_groups,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    resolved_alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
        acknowledged_at=timezone.now() + timezone.timedelta(hours=1),
        resolved_at=timezone.now() + timezone.timedelta(hours=2),
        resolved=True,
        acknowledged=True,
        resolved_by_user=first_user,
        acknowledged_by_user=second_user,
    )
    make_alert(alert_group=resolved_alert_group, raw_request_data=alert_raw_request_data)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?resolved_by={first_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1

    second_response = client.get(
        url + f"?resolved_by={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert second_response.status_code == status.HTTP_200_OK
    assert len(second_response.data["results"]) == 0


@pytest.mark.django_db
def test_get_filter_resolved_by_multiple_values(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolved_ack_new_silenced_alert_groups,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)
    third_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    def make_resolved_by_user_alert_group(user):
        resolved_alert_group = make_alert_group(
            alert_receive_channel,
            channel_filter=default_channel_filter,
            acknowledged_at=timezone.now() + timezone.timedelta(hours=1),
            resolved_at=timezone.now() + timezone.timedelta(hours=2),
            resolved=True,
            acknowledged=True,
            resolved_by_user=user,
            acknowledged_by_user=user,
        )
        make_alert(alert_group=resolved_alert_group, raw_request_data=alert_raw_request_data)

    make_resolved_by_user_alert_group(first_user)
    make_resolved_by_user_alert_group(second_user)
    make_resolved_by_user_alert_group(third_user)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?resolved_by={first_user.public_primary_key}&" f"resolved_by={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_acknowledged_by(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolved_ack_new_silenced_alert_groups,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    acknowledged_alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
        acknowledged_at=timezone.now() + timezone.timedelta(hours=1),
        resolved_at=timezone.now() + timezone.timedelta(hours=2),
        acknowledged=True,
        acknowledged_by_user=first_user,
    )
    make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?acknowledged_by={first_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1

    second_response = client.get(
        url + f"?acknowledged_by={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert second_response.status_code == status.HTTP_200_OK
    assert len(second_response.data["results"]) == 0


@pytest.mark.django_db
def test_get_filter_acknowledged_by_multiple_values(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolved_ack_new_silenced_alert_groups,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)
    third_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    def make_acknowledged_by_user_alert_group(user):
        acknowledged_alert_group = make_alert_group(
            alert_receive_channel,
            channel_filter=default_channel_filter,
            acknowledged_at=timezone.now() + timezone.timedelta(hours=1),
            resolved_at=timezone.now() + timezone.timedelta(hours=2),
            acknowledged=True,
            acknowledged_by_user=user,
        )
        make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)

    make_acknowledged_by_user_alert_group(first_user)
    make_acknowledged_by_user_alert_group(second_user)
    make_acknowledged_by_user_alert_group(third_user)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?acknowledged_by={first_user.public_primary_key}" f"&acknowledged_by={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_silenced_by(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolved_ack_new_silenced_alert_groups,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    silenced_alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
        silenced_at=timezone.now() + timezone.timedelta(hours=1),
        silenced=True,
        silenced_by_user=first_user,
    )
    make_alert(alert_group=silenced_alert_group, raw_request_data=alert_raw_request_data)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?silenced_by={first_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1

    second_response = client.get(
        url + f"?silenced_by={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert second_response.status_code == status.HTTP_200_OK
    assert len(second_response.data["results"]) == 0


@pytest.mark.django_db
def test_get_filter_silenced_by_multiple_values(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolved_ack_new_silenced_alert_groups,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)
    third_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    def make_silenced_by_user_alert_group(user):
        acknowledged_alert_group = make_alert_group(
            alert_receive_channel,
            channel_filter=default_channel_filter,
            silenced_at=timezone.now() + timezone.timedelta(hours=1),
            silenced=True,
            silenced_by_user=user,
        )
        make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)

    make_silenced_by_user_alert_group(first_user)
    make_silenced_by_user_alert_group(second_user)
    make_silenced_by_user_alert_group(third_user)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?silenced_by={first_user.public_primary_key}&silenced_by={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_invitees_are(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
    )
    make_alert(alert_group=alert_group, raw_request_data={})
    alert_group.log_records.create(
        type=AlertGroupLogRecord.TYPE_ESCALATION_TRIGGERED,
        author=first_user,
    )

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?invitees_are={first_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1

    second_response = client.get(
        url + f"?invitees_are={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert second_response.status_code == status.HTTP_200_OK
    assert len(second_response.data["results"]) == 0


@pytest.mark.django_db
def test_get_filter_invitees_are_multiple_values(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)
    third_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    def make_alert_group_with_invitee(user):
        alert_group = make_alert_group(
            alert_receive_channel,
            channel_filter=default_channel_filter,
        )
        make_alert(alert_group=alert_group, raw_request_data={})

        alert_group.log_records.create(
            type=AlertGroupLogRecord.TYPE_ESCALATION_TRIGGERED,
            author=user,
        )

    make_alert_group_with_invitee(first_user)
    make_alert_group_with_invitee(second_user)
    make_alert_group_with_invitee(third_user)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?invitees_are={first_user.public_primary_key}" f"&invitees_are={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_invitees_are_ag_with_multiple_logs(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
    )
    make_alert(alert_group=alert_group, raw_request_data={})

    alert_group.log_records.create(
        type=AlertGroupLogRecord.TYPE_ESCALATION_TRIGGERED,
        author=first_user,
    )

    alert_group.log_records.create(
        type=AlertGroupLogRecord.TYPE_ESCALATION_TRIGGERED,
        author=second_user,
    )

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?invitees_are={first_user.public_primary_key}" f"&invitees_are={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1


@pytest.mark.django_db
def test_get_filter_mine(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    acknowledged_alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
        acknowledged_at=timezone.now() + timezone.timedelta(hours=1),
        resolved_at=timezone.now() + timezone.timedelta(hours=2),
        acknowledged=True,
        acknowledged_by_user=first_user,
    )
    make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)

    # other alert group
    make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
    )
    make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + "?mine=true",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1

    second_response = client.get(
        url + "?mine=false",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert second_response.status_code == status.HTTP_200_OK
    assert len(second_response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_involved_users(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()

    organization, first_user, token = make_organization_and_user_with_plugin_token()
    second_user = make_user_for_organization(organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    default_channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    acknowledged_alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
        acknowledged_at=timezone.now() + timezone.timedelta(hours=1),
        resolved_at=timezone.now() + timezone.timedelta(hours=2),
        acknowledged=True,
        acknowledged_by_user=first_user,
    )
    make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)

    # other alert group
    other_alert_group = make_alert_group(
        alert_receive_channel,
        channel_filter=default_channel_filter,
    )
    make_alert(alert_group=acknowledged_alert_group, raw_request_data=alert_raw_request_data)
    # second user was notified
    other_alert_group.personal_log_records.create(
        type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_TRIGGERED,
        author=second_user,
    )

    url = reverse("api-internal:alertgroup-list")

    first_response = client.get(
        url + f"?acknowledged_by={first_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert first_response.status_code == status.HTTP_200_OK
    assert len(first_response.data["results"]) == 1

    second_response = client.get(
        url
        + f"?involved_users_are={first_user.public_primary_key}&involved_users_are={second_user.public_primary_key}",
        format="json",
        **make_user_auth_headers(first_user, token),
    )
    assert second_response.status_code == status.HTTP_200_OK
    assert len(second_response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_with_resolution_note(
    alert_group_internal_api_setup,
    make_resolution_note,
    make_user_auth_headers,
):
    user, token, alert_groups = alert_group_internal_api_setup
    res_alert_group, ack_alert_group, _, _ = alert_groups
    client = APIClient()

    url = reverse("api-internal:alertgroup-list")

    # there are no alert groups with resolution_notes
    response = client.get(url + "?with_resolution_note=true", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 0

    response = client.get(url + "?with_resolution_note=false", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 4

    # add resolution_notes to two of four alert groups
    make_resolution_note(res_alert_group)
    make_resolution_note(ack_alert_group)

    response = client.get(url + "?with_resolution_note=true", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 2

    response = client.get(url + "?with_resolution_note=false", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_with_resolution_note_after_delete_resolution_note(
    alert_group_internal_api_setup,
    make_resolution_note,
    make_user_auth_headers,
):
    user, token, alert_groups = alert_group_internal_api_setup
    res_alert_group, ack_alert_group, _, _ = alert_groups
    client = APIClient()

    url = reverse("api-internal:alertgroup-list")

    # add resolution note to two alert group
    resolution_note_res_alert_group = make_resolution_note(res_alert_group)
    make_resolution_note(ack_alert_group)

    # delete resolution note message using soft delete
    resolution_note_res_alert_group.delete()
    resolution_note_res_alert_group.refresh_from_db()
    assert resolution_note_res_alert_group.deleted_at is not None

    response = client.get(url + "?with_resolution_note=true", format="json", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1


@pytest.mark.django_db
def test_get_filter_escalation_chain(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_escalation_chain,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()
    organization, user, token = make_organization_and_user_with_plugin_token()

    alert_receive_channel = make_alert_receive_channel(organization)

    escalation_chain_1 = make_escalation_chain(organization=organization)
    escalation_chain_2 = make_escalation_chain(organization=organization)

    channel_filter_1 = make_channel_filter(alert_receive_channel, escalation_chain=escalation_chain_1, is_default=True)
    channel_filter_2 = make_channel_filter(alert_receive_channel, escalation_chain=escalation_chain_2, is_default=False)

    alert_group_1 = make_alert_group(alert_receive_channel, channel_filter=channel_filter_1)
    make_alert(alert_group=alert_group_1, raw_request_data=alert_raw_request_data)

    alert_group_2 = make_alert_group(alert_receive_channel, channel_filter=channel_filter_2)
    make_alert(alert_group=alert_group_2, raw_request_data=alert_raw_request_data)

    url = reverse("api-internal:alertgroup-list")

    # check when a single escalation chain is passed
    response = client.get(
        url + f"?escalation_chain={escalation_chain_1.public_primary_key}", **make_user_auth_headers(user, token)
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1
    assert response.data["results"][0]["pk"] == alert_group_1.public_primary_key

    # check when multiple escalation chains are passed
    response = client.get(
        url
        + f"?escalation_chain={escalation_chain_1.public_primary_key}&escalation_chain={escalation_chain_2.public_primary_key}",
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 2


@pytest.mark.django_db
def test_get_filter_by_teams(
    make_organization_and_user_with_plugin_token,
    make_team,
    make_alert_receive_channel,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    client = APIClient()
    organization, user, token = make_organization_and_user_with_plugin_token()
    team1 = make_team(organization)
    team2 = make_team(organization)

    alert_receive_channel_0 = make_alert_receive_channel(organization)
    alert_receive_channel_1 = make_alert_receive_channel(organization, team=team1)
    alert_receive_channel_2 = make_alert_receive_channel(organization, team=team2)

    alert_group_0 = make_alert_group(alert_receive_channel_0)
    make_alert(alert_group=alert_group_0, raw_request_data=alert_raw_request_data)

    alert_group_1 = make_alert_group(alert_receive_channel_1)
    make_alert(alert_group=alert_group_1, raw_request_data=alert_raw_request_data)

    alert_group_2 = make_alert_group(alert_receive_channel_2)
    make_alert(alert_group=alert_group_2, raw_request_data=alert_raw_request_data)

    url = reverse("api-internal:alertgroup-list")

    # check no team is given
    response = client.get(url, **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 3
    assert {ag["pk"] for ag in response.data["results"]} == {
        alert_group_0.public_primary_key,
        alert_group_1.public_primary_key,
        alert_group_2.public_primary_key,
    }

    # check the "No team" case
    response = client.get(url + "?team=null", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1
    assert {ag["pk"] for ag in response.data["results"]} == {alert_group_0.public_primary_key}

    # check the "No team" + other team case
    response = client.get(url + f"?team=null&team={team2.public_primary_key}", **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 2
    assert {ag["pk"] for ag in response.data["results"]} == {
        alert_group_0.public_primary_key,
        alert_group_2.public_primary_key,
    }

    # check the multiple teams case
    response = client.get(
        url + f"?team={team1.public_primary_key}&team={team2.public_primary_key}", **make_user_auth_headers(user, token)
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 2
    assert {ag["pk"] for ag in response.data["results"]} == {
        alert_group_1.public_primary_key,
        alert_group_2.public_primary_key,
    }


@pytest.mark.django_db
def test_get_filter_labels(
    make_organization_and_user_with_plugin_token,
    make_user_for_organization,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_alert_group_label_association,
    make_user_auth_headers,
):
    organization, user, token = make_organization_and_user_with_plugin_token()

    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    alert_groups = []
    for _ in range(3):
        alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
        make_alert(alert_group=alert_group, raw_request_data=alert_raw_request_data)
        alert_groups.append(alert_group)

    make_alert_group_label_association(organization, alert_groups[0], key_name="a", value_name="b")
    make_alert_group_label_association(organization, alert_groups[0], key_name="c", value_name="d")
    make_alert_group_label_association(organization, alert_groups[1], key_name="a", value_name="b")
    make_alert_group_label_association(organization, alert_groups[2], key_name="c", value_name="d")

    client = APIClient()
    url = reverse("api-internal:alertgroup-list")

    response = client.get(
        url + "?label=a:b&label=c:d",
        format="json",
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.json()["results"]) == 1
    assert response.json()["results"][0]["pk"] == alert_groups[0].public_primary_key


@pytest.mark.django_db
def test_get_filter_by_related_incident(
    alert_group_internal_api_setup, make_related_incident, make_alert_group, make_user_auth_headers
):
    user, token, alert_groups = alert_group_internal_api_setup

    alert_group = alert_groups[0]
    related_incident = make_related_incident("1", alert_group.channel.organization, alert_group.channel_filter)
    related_incident.attached_alert_groups.add(alert_group)

    client = APIClient()
    url = reverse("api-internal:alertgroup-list")
    response = client.get(
        url + "?has_related_incident=true",
        format="json",
        **make_user_auth_headers(user, token),
    )

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1

    response = client.get(
        url + "?has_related_incident=false",
        format="json",
        **make_user_auth_headers(user, token),
    )

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 3


@pytest.mark.django_db
def test_get_title_search(
    settings,
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    settings.FEATURE_ALERT_GROUP_SEARCH_ENABLED = True
    settings.FEATURE_ALERT_GROUP_SEARCH_CUTOFF_DAYS = 30
    organization, user, token = make_organization_and_user_with_plugin_token()

    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    alert_groups = []
    for i in range(3):
        alert_group = make_alert_group(
            alert_receive_channel, channel_filter=channel_filter, web_title_cache=f"testing {i+1}"
        )
        # alert groups starting every months going back
        alert_group.started_at = timezone.now() - timezone.timedelta(days=10 + 30 * i)
        alert_group.save(update_fields=["started_at"])
        make_alert(alert_group=alert_group, raw_request_data=alert_raw_request_data)
        alert_groups.append(alert_group)

    client = APIClient()
    url = reverse("api-internal:alertgroup-list")

    # check that the search returns alert groups started in the last 30 days
    response = client.get(
        url + "?search=testing",
        format="json",
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.json()["results"]) == 1
    assert response.json()["results"][0]["pk"] == alert_groups[0].public_primary_key

    # check that the search returns alert groups started in the last 30 days of specified date range
    response = client.get(
        url
        + "?search=testing&started_at={}_{}".format(
            (timezone.now() - timezone.timedelta(days=500)).strftime(DateRangeFilterMixin.DATE_FORMAT),
            (timezone.now() - timezone.timedelta(days=30)).strftime(DateRangeFilterMixin.DATE_FORMAT),
        ),
        format="json",
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert len(response.json()["results"]) == 1
    assert response.json()["results"][0]["pk"] == alert_groups[1].public_primary_key


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_acknowledge_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()

    url = reverse("api-internal:alertgroup-acknowledge", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.acknowledge",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_unacknowledge_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-unacknowledge", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.unacknowledge",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_resolve_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-resolve", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.resolve",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_unresolve_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-unresolve", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.unresolve",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_silence_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-silence", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.silence",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_unsilence_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-unsilence", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.unsilence",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_attach_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-attach", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.attach",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_unattach_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-unattach", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.unattach",
        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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_list_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-list")

    with patch(
        "apps.api.views.alert_group.AlertGroupView.list",
        return_value=Response(
            status=status.HTTP_200_OK,
        ),
    ):
        response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.status_code == expected_status


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_stats_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-stats")

    with patch(
        "apps.api.views.alert_group.AlertGroupView.stats",
        return_value=Response(
            status=status.HTTP_200_OK,
        ),
    ):
        response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.status_code == expected_status


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_bulk_action_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-bulk-action")

    with patch(
        "apps.api.views.alert_group.AlertGroupView.bulk_action", 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
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_filters_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-filters")

    with patch(
        "apps.api.views.alert_group.AlertGroupView.filters",
        return_value=Response(
            status=status.HTTP_200_OK,
        ),
    ):
        response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.status_code == expected_status


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_detail_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.retrieve",
        return_value=Response(
            status=status.HTTP_200_OK,
        ),
    ):
        response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.status_code == expected_status


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_escalation_snapshot_permissions(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    role,
    expected_status,
):
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    organization = new_alert_group.channel.organization
    user = make_user_for_organization(organization, role)

    client = APIClient()
    url = reverse("api-internal:alertgroup-escalation-snapshot", kwargs={"pk": new_alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.escalation_snapshot",
        return_value=Response(
            status=status.HTTP_200_OK,
        ),
    ):
        response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.status_code == expected_status


@pytest.mark.django_db
def test_silence(alert_group_internal_api_setup, make_user_auth_headers):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups
    url = reverse("api-internal:alertgroup-silence", kwargs={"pk": new_alert_group.public_primary_key})

    silence_delay = timezone.timedelta(seconds=60)
    response = client.post(
        url, data={"delay": silence_delay.seconds}, format="json", **make_user_auth_headers(user, token)
    )

    assert response.status_code == status.HTTP_200_OK

    new_alert_group.refresh_from_db()
    assert new_alert_group.silenced_until is not None

    assert new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_SILENCE, author=user, silence_delay=silence_delay
    ).exists()


@pytest.mark.django_db
def test_unsilence(
    alert_group_internal_api_setup,
    make_user_auth_headers,
):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups

    silence_url = reverse("api-internal:alertgroup-silence", kwargs={"pk": new_alert_group.public_primary_key})
    unsilence_url = reverse("api-internal:alertgroup-unsilence", kwargs={"pk": new_alert_group.public_primary_key})

    # silence alert group
    silence_delay = timezone.timedelta(seconds=10000)
    client.post(
        silence_url, data={"delay": silence_delay.seconds}, format="json", **make_user_auth_headers(user, token)
    )

    # unsnooze alert group
    response = client.post(unsilence_url, **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK

    new_alert_group.refresh_from_db()
    assert new_alert_group.silenced_until is None

    assert new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_SILENCE,
        author=user,
    ).exists()


@pytest.mark.django_db
def test_unpage_user(
    alert_group_internal_api_setup,
    make_user,
    make_user_auth_headers,
):
    client = APIClient()
    user, token, _ = alert_group_internal_api_setup
    other_user = make_user(organization=user.organization)

    alert_group = direct_paging(user.organization, user, "testtesttest", users=[(other_user, False)])
    paged_users = alert_group.get_paged_users()

    assert paged_users[0]["pk"] == other_user.public_primary_key

    url = reverse("api-internal:alertgroup-unpage-user", kwargs={"pk": alert_group.public_primary_key})
    response = client.post(url, data={"user_id": other_user.public_primary_key}, **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK

    alert_group.refresh_from_db()
    assert alert_group.silenced_until is None
    assert alert_group.get_paged_users() == []

    unpage_user_log_record = alert_group.log_records.get(
        type=AlertGroupLogRecord.TYPE_UNPAGE_USER,
        author=user,
    )

    assert unpage_user_log_record.reason == f"{user.username} unpaged user {other_user.username}"
    assert unpage_user_log_record.step_specific_info == {"user": other_user.public_primary_key}


@pytest.mark.django_db
def test_invalid_bulk_action(
    make_user_auth_headers,
    alert_group_internal_api_setup,
):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups

    url = reverse("api-internal:alertgroup-bulk-action")

    response = client.post(
        url,
        data={
            "alert_group_pks": [alert_group.public_primary_key for alert_group in alert_groups],
            "action": "invalid_action",
        },
        format="json",
        **make_user_auth_headers(user, token),
    )

    assert response.status_code == status.HTTP_400_BAD_REQUEST


@patch("apps.alerts.tasks.send_alert_group_signal.send_alert_group_signal.delay", return_value=None)
@patch("apps.alerts.tasks.send_update_log_report_signal.send_update_log_report_signal.apply_async", return_value=None)
@patch("apps.alerts.models.AlertGroup.start_escalation_if_needed", return_value=None)
@pytest.mark.django_db
def test_bulk_action_restart(
    mocked_alert_group_signal_task,
    mocked_log_report_signal_task,
    mocked_start_escalate_alert,
    make_user_auth_headers,
    alert_group_internal_api_setup,
    django_capture_on_commit_callbacks,
):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, acked_alert_group, new_alert_group, silenced_alert_group = alert_groups

    url = reverse("api-internal:alertgroup-bulk-action")

    assert not resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_RESOLVED,
        author=user,
    ).exists()

    assert not acked_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_ACK,
        author=user,
    ).exists()

    assert not silenced_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_SILENCE,
        author=user,
    ).exists()

    with django_capture_on_commit_callbacks(execute=True) as callbacks:
        # restart alert groups
        response = client.post(
            url,
            data={
                "alert_group_pks": [alert_group.public_primary_key for alert_group in alert_groups],
                "action": AlertGroup.RESTART,
            },
            format="json",
            **make_user_auth_headers(user, token),
        )

    assert response.status_code == status.HTTP_200_OK
    assert len(callbacks) == 3

    assert resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_RESOLVED,
        author=user,
    ).exists()

    assert acked_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_ACK,
        author=user,
    ).exists()

    assert silenced_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_SILENCE,
        author=user,
    ).exists()

    assert mocked_alert_group_signal_task.called
    assert mocked_log_report_signal_task.called
    assert mocked_start_escalate_alert.called


@patch("apps.alerts.tasks.send_alert_group_signal.send_alert_group_signal.delay", return_value=None)
@patch("apps.alerts.tasks.send_update_log_report_signal.send_update_log_report_signal.apply_async", return_value=None)
@pytest.mark.django_db
def test_bulk_action_acknowledge(
    mocked_alert_group_signal_task,
    mocked_log_report_signal_task,
    make_user_auth_headers,
    alert_group_internal_api_setup,
    django_capture_on_commit_callbacks,
):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, acked_alert_group, new_alert_group, _ = alert_groups

    url = reverse("api-internal:alertgroup-bulk-action")

    assert not new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_RESOLVED,
        author=user,
    ).exists()

    with django_capture_on_commit_callbacks(execute=True) as callbacks:
        # acknowledge alert groups
        response = client.post(
            url,
            data={
                "alert_group_pks": [alert_group.public_primary_key for alert_group in alert_groups],
                "action": AlertGroup.ACKNOWLEDGE,
            },
            format="json",
            **make_user_auth_headers(user, token),
        )

    assert response.status_code == status.HTTP_200_OK
    assert len(callbacks) == 3

    assert new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_ACK,
        author=user,
    ).exists()

    assert resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_RESOLVED,
        author=user,
    ).exists()

    assert resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_ACK,
        author=user,
    ).exists()

    assert not acked_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_ACK,
        author=user,
    ).exists()

    assert mocked_alert_group_signal_task.called
    assert mocked_log_report_signal_task.called


@patch("apps.alerts.tasks.send_alert_group_signal.send_alert_group_signal.delay", return_value=None)
@patch("apps.alerts.tasks.send_update_log_report_signal.send_update_log_report_signal.apply_async", return_value=None)
@pytest.mark.django_db
def test_bulk_action_resolve(
    mocked_alert_group_signal_task,
    mocked_log_report_signal_task,
    make_user_auth_headers,
    alert_group_internal_api_setup,
    django_capture_on_commit_callbacks,
):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, acked_alert_group, new_alert_group, _ = alert_groups

    url = reverse("api-internal:alertgroup-bulk-action")

    assert not new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_RESOLVED,
        author=user,
    ).exists()

    with django_capture_on_commit_callbacks(execute=True) as callbacks:
        # resolve alert groups
        response = client.post(
            url,
            data={
                "alert_group_pks": [alert_group.public_primary_key for alert_group in alert_groups],
                "action": AlertGroup.RESOLVE,
            },
            format="json",
            **make_user_auth_headers(user, token),
        )

    assert response.status_code == status.HTTP_200_OK
    assert len(callbacks) == 3

    assert new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_RESOLVED,
        author=user,
    ).exists()

    assert acked_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_RESOLVED,
        author=user,
    ).exists()

    assert not resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_RESOLVED,
        author=user,
    ).exists()

    assert mocked_alert_group_signal_task.called
    assert mocked_log_report_signal_task.called


@patch("apps.alerts.tasks.send_alert_group_signal.send_alert_group_signal.delay", return_value=None)
@patch("apps.alerts.tasks.send_update_log_report_signal.send_update_log_report_signal.apply_async", return_value=None)
@patch("apps.alerts.models.AlertGroup.start_unsilence_task", return_value=None)
@pytest.mark.django_db
def test_bulk_action_silence(
    mocked_alert_group_signal_task,
    mocked_log_report_signal_task,
    mocked_start_unsilence_task,
    make_user_auth_headers,
    alert_group_internal_api_setup,
    django_capture_on_commit_callbacks,
):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, acked_alert_group, new_alert_group, silenced_alert_groups = alert_groups

    url = reverse("api-internal:alertgroup-bulk-action")

    assert not new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_SILENCE,
        author=user,
    ).exists()

    with django_capture_on_commit_callbacks(execute=True) as callbacks:
        # silence alert groups
        response = client.post(
            url,
            data={
                "alert_group_pks": [alert_group.public_primary_key for alert_group in alert_groups],
                "action": AlertGroup.SILENCE,
                "delay": 180,
            },
            format="json",
            **make_user_auth_headers(user, token),
        )

    assert response.status_code == status.HTTP_200_OK
    assert len(callbacks) == 4

    assert new_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_SILENCE,
        author=user,
    ).exists()

    new_alert_group.refresh_from_db()
    assert new_alert_group.silenced

    assert acked_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_ACK,
        author=user,
    ).exists()

    assert acked_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_SILENCE,
        author=user,
    ).exists()

    assert resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_RESOLVED,
        author=user,
    ).exists()

    assert resolved_alert_group.log_records.filter(
        type=AlertGroupLogRecord.TYPE_SILENCE,
        author=user,
    ).exists()

    assert silenced_alert_groups.log_records.filter(
        type=AlertGroupLogRecord.TYPE_UN_SILENCE,
        author=user,
    ).exists()

    assert silenced_alert_groups.log_records.filter(
        type=AlertGroupLogRecord.TYPE_SILENCE,
        author=user,
    ).exists()

    assert mocked_alert_group_signal_task.called
    assert mocked_log_report_signal_task.called
    assert mocked_start_unsilence_task.called


@pytest.mark.django_db
def test_alert_group_status_field(
    make_user_auth_headers,
    alert_group_internal_api_setup,
):
    client = APIClient()

    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, acked_alert_group, new_alert_group, silenced_alert_group = alert_groups

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": new_alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.json()["status"] == AlertGroup.NEW

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": acked_alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.json()["status"] == AlertGroup.ACKNOWLEDGED

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": resolved_alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.json()["status"] == AlertGroup.RESOLVED

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": silenced_alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.json()["status"] == AlertGroup.SILENCED


@pytest.mark.django_db
@pytest.mark.parametrize(
    "role,expected_status",
    [
        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
        (LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
    ],
)
def test_alert_group_preview_template_permissions(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_user_auth_headers,
    role,
    expected_status,
    make_alert_group,
    make_alert,
):
    organization, user, token = make_organization_and_user_with_plugin_token(role=role)
    alert_receive_channel = make_alert_receive_channel(organization)
    alert_group = make_alert_group(alert_receive_channel)
    make_alert(alert_group=alert_group, raw_request_data=alert_receive_channel.config.example_payload)

    client = APIClient()
    url = reverse("api-internal:alertgroup-preview-template", kwargs={"pk": alert_group.public_primary_key})

    with patch(
        "apps.api.views.alert_group.AlertGroupView.preview_template",
        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_alert_group_preview_body_non_existent_template_var(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_user_auth_headers,
    make_alert_group,
    make_alert,
):
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    alert_group = make_alert_group(alert_receive_channel)
    make_alert(alert_group=alert_group, raw_request_data=alert_receive_channel.config.example_payload)
    client = APIClient()
    url = reverse("api-internal:alertgroup-preview-template", kwargs={"pk": alert_group.public_primary_key})

    data = {"template_name": "testonly_title_template", "template_body": "foobar: {{ foobar.does_not_exist }}"}
    response = client.post(url, data, format="json", **make_user_auth_headers(user, token))

    # Return errors as preview body instead of None
    assert response.status_code == status.HTTP_200_OK
    assert response.json()["preview"] == "Template Warning: 'foobar' is undefined"


@pytest.mark.django_db
def test_alert_group_preview_body_invalid_template_syntax(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_user_auth_headers,
    make_alert_group,
    make_alert,
):
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    alert_group = make_alert_group(alert_receive_channel)
    make_alert(alert_group=alert_group, raw_request_data=alert_receive_channel.config.example_payload)
    client = APIClient()
    url = reverse("api-internal:alertgroup-preview-template", kwargs={"pk": alert_group.public_primary_key})

    data = {"template_name": "testonly_title_template", "template_body": "{{'' if foo is None else foo}}"}
    response = client.post(url, data, format="json", **make_user_auth_headers(user, token))

    # Errors now returned preview content
    assert response.status_code == status.HTTP_200_OK
    assert response.data["preview"] == "Template Error: No test named 'None' found."


@pytest.mark.django_db
def test_grouped_alerts(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    alert_group = make_alert_group(alert_receive_channel)

    # create 101 alerts and check that only 100 are returned
    for i in range(101):
        make_alert(
            alert_group=alert_group,
            created_at=timezone.datetime.min + timezone.timedelta(minutes=i),
            raw_request_data=alert_receive_channel.config.example_payload,
        )

    client = APIClient()
    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})

    response = client.get(url, **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK
    assert len(response.json()["alerts"]) == 100

    first_alert_created_at = response.json()["alerts"][0]["created_at"]
    last_alert_created_at = response.json()["alerts"][-1]["created_at"]

    first_alert_created_at = timezone.datetime.strptime(first_alert_created_at, "%Y-%m-%dT%H:%M:%S.%fZ")
    last_alert_created_at = timezone.datetime.strptime(last_alert_created_at, "%Y-%m-%dT%H:%M:%S.%fZ")

    assert first_alert_created_at > last_alert_created_at


@pytest.mark.django_db
def test_alert_group_paged_users(
    make_user_for_organization,
    make_user_auth_headers,
    alert_group_internal_api_setup,
):
    client = APIClient()

    user, token, alert_groups = alert_group_internal_api_setup
    _, _, new_alert_group, _ = alert_groups

    user1 = make_user_for_organization(user.organization)
    user2 = make_user_for_organization(user.organization)

    # add paging log records
    new_alert_group.log_records.create(
        type=AlertGroupLogRecord.TYPE_DIRECT_PAGING,
        author=user,
        reason="paged user",
        step_specific_info={"user": user1.public_primary_key},
    )
    new_alert_group.log_records.create(
        type=AlertGroupLogRecord.TYPE_DIRECT_PAGING,
        author=user,
        reason="paged user",
        step_specific_info={"user": user2.public_primary_key},
    )
    new_alert_group.log_records.create(
        type=AlertGroupLogRecord.TYPE_UNPAGE_USER,
        author=user,
        reason="unpaged user",
        step_specific_info={"user": user1.public_primary_key},
    )

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": new_alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.json()["paged_users"] == [
        {
            "avatar": user2.avatar_url,
            "avatar_full": user2.avatar_full_url(user.organization),
            "id": user2.pk,
            "pk": user2.public_primary_key,
            "important": None,
            "name": user2.name,
            "username": user2.username,
            "teams": [],
        }
    ]


@pytest.mark.django_db
def test_alert_group_resolve_resolution_note(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)
    new_alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
    make_alert(alert_group=new_alert_group, raw_request_data=alert_raw_request_data)

    organization.is_resolution_note_required = True
    organization.save()

    client = APIClient()
    url = reverse("api-internal:alertgroup-resolve", kwargs={"pk": new_alert_group.public_primary_key})

    response = client.post(url, format="json", **make_user_auth_headers(user, token))
    # check that resolution note is required
    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert response.json()["code"] == AlertGroupAPIError.RESOLUTION_NOTE_REQUIRED.value

    with patch(
        "apps.alerts.tasks.send_update_resolution_note_signal.send_update_resolution_note_signal.apply_async"
    ) as mock_signal:
        url = reverse("api-internal:alertgroup-resolve", kwargs={"pk": new_alert_group.public_primary_key})
        response = client.post(
            url, format="json", data={"resolution_note": "hi"}, **make_user_auth_headers(user, token)
        )
        assert response.status_code == status.HTTP_200_OK

        assert new_alert_group.has_resolution_notes
        assert mock_signal.called


@pytest.mark.django_db
def test_alert_group_resolve_resolution_note_mobile_app(
    make_organization_and_user,
    make_mobile_app_auth_token_for_user,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    organization, user = make_organization_and_user()
    organization.is_resolution_note_required = True
    organization.save()
    _, token = make_mobile_app_auth_token_for_user(user, organization)

    alert_receive_channel = make_alert_receive_channel(organization)
    alert_group = make_alert_group(alert_receive_channel)

    client = APIClient()
    url = reverse("api-internal:alertgroup-resolve", kwargs={"pk": alert_group.public_primary_key})
    response = client.post(url, format="json", data={"resolution_note": "hi"}, HTTP_AUTHORIZATION=token)

    assert response.status_code == status.HTTP_200_OK
    assert alert_group.resolution_notes.get().source == ResolutionNote.Source.MOBILE_APP


@pytest.mark.parametrize("source", ResolutionNote.Source)
@pytest.mark.django_db
def test_timeline_resolution_note_source(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_resolution_note_slack_message,
    make_resolution_note,
    make_user_auth_headers,
    source,
):
    """The 'type' field in timeline items should hold the source of the resolution note"""
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)
    alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
    make_alert(alert_group=alert_group, raw_request_data=alert_raw_request_data)

    # Create resolution note
    resolution_note_slack_message = make_resolution_note_slack_message(
        alert_group=alert_group, user=user, added_by_user=user, text="resolution note"
    )
    make_resolution_note(
        alert_group=alert_group, author=user, resolution_note_slack_message=resolution_note_slack_message, source=source
    )

    client = APIClient()
    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})
    response = client.get(url, **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["render_after_resolve_report_json"][0]["type"] == source.value


@pytest.mark.django_db
def test_timeline_api_action(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    """Check that the timeline API returns the correct actions when using AlertSource.WEB vs ActionSource.API"""
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)
    alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
    make_alert(alert_group=alert_group, raw_request_data=alert_raw_request_data)

    alert_group.acknowledge_by_user_or_backsync(user, action_source=ActionSource.WEB)
    alert_group.resolve_by_user_or_backsync(user, action_source=ActionSource.API)

    client = APIClient()
    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})
    response = client.get(url, **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["render_after_resolve_report_json"][0]["action"] == "acknowledged by {{author}}"
    assert response.json()["render_after_resolve_report_json"][1]["action"] == "resolved by API"


@pytest.mark.django_db
def test_wipe_clears_cache(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    """Check that internal API cache is cleared when wiping an alert group"""
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)
    alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
    alert = make_alert(alert_group=alert_group, raw_request_data=alert_raw_request_data)

    # Populate cache
    client = APIClient()
    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})
    response = client.get(url, **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK

    # Wipe alert group
    wipe(alert_group.pk, user.pk)

    # Check that cache is cleared for alert group
    alert_group_cache_keys = [
        AlertGroupFieldsCacheSerializerMixin.calculate_cache_key(field_name, alert_group)
        for field_name in AlertGroupFieldsCacheSerializerMixin.ALL_FIELD_NAMES
    ]
    assert not any([cache.get(key) for key in alert_group_cache_keys])

    # Check that cache is cleared for alert
    alert_cache_keys = [
        AlertFieldsCacheSerializerMixin.calculate_cache_key(field_name, alert)
        for field_name in AlertFieldsCacheSerializerMixin.ALL_FIELD_NAMES
    ]
    assert not any([cache.get(key) for key in alert_cache_keys])


@patch("apps.api.views.alert_group.delete_alert_group.apply_async")
@pytest.mark.django_db
def test_delete(mock_delete_alert_group, make_user_auth_headers, alert_group_internal_api_setup):
    client = APIClient()
    user, token, alert_groups = alert_group_internal_api_setup
    resolved_alert_group, acked_alert_group, new_alert_group, _ = alert_groups

    auth_headers = make_user_auth_headers(user, token)

    for alert_group in [resolved_alert_group, acked_alert_group, new_alert_group]:
        mock_delete_alert_group.reset_mock()

        url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})
        response = client.delete(url, **auth_headers)

        assert response.status_code == status.HTTP_204_NO_CONTENT
        mock_delete_alert_group.assert_called_once_with((alert_group.pk, user.pk))

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": "potato"})
    response = client.delete(url, **auth_headers)
    assert response.status_code == status.HTTP_404_NOT_FOUND


@pytest.mark.django_db
def test_alert_group_list_labels(
    alert_group_internal_api_setup,
    make_alert_group_label_association,
    make_alert_receive_channel,
    make_alert_group,
    make_user_auth_headers,
):
    user, token, alert_groups = alert_group_internal_api_setup
    make_alert_group_label_association(user.organization, alert_groups[0], key_name="a", value_name="b")

    client = APIClient()
    url = reverse("api-internal:alertgroup-list")
    response = client.get(url, **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["results"][-1]["labels"] == [
        {"key": {"id": "a", "name": "a"}, "value": {"id": "b", "name": "b"}}
    ]


@pytest.mark.django_db
def test_alert_group_external_urls(
    make_organization_and_user_with_plugin_token,
    make_user_auth_headers,
    make_alert_receive_channel,
    make_alert_receive_channel_connection,
    make_alert_group,
):
    client = APIClient()

    integration_config = AlertReceiveChannel._config[0]
    integration_config.get_url = Mock(return_value="https://some-url")

    organization, user, token = make_organization_and_user_with_plugin_token()
    source_alert_receive_channel = make_alert_receive_channel(organization, integration=integration_config.slug)
    alert_receive_channel = make_alert_receive_channel(organization)
    make_alert_receive_channel_connection(source_alert_receive_channel, alert_receive_channel)

    alert_group = make_alert_group(alert_receive_channel)

    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    assert response.json()["external_urls"] == []

    # create external IDs
    external_id = AlertGroupExternalID.objects.create(
        source_alert_receive_channel=source_alert_receive_channel, alert_group=alert_group, value="test123"
    )
    url = reverse("api-internal:alertgroup-detail", kwargs={"pk": alert_group.public_primary_key})
    response = client.get(url, format="json", **make_user_auth_headers(user, token))
    expected = {
        "integration": source_alert_receive_channel.public_primary_key,
        "integration_type": source_alert_receive_channel.integration,
        "external_id": external_id.value,
        "url": "https://some-url",
    }
    assert response.json()["external_urls"] == [expected]


@pytest.mark.django_db
def test_filter_default_started_at(
    make_organization_and_user_with_plugin_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_alert_group,
    make_alert,
    make_user_auth_headers,
):
    organization, user, token = make_organization_and_user_with_plugin_token()
    alert_receive_channel = make_alert_receive_channel(organization)
    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)

    old_alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
    old_alert_group.started_at = timezone.now() - timezone.timedelta(days=31)
    old_alert_group.save(update_fields=["started_at"])
    make_alert(alert_group=old_alert_group, raw_request_data=alert_raw_request_data)

    new_alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
    make_alert(alert_group=new_alert_group, raw_request_data=alert_raw_request_data)

    client = APIClient()

    # by default, no alert groups older than 30 days should be listed
    response = client.get(reverse("api-internal:alertgroup-list"), **make_user_auth_headers(user, token))
    assert response.status_code == status.HTTP_200_OK
    assert len(response.json()["results"]) == 1
    assert response.json()["results"][0]["pk"] == new_alert_group.public_primary_key

    # but it should still be possible to retrieve an old alert group by its ID
    response = client.get(
        reverse("api-internal:alertgroup-detail", kwargs={"pk": old_alert_group.public_primary_key}),
        **make_user_auth_headers(user, token),
    )
    assert response.status_code == status.HTTP_200_OK
    assert response.json()["pk"] == old_alert_group.public_primary_key


@pytest.mark.django_db
def test_alert_group_affected_services(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    make_alert_group_label_association,
    settings,
):
    settings.FEATURE_SERVICE_DEPENDENCIES_ENABLED = True
    _, token, alert_groups = alert_group_internal_api_setup
    resolved_ag, ack_ag, new_ag, silenced_ag = alert_groups
    organization = new_ag.channel.organization
    user = make_user_for_organization(organization)

    # set firing alert group service label
    make_alert_group_label_association(organization, new_ag, key_name="service_name", value_name="service-a")
    # set other service name labels for other alert groups
    make_alert_group_label_association(organization, ack_ag, key_name="service_name", value_name="service-2")
    make_alert_group_label_association(organization, resolved_ag, key_name="service_name", value_name="service-3")
    make_alert_group_label_association(organization, silenced_ag, key_name="service_name", value_name="service-4")

    client = APIClient()
    url = reverse("api-internal:alertgroup-filter-affected-services")

    url = f"{url}?service=service-1&service=service-2&service=service-3&service=service-a"
    response = client.get(url, format="json", **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_200_OK
    expected = [
        {
            "name": "service-2",
            "service_url": "a/grafana-slo-app/service/service-2",
            "alert_groups_url": "a/grafana-oncall-app/alert-groups?status=0&status=1&started_at=now-7d_now&label=service_name:service-2",
        },
        {
            "name": "service-a",
            "service_url": "a/grafana-slo-app/service/service-a",
            "alert_groups_url": "a/grafana-oncall-app/alert-groups?status=0&status=1&started_at=now-7d_now&label=service_name:service-a",
        },
    ]
    assert response.json() == expected


@pytest.mark.django_db
def test_alert_group_service_dependencies_feature_not_enabled(
    alert_group_internal_api_setup,
    make_user_for_organization,
    make_user_auth_headers,
    make_alert_group_label_association,
    settings,
):
    settings.FEATURE_SERVICE_DEPENDENCIES_ENABLED = False
    _, token, alert_groups = alert_group_internal_api_setup
    _, _, new_ag, _ = alert_groups
    organization = new_ag.channel.organization
    user = make_user_for_organization(organization)

    # set firing alert group service label
    make_alert_group_label_association(organization, new_ag, key_name="service_name", value_name="service-a")

    client = APIClient()
    url = reverse("api-internal:alertgroup-filter-affected-services")

    url = f"{url}?service=service-1"
    response = client.get(url, format="json", **make_user_auth_headers(user, token))

    assert response.status_code == status.HTTP_404_NOT_FOUND