1 # Copyright 2019 Preethika P(Tata Elxsi)
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 __author__
= "Preethika P,preethika.p@tataelxsi.co.in"
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
, cpu_utilization
, users
, load
, empty
28 # from osm_nbi.tests.pmjob_mocks.response import prom_res
31 class PmJobsTopicTest(asynctest
.TestCase
):
35 self
.pmjobs_topic
= PmJobsTopic(self
.db
, host
="prometheus", port
=9091)
36 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
37 self
.db
.create_list("vnfds", yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
))
38 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
39 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
40 self
.nsr
= self
.db
.get_list("nsrs")[0]
41 self
.nsr_id
= self
.nsr
["_id"]
42 project_id
= self
.nsr
["_admin"]["projects_write"]
43 """metric_check_list contains the vnf metric name used in descriptor i.e users,load"""
44 self
.metric_check_list
= ['cpu_utilization', 'average_memory_utilization', 'disk_read_ops',
45 'disk_write_ops', 'disk_read_bytes', 'disk_write_bytes',
46 'packets_dropped', 'packets_sent', 'packets_received', 'users', 'load']
47 self
.session
= {"username": "admin", "project_id": project_id
, "method": None,
48 "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
50 def set_get_mock_res(self
, mock_res
, ns_id
, metric_list
):
51 site
= "http://prometheus:9091/api/v1/query?query=osm_metric_name{ns_id='nsr'}"
52 site
= re
.sub(r
'nsr', ns_id
, site
)
53 for metric
in metric_list
:
54 endpoint
= re
.sub(r
'metric_name', metric
, site
)
55 if metric
== 'cpu_utilization':
56 response
= yaml
.load(cpu_utilization
, Loader
=yaml
.Loader
)
57 elif metric
== 'users':
58 response
= yaml
.load(users
, Loader
=yaml
.Loader
)
59 elif metric
== 'load':
60 response
= yaml
.load(load
, Loader
=yaml
.Loader
)
62 response
= yaml
.load(empty
, Loader
=yaml
.Loader
)
63 mock_res
.get(endpoint
, payload
=response
)
65 def test_get_vnf_metric_list(self
):
66 with self
.subTest("Test case1 failed in test_get_vnf_metric_list"):
67 metric_list
= self
.pmjobs_topic
._get
_vnf
_metric
_list
(self
.nsr_id
)
68 self
.assertCountEqual(metric_list
, self
.metric_check_list
,
69 "VNF metric list is not correctly fetched")
70 with self
.subTest("Test case2 failed in test_get_vnf_metric_list"):
71 wrong_ns_id
= "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa"
72 with self
.assertRaises(EngineException
, msg
="ns not found") as e
:
73 self
.pmjobs_topic
._get
_vnf
_metric
_list
(wrong_ns_id
)
74 self
.assertEqual(e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code")
75 self
.assertIn("NS not found with id {}".format(wrong_ns_id
),
76 str(e
.exception
), "Wrong exception text")
78 # async def test_prom_metric_request(self):
79 # with self.subTest("Test case1 failed in test_prom"):
80 # prom_response = yaml.load(prom_res, Loader=yaml.Loader)
81 # with aioresponses() as mock_res:
82 # self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list)
83 # result = await self.pmjobs_topic._prom_metric_request(self.nsr_id, self.metric_check_list)
84 # self.assertCountEqual(result, prom_response, "Metric Data is valid")
85 # with self.subTest("Test case2 failed in test_prom"):
86 # with self.assertRaises(EngineException, msg="Prometheus not reachable") as e:
87 # await self.pmjobs_topic._prom_metric_request(self.nsr_id, self.metric_check_list)
88 # self.assertIn("Connection to ", str(e.exception), "Wrong exception text")
91 with self
.subTest("Test case1 failed in test_show"):
92 show_response
= yaml
.load(show_res
, Loader
=yaml
.Loader
)
93 with
aioresponses() as mock_res
:
94 self
.set_get_mock_res(mock_res
, self
.nsr_id
, self
.metric_check_list
)
95 result
= self
.pmjobs_topic
.show(self
.session
, self
.nsr_id
)
96 self
.assertEqual(len(result
['entries']), 3, "Number of metrics returned")
97 self
.assertCountEqual(result
, show_response
, "Response is valid")
98 with self
.subTest("Test case2 failed in test_show"):
99 wrong_ns_id
= "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa"
100 with
aioresponses() as mock_res
:
101 self
.set_get_mock_res(mock_res
, wrong_ns_id
, self
.metric_check_list
)
102 with self
.assertRaises(EngineException
, msg
="ns not found") as e
:
103 self
.pmjobs_topic
.show(self
.session
, wrong_ns_id
)
104 self
.assertEqual(e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code")
105 self
.assertIn("NS not found with id {}".format(wrong_ns_id
), str(e
.exception
),
106 "Wrong exception text")