centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
oncall-engine/engine/apps/webhooks/tests/test_trigger_webhook.py
Michael Derynck cef748ed4c
Add users_to_be_notified to new webhooks payload (#1798) 
- Change FIRING trigger for webhooks to be sent after escalation
snapshot has been computed
- Extract users from `notify_to_users_queue` and `notify_schedule` from
escalation snapshot to populate `users_to_be_notified` in webhook
payload
2023-04-20 16:13:48 +00:00

514 lines
20 KiB
Python
Raw Blame History

import json
from unittest.mock import call, patch
import pytest
from django.utils import timezone
from apps.alerts.models import AlertGroupLogRecord, EscalationPolicy
from apps.base.models import UserNotificationPolicyLogRecord
from apps.public_api.serializers import IncidentSerializer
from apps.webhooks.models import Webhook
from apps.webhooks.tasks import execute_webhook, send_webhook_event
from apps.webhooks.tasks.trigger_webhook import NOT_FROM_SELECTED_INTEGRATION
class MockResponse:
def __init__(self, status_code=200):
self.status_code = status_code
def json(self):
return {"response": self.status_code}
@pytest.mark.django_db
def test_send_webhook_event_filters(
make_organization, make_team, make_alert_receive_channel, make_alert_group, make_custom_webhook
):
organization = make_organization()
other_organization = make_organization()
other_team = make_team(organization)
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(alert_receive_channel)
webhooks = {}
for trigger_type, _ in Webhook.TRIGGER_TYPES:
webhooks[trigger_type] = make_custom_webhook(
organization=organization,
trigger_type=trigger_type,
)
other_team_webhook = make_custom_webhook(
organization=organization, team=other_team, trigger_type=Webhook.TRIGGER_ACKNOWLEDGE
)
other_org_webhook = make_custom_webhook(organization=other_organization, trigger_type=Webhook.TRIGGER_FIRING)
for trigger_type, _ in Webhook.TRIGGER_TYPES:
with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
send_webhook_event(trigger_type, alert_group.pk, organization_id=organization.pk)
assert mock_execute.call_args == call((webhooks[trigger_type].pk, alert_group.pk, None, None))
# other team
alert_receive_channel = make_alert_receive_channel(organization, team=other_team)
alert_group = make_alert_group(alert_receive_channel)
with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
send_webhook_event(
Webhook.TRIGGER_ACKNOWLEDGE, alert_group.pk, organization_id=organization.pk, team_id=other_team.pk
)
assert mock_execute.call_args == call((other_team_webhook.pk, alert_group.pk, None, None))
# other org
alert_receive_channel = make_alert_receive_channel(other_organization)
alert_group = make_alert_group(alert_receive_channel)
with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
send_webhook_event(Webhook.TRIGGER_FIRING, alert_group.pk, organization_id=other_organization.pk)
assert mock_execute.call_args == call((other_org_webhook.pk, alert_group.pk, None, None))
@pytest.mark.django_db
def test_execute_webhook_disabled(
make_organization, make_team, make_alert_receive_channel, make_alert_group, make_custom_webhook
):
organization = make_organization()
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(alert_receive_channel)
make_custom_webhook(organization=organization, trigger_type=Webhook.TRIGGER_FIRING)
make_custom_webhook(organization=organization, trigger_type=Webhook.TRIGGER_FIRING, is_webhook_enabled=False)
with patch("apps.webhooks.tasks.trigger_webhook.execute_webhook.apply_async") as mock_execute:
send_webhook_event(Webhook.TRIGGER_FIRING, alert_group.pk, organization_id=organization.pk)
mock_execute.assert_called_once()
@pytest.mark.django_db
def test_execute_webhook_integration_filter_not_matching(
make_organization, make_team, make_alert_receive_channel, make_alert_group, make_custom_webhook
):
organization = make_organization()
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(alert_receive_channel)
webhook = make_custom_webhook(
organization=organization, trigger_type=Webhook.TRIGGER_FIRING, integration_filter=["does-not-match"]
)
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
execute_webhook(webhook.pk, alert_group.pk, None, None)
assert not mock_requests.post.called
# check log should exist but have no status code
assert (
webhook.responses.count() == 1
and webhook.responses.first().status_code is None
and webhook.responses.first().request_trigger == NOT_FROM_SELECTED_INTEGRATION
)
@pytest.mark.django_db
def test_execute_webhook_integration_filter_matching(
make_organization, make_team, make_alert_receive_channel, make_alert_group, make_custom_webhook
):
organization = make_organization()
alert_receive_channel = make_alert_receive_channel(organization, public_primary_key="test-integration-1")
alert_group = make_alert_group(alert_receive_channel)
webhook = make_custom_webhook(
organization=organization,
trigger_type=Webhook.TRIGGER_FIRING,
integration_filter=["test-integration-1"],
# Check we get past integration filter but exit early to keep test simple
trigger_template="False",
)
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
execute_webhook(webhook.pk, alert_group.pk, None, None)
assert not mock_requests.post.called
# check log should exist but have no status code
assert (
webhook.responses.count() == 1
and webhook.responses.first().status_code is None
# Matches evaluated trigger_template
and webhook.responses.first().request_trigger == "False"
)
@pytest.mark.django_db
def test_execute_webhook_ok(
make_organization, make_user_for_organization, make_alert_receive_channel, make_alert_group, make_custom_webhook
):
organization = make_organization()
user = make_user_for_organization(organization)
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(
alert_receive_channel, acknowledged_at=timezone.now(), acknowledged=True, acknowledged_by=user.pk
)
webhook = make_custom_webhook(
organization=organization,
url="https://something/{{ alert_group_id }}/",
http_method="POST",
trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
trigger_template="{{{{ alert_group.integration_id == '{}' }}}}".format(
alert_receive_channel.public_primary_key
),
headers='{"some-header": "{{ alert_group_id }}"}',
data='{"value": "{{ alert_group_id }}"}',
forward_all=False,
)
mock_response = MockResponse()
with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
mock_gethostbyname.return_value = "8.8.8.8"
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
mock_requests.post.return_value = mock_response
execute_webhook(webhook.pk, alert_group.pk, user.pk, None)
assert mock_requests.post.called
expected_call = call(
"https://something/{}/".format(alert_group.public_primary_key),
timeout=10,
headers={"some-header": alert_group.public_primary_key},
json={"value": alert_group.public_primary_key},
)
assert mock_requests.post.call_args == expected_call
# check logs
log = webhook.responses.all()[0]
assert log.status_code == 200
assert log.content == json.dumps(mock_response.json())
assert log.request_data == json.dumps({"value": alert_group.public_primary_key})
assert log.request_headers == json.dumps({"some-header": alert_group.public_primary_key})
assert log.url == "https://something/{}/".format(alert_group.public_primary_key)
# check log record
log_record = alert_group.log_records.last()
assert log_record.type == AlertGroupLogRecord.TYPE_CUSTOM_BUTTON_TRIGGERED
expected_info = {
"trigger": "acknowledge",
"webhook_id": webhook.public_primary_key,
"webhook_name": webhook.name,
}
assert log_record.step_specific_info == expected_info
assert log_record.escalation_policy is None
assert log_record.escalation_policy_step is None
assert log_record.rendered_log_line_action() == f"outgoing webhook `{webhook.name}` triggered by acknowledge"
@pytest.mark.django_db
def test_execute_webhook_via_escalation_ok(
make_organization,
make_user_for_organization,
make_alert_receive_channel,
make_alert_group,
make_custom_webhook,
make_escalation_chain,
make_escalation_policy,
):
organization = make_organization()
user = make_user_for_organization(organization)
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(
alert_receive_channel, acknowledged_at=timezone.now(), acknowledged=True, acknowledged_by=user.pk
)
webhook = make_custom_webhook(
organization=organization,
url="https://something/{{ alert_group_id }}/",
http_method="POST",
trigger_type=Webhook.TRIGGER_ESCALATION_STEP,
trigger_template="{{{{ alert_group.integration_id == '{}' }}}}".format(
alert_receive_channel.public_primary_key
),
headers='{"some-header": "{{ alert_group_id }}"}',
data='{"value": "{{ alert_group_id }}"}',
forward_all=False,
)
escalation_chain = make_escalation_chain(organization)
escalation_policy = make_escalation_policy(
escalation_chain=escalation_chain,
escalation_policy_step=EscalationPolicy.STEP_TRIGGER_CUSTOM_WEBHOOK,
custom_webhook=webhook,
)
mock_response = MockResponse()
with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
mock_gethostbyname.return_value = "8.8.8.8"
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
mock_requests.post.return_value = mock_response
execute_webhook(webhook.pk, alert_group.pk, user.pk, escalation_policy.pk)
assert mock_requests.post.called
# check log record
log_record = alert_group.log_records.last()
assert log_record.type == AlertGroupLogRecord.TYPE_CUSTOM_BUTTON_TRIGGERED
expected_info = {
"trigger": "escalation",
"webhook_id": webhook.public_primary_key,
"webhook_name": webhook.name,
}
assert log_record.step_specific_info == expected_info
assert log_record.escalation_policy == escalation_policy
assert log_record.escalation_policy_step == EscalationPolicy.STEP_TRIGGER_CUSTOM_WEBHOOK
assert log_record.rendered_log_line_action() == f"outgoing webhook `{webhook.name}` triggered by escalation"
@pytest.mark.django_db
def test_execute_webhook_ok_forward_all(
make_organization,
make_user_for_organization,
make_alert_receive_channel,
make_alert_group,
make_user_notification_policy_log_record,
make_custom_webhook,
):
organization = make_organization()
user = make_user_for_organization(organization)
notified_user = make_user_for_organization(organization)
other_user = make_user_for_organization(organization)
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(
alert_receive_channel, acknowledged_at=timezone.now(), acknowledged=True, acknowledged_by=user.pk
)
for i in range(3):
make_user_notification_policy_log_record(
author=notified_user,
alert_group=alert_group,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_SUCCESS,
)
make_user_notification_policy_log_record(
author=other_user,
alert_group=alert_group,
type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED,
)
webhook = make_custom_webhook(
organization=organization,
url="https://something/{{ alert_group_id }}/",
http_method="POST",
trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
forward_all=True,
)
mock_response = MockResponse()
with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
mock_gethostbyname.return_value = "8.8.8.8"
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
mock_requests.post.return_value = mock_response
execute_webhook(webhook.pk, alert_group.pk, user.pk, None)
assert mock_requests.post.called
expected_data = {
"event": {
"type": "acknowledge",
"time": alert_group.acknowledged_at.isoformat(),
},
"user": {
"id": user.public_primary_key,
"username": user.username,
"email": user.email,
},
"integration": {
"id": alert_receive_channel.public_primary_key,
"type": alert_receive_channel.integration,
"name": alert_receive_channel.short_name,
"team": None,
},
"notified_users": [
{
"id": notified_user.public_primary_key,
"username": notified_user.username,
"email": notified_user.email,
}
],
"alert_group": IncidentSerializer(alert_group).data,
"alert_group_id": alert_group.public_primary_key,
"alert_payload": "",
"users_to_be_notified": [],
}
expected_call = call(
"https://something/{}/".format(alert_group.public_primary_key),
timeout=10,
headers={},
json=expected_data,
)
assert mock_requests.post.call_args == expected_call
# check logs
log = webhook.responses.all()[0]
assert log.status_code == 200
assert log.content == json.dumps(mock_response.json())
assert json.loads(log.request_data) == expected_data
assert log.url == "https://something/{}/".format(alert_group.public_primary_key)
@pytest.mark.django_db
def test_execute_webhook_using_responses_data(
make_organization,
make_user_for_organization,
make_alert_receive_channel,
make_alert_group,
make_custom_webhook,
make_webhook_response,
):
organization = make_organization()
user = make_user_for_organization(organization)
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(
alert_receive_channel, acknowledged_at=timezone.now(), acknowledged=True, acknowledged_by=user.pk
)
webhook = make_custom_webhook(
organization=organization,
url='https://something/{{ responses["response-1"].id }}/',
http_method="POST",
trigger_type=Webhook.TRIGGER_RESOLVE,
data='{"value": "{{ responses["response-2"].status }}"}',
forward_all=False,
)
# add previous webhook responses for the related alert group
make_webhook_response(
alert_group=alert_group,
webhook=make_custom_webhook(
organization=organization,
public_primary_key="response-1",
),
trigger_type=Webhook.TRIGGER_FIRING,
status_code=200,
content=json.dumps({"id": "third-party-id"}),
)
make_webhook_response(
alert_group=alert_group,
webhook=make_custom_webhook(
organization=organization,
public_primary_key="response-2",
),
trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
status_code=200,
content=json.dumps({"id": "third-party-id", "status": "updated"}),
)
# webhook wasn't executed because of some error, there is no content or status_code
make_webhook_response(
alert_group=alert_group,
webhook=make_custom_webhook(
organization=organization,
public_primary_key="response-3",
),
trigger_type=Webhook.TRIGGER_SILENCE,
content=None,
status_code=None,
)
mock_response = MockResponse()
with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
mock_gethostbyname.return_value = "8.8.8.8"
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
mock_requests.post.return_value = mock_response
execute_webhook(webhook.pk, alert_group.pk, user.pk, None)
assert mock_requests.post.called
expected_data = {"value": "updated"}
expected_call = call(
"https://something/third-party-id/",
timeout=10,
headers={},
json=expected_data,
)
assert mock_requests.post.call_args == expected_call
# check logs
log = webhook.responses.all()[0]
assert log.status_code == 200
assert log.content == json.dumps(mock_response.json())
assert json.loads(log.request_data) == expected_data
assert log.url == "https://something/third-party-id/"
@pytest.mark.django_db
def test_execute_webhook_trigger_false(
make_organization, make_alert_receive_channel, make_alert_group, make_custom_webhook
):
organization = make_organization()
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(alert_receive_channel, acknowledged_at=timezone.now(), acknowledged=True)
webhook = make_custom_webhook(
organization=organization,
url="https://something/{{ alert_id }}/",
http_method="POST",
trigger_type=Webhook.TRIGGER_ACKNOWLEDGE,
trigger_template="{{ integration_id == 'the-integration' }}",
)
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
execute_webhook(webhook.pk, alert_group.pk, None, None)
assert not mock_requests.post.called
# check log should exist but have no status
assert webhook.responses.count() == 1 and webhook.responses.first().status_code is None
@pytest.mark.django_db
@pytest.mark.parametrize(
"field_name,value,log_field_name,expected_error",
[
(
"url",
"https://myserver/{{ }}/triggered",
"url",
"URL - Template Error: Expected an expression, got 'end of print statement'",
),
(
"trigger_template",
"{{ }}",
"request_trigger",
"Trigger - Template Error: Expected an expression, got 'end of print statement'",
),
("headers", '"{{foo|invalid}}"', "request_headers", "Headers - Template Error: No filter named 'invalid'."),
(
"data",
"{{ }}",
"request_data",
"Data - Template Error: Expected an expression, got 'end of print statement'",
),
],
)
def test_execute_webhook_errors(
make_organization,
make_alert_receive_channel,
make_alert_group,
make_custom_webhook,
field_name,
value,
log_field_name,
expected_error,
):
organization = make_organization()
alert_receive_channel = make_alert_receive_channel(organization)
alert_group = make_alert_group(alert_receive_channel, resolved_at=timezone.now(), resolved=True)
extra_kwargs = {field_name: value}
if "url" not in extra_kwargs:
extra_kwargs["url"] = "https://something.cool/"
webhook = make_custom_webhook(
organization=organization,
http_method="POST",
trigger_type=Webhook.TRIGGER_RESOLVE,
forward_all=False,
**extra_kwargs,
)
with patch("apps.webhooks.utils.socket.gethostbyname") as mock_gethostbyname:
# make it a valid URL when resolving name
mock_gethostbyname.return_value = "8.8.8.8"
with patch("apps.webhooks.models.webhook.requests") as mock_requests:
execute_webhook(webhook.pk, alert_group.pk, None, None)
assert not mock_requests.post.called
log = webhook.responses.all()[0]
assert log.status_code is None
assert log.content is None
error = getattr(log, log_field_name)
assert error == expected_error
# check log record
log_record = alert_group.log_records.last()
assert log_record.type == AlertGroupLogRecord.ERROR_ESCALATION_TRIGGER_CUSTOM_WEBHOOK_ERROR
expected_info = {
"trigger": "resolve",
"webhook_id": webhook.public_primary_key,
"webhook_name": webhook.name,
}
assert log_record.step_specific_info == expected_info
assert log_record.reason == expected_error
assert (
log_record.rendered_log_line_action() == f"skipped resolve outgoing webhook `{webhook.name}`: {expected_error}"
)
Reference in a new issue View git blame Copy permalink