2 # -*- coding: utf-8 -*-
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
17 __author__
= "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
18 __date__
= "2020-06-17"
20 from copy
import deepcopy
22 from unittest
import TestCase
23 from unittest
.mock
import patch
, Mock
24 from osm_nbi
.base_topic
import (
27 NBIBadArgumentsException
,
28 detect_descriptor_usage
,
29 update_descriptor_usage_state
,
31 from osm_common
import dbbase
32 from osm_nbi
.tests
.test_pkg_descriptors
import db_vnfds_text
, db_nsds_text
35 db_vnfd_content
= yaml
.safe_load(db_vnfds_text
)[0]
36 db_nsd_content
= yaml
.safe_load(db_nsds_text
)[0]
39 class Test_BaseTopic(TestCase
):
42 cls
.test_name
= "test-base-topic"
45 def tearDownClass(cls
):
49 self
.db
= Mock(dbbase
.DbBase())
51 def test_update_input_with_kwargs(self
):
53 # (descriptor content, kwargs, expected descriptor (None=fails), message)
55 {"a": {"none": None}},
57 {"a": {"none": None, "b": {"num": "v"}}},
61 {"a": {"none": None}},
63 {"a": {"none": {"num": "v"}}},
64 "create dict over none",
67 {"a": {"b": {"num": 4}}},
69 {"a": {"b": {"num": "v"}}},
73 {"a": {"b": {"num": 4}}},
75 {"a": {"b": {"num": {"c": {"d": "v"}}}}},
76 "create dict over number",
79 {"a": {"b": {"num": 4}}},
82 "replace dict with a string",
85 {"a": {"b": {"num": 4}}},
88 "replace dict with None",
91 {"a": [{"b": {"num": 4}}]},
94 "create dict over list should fail",
97 {"a": [{"b": {"num": 4}}]},
99 {"a": [{"b": {"num": "v"}}]},
103 {"a": [{"b": {"num": 4}}]},
105 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
108 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
112 {"a": [[4, None, "v"]]},
113 "expand nested list",
118 {"a": [[4], None, {"2": "v"}]},
119 "expand list and add number key",
121 ({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
123 for desc
, kwargs
, expected
, message
in test_set
:
126 EngineException
, BaseTopic
._update
_input
_with
_kwargs
, desc
, kwargs
129 BaseTopic
._update
_input
_with
_kwargs
(desc
, kwargs
)
130 self
.assertEqual(desc
, expected
, message
)
132 def test_detect_descriptor_usage_empty_descriptor(self
):
134 db_collection
= "vnfds"
135 with self
.assertRaises(EngineException
) as error
:
136 detect_descriptor_usage(descriptor
, db_collection
, self
.db
)
138 "Argument is mandatory and can not be empty, Bad arguments: descriptor",
140 "Error message is wrong.",
142 self
.db
.get_list
.assert_not_called()
144 def test_detect_descriptor_usage_empty_db_argument(self
):
145 descriptor
= deepcopy(db_vnfd_content
)
146 db_collection
= "vnfds"
148 with self
.assertRaises(EngineException
) as error
:
149 detect_descriptor_usage(descriptor
, db_collection
, db
)
151 "A valid DB object should be provided, Bad arguments: db",
153 "Error message is wrong.",
155 self
.db
.get_list
.assert_not_called()
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
)]
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"]}
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"
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"]})
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()
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
)
193 "db_collection should be equal to vnfds or nsds, db_collection",
198 self
.db
.get_list
.assert_not_called()
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
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
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
)
230 "db_collection should be equal to vnfds or nsds, db_collection",
234 self
.db
.set_one
.assert_not_called()
237 if __name__
== "__main__":