Coverage for osm_nbi/tests/test_pmjobs_topic.py: 100%

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 

18import asynctest 

19import yaml 

20import re 

21from aioresponses import aioresponses 

22from http import HTTPStatus 

23from osm_nbi.engine import EngineException 

24from osm_common.dbmemory import DbMemory 

25from osm_nbi.pmjobs_topics import PmJobsTopic 

26from osm_nbi.tests.test_db_descriptors import ( 

27 db_nsds_text, 

28 db_vnfds_text, 

29 db_nsrs_text, 

30 db_vnfrs_text, 

31) 

32from osm_nbi.tests.pmjob_mocks.response import ( 

33 show_res, 

34 prom_res, 

35 cpu_utilization, 

36 users, 

37 load, 

38 empty, 

39) 

40 

41 

42class PmJobsTopicTest(asynctest.TestCase): 

43 def setUp(self): 

44 self.db = DbMemory() 

45 self.pmjobs_topic = PmJobsTopic(self.db, host="prometheus", port=9091) 

46 self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) 

47 self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) 

48 self.db.create_list("vnfrs", yaml.safe_load(db_vnfrs_text)) 

49 self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 

50 self.nsr = self.db.get_list("nsrs")[0] 

51 self.nsr_id = self.nsr["_id"] 

52 project_id = self.nsr["_admin"]["projects_write"] 

53 """metric_check_list contains the vnf metric name used in descriptor i.e users,load""" 

54 self.metric_check_list = [ 

55 "cpu_utilization", 

56 "average_memory_utilization", 

57 "disk_read_ops", 

58 "disk_write_ops", 

59 "disk_read_bytes", 

60 "disk_write_bytes", 

61 "packets_dropped", 

62 "packets_sent", 

63 "packets_received", 

64 "users", 

65 "load", 

66 ] 

67 self.session = { 

68 "username": "admin", 

69 "project_id": project_id, 

70 "method": None, 

71 "admin": True, 

72 "force": False, 

73 "public": False, 

74 "allow_show_user_project_role": True, 

75 } 

76 

77 def set_get_mock_res(self, mock_res, ns_id, metric_list): 

78 site = "http://prometheus:9091/api/v1/query?query=osm_metric_name{ns_id='nsr'}" 

79 site = re.sub(r"nsr", ns_id, site) 

80 for metric in metric_list: 

81 endpoint = re.sub(r"metric_name", metric, site) 

82 if metric == "cpu_utilization": 

83 response = yaml.safe_load(cpu_utilization) 

84 elif metric == "users": 

85 response = yaml.safe_load(users) 

86 elif metric == "load": 

87 response = yaml.safe_load(load) 

88 else: 

89 response = yaml.safe_load(empty) 

90 mock_res.get(endpoint, payload=response) 

91 

92 async def test_prom_metric_request(self): 

93 with self.subTest("Test case1 failed in test_prom"): 

94 prom_response = yaml.safe_load(prom_res) 

95 with aioresponses() as mock_res: 

96 self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list) 

97 result = await self.pmjobs_topic._prom_metric_request( 

98 self.nsr_id, self.metric_check_list 

99 ) 

100 self.assertCountEqual(result, prom_response, "Metric Data is valid") 

101 with self.subTest("Test case2 failed in test_prom"): 

102 with self.assertRaises( 

103 EngineException, msg="Prometheus not reachable" 

104 ) as e: 

105 await self.pmjobs_topic._prom_metric_request( 

106 self.nsr_id, self.metric_check_list 

107 ) 

108 self.assertIn("Connection to ", str(e.exception), "Wrong exception text") 

109 

110 def test_show(self): 

111 with self.subTest("Test case1 failed in test_show"): 

112 show_response = yaml.safe_load(show_res) 

113 with aioresponses() as mock_res: 

114 self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list) 

115 result = self.pmjobs_topic.show(self.session, self.nsr_id) 

116 self.assertEqual(len(result["entries"]), 1, "Number of metrics returned") 

117 self.assertCountEqual(result, show_response, "Response is valid") 

118 with self.subTest("Test case2 failed in test_show"): 

119 wrong_ns_id = "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa" 

120 with aioresponses() as mock_res: 

121 self.set_get_mock_res(mock_res, wrong_ns_id, self.metric_check_list) 

122 with self.assertRaises(EngineException, msg="ns not found") as e: 

123 self.pmjobs_topic.show(self.session, wrong_ns_id) 

124 self.assertEqual( 

125 e.exception.http_code, 

126 HTTPStatus.NOT_FOUND, 

127 "Wrong HTTP status code", 

128 ) 

129 self.assertIn( 

130 "NS not found with id {}".format(wrong_ns_id), 

131 str(e.exception), 

132 "Wrong exception text", 

133 )