Add incident API client (#5040)

Related to https://github.com/grafana/oncall-private/issues/2831

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

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"