Grafana Config End-to-End: Credential validation

mercury/grafanaAPI/grafana_api.py

@@ -2,6 +2,8 @@
 import json
 import requests
 from mercury.models import GFConfig
+import string
+import random
 
 TOKEN = "eyJrIjoiRTQ0cmNGcXRybkZlUUNZWmRvdFI0UlMwdFVYVUt3bzgiLCJuIjoia2V5IiwiaWQiOjF9"
 HOST = "https://dbc291.grafana.net"
@@ -76,6 +78,35 @@ def __init__(self, host=None, token=None):
         self.base_panel_width = 15
         self.base_panel_height = 12
 
+    def generate_random_string(self, length):
+        """
+        Generates a random string of letters of length=length
+        :param length: Target length for the random string
+        :return: Random string
+        """
+        letters = string.ascii_lowercase
+        return "".join(random.choice(letters) for i in range(length))
+
+    def validate_credentials(self):
+        """
+        Validates current set of grafana API credentials (hostname and API token).
+        Attempts to create and delete a dashboard. If there is any failure,
+        a ValueError will be raised which can be caught by the caller. If the
+        dashboard is created successfully, True is returned.
+        :return: True if a dashboard could be created using these API credentials,
+        False otherwise.
+        """
+        dashboard_name = self.generate_random_string(10)
+
+        try:
+            self.create_dashboard(dashboard_name)
+        except ValueError as error:
+            raise ValueError(f"Grafana API validation failed: {error}")
+
+        self.delete_dashboard_by_name(dashboard_name)
+
+        return True
+
     def get_dashboard_with_uid(self, uid):
         """
         :param uid: uid of the target dashboard

mercury/tests/test_grafana.py

@@ -7,8 +7,6 @@
 import requests
 import os
 import datetime
-import random
-import string
 
 
 # default host and token, use this if user did not provide anything
@@ -93,9 +91,8 @@ def setUp(self):
         self.grafana = Grafana(HOST, TOKEN)
 
         # Create random name to be used for event and datasource
-        letters = string.ascii_lowercase
-        self.event_name = "".join(random.choice(letters) for i in range(10))
-        self.datasource_name = "".join(random.choice(letters) for i in range(10))
+        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)
@@ -133,8 +130,7 @@ def test_get_dashboard_exists(self):
         self.assertTrue(fetched_dashboard["dashboard"]["title"], self.event_name)
 
     def test_get_dashboard_fail(self):
-        letters = string.ascii_lowercase
-        uid = "".join(random.choice(letters) for i in range(10))
+        uid = self.grafana.generate_random_string(10)
 
         fetched_dashboard = self.grafana.get_dashboard_with_uid(uid)
 
@@ -177,6 +173,25 @@ def test_create_grafana_dashboard_fail_permissions(self):
         with self.assertRaisesMessage(ValueError, expected_message):
             self.grafana.create_dashboard(self.event_name)
 
+    def test_validate_credentials_success(self):
+        self.assertTrue(self.grafana.validate_credentials())
+
+    def test_validate_credentials_fail_authorization(self):
+        self.grafana.api_token = "abcde"  # invalidate API token
+
+        expected_message = "Grafana API validation failed: Invalid API key"
+        with self.assertRaisesMessage(ValueError, expected_message):
+            self.grafana.validate_credentials()
+
+    def test_validate_credentials_fail_permissions(self):
+        self.grafana.api_token = EDITOR_TOKEN  # API token with Editor permissions
+
+        expected_message = (
+            "Grafana API validation failed: Access denied - " "check API permissions"
+        )
+        with self.assertRaisesMessage(ValueError, expected_message):
+            self.grafana.validate_credentials()
+
     def test_create_grafana_dashboard_fail_duplicate_title(self):
         dashboard = self.grafana.create_dashboard(self.event_name)
         self.assertTrue(dashboard)
@@ -205,8 +220,7 @@ def test_delete_grafana_dashboard(self):
         self.assertEquals(response.json()["message"], "Dashboard not found")
 
     def test_delete_grafana_dashboard_fail(self):
-        letters = string.ascii_lowercase
-        uid = "".join(random.choice(letters) for i in range(10))
+        uid = self.grafana.generate_random_string(10)
 
         # should return false if dashboard doesn't exist
         deleted_dashboard = self.grafana.delete_dashboard(uid)
@@ -299,8 +313,7 @@ def test_add_multiple_panels(self):
             self.assertTrue(dashboard_info["dashboard"]["panels"][i]["title"] == name)
 
     def test_add_panel_fail_invalid_uid(self):
-        letters = string.ascii_lowercase
-        uid = "".join(random.choice(letters) for i in range(10))
+        uid = self.grafana.generate_random_string(10)
 
         self.sim.createOrResetASensorTypeFromPresets(0)
         self.sim.createASensorFromPresets(0)

mercury/views/gf_config.py

@@ -40,13 +40,13 @@ def post(self, request, *args, **kwargs):
                 gf_token=request.POST.get("gf_token"),
             )
 
+            # Create Grafana instance with host and token
+            grafana = Grafana(config_data.gf_host, config_data.gf_token)
             try:
-                # Create Grafana instance with host and token
-                grafana = Grafana(config_data.gf_host, config_data.gf_token)
+                grafana.validate_credentials()
                 config_data.gf_current = True
                 # Only save the config if credentials were validated
                 config_data.save()
-
             except ValueError as error:
                 messages.error(request, f"Grafana initial set up failed: {error}")