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
, prom_res
, cpu_utilization
, users
, load
, empty
30 class PmJobsTopicTest(asynctest
.TestCase
):
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}
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
)
61 response
= yaml
.load(empty
, Loader
=yaml
.Loader
)
62 mock_res
.get(endpoint
, payload
=response
)
64 async def test_prom_metric_request(self
):
65 with self
.subTest("Test case1 failed in test_prom"):
66 prom_response
= yaml
.load(prom_res
, Loader
=yaml
.Loader
)
67 with
aioresponses() as mock_res
:
68 self
.set_get_mock_res(mock_res
, self
.nsr_id
, self
.metric_check_list
)
69 result
= await self
.pmjobs_topic
._prom
_metric
_request
(self
.nsr_id
, self
.metric_check_list
)
70 self
.assertCountEqual(result
, prom_response
, "Metric Data is valid")
71 with self
.subTest("Test case2 failed in test_prom"):
72 with self
.assertRaises(EngineException
, msg
="Prometheus not reachable") as e
:
73 await self
.pmjobs_topic
._prom
_metric
_request
(self
.nsr_id
, self
.metric_check_list
)
74 self
.assertIn("Connection to ", str(e
.exception
), "Wrong exception text")
77 with self
.subTest("Test case1 failed in test_show"):
78 show_response
= yaml
.load(show_res
, Loader
=yaml
.Loader
)
79 with
aioresponses() as mock_res
:
80 self
.set_get_mock_res(mock_res
, self
.nsr_id
, self
.metric_check_list
)
81 result
= self
.pmjobs_topic
.show(self
.session
, self
.nsr_id
)
82 self
.assertEqual(len(result
['entries']), 1, "Number of metrics returned")
83 self
.assertCountEqual(result
, show_response
, "Response is valid")
84 with self
.subTest("Test case2 failed in test_show"):
85 wrong_ns_id
= "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa"
86 with
aioresponses() as mock_res
:
87 self
.set_get_mock_res(mock_res
, wrong_ns_id
, self
.metric_check_list
)
88 with self
.assertRaises(EngineException
, msg
="ns not found") as e
:
89 self
.pmjobs_topic
.show(self
.session
, wrong_ns_id
)
90 self
.assertEqual(e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code")
91 self
.assertIn("NS not found with id {}".format(wrong_ns_id
), str(e
.exception
),
92 "Wrong exception text")