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 (
32 from osm_nbi
.tests
.pmjob_mocks
.response
import (
42 class PmJobsTopicTest(asynctest
.TestCase
):
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
= [
56 "average_memory_utilization",
69 "project_id": project_id
,
74 "allow_show_user_project_role": True,
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
)
89 response
= yaml
.safe_load(empty
)
90 mock_res
.get(endpoint
, payload
=response
)
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
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"
105 await self
.pmjobs_topic
._prom
_metric
_request
(
106 self
.nsr_id
, self
.metric_check_list
108 self
.assertIn("Connection to ", str(e
.exception
), "Wrong exception text")
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
)
125 e
.exception
.http_code
,
126 HTTPStatus
.NOT_FOUND
,
127 "Wrong HTTP status code",
130 "NS not found with id {}".format(wrong_ns_id
),
132 "Wrong exception text",