1 |
|
#! /usr/bin/python3 |
2 |
|
# -*- coding: utf-8 -*- |
3 |
|
|
4 |
|
# Licensed under the Apache License, Version 2.0 (the "License"); |
5 |
|
# you may not use this file except in compliance with the License. |
6 |
|
# You may obtain a copy of the License at |
7 |
|
# |
8 |
|
# http://www.apache.org/licenses/LICENSE-2.0 |
9 |
|
# |
10 |
|
# Unless required by applicable law or agreed to in writing, software |
11 |
|
# distributed under the License is distributed on an "AS IS" BASIS, |
12 |
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or |
13 |
|
# implied. |
14 |
|
# See the License for the specific language governing permissions and |
15 |
|
# limitations under the License. |
16 |
|
|
17 |
1 |
__author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com" |
18 |
1 |
__date__ = "2020-06-17" |
19 |
|
|
20 |
1 |
from copy import deepcopy |
21 |
1 |
import unittest |
22 |
1 |
from unittest import TestCase |
23 |
1 |
from unittest.mock import patch, Mock |
24 |
1 |
from osm_nbi.base_topic import ( |
25 |
|
BaseTopic, |
26 |
|
EngineException, |
27 |
|
NBIBadArgumentsException, |
28 |
|
detect_descriptor_usage, |
29 |
|
update_descriptor_usage_state, |
30 |
|
) |
31 |
1 |
from osm_common import dbbase |
32 |
1 |
from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text |
33 |
1 |
import yaml |
34 |
|
|
35 |
1 |
db_vnfd_content = yaml.safe_load(db_vnfds_text)[0] |
36 |
1 |
db_nsd_content = yaml.safe_load(db_nsds_text)[0] |
37 |
|
|
38 |
|
|
39 |
1 |
class Test_BaseTopic(TestCase): |
40 |
1 |
@classmethod |
41 |
1 |
def setUpClass(cls): |
42 |
1 |
cls.test_name = "test-base-topic" |
43 |
|
|
44 |
1 |
@classmethod |
45 |
1 |
def tearDownClass(cls): |
46 |
1 |
pass |
47 |
|
|
48 |
1 |
def setUp(self): |
49 |
1 |
self.db = Mock(dbbase.DbBase()) |
50 |
|
|
51 |
1 |
def test_update_input_with_kwargs(self): |
52 |
1 |
test_set = ( |
53 |
|
# (descriptor content, kwargs, expected descriptor (None=fails), message) |
54 |
|
( |
55 |
|
{"a": {"none": None}}, |
56 |
|
{"a.b.num": "v"}, |
57 |
|
{"a": {"none": None, "b": {"num": "v"}}}, |
58 |
|
"create dict", |
59 |
|
), |
60 |
|
( |
61 |
|
{"a": {"none": None}}, |
62 |
|
{"a.none.num": "v"}, |
63 |
|
{"a": {"none": {"num": "v"}}}, |
64 |
|
"create dict over none", |
65 |
|
), |
66 |
|
( |
67 |
|
{"a": {"b": {"num": 4}}}, |
68 |
|
{"a.b.num": "v"}, |
69 |
|
{"a": {"b": {"num": "v"}}}, |
70 |
|
"replace_number", |
71 |
|
), |
72 |
|
( |
73 |
|
{"a": {"b": {"num": 4}}}, |
74 |
|
{"a.b.num.c.d": "v"}, |
75 |
|
{"a": {"b": {"num": {"c": {"d": "v"}}}}}, |
76 |
|
"create dict over number", |
77 |
|
), |
78 |
|
( |
79 |
|
{"a": {"b": {"num": 4}}}, |
80 |
|
{"a.b": "v"}, |
81 |
|
{"a": {"b": "v"}}, |
82 |
|
"replace dict with a string", |
83 |
|
), |
84 |
|
( |
85 |
|
{"a": {"b": {"num": 4}}}, |
86 |
|
{"a.b": None}, |
87 |
|
{"a": {}}, |
88 |
|
"replace dict with None", |
89 |
|
), |
90 |
|
( |
91 |
|
{"a": [{"b": {"num": 4}}]}, |
92 |
|
{"a.b.num": "v"}, |
93 |
|
None, |
94 |
|
"create dict over list should fail", |
95 |
|
), |
96 |
|
( |
97 |
|
{"a": [{"b": {"num": 4}}]}, |
98 |
|
{"a.0.b.num": "v"}, |
99 |
|
{"a": [{"b": {"num": "v"}}]}, |
100 |
|
"set list", |
101 |
|
), |
102 |
|
( |
103 |
|
{"a": [{"b": {"num": 4}}]}, |
104 |
|
{"a.3.b.num": "v"}, |
105 |
|
{"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, |
106 |
|
"expand list", |
107 |
|
), |
108 |
|
({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"), |
109 |
|
( |
110 |
|
{"a": [[4]]}, |
111 |
|
{"a.0.2": "v"}, |
112 |
|
{"a": [[4, None, "v"]]}, |
113 |
|
"expand nested list", |
114 |
|
), |
115 |
|
( |
116 |
|
{"a": [[4]]}, |
117 |
|
{"a.2.2": "v"}, |
118 |
|
{"a": [[4], None, {"2": "v"}]}, |
119 |
|
"expand list and add number key", |
120 |
|
), |
121 |
|
({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"), |
122 |
|
) |
123 |
1 |
for desc, kwargs, expected, message in test_set: |
124 |
1 |
if expected is None: |
125 |
1 |
self.assertRaises( |
126 |
|
EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs |
127 |
|
) |
128 |
|
else: |
129 |
1 |
BaseTopic._update_input_with_kwargs(desc, kwargs) |
130 |
1 |
self.assertEqual(desc, expected, message) |
131 |
|
|
132 |
1 |
def test_detect_descriptor_usage_empty_descriptor(self): |
133 |
1 |
descriptor = {} |
134 |
1 |
db_collection = "vnfds" |
135 |
1 |
with self.assertRaises(EngineException) as error: |
136 |
1 |
detect_descriptor_usage(descriptor, db_collection, self.db) |
137 |
0 |
self.assertIn( |
138 |
|
"Argument is mandatory and can not be empty, Bad arguments: descriptor", |
139 |
|
error, |
140 |
|
"Error message is wrong.", |
141 |
|
) |
142 |
1 |
self.db.get_list.assert_not_called() |
143 |
|
|
144 |
1 |
def test_detect_descriptor_usage_empty_db_argument(self): |
145 |
1 |
descriptor = deepcopy(db_vnfd_content) |
146 |
1 |
db_collection = "vnfds" |
147 |
1 |
db = None |
148 |
1 |
with self.assertRaises(EngineException) as error: |
149 |
1 |
detect_descriptor_usage(descriptor, db_collection, db) |
150 |
0 |
self.assertIn( |
151 |
|
"A valid DB object should be provided, Bad arguments: db", |
152 |
|
error, |
153 |
|
"Error message is wrong.", |
154 |
|
) |
155 |
1 |
self.db.get_list.assert_not_called() |
156 |
|
|
157 |
1 |
def test_detect_descriptor_usage_which_is_in_use(self): |
158 |
1 |
descriptor = deepcopy(db_vnfd_content) |
159 |
1 |
db_collection = "vnfds" |
160 |
1 |
self.db.get_list.side_effect = [deepcopy(db_vnfd_content)] |
161 |
1 |
expected = True |
162 |
1 |
result = detect_descriptor_usage(descriptor, db_collection, self.db) |
163 |
1 |
self.assertEqual(result, expected, "wrong result") |
164 |
1 |
self.db.get_list.assert_called_once_with( |
165 |
|
"vnfrs", {"vnfd-id": descriptor["_id"]} |
166 |
|
) |
167 |
|
|
168 |
1 |
def test_detect_descriptor_usage_which_is_not_in_use(self): |
169 |
1 |
descriptor = deepcopy(db_nsd_content) |
170 |
1 |
self.db.get_list.return_value = [] |
171 |
1 |
db_collection = "nsds" |
172 |
1 |
expected = None |
173 |
1 |
result = detect_descriptor_usage(descriptor, db_collection, self.db) |
174 |
1 |
self.assertEqual(result, expected, "wrong result") |
175 |
1 |
self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]}) |
176 |
|
|
177 |
1 |
def test_detect_descriptor_usage_wrong_desc_format(self): |
178 |
1 |
descriptor = deepcopy(db_nsd_content) |
179 |
1 |
descriptor.pop("_id") |
180 |
1 |
db_collection = "nsds" |
181 |
1 |
with self.assertRaises(EngineException) as error: |
182 |
1 |
detect_descriptor_usage(descriptor, db_collection, self.db) |
183 |
0 |
self.assertIn("KeyError", error, "wrong error type") |
184 |
1 |
self.db.get_list.assert_not_called() |
185 |
|
|
186 |
1 |
def test_detect_descriptor_usage_wrong_db_collection(self): |
187 |
1 |
descriptor = deepcopy(db_vnfd_content) |
188 |
1 |
descriptor.pop("_id") |
189 |
1 |
db_collection = "vnf" |
190 |
1 |
with self.assertRaises(EngineException) as error: |
191 |
1 |
detect_descriptor_usage(descriptor, db_collection, self.db) |
192 |
0 |
self.assertIn( |
193 |
|
"db_collection should be equal to vnfds or nsds, db_collection", |
194 |
|
error, |
195 |
|
"wrong error type", |
196 |
|
) |
197 |
|
|
198 |
1 |
self.db.get_list.assert_not_called() |
199 |
|
|
200 |
1 |
@patch("osm_nbi.base_topic.detect_descriptor_usage") |
201 |
1 |
def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage): |
202 |
1 |
db_collection = "vnfds" |
203 |
1 |
descriptor = deepcopy(db_vnfd_content) |
204 |
1 |
mock_descriptor_usage.return_value = True |
205 |
1 |
descriptor_update = {"_admin.usageState": "IN_USE"} |
206 |
1 |
update_descriptor_usage_state(descriptor, db_collection, self.db) |
207 |
1 |
self.db.set_one.assert_called_once_with( |
208 |
|
db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update |
209 |
|
) |
210 |
|
|
211 |
1 |
@patch("osm_nbi.base_topic.detect_descriptor_usage") |
212 |
1 |
def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage): |
213 |
1 |
db_collection = "nsds" |
214 |
1 |
descriptor = deepcopy(db_nsd_content) |
215 |
1 |
mock_descriptor_usage.return_value = False |
216 |
1 |
descriptor_update = {"_admin.usageState": "NOT_IN_USE"} |
217 |
1 |
update_descriptor_usage_state(descriptor, db_collection, self.db) |
218 |
1 |
self.db.set_one.assert_called_once_with( |
219 |
|
db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update |
220 |
|
) |
221 |
|
|
222 |
1 |
@patch("osm_nbi.base_topic.detect_descriptor_usage") |
223 |
1 |
def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage): |
224 |
1 |
db_collection = "nsd" |
225 |
1 |
descriptor = deepcopy(db_nsd_content) |
226 |
1 |
mock_descriptor_usage.side_effect = NBIBadArgumentsException |
227 |
1 |
with self.assertRaises(EngineException) as error: |
228 |
1 |
update_descriptor_usage_state(descriptor, db_collection, self.db) |
229 |
0 |
self.assertIn( |
230 |
|
"db_collection should be equal to vnfds or nsds, db_collection", |
231 |
|
error, |
232 |
|
"wrong error type", |
233 |
|
) |
234 |
1 |
self.db.set_one.assert_not_called() |
235 |
|
|
236 |
|
|
237 |
1 |
if __name__ == "__main__": |
238 |
0 |
unittest.main() |