osm_nbi/tests/test_pmjobs_topic.py

   1 # Copyright 2019 Preethika P(Tata Elxsi)
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 __author__ = "Preethika P,preethika.p@tataelxsi.co.in"
  17
  18 import asynctest
  19 import yaml
  20 import re
  21 from aioresponses import aioresponses
  22 from http import HTTPStatus
  23 from osm_nbi.engine import EngineException
  24 from osm_common.dbmemory import DbMemory
  25 from osm_nbi.pmjobs_topics import PmJobsTopic
  26 from osm_nbi.tests.test_db_descriptors import db_nsds_text, db_vnfds_text, db_nsrs_text, db_vnfrs_text
  27 from osm_nbi.tests.pmjob_mocks.response import show_res, prom_res, cpu_utilization, users, load, empty
  28
  29
  30 class PmJobsTopicTest(asynctest.TestCase):
  31
  32     def setUp(self):
  33         self.db = DbMemory()
  34         self.pmjobs_topic = PmJobsTopic(self.db, host="prometheus", port=9091)
  35         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
  36         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
  37         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
  38         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
  39         self.nsr = self.db.get_list("nsrs")[0]
  40         self.nsr_id = self.nsr["_id"]
  41         project_id = self.nsr["_admin"]["projects_write"]
  42         """metric_check_list contains the vnf metric name used in descriptor i.e users,load"""
  43         self.metric_check_list = ['cpu_utilization', 'average_memory_utilization', 'disk_read_ops',
  44                                   'disk_write_ops', 'disk_read_bytes', 'disk_write_bytes',
  45                                   'packets_dropped', 'packets_sent', 'packets_received', 'users', 'load']
  46         self.session = {"username": "admin", "project_id": project_id, "method": None,
  47                         "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
  48
  49     def set_get_mock_res(self, mock_res, ns_id, metric_list):
  50         site = "http://prometheus:9091/api/v1/query?query=osm_metric_name{ns_id='nsr'}"
  51         site = re.sub(r'nsr', ns_id, site)
  52         for metric in metric_list:
  53             endpoint = re.sub(r'metric_name', metric, site)
  54             if metric == 'cpu_utilization':
  55                 response = yaml.load(cpu_utilization, Loader=yaml.Loader)
  56             elif metric == 'users':
  57                 response = yaml.load(users, Loader=yaml.Loader)
  58             elif metric == 'load':
  59                 response = yaml.load(load, Loader=yaml.Loader)
  60             else:
  61                 response = yaml.load(empty, Loader=yaml.Loader)
  62             mock_res.get(endpoint, payload=response)
  63
  64     def test_get_vnf_metric_list(self):
  65         with self.subTest("Test case1 failed in test_get_vnf_metric_list"):
  66             metric_list = self.pmjobs_topic._get_vnf_metric_list(self.nsr_id)
  67             self.assertCountEqual(metric_list, self.metric_check_list,
  68                                   "VNF metric list is not correctly fetched")
  69         with self.subTest("Test case2 failed in test_get_vnf_metric_list"):
  70             wrong_ns_id = "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa"
  71             with self.assertRaises(EngineException, msg="ns not found") as e:
  72                 self.pmjobs_topic._get_vnf_metric_list(wrong_ns_id)
  73             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
  74             self.assertIn("NS not found with id {}".format(wrong_ns_id),
  75                           str(e.exception), "Wrong exception text")
  76
  77     async def test_prom_metric_request(self):
  78         with self.subTest("Test case1 failed in test_prom"):
  79             prom_response = yaml.load(prom_res, Loader=yaml.Loader)
  80             with aioresponses() as mock_res:
  81                 self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list)
  82                 result = await self.pmjobs_topic._prom_metric_request(self.nsr_id, self.metric_check_list)
  83             self.assertCountEqual(result, prom_response, "Metric Data is valid")
  84         with self.subTest("Test case2 failed in test_prom"):
  85             with self.assertRaises(EngineException, msg="Prometheus not reachable") as e:
  86                 await self.pmjobs_topic._prom_metric_request(self.nsr_id, self.metric_check_list)
  87             self.assertIn("Connection to ", str(e.exception), "Wrong exception text")
  88
  89     def test_show(self):
  90         with self.subTest("Test case1 failed in test_show"):
  91             show_response = yaml.load(show_res, Loader=yaml.Loader)
  92             with aioresponses() as mock_res:
  93                 self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list)
  94                 result = self.pmjobs_topic.show(self.session, self.nsr_id)
  95             self.assertEqual(len(result['entries']), 3, "Number of metrics returned")
  96             self.assertCountEqual(result, show_response, "Response is valid")
  97         with self.subTest("Test case2 failed in test_show"):
  98             wrong_ns_id = "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa"
  99             with aioresponses() as mock_res:
 100                 self.set_get_mock_res(mock_res, wrong_ns_id, self.metric_check_list)
 101                 with self.assertRaises(EngineException, msg="ns not found") as e:
 102                     self.pmjobs_topic.show(self.session, wrong_ns_id)
 103                 self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 104                 self.assertIn("NS not found with id {}".format(wrong_ns_id), str(e.exception),
 105                               "Wrong exception text")