Updating python dependencies
[osm/NBI.git] / osm_nbi / tests / test_base_topic.py
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 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
18 __date__ = "2020-06-17"
19
20 from copy import deepcopy
21 import unittest
22 from unittest import TestCase
23 from unittest.mock import patch, Mock
24 from osm_nbi.base_topic import (
25 BaseTopic,
26 EngineException,
27 NBIBadArgumentsException,
28 detect_descriptor_usage,
29 update_descriptor_usage_state,
30 )
31 from osm_common import dbbase
32 from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
33 import yaml
34
35 db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
36 db_nsd_content = yaml.safe_load(db_nsds_text)[0]
37
38
39 class Test_BaseTopic(TestCase):
40 @classmethod
41 def setUpClass(cls):
42 cls.test_name = "test-base-topic"
43
44 @classmethod
45 def tearDownClass(cls):
46 pass
47
48 def setUp(self):
49 self.db = Mock(dbbase.DbBase())
50
51 def test_update_input_with_kwargs(self):
52 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 for desc, kwargs, expected, message in test_set:
124 if expected is None:
125 self.assertRaises(
126 EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs
127 )
128 else:
129 BaseTopic._update_input_with_kwargs(desc, kwargs)
130 self.assertEqual(desc, expected, message)
131
132 def test_detect_descriptor_usage_empty_descriptor(self):
133 descriptor = {}
134 db_collection = "vnfds"
135 with self.assertRaises(EngineException) as error:
136 detect_descriptor_usage(descriptor, db_collection, self.db)
137 self.assertIn(
138 "Argument is mandatory and can not be empty, Bad arguments: descriptor",
139 error,
140 "Error message is wrong.",
141 )
142 self.db.get_list.assert_not_called()
143
144 def test_detect_descriptor_usage_empty_db_argument(self):
145 descriptor = deepcopy(db_vnfd_content)
146 db_collection = "vnfds"
147 db = None
148 with self.assertRaises(EngineException) as error:
149 detect_descriptor_usage(descriptor, db_collection, db)
150 self.assertIn(
151 "A valid DB object should be provided, Bad arguments: db",
152 error,
153 "Error message is wrong.",
154 )
155 self.db.get_list.assert_not_called()
156
157 def test_detect_descriptor_usage_which_is_in_use(self):
158 descriptor = deepcopy(db_vnfd_content)
159 db_collection = "vnfds"
160 self.db.get_list.side_effect = [deepcopy(db_vnfd_content)]
161 expected = True
162 result = detect_descriptor_usage(descriptor, db_collection, self.db)
163 self.assertEqual(result, expected, "wrong result")
164 self.db.get_list.assert_called_once_with(
165 "vnfrs", {"vnfd-id": descriptor["_id"]}
166 )
167
168 def test_detect_descriptor_usage_which_is_not_in_use(self):
169 descriptor = deepcopy(db_nsd_content)
170 self.db.get_list.return_value = []
171 db_collection = "nsds"
172 expected = None
173 result = detect_descriptor_usage(descriptor, db_collection, self.db)
174 self.assertEqual(result, expected, "wrong result")
175 self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]})
176
177 def test_detect_descriptor_usage_wrong_desc_format(self):
178 descriptor = deepcopy(db_nsd_content)
179 descriptor.pop("_id")
180 db_collection = "nsds"
181 with self.assertRaises(EngineException) as error:
182 detect_descriptor_usage(descriptor, db_collection, self.db)
183 self.assertIn("KeyError", error, "wrong error type")
184 self.db.get_list.assert_not_called()
185
186 def test_detect_descriptor_usage_wrong_db_collection(self):
187 descriptor = deepcopy(db_vnfd_content)
188 descriptor.pop("_id")
189 db_collection = "vnf"
190 with self.assertRaises(EngineException) as error:
191 detect_descriptor_usage(descriptor, db_collection, self.db)
192 self.assertIn(
193 "db_collection should be equal to vnfds or nsds, db_collection",
194 error,
195 "wrong error type",
196 )
197
198 self.db.get_list.assert_not_called()
199
200 @patch("osm_nbi.base_topic.detect_descriptor_usage")
201 def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage):
202 db_collection = "vnfds"
203 descriptor = deepcopy(db_vnfd_content)
204 mock_descriptor_usage.return_value = True
205 descriptor_update = {"_admin.usageState": "IN_USE"}
206 update_descriptor_usage_state(descriptor, db_collection, self.db)
207 self.db.set_one.assert_called_once_with(
208 db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
209 )
210
211 @patch("osm_nbi.base_topic.detect_descriptor_usage")
212 def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage):
213 db_collection = "nsds"
214 descriptor = deepcopy(db_nsd_content)
215 mock_descriptor_usage.return_value = False
216 descriptor_update = {"_admin.usageState": "NOT_IN_USE"}
217 update_descriptor_usage_state(descriptor, db_collection, self.db)
218 self.db.set_one.assert_called_once_with(
219 db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
220 )
221
222 @patch("osm_nbi.base_topic.detect_descriptor_usage")
223 def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage):
224 db_collection = "nsd"
225 descriptor = deepcopy(db_nsd_content)
226 mock_descriptor_usage.side_effect = NBIBadArgumentsException
227 with self.assertRaises(EngineException) as error:
228 update_descriptor_usage_state(descriptor, db_collection, self.db)
229 self.assertIn(
230 "db_collection should be equal to vnfds or nsds, db_collection",
231 error,
232 "wrong error type",
233 )
234 self.db.set_one.assert_not_called()
235
236
237 if __name__ == "__main__":
238 unittest.main()