oncall-engine/engine/apps/public_api/tests/test_integrations.py

import httpretty
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient

from apps.alerts.models import AlertReceiveChannel
from apps.api import permissions
from apps.auth_token.tests.helpers import setup_service_account_api_mocks
from apps.base.tests.messaging_backend import TestOnlyBackend

TEST_MESSAGING_BACKEND_FIELD = TestOnlyBackend.backend_id.lower()


@pytest.mark.django_db
def test_get_list_integrations(
    make_organization_and_user_with_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_integration_heartbeat,
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana", description_short="Some description")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    expected_response = {
        "count": 1,
        "next": None,
        "previous": None,
        "results": [
            {
                "id": integration.public_primary_key,
                "team_id": None,
                "name": "grafana",
                "description_short": "Some description",
                "link": integration.integration_url,
                "inbound_email": None,
                "type": "grafana",
                "default_route": {
                    "escalation_chain_id": None,
                    "id": default_channel_filter.public_primary_key,
                    "slack": {"channel_id": None, "enabled": True},
                    "telegram": {"id": None, "enabled": False},
                    TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
                },
                "heartbeat": {
                    "link": f"{integration.integration_url}heartbeat/",
                },
                "templates": {
                    "grouping_key": None,
                    "resolve_signal": None,
                    "acknowledge_signal": None,
                    "source_link": None,
                    "slack": {"title": None, "message": None, "image_url": None},
                    "web": {"title": None, "message": None, "image_url": None},
                    "sms": {
                        "title": None,
                    },
                    "phone_call": {
                        "title": None,
                    },
                    "telegram": {
                        "title": None,
                        "message": None,
                        "image_url": None,
                    },
                    TEST_MESSAGING_BACKEND_FIELD: {
                        "title": None,
                        "message": None,
                        "image_url": None,
                    },
                },
                "maintenance_mode": None,
                "maintenance_started_at": None,
                "maintenance_end_at": None,
            }
        ],
        "current_page_number": 1,
        "page_size": 50,
        "total_pages": 1,
    }
    url = reverse("api-public:integrations-list")
    response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")

    assert response.status_code == status.HTTP_200_OK
    assert response.json() == expected_response


@pytest.mark.django_db
def test_create_integration(
    make_organization_and_user_with_token,
    make_escalation_chain,
):
    organization, _, token = make_organization_and_user_with_token()
    make_escalation_chain(organization)

    client = APIClient()
    data_for_create = {
        "type": "grafana",
        "name": "grafana_created",
        "team_id": None,
    }
    url = reverse("api-public:integrations-list")
    response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_201_CREATED


@pytest.mark.django_db
@httpretty.activate(verbose=True, allow_net_connect=False)
def test_create_integration_via_service_account(
    make_organization,
    make_service_account_for_organization,
    make_token_for_service_account,
    make_escalation_chain,
):
    organization = make_organization(grafana_url="http://grafana.test")
    service_account = make_service_account_for_organization(organization)
    token_string = "glsa_token"
    make_token_for_service_account(service_account, token_string)
    make_escalation_chain(organization)

    perms = {
        permissions.RBACPermission.Permissions.INTEGRATIONS_WRITE.value: ["*"],
    }
    setup_service_account_api_mocks(organization.grafana_url, perms)

    client = APIClient()
    data_for_create = {
        "type": "grafana",
        "name": "grafana_created",
        "team_id": None,
    }
    url = reverse("api-public:integrations-list")
    response = client.post(
        url,
        data=data_for_create,
        format="json",
        HTTP_AUTHORIZATION=f"{token_string}",
        HTTP_X_GRAFANA_URL=organization.grafana_url,
    )
    assert response.status_code == status.HTTP_201_CREATED
    integration = AlertReceiveChannel.objects.get(public_primary_key=response.data["id"])
    assert integration.service_account == service_account


@pytest.mark.django_db
def test_integration_name_uniqueness(
    make_organization_and_user_with_token,
    make_team,
):
    organization, _, token = make_organization_and_user_with_token()

    client = APIClient()
    data = {
        "type": "grafana",
        "name": "grafana_created",
        "team_id": None,
    }
    url = reverse("api-public:integrations-list")
    response = client.post(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_201_CREATED
    integration_pk = response.data["id"]

    # cannot create another one with the same name
    response = client.post(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST

    # but name can be reused in a different team
    another_team = make_team(organization)
    data["team_id"] = another_team.public_primary_key
    response = client.post(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_201_CREATED

    # update works
    url = reverse("api-public:integrations-detail", args=[integration_pk])
    data["team_id"] = None
    data["description"] = "some description"
    response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK

    # but updating team will fail if name exists
    data["team_id"] = another_team.public_primary_key
    response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_integration_name_duplicated(
    make_organization_and_user_with_token,
    make_alert_receive_channel,
    make_user_auth_headers,
):
    # this could happen in case a team is removed and integrations are set to have "no team"
    organization, _, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization)
    # duplicated name integration
    make_alert_receive_channel(organization, verbal_name=integration.verbal_name)

    client = APIClient()

    # updating team will require changing the name or the team
    data = {"name": integration.verbal_name}
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST

    # but updating team will fail if name exists
    data["name"] = "a new name"
    response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK


@pytest.mark.django_db
def test_integration_uniqueness_validation_optional_team_field(
    make_organization_and_user_with_token,
    make_alert_receive_channel,
    make_team,
):
    organization, _, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization)
    team = make_team(organization)
    integration_with_team = make_alert_receive_channel(organization, verbal_name=integration.verbal_name, team=team)

    # updating works if team is not present
    client = APIClient()
    data = {
        "type": "grafana",
        "description_short": "updated_description",
    }

    url = reverse("api-public:integrations-detail", args=[integration_with_team.public_primary_key])
    response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK

    # updating works if team is present
    data = {"type": "grafana", "description_short": "updated_description", "team_id": team.public_primary_key}
    response = client.put(url, data=data, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK


@pytest.mark.django_db
def test_create_integrations_with_none_templates(
    make_organization_and_user_with_token,
    make_escalation_chain,
):
    organization, _, token = make_organization_and_user_with_token()
    make_escalation_chain(organization)

    client = APIClient()
    data_for_create = {
        "type": "grafana",
        "team_id": None,
        "name": "grafana_created",
        "templates": {
            "grouping_key": None,
            "resolve_signal": None,
            "acknowledge_signal": None,
            "slack": None,
            "web": None,
            "sms": None,
            "phone_call": None,
            "telegram": None,
        },
    }

    url = reverse("api-public:integrations-list")

    response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_201_CREATED


@pytest.mark.django_db
def test_create_integration_with_invalid_type(
    make_organization_and_user_with_token,
):
    _, _, token = make_organization_and_user_with_token()

    client = APIClient()
    data_for_create = {
        "type": "this_is_invalid_integration_type",
        "name": "grafana_created",
        "team_id": None,
    }
    url = reverse("api-public:integrations-list")
    response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_integration_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"grouping_key": "ip_addr", "slack": {"title": "Incident"}}}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": "ip_addr",
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": "Incident", "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_update_integration_template_messaging_backend(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"grouping_key": "ip_addr", TEST_MESSAGING_BACKEND_FIELD: {"title": "Incident"}}}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": "ip_addr",
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": "Incident",
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_update_invalid_integration_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"slack": {"title": "{%  invalid jinja template }}"}}}
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_resolve_signal_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"resolve_signal": "resig"}}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": None,
            "resolve_signal": "resig",
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_update_invalid_resolve_signal_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"resolve_signal": "{%  invalid jinja template }}"}}
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_empty_grouping_key_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"grouping_key": {}}}
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_invalid_flat_web_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"web": "invalid_web_template"}}
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_sms_template_with_empty_dict(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"sms": {}}}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": None,
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_update_integration_name(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"name": "grafana_updated"}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana_updated",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": None,
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_update_integration_name_and_description_short(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana", description_short="Some description")
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"name": "grafana_updated"}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana_updated",
        "description_short": "Some description",
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": None,
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_set_default_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, verbal_name="grafana")
    integration.slack_title_template = "updated_template"
    integration.grouping_id_template = "updated_template"
    integration.save()
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"grouping_key": None, "slack": {"title": None}}}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": None,
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": None,
                "image_url": None,
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_set_default_messaging_backend_template(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(
        organization,
        verbal_name="grafana",
        messaging_backends_templates={
            "TESTONLY": {"title": "the-title", "message": "the-message", "image_url": "the-image-url"}
        },
    )
    default_channel_filter = make_channel_filter(integration, is_default=True)
    make_integration_heartbeat(integration)

    client = APIClient()
    data_for_update = {"templates": {"testonly": {"title": None}}}
    expected_response = {
        "id": integration.public_primary_key,
        "team_id": None,
        "name": "grafana",
        "description_short": None,
        "link": integration.integration_url,
        "inbound_email": None,
        "type": "grafana",
        "default_route": {
            "escalation_chain_id": None,
            "id": default_channel_filter.public_primary_key,
            "slack": {"channel_id": None, "enabled": True},
            "telegram": {"id": None, "enabled": False},
            TEST_MESSAGING_BACKEND_FIELD: {"id": None, "enabled": False},
        },
        "heartbeat": {
            "link": f"{integration.integration_url}heartbeat/",
        },
        "templates": {
            "grouping_key": None,
            "resolve_signal": None,
            "acknowledge_signal": None,
            "source_link": None,
            "slack": {"title": None, "message": None, "image_url": None},
            "web": {"title": None, "message": None, "image_url": None},
            "sms": {
                "title": None,
            },
            "phone_call": {
                "title": None,
            },
            "telegram": {
                "title": None,
                "message": None,
                "image_url": None,
            },
            TEST_MESSAGING_BACKEND_FIELD: {
                "title": None,
                "message": "the-message",
                "image_url": "the-image-url",
            },
        },
        "maintenance_mode": None,
        "maintenance_started_at": None,
        "maintenance_end_at": None,
    }
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data == expected_response


@pytest.mark.django_db
def test_get_list_integrations_link_and_inbound_email(
    make_organization_and_user_with_token,
    make_alert_receive_channel,
    make_channel_filter,
    make_integration_heartbeat,
    settings,
):
    """
    Check that "link" and "inbound_email" fields are populated correctly for different integration types.
    """

    settings.BASE_URL = "https://test.com"
    settings.INBOUND_EMAIL_DOMAIN = "test.com"

    organization, user, token = make_organization_and_user_with_token()

    for integration in AlertReceiveChannel._config:
        make_alert_receive_channel(organization, integration=integration.slug, token="test123")

    client = APIClient()
    url = reverse("api-public:integrations-list")

    response = client.get(url, HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK

    for integration in response.json()["results"]:
        integration_type, integration_link, integration_inbound_email = (
            integration["type"],
            integration["link"],
            integration["inbound_email"],
        )

        if integration_type in [
            AlertReceiveChannel.INTEGRATION_MANUAL,
            AlertReceiveChannel.INTEGRATION_MAINTENANCE,
        ]:
            assert integration_link is None
            assert integration_inbound_email is None
        elif integration_type == AlertReceiveChannel.INTEGRATION_INBOUND_EMAIL:
            assert integration_link is None
            assert integration_inbound_email == "test123@test.com"
        else:
            assert integration_link == f"https://test.com/integrations/v1/{integration_type}/test123/"
            assert integration_inbound_email is None


@pytest.mark.django_db
def test_create_integration_default_route(
    make_organization_and_user_with_token,
    make_escalation_chain,
):
    organization, _, token = make_organization_and_user_with_token()
    escalation_chain = make_escalation_chain(organization)

    client = APIClient()
    data_for_create = {
        "type": "grafana",
        "name": "grafana_created",
        "team_id": None,
        "default_route": {"escalation_chain_id": escalation_chain.public_primary_key},
    }
    url = reverse("api-public:integrations-list")
    response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_201_CREATED
    assert response.data["default_route"]["escalation_chain_id"] == escalation_chain.public_primary_key


@pytest.mark.django_db
def test_update_integration_default_route(
    make_organization_and_user_with_token, make_escalation_chain, make_alert_receive_channel, make_channel_filter
):
    organization, _, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization)
    make_channel_filter(integration, is_default=True)
    escalation_chain = make_escalation_chain(organization)

    client = APIClient()
    data_for_update = {
        "default_route": {"escalation_chain_id": escalation_chain.public_primary_key},
    }

    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")

    assert response.status_code == status.HTTP_200_OK
    assert response.data["default_route"]["escalation_chain_id"] == escalation_chain.public_primary_key


@pytest.mark.django_db
def test_create_integration_default_route_with_slack_field(
    make_organization_and_user_with_token,
    make_escalation_chain,
):
    organization, _, token = make_organization_and_user_with_token()
    escalation_chain = make_escalation_chain(organization)

    client = APIClient()
    data_for_create = {
        "type": "grafana",
        "name": "grafana_created",
        "team_id": None,
        "default_route": {
            "escalation_chain_id": escalation_chain.public_primary_key,
            "slack": {"channel_id": "TEST_SLACK_ID"},
        },
    }
    url = reverse("api-public:integrations-list")
    response = client.post(url, data=data_for_create, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert response.data["detail"] == "Slack isn't connected to this workspace"


@pytest.mark.django_db
def test_update_integration_default_route_with_slack_field(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter
):
    organization, _, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization)
    make_channel_filter(integration, is_default=True)

    client = APIClient()
    data_for_update = {
        "default_route": {"slack": {"channel_id": "TEST_SLACK_ID"}},
    }

    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")

    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert response.data["detail"] == "Slack isn't connected to this workspace"


@pytest.mark.django_db
def test_cant_create_integrations_direct_paging(
    make_organization_and_user_with_token, make_team, make_alert_receive_channel, make_user_auth_headers
):
    organization, _, token = make_organization_and_user_with_token()

    client = APIClient()
    url = reverse("api-public:integrations-list")
    response = client.post(url, data={"type": "direct_paging"}, format="json", HTTP_AUTHORIZATION=token)
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_integrations_direct_paging(
    make_organization_and_user_with_token, make_team, make_alert_receive_channel, make_user_auth_headers
):
    organization, _, token = make_organization_and_user_with_token()
    team = make_team(organization)

    integration = make_alert_receive_channel(
        organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, team=None
    )
    make_alert_receive_channel(organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, team=team)

    client = APIClient()
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])

    # Move direct paging integration from "No team" to team
    response = client.put(url, data={"team_id": team.public_primary_key}, format="json", HTTP_AUTHORIZATION=token)

    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert response.data["detail"] == AlertReceiveChannel.DuplicateDirectPagingError.DETAIL


@pytest.mark.django_db
def test_cant_delete_direct_paging_integration(make_organization_and_user_with_token, make_alert_receive_channel):
    organization, user, token = make_organization_and_user_with_token()
    integration = make_alert_receive_channel(organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING)

    client = APIClient()
    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
    response = client.delete(url, HTTP_AUTHORIZATION=token)
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_get_integration_type_legacy(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    am = make_alert_receive_channel(
        organization, verbal_name="AMV2", integration=AlertReceiveChannel.INTEGRATION_ALERTMANAGER
    )
    legacy_am = make_alert_receive_channel(
        organization, verbal_name="AMV2", integration=AlertReceiveChannel.INTEGRATION_LEGACY_ALERTMANAGER
    )

    client = APIClient()
    url = reverse("api-public:integrations-detail", args=[am.public_primary_key])
    response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data["type"] == "alertmanager"

    url = reverse("api-public:integrations-detail", args=[legacy_am.public_primary_key])
    response = client.get(url, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data["type"] == "alertmanager"


@pytest.mark.django_db
def test_create_integration_type_legacy(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()

    client = APIClient()
    url = reverse("api-public:integrations-list")
    response = client.post(url, data={"type": "alertmanager"}, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_201_CREATED
    assert response.data["type"] == "alertmanager"

    response = client.post(url, data={"type": "legacy_alertmanager"}, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_400_BAD_REQUEST


@pytest.mark.django_db
def test_update_integration_type_legacy(
    make_organization_and_user_with_token, make_alert_receive_channel, make_channel_filter, make_integration_heartbeat
):
    organization, user, token = make_organization_and_user_with_token()
    am = make_alert_receive_channel(
        organization, verbal_name="AMV2", integration=AlertReceiveChannel.INTEGRATION_ALERTMANAGER
    )
    legacy_am = make_alert_receive_channel(
        organization, verbal_name="AMV2", integration=AlertReceiveChannel.INTEGRATION_LEGACY_ALERTMANAGER
    )

    data_for_update = {"type": "alertmanager", "description_short": "Updated description"}

    client = APIClient()
    url = reverse("api-public:integrations-detail", args=[am.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data["type"] == "alertmanager"
    assert response.data["description_short"] == "Updated description"

    url = reverse("api-public:integrations-detail", args=[legacy_am.public_primary_key])
    response = client.put(url, data=data_for_update, format="json", HTTP_AUTHORIZATION=f"{token}")
    assert response.status_code == status.HTTP_200_OK
    assert response.data["description_short"] == "Updated description"
    assert response.data["type"] == "alertmanager"