Reformat NBI to standardized format

[osm/NBI.git] / osm_nbi / tests / test_pmjobs_topic.py

diff --git a/osm_nbi/tests/test_pmjobs_topic.py b/osm_nbi/tests/test_pmjobs_topic.py
index 84bf048..231818b 100644 (file)
--- a/osm_nbi/tests/test_pmjobs_topic.py
+++ b/osm_nbi/tests/test_pmjobs_topic.py
@@ -23,12 +23,23 @@ from http import HTTPStatus
 from osm_nbi.engine import EngineException
 from osm_common.dbmemory import DbMemory
 from osm_nbi.pmjobs_topics import PmJobsTopic
-from osm_nbi.tests.test_db_descriptors import db_nsds_text, db_vnfds_text, db_nsrs_text, db_vnfrs_text
-from osm_nbi.tests.pmjob_mocks.response import show_res, prom_res, cpu_utilization, users, load, empty
+from osm_nbi.tests.test_db_descriptors import (
+    db_nsds_text,
+    db_vnfds_text,
+    db_nsrs_text,
+    db_vnfrs_text,
+)
+from osm_nbi.tests.pmjob_mocks.response import (
+    show_res,
+    prom_res,
+    cpu_utilization,
+    users,
+    load,
+    empty,
+)
 
 
 class PmJobsTopicTest(asynctest.TestCase):
-
     def setUp(self):
         self.db = DbMemory()
         self.pmjobs_topic = PmJobsTopic(self.db, host="prometheus", port=9091)
@@ -40,22 +51,39 @@ class PmJobsTopicTest(asynctest.TestCase):
         self.nsr_id = self.nsr["_id"]
         project_id = self.nsr["_admin"]["projects_write"]
         """metric_check_list contains the vnf metric name used in descriptor i.e users,load"""
-        self.metric_check_list = ['cpu_utilization', 'average_memory_utilization', 'disk_read_ops',
-                                  'disk_write_ops', 'disk_read_bytes', 'disk_write_bytes',
-                                  'packets_dropped', 'packets_sent', 'packets_received', 'users', 'load']
-        self.session = {"username": "admin", "project_id": project_id, "method": None,
-                        "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
+        self.metric_check_list = [
+            "cpu_utilization",
+            "average_memory_utilization",
+            "disk_read_ops",
+            "disk_write_ops",
+            "disk_read_bytes",
+            "disk_write_bytes",
+            "packets_dropped",
+            "packets_sent",
+            "packets_received",
+            "users",
+            "load",
+        ]
+        self.session = {
+            "username": "admin",
+            "project_id": project_id,
+            "method": None,
+            "admin": True,
+            "force": False,
+            "public": False,
+            "allow_show_user_project_role": True,
+        }
 
     def set_get_mock_res(self, mock_res, ns_id, metric_list):
         site = "http://prometheus:9091/api/v1/query?query=osm_metric_name{ns_id='nsr'}"
-        site = re.sub(r'nsr', ns_id, site)
+        site = re.sub(r"nsr", ns_id, site)
         for metric in metric_list:
-            endpoint = re.sub(r'metric_name', metric, site)
-            if metric == 'cpu_utilization':
+            endpoint = re.sub(r"metric_name", metric, site)
+            if metric == "cpu_utilization":
                 response = yaml.load(cpu_utilization, Loader=yaml.Loader)
-            elif metric == 'users':
+            elif metric == "users":
                 response = yaml.load(users, Loader=yaml.Loader)
-            elif metric == 'load':
+            elif metric == "load":
                 response = yaml.load(load, Loader=yaml.Loader)
             else:
                 response = yaml.load(empty, Loader=yaml.Loader)
@@ -66,11 +94,17 @@ class PmJobsTopicTest(asynctest.TestCase):
             prom_response = yaml.load(prom_res, Loader=yaml.Loader)
             with aioresponses() as mock_res:
                 self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list)
-                result = await self.pmjobs_topic._prom_metric_request(self.nsr_id, self.metric_check_list)
+                result = await self.pmjobs_topic._prom_metric_request(
+                    self.nsr_id, self.metric_check_list
+                )
             self.assertCountEqual(result, prom_response, "Metric Data is valid")
         with self.subTest("Test case2 failed in test_prom"):
-            with self.assertRaises(EngineException, msg="Prometheus not reachable") as e:
-                await self.pmjobs_topic._prom_metric_request(self.nsr_id, self.metric_check_list)
+            with self.assertRaises(
+                EngineException, msg="Prometheus not reachable"
+            ) as e:
+                await self.pmjobs_topic._prom_metric_request(
+                    self.nsr_id, self.metric_check_list
+                )
             self.assertIn("Connection to ", str(e.exception), "Wrong exception text")
 
     def test_show(self):
@@ -79,7 +113,7 @@ class PmJobsTopicTest(asynctest.TestCase):
             with aioresponses() as mock_res:
                 self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list)
                 result = self.pmjobs_topic.show(self.session, self.nsr_id)
-            self.assertEqual(len(result['entries']), 1, "Number of metrics returned")
+            self.assertEqual(len(result["entries"]), 1, "Number of metrics returned")
             self.assertCountEqual(result, show_response, "Response is valid")
         with self.subTest("Test case2 failed in test_show"):
             wrong_ns_id = "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa"
@@ -87,6 +121,13 @@ class PmJobsTopicTest(asynctest.TestCase):
                 self.set_get_mock_res(mock_res, wrong_ns_id, self.metric_check_list)
                 with self.assertRaises(EngineException, msg="ns not found") as e:
                     self.pmjobs_topic.show(self.session, wrong_ns_id)
-                self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
-                self.assertIn("NS not found with id {}".format(wrong_ns_id), str(e.exception),
-                              "Wrong exception text")
+                self.assertEqual(
+                    e.exception.http_code,
+                    HTTPStatus.NOT_FOUND,
+                    "Wrong HTTP status code",
+                )
+                self.assertIn(
+                    "NS not found with id {}".format(wrong_ns_id),
+                    str(e.exception),
+                    "Wrong exception text",
+                )