Grafana API improvements

mercury/grafanaAPI/grafana_api.py

@@ -1,22 +1,19 @@
 import os
 import json
 import requests
-from mercury.models import GFConfig
 import string
 import random
 
 TOKEN = "eyJrIjoiRTQ0cmNGcXRybkZlUUNZWmRvdFI0UlMwdFVYVUt3bzgiLCJuIjoia2V5IiwiaWQiOjF9"
 HOST = "https://dbc291.grafana.net"
-DASHBOARD_UID = "9UF7VluWz"
-DB_GRAFANA_NAME = "Heroku PostgreSQL (sextants-telemetry)"
 DB_HOSTNAME = "ec2-35-168-54-239.compute-1.amazonaws.com:5432"
 DB_NAME = "d76k4515q6qv"
 DB_USERNAME = "qvqhuplbiufdyq"
 DB_PASSWORD = "f45a1cfe8458ff9236ead8a7943eba31dcef761471e0d6d62b043b4e3d2e10e5"
 
 
 class Grafana:
-    def __init__(self, host=None, token=None):
+    def __init__(self, gf_config=None):
         """
 
         Initialize parameters needed to use the API: hostname, admin-level API token,
@@ -31,47 +28,28 @@ def __init__(self, host=None, token=None):
         :param host: Grafana hostname, e.g. https://dbc291.grafana.net
         :param token: API key with admin-level permissions
         """
-        gf_config = GFConfig.objects.filter(gf_current=True).first()
         if gf_config:
             self.hostname = gf_config.gf_host
             self.api_token = gf_config.gf_token
             self.database_hostname = gf_config.gf_host
             self.database_name = gf_config.gf_db_name
             self.database_username = gf_config.gf_db_username
             self.database_password = gf_config.gf_db_pw
-            self.database_grafana_name = gf_config.gf_db_grafana_name
         else:
-            # for test purposes, a test case should init this class with credentials
-            self.hostname = host
-            self.api_token = token
+            # for test purposes
+            self.hostname = HOST
+            self.api_token = TOKEN
             self.database_hostname = DB_HOSTNAME
             self.database_name = DB_NAME
             self.database_username = DB_USERNAME
             self.database_password = DB_PASSWORD
-            self.database_grafana_name = DB_GRAFANA_NAME
-
-        # URLs for all Grafana API endpoints
-        self.urls = {
-            "dashboard_post": "api/dashboards/db",
-            "dashboard_get": "api/dashboards",
-            "dashboard_home": "api/dashboards/home",
-            "dashboard_uid": "api/dashboards/uid",
-            "search": "api/search?",
-            "datasources": "api/datasources",
-            "datasource_name": "api/datasources/name",
-        }
 
         # Grafana API endpoints constructed with hostname + url
         self.endpoints = {
-            "dashboard_post": os.path.join(self.hostname, self.urls["dashboard_post"]),
-            "dashboard_get": os.path.join(self.hostname, self.urls["dashboard_get"]),
-            "dashboard_home": os.path.join(self.hostname, self.urls["dashboard_home"]),
-            "dashboard_uid": os.path.join(self.hostname, self.urls["dashboard_uid"]),
-            "search": os.path.join(self.hostname, self.urls["search"]),
-            "datasources": os.path.join(self.hostname, self.urls["datasources"]),
-            "datasource_name": os.path.join(
-                self.hostname, self.urls["datasource_name"]
-            ),
+            "dashboards": os.path.join(self.hostname, "api/dashboards/db"),
+            "dashboard_uid": os.path.join(self.hostname, "api/dashboards/uid"),
+            "datasources": os.path.join(self.hostname, "api/datasources"),
+            "datasource_name": os.path.join(self.hostname, "api/datasources/name"),
         }
 
         # Default panel sizes
@@ -80,7 +58,7 @@ def __init__(self, host=None, token=None):
 
     def generate_random_string(self, length):
         """
-        Generates a random string of letters of length=length
+        Generates a random string of letters of given length.
         :param length: Target length for the random string
         :return: Random string
         """
@@ -111,7 +89,7 @@ def get_dashboard_with_uid(self, uid):
         """
         :param uid: uid of the target dashboard
         :return:
-        Returns dashboard dictionary for given uid:
+        Returns dashboard dictionary for given uid or None if no dashboard was found.
         e.g. {
             'meta':
                 {
@@ -132,8 +110,6 @@ def get_dashboard_with_uid(self, uid):
                 'version': 1
             }
         }
-
-        Returns None if no dashboard is found.
         """
         headers = {"Content-Type": "application/json"}
         endpoint = os.path.join(self.endpoints["dashboard_uid"], uid)
@@ -182,7 +158,7 @@ def create_dashboard(self, title="Sensors"):
 
         # Prepare post request
         response = requests.post(
-            url=self.endpoints["dashboard_post"],
+            url=self.endpoints["dashboards"],
             auth=("api_key", self.api_token),
             json=dashboard_base,
         )
@@ -423,7 +399,7 @@ def add_panel(self, sensor, event, dashboard_uid):
         # POST updated dashboard
         headers = {"Content-Type": "application/json"}
         response = requests.post(
-            self.endpoints["dashboard_post"],
+            self.endpoints["dashboards"],
             data=json.dumps(updated_dashboard),
             headers=headers,
             auth=("api_key", self.api_token),
@@ -483,7 +459,7 @@ def create_panel_dict(self, panel_id, fields, panel_sql_query, title, x, y):
             "bars": False,
             "dashLength": 10,
             "dashes": False,
-            "datasource": self.database_grafana_name,
+            "datasource": self.database_name,
             "fill": 1,
             "fillGradient": 0,
             "gridPos": {"h": 9, "w": 12, "x": x, "y": y},

mercury/tests/test_gf_configs.py

@@ -0,0 +1,116 @@
+from django.test import TestCase
+from django.urls import reverse
+from mercury.models import EventCodeAccess, GFConfig
+from mercury.grafanaAPI.grafana_api import Grafana
+
+
+# default host and token, use this if user did not provide anything
+HOST = "https://mercurytests.grafana.net"
+# this token has Admin level permissions
+TOKEN = "eyJrIjoiQzFMemVOQ0RDUExIcTdhbEluS0hPMDJTZXdKMWQyYTEiLCJuIjoiYXBpX2tleTIiLCJpZCI6MX0="
+# this token has Editor level permissions
+EDITOR_TOKEN = (
+    "eyJrIjoibHlrZ2JWY0pnQk94b1YxSGYzd0NJ"
+    "ZUdZa3JBeWZIT3QiLCJuIjoiZWRpdG9yX2tleSIsImlkIjoxfQ=="
+)
+
+
+class TestGFConfig(TestCase):
+    TESTCODE = "testcode"
+
+    def setUp(self):
+        self.login_url = "mercury:EventAccess"
+        self.sensor_url = "mercury:sensor"
+        self.event_url = "mercury:events"
+        self.config_url = "mercury:gfconfig"
+        test_code = EventCodeAccess(event_code="testcode", enabled=True)
+        test_code.save()
+        # Login
+        self._get_with_event_code(self.sensor_url, self.TESTCODE)
+
+        self.gfconfig = GFConfig.objects.create(
+            gf_name="Test", gf_host=HOST, gf_token=TOKEN, gf_current=True
+        )
+        self.gfconfig.save()
+        # Create fresh grafana object
+        self.grafana = Grafana(self.gfconfig)
+
+        # Create random name to be used for event and datasource
+        self.event_name = self.grafana.generate_random_string(10)
+        self.datasource_name = self.grafana.generate_random_string(10)
+
+        # Clear existing dashboard and datasource
+        self.grafana.delete_dashboard_by_name(self.event_name)
+        self.grafana.delete_datasource_by_name(self.datasource_name)
+
+    def tearDown(self):
+        # Create fresh grafana instance (in case test invalidated any tokens, etc.)
+        self.grafana = Grafana(self.gfconfig)
+
+        # Clear all of the created dashboards
+        self.grafana.delete_dashboard_by_name(self.event_name)
+        self.grafana.delete_datasource_by_name(self.datasource_name)
+
+    def _get_with_event_code(self, url, event_code):
+        self.client.get(reverse(self.login_url))
+        self.client.post(reverse(self.login_url), data={"eventcode": event_code})
+        response = self.client.get(reverse(url))
+        session = self.client.session
+        return response, session
+
+    def test_config_view_get_success(self):
+        response = self.client.get(reverse(self.config_url))
+        self.assertEqual(200, response.status_code)
+
+    def test_config_post_success(self):
+        response = self.client.post(
+            reverse(self.config_url),
+            data={
+                "submit": "",
+                "gf_name": "Test Grafana Instance",
+                "gf_host": HOST,
+                "gf_token": TOKEN,
+            },
+        )
+        self.assertEqual(200, response.status_code)
+
+        gfconfig = GFConfig.objects.filter(gf_name="Test Grafana Instance")
+        self.assertTrue(gfconfig.count() > 0)
+        self.assertTrue(gfconfig[0].gf_name == "Test Grafana Instance")
+        self.assertTrue(gfconfig[0].gf_host == HOST)
+        self.assertTrue(gfconfig[0].gf_token == TOKEN)
+
+    def test_config_post_fail_invalid_API_key(self):
+        response = self.client.post(
+            reverse(self.config_url),
+            data={
+                "submit": "",
+                "gf_name": "Test Grafana Instance",
+                "gf_host": HOST,
+                "gf_token": "abcde",
+            },
+        )
+        self.assertEqual(200, response.status_code)
+        self.assertContains(response, "Grafana API validation failed: Invalid API key")
+
+        gfconfig = GFConfig.objects.filter(gf_name="Test Grafana Instance")
+        self.assertTrue(gfconfig.count() == 0)
+
+    def test_config_post_fail_insufficient_permissions(self):
+        response = self.client.post(
+            reverse(self.config_url),
+            data={
+                "submit": "",
+                "gf_name": "Test Grafana Instance",
+                "gf_host": HOST,
+                "gf_token": EDITOR_TOKEN,
+            },
+        )
+        self.assertEqual(200, response.status_code)
+        self.assertContains(
+            response,
+            "Grafana API validation failed: Access denied - check API permissions",
+        )
+
+        gfconfig = GFConfig.objects.filter(gf_name="Test Grafana Instance")
+        self.assertTrue(gfconfig.count() == 0)