diff --git a/engine/common/incident_api/__init__.py b/engine/common/incident_api/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/engine/common/incident_api/client.py b/engine/common/incident_api/client.py new file mode 100644 index 00000000..41d2b28c --- /dev/null +++ b/engine/common/incident_api/client.py @@ -0,0 +1,166 @@ +import typing +from json import JSONDecodeError +from urllib.parse import urljoin + +import requests +from django.conf import settings + + +class IncidentDetails(typing.TypedDict): + # https://grafana.com/docs/grafana-cloud/alerting-and-irm/irm/incident/api/reference/#getincidentresponse + createdByUser: typing.Dict + createdTime: str + durationSeconds: int + heroImagePath: str + incidentEnd: str + incidentID: str + incidentMembership: typing.Dict + incidentStart: str + isDrill: bool + labels: typing.List[dict] + overviewURL: str + severity: str + status: str + summary: str + taskList: typing.Dict + title: str + + +class SeverityDetails(typing.TypedDict): + severityID: str + orgID: str + displayLabel: str + level: int + iconName: str + description: str + darkColor: str + lightColor: str + archivedAt: str + archivedByUserID: str | None + deletedAt: str + deletedByUserID: str | None + createdAt: str + updatedAt: str + + +class ActivityItemDetails(typing.TypedDict): + # https://grafana.com/docs/grafana-cloud/alerting-and-irm/irm/incident/api/reference/#addactivityresponse + activityItemID: str + activityKind: str + attachments: typing.List[dict] + body: str + createdTime: str + eventTime: str + fieldValues: typing.Dict[str, str] + immutable: bool + incidentID: str + relevance: str + subjectUser: typing.Dict[str, str] + tags: typing.List[str] + url: str + user: typing.Dict[str, str] + + +class IncidentAPIException(Exception): + def __init__(self, status, url, msg="", method="GET"): + self.url = url + self.status = status + self.method = method + self.msg = msg + + def __str__(self): + return f"IncidentAPIException: status={self.status} url={self.url} method={self.method}" + + +TIMEOUT = 5 +DEFAULT_INCIDENT_SEVERITY = "pending" +DEFAULT_INCIDENT_STATUS = "active" +DEFAULT_ACTIVITY_KIND = "userNote" + + +class IncidentAPIClient: + INCIDENT_BASE_PATH = "/api/plugins/grafana-incident-app/resources/" + + def __init__(self, api_url: str, api_token: str) -> None: + self.api_token = api_token + self.api_url = urljoin(api_url, self.INCIDENT_BASE_PATH) + + @property + def _request_headers(self): + return {"User-Agent": settings.GRAFANA_COM_USER_AGENT, "Authorization": f"Bearer {self.api_token}"} + + def _check_response(self, response: requests.models.Response): + message = None + + if 400 <= response.status_code < 500: + try: + error_data = response.json() + message = error_data.get("error", response.reason) + except JSONDecodeError: + message = response.reason + elif 500 <= response.status_code < 600: + message = response.reason + + if message: + raise IncidentAPIException( + status=response.status_code, + url=response.request.url, + msg=message, + method=response.request.method, + ) + + def create_incident( + self, + title: str, + severity: str = DEFAULT_INCIDENT_SEVERITY, + status: str = DEFAULT_INCIDENT_STATUS, + attachCaption: str = "", + attachURL: str = "", + ) -> typing.Tuple[IncidentDetails, requests.models.Response]: + endpoint = "api/v1/IncidentsService.CreateIncident" + url = self.api_url + endpoint + # NOTE: invalid severity will raise a 500 error + response = requests.post( + url, + json={ + "title": title, + "severity": severity, + "attachCaption": attachCaption, + "attachURL": attachURL, + "status": status, + }, + timeout=TIMEOUT, + headers=self._request_headers, + ) + self._check_response(response) + return response.json().get("incident"), response + + def get_incident(self, incident_id: str) -> typing.Tuple[IncidentDetails, requests.models.Response]: + endpoint = "api/v1/IncidentsService.GetIncident" + url = self.api_url + endpoint + response = requests.post(url, json={"incidentID": incident_id}, timeout=TIMEOUT, headers=self._request_headers) + self._check_response(response) + return response.json().get("incident"), response + + def get_severities(self) -> typing.Tuple[typing.List[SeverityDetails], requests.models.Response]: + # NOTE: internal endpoint + endpoint = "api/SeveritiesService.GetOrgSeverities" + url = self.api_url + endpoint + # pass empty json payload otherwise it will return a 500 response + response = requests.post(url, timeout=TIMEOUT, headers=self._request_headers, json={}) + self._check_response(response) + return response.json().get("severities"), response + + def add_activity( + self, incident_id: str, body: str, kind: str = DEFAULT_ACTIVITY_KIND + ) -> typing.Tuple[ActivityItemDetails, requests.models.Response]: + endpoint = "api/v1/ActivityService.AddActivity" + url = self.api_url + endpoint + response = requests.post( + url, + json={"incidentID": incident_id, "activityKind": kind, "body": body}, + timeout=TIMEOUT, + headers=self._request_headers, + ) + self._check_response(response) + return response.json().get("activityItem"), response diff --git a/engine/common/incident_api/tests/__init__.py b/engine/common/incident_api/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/engine/common/incident_api/tests/test_client.py b/engine/common/incident_api/tests/test_client.py new file mode 100644 index 00000000..2fffad1d --- /dev/null +++ b/engine/common/incident_api/tests/test_client.py @@ -0,0 +1,190 @@ +import json + +import httpretty +import pytest +from rest_framework import status + +from common.incident_api.client import ( + DEFAULT_ACTIVITY_KIND, + DEFAULT_INCIDENT_SEVERITY, + DEFAULT_INCIDENT_STATUS, + IncidentAPIClient, + IncidentAPIException, +) + + +@httpretty.activate(verbose=True, allow_net_connect=False) +def test_create_incident_expected_request(): + stack_url = "https://foobar.grafana.net" + api_token = "asdfasdfasdfasdf" + client = IncidentAPIClient(stack_url, api_token) + url = f"{stack_url}{client.INCIDENT_BASE_PATH}api/v1/IncidentsService.CreateIncident" + response_data = { + "error": "", + "incident": { + "incidentID": "123", + }, + } + mock_response = httpretty.Response(json.dumps(response_data), status=status.HTTP_200_OK) + httpretty.register_uri(httpretty.POST, url, responses=[mock_response]) + + title = "title" + severity = "severity" + attachCaption = "attachCaption" + attachURL = "http://some.url" + incident_status = "active" + data, response = client.create_incident(title, severity, incident_status, attachCaption, attachURL) + + assert data == response_data["incident"] + assert response.status_code == status.HTTP_200_OK + last_request = httpretty.last_request() + assert last_request.headers["Authorization"] == f"Bearer {api_token}" + assert last_request.method == "POST" + assert last_request.url == url + assert json.loads(last_request.body) == { + "title": title, + "severity": severity, + "attachCaption": attachCaption, + "attachURL": attachURL, + "status": incident_status, + } + + # test using defaults + data, response = client.create_incident(title) + + assert data == response_data["incident"] + assert response.status_code == status.HTTP_200_OK + last_request = httpretty.last_request() + assert json.loads(last_request.body) == { + "title": title, + "severity": DEFAULT_INCIDENT_SEVERITY, + "attachCaption": "", + "attachURL": "", + "status": DEFAULT_INCIDENT_STATUS, + } + + +@httpretty.activate(verbose=True, allow_net_connect=False) +def test_get_incident_expected_request(): + stack_url = "https://foobar.grafana.net" + api_token = "asdfasdfasdfasdf" + client = IncidentAPIClient(stack_url, api_token) + url = f"{stack_url}{client.INCIDENT_BASE_PATH}api/v1/IncidentsService.GetIncident" + incident_id = "123" + response_data = { + "error": "", + "incident": { + "incidentID": incident_id, + }, + } + mock_response = httpretty.Response(json.dumps(response_data), status=status.HTTP_200_OK) + httpretty.register_uri(httpretty.POST, url, responses=[mock_response]) + + data, response = client.get_incident(incident_id) + + assert data == response_data["incident"] + assert response.status_code == status.HTTP_200_OK + last_request = httpretty.last_request() + assert last_request.headers["Authorization"] == f"Bearer {api_token}" + assert last_request.method == "POST" + assert last_request.url == url + + +@httpretty.activate(verbose=True, allow_net_connect=False) +def test_get_severities_expected_request(): + stack_url = "https://foobar.grafana.net" + api_token = "asdfasdfasdfasdf" + client = IncidentAPIClient(stack_url, api_token) + url = f"{stack_url}{client.INCIDENT_BASE_PATH}api/SeveritiesService.GetOrgSeverities" + response_data = { + "error": "", + "severities": [ + {"severityID": "abc", "orgID": "1", "displayLabel": "Pending", "level": -1}, + {"severityID": "def", "orgID": "1", "displayLabel": "Critical", "level": 1}, + ], + } + mock_response = httpretty.Response(json.dumps(response_data), status=status.HTTP_200_OK) + httpretty.register_uri(httpretty.POST, url, responses=[mock_response]) + + data, response = client.get_severities() + + assert data == response_data["severities"] + assert response.status_code == status.HTTP_200_OK + last_request = httpretty.last_request() + assert last_request.headers["Authorization"] == f"Bearer {api_token}" + assert last_request.method == "POST" + assert last_request.url == url + assert json.loads(last_request.body) == {} + + +@httpretty.activate(verbose=True, allow_net_connect=False) +def test_add_activity_expected_request(): + stack_url = "https://foobar.grafana.net" + api_token = "asdfasdfasdfasdf" + client = IncidentAPIClient(stack_url, api_token) + url = f"{stack_url}{client.INCIDENT_BASE_PATH}api/v1/ActivityService.AddActivity" + incident_id = "123" + content = "some content" + response_data = { + "error": "", + "activityItem": { + "activityItemID": "activity-item-theID", + "incidentID": incident_id, + "user": { + "userID": "grafana-incident:user-user-id", + "name": "Service Account: extsvc-grafana-oncall-app", + }, + "createdTime": "2024-09-18T14:06:47.57795Z", + "activityKind": "userNote", + "body": content, + }, + } + mock_response = httpretty.Response(json.dumps(response_data), status=status.HTTP_200_OK) + httpretty.register_uri(httpretty.POST, url, responses=[mock_response]) + + data, response = client.add_activity(incident_id, content) + + assert data == response_data["activityItem"] + assert response.status_code == status.HTTP_200_OK + last_request = httpretty.last_request() + assert last_request.headers["Authorization"] == f"Bearer {api_token}" + assert last_request.method == "POST" + assert last_request.url == url + assert json.loads(last_request.body) == { + "incidentID": incident_id, + "activityKind": DEFAULT_ACTIVITY_KIND, + "body": content, + } + + +@pytest.mark.parametrize( + "endpoint, client_method_name, args", + [ + ("api/v1/IncidentsService.CreateIncident", "create_incident", ("title",)), + ("api/v1/IncidentsService.GetIncident", "get_incident", ("incident-id",)), + ("api/SeveritiesService.GetOrgSeverities", "get_severities", ()), + ("api/v1/ActivityService.AddActivity", "add_activity", ("incident-id", "content")), + ], +) +@httpretty.activate(verbose=True, allow_net_connect=False) +def test_error_handling(endpoint, client_method_name, args): + stack_url = "https://foobar.grafana.net" + api_token = "asdfasdfasdfasdf" + client = IncidentAPIClient(stack_url, api_token) + url = f"{stack_url}{client.INCIDENT_BASE_PATH}{endpoint}" + response_data = { + "error": "There was an error", + } + for error_code in (status.HTTP_400_BAD_REQUEST, status.HTTP_500_INTERNAL_SERVER_ERROR): + mock_response = httpretty.Response(json.dumps(response_data), status=error_code) + httpretty.register_uri(httpretty.POST, url, responses=[mock_response]) + with pytest.raises(IncidentAPIException) as excinfo: + client_method = getattr(client, client_method_name) + client_method(*args) + assert excinfo.value.status == error_code + expected_error = ( + response_data["error"] if error_code == status.HTTP_400_BAD_REQUEST else "Internal Server Error" + ) + assert excinfo.value.msg == expected_error + assert excinfo.value.url == url + assert excinfo.value.method == "POST"