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 |
1 |
__author__ = "Preethika P,preethika.p@tataelxsi.co.in" |
17 |
|
|
18 |
1 |
import asynctest |
19 |
1 |
import yaml |
20 |
1 |
import re |
21 |
1 |
from aioresponses import aioresponses |
22 |
1 |
from http import HTTPStatus |
23 |
1 |
from osm_nbi.engine import EngineException |
24 |
1 |
from osm_common.dbmemory import DbMemory |
25 |
1 |
from osm_nbi.pmjobs_topics import PmJobsTopic |
26 |
1 |
from osm_nbi.tests.test_db_descriptors import ( |
27 |
|
db_nsds_text, |
28 |
|
db_vnfds_text, |
29 |
|
db_nsrs_text, |
30 |
|
db_vnfrs_text, |
31 |
|
) |
32 |
1 |
from 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 |
|
|
42 |
1 |
class PmJobsTopicTest(asynctest.TestCase): |
43 |
1 |
def setUp(self): |
44 |
1 |
self.db = DbMemory() |
45 |
1 |
self.pmjobs_topic = PmJobsTopic(self.db, host="prometheus", port=9091) |
46 |
1 |
self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) |
47 |
1 |
self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) |
48 |
1 |
self.db.create_list("vnfrs", yaml.safe_load(db_vnfrs_text)) |
49 |
1 |
self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) |
50 |
1 |
self.nsr = self.db.get_list("nsrs")[0] |
51 |
1 |
self.nsr_id = self.nsr["_id"] |
52 |
1 |
project_id = self.nsr["_admin"]["projects_write"] |
53 |
1 |
"""metric_check_list contains the vnf metric name used in descriptor i.e users,load""" |
54 |
1 |
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 |
1 |
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 |
1 |
def set_get_mock_res(self, mock_res, ns_id, metric_list): |
78 |
1 |
site = "http://prometheus:9091/api/v1/query?query=osm_metric_name{ns_id='nsr'}" |
79 |
1 |
site = re.sub(r"nsr", ns_id, site) |
80 |
1 |
for metric in metric_list: |
81 |
1 |
endpoint = re.sub(r"metric_name", metric, site) |
82 |
1 |
if metric == "cpu_utilization": |
83 |
1 |
response = yaml.safe_load(cpu_utilization) |
84 |
1 |
elif metric == "users": |
85 |
1 |
response = yaml.safe_load(users) |
86 |
1 |
elif metric == "load": |
87 |
1 |
response = yaml.safe_load(load) |
88 |
|
else: |
89 |
1 |
response = yaml.safe_load(empty) |
90 |
1 |
mock_res.get(endpoint, payload=response) |
91 |
|
|
92 |
1 |
async def test_prom_metric_request(self): |
93 |
1 |
with self.subTest("Test case1 failed in test_prom"): |
94 |
1 |
prom_response = yaml.safe_load(prom_res) |
95 |
1 |
with aioresponses() as mock_res: |
96 |
1 |
self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list) |
97 |
1 |
result = await self.pmjobs_topic._prom_metric_request( |
98 |
|
self.nsr_id, self.metric_check_list |
99 |
|
) |
100 |
1 |
self.assertCountEqual(result, prom_response, "Metric Data is valid") |
101 |
1 |
with self.subTest("Test case2 failed in test_prom"): |
102 |
1 |
with self.assertRaises( |
103 |
|
EngineException, msg="Prometheus not reachable" |
104 |
|
) as e: |
105 |
1 |
await self.pmjobs_topic._prom_metric_request( |
106 |
|
self.nsr_id, self.metric_check_list |
107 |
|
) |
108 |
1 |
self.assertIn("Connection to ", str(e.exception), "Wrong exception text") |
109 |
|
|
110 |
1 |
def test_show(self): |
111 |
1 |
with self.subTest("Test case1 failed in test_show"): |
112 |
1 |
show_response = yaml.safe_load(show_res) |
113 |
1 |
with aioresponses() as mock_res: |
114 |
1 |
self.set_get_mock_res(mock_res, self.nsr_id, self.metric_check_list) |
115 |
1 |
result = self.pmjobs_topic.show(self.session, self.nsr_id) |
116 |
1 |
self.assertEqual(len(result["entries"]), 1, "Number of metrics returned") |
117 |
1 |
self.assertCountEqual(result, show_response, "Response is valid") |
118 |
1 |
with self.subTest("Test case2 failed in test_show"): |
119 |
1 |
wrong_ns_id = "88d90b0c-faff-4bbc-cccc-aaaaaaaaaaaa" |
120 |
1 |
with aioresponses() as mock_res: |
121 |
1 |
self.set_get_mock_res(mock_res, wrong_ns_id, self.metric_check_list) |
122 |
1 |
with self.assertRaises(EngineException, msg="ns not found") as e: |
123 |
1 |
self.pmjobs_topic.show(self.session, wrong_ns_id) |
124 |
1 |
self.assertEqual( |
125 |
|
e.exception.http_code, |
126 |
|
HTTPStatus.NOT_FOUND, |
127 |
|
"Wrong HTTP status code", |
128 |
|
) |
129 |
1 |
self.assertIn( |
130 |
|
"NS not found with id {}".format(wrong_ns_id), |
131 |
|
str(e.exception), |
132 |
|
"Wrong exception text", |
133 |
|
) |