centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oncall-engine/engine/apps/api/tests/test_webhooks.py
Innokentii Konstantinov acd0c44c33
Support prescribed labels (#3848) 
# What this PR does

**Cleanup label typing:**
1. LabelParam -> two separate types LabekKey and LabelValue 
2. LabelData -> renamed to LabelPair. 
3. LabelKeyData -> renamed to LabelOption
Data is not giving any info about what this type represents. 
4. Remove LabelsData and LabelsKeysData types. They are just list of
types listed above and with new naming it feels obsolete.
5. ValueData removed. LabelPair is used instead.
6. Rework AlertGroupCustomLabel to use LabelKey type for key to make
type system more consistent. Name model type AlertGroupCustomLabel**DB**
and api type AlertGroupCustomLabel**API** to clearly distinguish them.

**Split update_labels_cache into two tasks** update_label_option_cache
and update_label_pairs_cache.
Original task was expecting array of LabelsData (now it's LabelPair) OR
one LabelKeyData ( now it's LabelOption). I believe having one function
with two sp different argument types makes it more complicated for
understanding.


**Make OnCall backend support prescribed labels**. OnCall will sync and
store "prescribed" field for key and values, so Label dropdown able to
disable editing for certain labels.

## Which issue(s) this PR fixes

## Checklist

- [x] Unit, integration, and e2e (if applicable) tests updated
- [ ] Documentation added (or `pr:no public docs` PR label added if not
required)
- [x] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not
required)

---------

Co-authored-by: Maxim Mordasov <maxim.mordasov@grafana.com>
Co-authored-by: Yulya Artyukhina <Ferril.darkdiver@gmail.com>
2024-02-20 14:42:51 +08:00

948 lines
32 KiB
Python
Raw Blame History

import json
from unittest.mock import patch
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.response import Response
from rest_framework.test import APIClient
from apps.api.permissions import LegacyAccessControlRole
from apps.api.views.webhooks import RECENT_RESPONSE_LIMIT, WEBHOOK_URL
from apps.webhooks.models import Webhook
from apps.webhooks.models.webhook import WEBHOOK_FIELD_PLACEHOLDER
TEST_URL = "https://some-url"
@pytest.fixture()
def webhook_internal_api_setup(make_organization_and_user_with_plugin_token, make_custom_webhook):
organization, user, token = make_organization_and_user_with_plugin_token()
webhook = make_custom_webhook(
name="some_webhook",
url="https://github.com/",
username="Chris Vanstras",
password="qwerty",
data='{"name": "{{ alert_payload }}"}',
authorization_header="auth_token",
organization=organization,
forward_all=False,
)
return user, token, webhook
@pytest.mark.django_db
def test_get_list_webhooks(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
expected_payload = [
{
"id": webhook.public_primary_key,
"name": "some_webhook",
"team": None,
"url": "https://github.com/",
"data": '{"name": "{{ alert_payload }}"}',
"username": "Chris Vanstras",
"password": WEBHOOK_FIELD_PLACEHOLDER,
"authorization_header": WEBHOOK_FIELD_PLACEHOLDER,
"forward_all": False,
"headers": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"labels": [],
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": "0",
"trigger_type_name": "Escalation step",
"preset": None,
}
]
response = client.get(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
assert response.json() == expected_payload
@pytest.mark.django_db
def test_get_detail_webhook(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
expected_payload = {
"id": webhook.public_primary_key,
"name": "some_webhook",
"team": None,
"url": "https://github.com/",
"data": '{"name": "{{ alert_payload }}"}',
"username": "Chris Vanstras",
"password": WEBHOOK_FIELD_PLACEHOLDER,
"authorization_header": WEBHOOK_FIELD_PLACEHOLDER,
"forward_all": False,
"headers": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"labels": [],
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": "0",
"trigger_type_name": "Escalation step",
"preset": None,
}
response = client.get(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
assert response.json() == expected_payload
@pytest.mark.django_db
def test_create_webhook(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
data = {
"name": "the_webhook",
"url": TEST_URL,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"team": None,
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
webhook = Webhook.objects.get(public_primary_key=response.json()["id"])
expected_response = data | {
"id": webhook.public_primary_key,
"data": None,
"username": None,
"password": None,
"authorization_header": None,
"forward_all": True,
"headers": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"labels": [],
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": str(data["trigger_type"]),
"trigger_type_name": "Alert Group Created",
"preset": None,
}
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == expected_response
# user creating the webhook is set
assert webhook.user == user
@pytest.mark.django_db
@pytest.mark.parametrize(
"field_name,value",
[
("data", '{"name": "{{ alert_payload }}"}'),
("headers", '"request-id": "{{ alert_payload.id }}"'),
("trigger_template", "integration_id == {{ some_var_value }}"),
("url", "https://myserver/{{ alert_payload.id }}/triggered"),
],
)
def test_create_valid_templated_field(webhook_internal_api_setup, make_user_auth_headers, field_name, value):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
data = {
"name": "webhook_with_valid_data",
"url": TEST_URL,
field_name: value,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"team": None,
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
# modify initial data by adding id and None for optional fields
webhook = Webhook.objects.get(public_primary_key=response.data["id"])
expected_response = data | {
"id": webhook.public_primary_key,
"username": None,
"password": None,
"authorization_header": None,
"forward_all": True,
"headers": None,
"data": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"labels": [],
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": str(data["trigger_type"]),
"trigger_type_name": "Alert Group Created",
"preset": None,
}
# update expected value for changed field
expected_response[field_name] = value
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == expected_response
@pytest.mark.django_db
@pytest.mark.parametrize(
"field_name,value",
[
("data", "{{%"),
("headers", '"request-id": "{{}}"'),
("trigger_template", "integration_id == {{}}"),
("url", "invalid-url/{{}}/triggered"),
],
)
def test_create_invalid_templated_field(webhook_internal_api_setup, make_user_auth_headers, field_name, value):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
data = {
"name": "webhook_with_valid_data",
"url": TEST_URL,
field_name: value,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"team": None,
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_update_webhook(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
data = {
"name": "github_button_updated",
"url": "https://github.com/",
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"team": None,
}
response = client.put(
url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token)
)
updated_instance = Webhook.objects.get(public_primary_key=webhook.public_primary_key)
assert response.status_code == status.HTTP_200_OK
assert updated_instance.name == "github_button_updated"
@pytest.mark.django_db
def test_delete_webhook(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
response = client.delete(url, **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.django_db
@pytest.mark.parametrize(
"role,expected_status",
[
(LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
(LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN),
(LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
(LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
],
)
def test_webhook_create_permissions(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
role,
expected_status,
):
_, user, token = make_organization_and_user_with_plugin_token(role)
client = APIClient()
url = reverse("api-internal:webhooks-list")
with patch(
"apps.api.views.webhooks.WebhooksView.create",
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_403_FORBIDDEN),
(LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
(LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
],
)
def test_webhook_update_permissions(
make_organization_and_user_with_plugin_token,
make_custom_webhook,
make_user_auth_headers,
role,
expected_status,
):
organization, user, token = make_organization_and_user_with_plugin_token(role)
webhook = make_custom_webhook(organization=organization)
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
with patch(
"apps.api.views.webhooks.WebhooksView.update",
return_value=Response(
status=status.HTTP_200_OK,
),
):
response = client.put(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == expected_status
response = client.patch(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_webhook_list_permissions(
make_organization_and_user_with_plugin_token,
make_custom_webhook,
make_user_auth_headers,
role,
expected_status,
):
organization, user, token = make_organization_and_user_with_plugin_token(role)
make_custom_webhook(organization=organization)
client = APIClient()
url = reverse("api-internal:webhooks-list")
with patch(
"apps.api.views.webhooks.WebhooksView.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_webhook_retrieve_permissions(
make_organization_and_user_with_plugin_token,
make_custom_webhook,
make_user_auth_headers,
role,
expected_status,
):
organization, user, token = make_organization_and_user_with_plugin_token(role)
webhook = make_custom_webhook(organization=organization)
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
with patch(
"apps.api.views.webhooks.WebhooksView.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_204_NO_CONTENT),
(LegacyAccessControlRole.EDITOR, status.HTTP_403_FORBIDDEN),
(LegacyAccessControlRole.VIEWER, status.HTTP_403_FORBIDDEN),
(LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
],
)
def test_webhook_delete_permissions(
make_organization_and_user_with_plugin_token,
make_custom_webhook,
make_user_auth_headers,
role,
expected_status,
):
organization, user, token = make_organization_and_user_with_plugin_token(role)
webhook = make_custom_webhook(organization=organization)
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
with patch(
"apps.api.views.webhooks.WebhooksView.destroy",
return_value=Response(
status=status.HTTP_204_NO_CONTENT,
),
):
response = client.delete(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == expected_status
@pytest.mark.django_db
def test_get_webhook_from_other_team_with_flag(
make_organization_and_user_with_plugin_token,
make_team,
make_user_auth_headers,
make_custom_webhook,
):
organization, user, token = make_organization_and_user_with_plugin_token()
team = make_team(organization)
webhook = make_custom_webhook(organization=organization, team=team)
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
url = f"{url}?from_organization=true"
response = client.get(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
@pytest.mark.django_db
def test_webhook_from_other_team_without_flag(
make_organization_and_user_with_plugin_token,
make_team,
make_user_auth_headers,
make_custom_webhook,
):
organization, user, token = make_organization_and_user_with_plugin_token()
team = make_team(organization)
webhook = make_custom_webhook(organization=organization, team=team)
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
response = client.get(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
@pytest.mark.django_db
def test_get_webhook_responses(
make_organization_and_user_with_plugin_token,
make_team,
make_user_auth_headers,
make_custom_webhook,
make_webhook_response,
):
organization, user, token = make_organization_and_user_with_plugin_token()
team = make_team(organization)
webhook = make_custom_webhook(
organization=organization, team=team, trigger_type=Webhook.TRIGGER_ALERT_GROUP_CREATED
)
for i in range(0, RECENT_RESPONSE_LIMIT + 1):
make_webhook_response(
webhook=webhook,
trigger_type=webhook.trigger_type,
status_code=200,
content=json.dumps({"id": "third-party-id"}),
event_data=json.dumps({"test": f"{i}"}),
)
client = APIClient()
url = reverse("api-internal:webhooks-responses", kwargs={"pk": webhook.public_primary_key})
response = client.get(url, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
assert len(response.data) == RECENT_RESPONSE_LIMIT
@pytest.mark.django_db
@pytest.mark.parametrize(
"test_template, test_payload, expected_result",
[
("https://test.com", None, "https://test.com"),
("https://test.com", "", "https://test.com"),
("{{ name }}", {"name": "test_1"}, "test_1"),
("{{ name }}", '{"name": "test_1"}', "test_1"),
],
)
def test_webhook_preview_template(
webhook_internal_api_setup, make_user_auth_headers, test_template, test_payload, expected_result
):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-preview-template", kwargs={"pk": webhook.public_primary_key})
data = {
"template_name": WEBHOOK_URL,
"template_body": test_template,
"payload": test_payload,
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
assert response.data["preview"] == expected_result
@pytest.mark.django_db
def test_webhook_field_masking(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
data = {
"name": "the_webhook",
"url": TEST_URL,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"team": None,
"password": "secret_password",
"authorization_header": "auth 1234",
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
webhook = Webhook.objects.get(public_primary_key=response.data["id"])
expected_response = data | {
"id": webhook.public_primary_key,
"data": None,
"username": None,
"password": WEBHOOK_FIELD_PLACEHOLDER,
"authorization_header": WEBHOOK_FIELD_PLACEHOLDER,
"forward_all": True,
"headers": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"labels": [],
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": str(data["trigger_type"]),
"trigger_type_name": "Alert Group Created",
"preset": None,
}
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == expected_response
assert webhook.password == data["password"]
assert webhook.authorization_header == data["authorization_header"]
assert webhook.user == user
@pytest.mark.django_db
def test_webhook_copy(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
data = {
"name": "the_webhook",
"url": TEST_URL,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"team": None,
"password": "secret_password",
"authorization_header": "auth 1234",
}
response1 = client.post(url, data, format="json", **make_user_auth_headers(user, token))
get_url = reverse("api-internal:webhooks-detail", kwargs={"pk": response1.data["id"]})
response2 = client.get(get_url, format="json", **make_user_auth_headers(user, token))
to_copy = response2.json()
to_copy["name"] = "copied_webhook"
response3 = client.post(url, to_copy, format="json", **make_user_auth_headers(user, token))
webhook = Webhook.objects.get(public_primary_key=response3.data["id"])
expected_response = data | {
"id": webhook.public_primary_key,
"name": to_copy["name"],
"data": None,
"username": None,
"password": WEBHOOK_FIELD_PLACEHOLDER,
"authorization_header": WEBHOOK_FIELD_PLACEHOLDER,
"forward_all": True,
"headers": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"labels": [],
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": str(data["trigger_type"]),
"trigger_type_name": "Alert Group Created",
"preset": None,
}
assert response3.status_code == status.HTTP_201_CREATED
assert response3.json() == expected_response
assert webhook.password == data["password"]
assert webhook.authorization_header == data["authorization_header"]
assert webhook.id != to_copy["id"]
assert webhook.user == user
@pytest.mark.django_db
def test_create_invalid_missing_fields(webhook_internal_api_setup, make_user_auth_headers):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-list")
data = {"url": TEST_URL, "trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED, "http_method": "POST"}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["name"][0] == "This field is required."
data = {"name": "test webhook 1", "trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED, "http_method": "POST"}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["url"][0] == "This field is required."
data = {"name": "test webhook 2", "url": TEST_URL, "http_method": "POST"}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["trigger_type"][0] == "This field is required."
data = {
"name": "test webhook 3",
"url": TEST_URL,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert (
response.json()["http_method"][0]
== "This field must be one of ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH']."
)
data = {
"name": "test webhook 3",
"url": TEST_URL,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "TOAST",
}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert (
response.json()["http_method"][0]
== "This field must be one of ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH']."
)
data = {"name": "test webhook 3", "url": TEST_URL, "trigger_type": 2000000, "http_method": "POST"}
response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["trigger_type"][0] == "This field is required."
@pytest.mark.django_db
def test_webhook_filter_by_labels(
make_organization_and_user_with_plugin_token,
make_custom_webhook,
make_webhook_label_association,
make_label_key_and_value,
make_user_auth_headers,
):
organization, user, token = make_organization_and_user_with_plugin_token()
webhook_with_label = make_custom_webhook(organization)
label = make_webhook_label_association(organization, webhook_with_label)
webhook_with_another_label = make_custom_webhook(organization)
another_label = make_webhook_label_association(organization, webhook_with_another_label)
not_attached_key, not_attached_value = make_label_key_and_value(organization)
client = APIClient()
# test filter by label, which is attached to only one webhook
url = reverse("api-internal:webhooks-list")
response = client.get(
f"{url}?label={label.key_id}:{label.value_id}",
content_type="application/json",
**make_user_auth_headers(user, token),
)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
assert response.json()[0]["id"] == webhook_with_label.public_primary_key
url = reverse("api-internal:webhooks-list")
response = client.get(
f"{url}?label={another_label.key_id}:{another_label.value_id}",
content_type="application/json",
**make_user_auth_headers(user, token),
)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
assert response.json()[0]["id"] == webhook_with_another_label.public_primary_key
# test filter by label which is not attached to any webhooks
response = client.get(
f"{url}?label={not_attached_key.id}:{not_attached_value.id}",
content_type="application/json",
**make_user_auth_headers(user, token),
)
assert len(response.json()) == 0
@pytest.mark.django_db
def test_update_webhook_labels(
webhook_internal_api_setup,
make_user_auth_headers,
):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
key_id = "testkey"
value_id = "testvalue"
data = {
"labels": [
{
"key": {"id": key_id, "name": "test", "prescribed": False},
"value": {"id": value_id, "name": "testv", "prescribed": False},
}
]
}
response = client.patch(
url,
data=json.dumps(data),
content_type="application/json",
**make_user_auth_headers(user, token),
)
webhook.refresh_from_db()
assert response.status_code == status.HTTP_200_OK
assert webhook.labels.count() == 1
label = webhook.labels.first()
assert label.key_id == key_id
assert label.value_id == value_id
response = client.patch(
url,
data=json.dumps({"labels": []}),
content_type="application/json",
**make_user_auth_headers(user, token),
)
webhook.refresh_from_db()
assert response.status_code == status.HTTP_200_OK
assert webhook.labels.count() == 0
@pytest.mark.django_db
def test_create_webhook_with_labels(
make_organization_and_user_with_plugin_token,
make_user_auth_headers,
):
organization, user, token = make_organization_and_user_with_plugin_token()
client = APIClient()
url = reverse("api-internal:webhooks-list")
key_id = "testkey"
value_id = "testvalue"
data = {
"name": "the_webhook",
"url": TEST_URL,
"trigger_type": Webhook.TRIGGER_ALERT_GROUP_CREATED,
"http_method": "POST",
"labels": [
{
"key": {"id": key_id, "name": "test", "prescribed": False},
"value": {"id": value_id, "name": "testv", "prescribed": False},
}
],
"team": None,
}
response = client.post(
url,
data=json.dumps(data),
content_type="application/json",
**make_user_auth_headers(user, token),
)
assert response.status_code == 201
webhook = Webhook.objects.get(public_primary_key=response.json()["id"])
expected_response = data | {
"id": webhook.public_primary_key,
"data": None,
"username": None,
"password": None,
"authorization_header": None,
"forward_all": True,
"headers": None,
"http_method": "POST",
"integration_filter": None,
"is_webhook_enabled": True,
"is_legacy": False,
"last_response_log": {
"request_data": "",
"request_headers": "",
"timestamp": None,
"content": "",
"status_code": None,
"request_trigger": "",
"url": "",
"event_data": "",
},
"trigger_template": None,
"trigger_type": str(data["trigger_type"]),
"trigger_type_name": "Alert Group Created",
"preset": None,
}
assert response.status_code == status.HTTP_201_CREATED
assert response.json() == expected_response
@pytest.mark.django_db
def test_update_webhook_labels_duplicate_key(
webhook_internal_api_setup,
make_user_auth_headers,
):
user, token, webhook = webhook_internal_api_setup
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
key_id = "testkey"
data = {
"labels": [
{"key": {"id": key_id, "name": "test"}, "value": {"id": "testvalue1", "name": "testv1"}},
{"key": {"id": key_id, "name": "test"}, "value": {"id": "testvalue2", "name": "testv2"}},
]
}
response = client.patch(
url,
data=json.dumps(data),
content_type="application/json",
**make_user_auth_headers(user, token),
)
webhook.refresh_from_db()
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert webhook.labels.count() == 0
@pytest.mark.django_db
def test_team_not_updated_if_not_in_data(
make_organization_and_user_with_plugin_token,
make_team,
make_custom_webhook,
make_user_auth_headers,
):
organization, user, token = make_organization_and_user_with_plugin_token()
team = make_team(organization)
webhook = make_custom_webhook(
name="some_webhook",
url="https://github.com/",
organization=organization,
forward_all=False,
team=team,
)
assert webhook.team == team
client = APIClient()
url = reverse("api-internal:webhooks-detail", kwargs={"pk": webhook.public_primary_key})
data = {"name": "renamed"}
response = client.put(url, data=data, format="json", **make_user_auth_headers(user, token))
assert response.status_code == status.HTTP_200_OK
assert response.json()["team"] == webhook.team.public_primary_key
webhook.refresh_from_db()
assert webhook.team == team
Reference in a new issue View git blame Copy permalink