return op_id on asynchronous delete
[osm/NBI.git] / osm_nbi / tests / test_instance_topics.py
1 #
2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
3 # not use this file except in compliance with the License. You may obtain
4 # a copy of the License at
5 #
6 # http://www.apache.org/licenses/LICENSE-2.0
7 #
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11 # License for the specific language governing permissions and limitations
12 # under the License.
13 #
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
16 ##
17
18 import unittest
19 from unittest.mock import Mock, mock_open # patch, MagicMock
20 from osm_common.dbbase import DbException
21 from osm_nbi.engine import EngineException
22 from osm_common.dbmemory import DbMemory
23 from osm_common.fsbase import FsBase
24 from osm_common.msgbase import MsgBase
25 from http import HTTPStatus
26 from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
27 from osm_nbi.tests.test_db_descriptors import db_vim_accounts_text, db_nsds_text, db_vnfds_text, db_nsrs_text,\
28 db_vnfrs_text
29 from copy import deepcopy
30 import yaml
31
32
33 class TestNsLcmOpTopic(unittest.TestCase):
34
35 def setUp(self):
36 self.db = DbMemory()
37 self.fs = Mock(FsBase())
38 self.fs.get_params.return_value = {"./fake/folder"}
39 self.fs.file_open = mock_open()
40 self.msg = Mock(MsgBase())
41 # create class
42 self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
43 self.nslcmop_topic.check_quota = Mock(return_value=None) # skip quota
44
45 self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
46 self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
47 self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
48 self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
49 self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
50 self.db.create = Mock(return_value="created_id")
51 self.db.set_one = Mock(return_value={"updated": 1})
52 self.nsd = self.db.get_list("nsds")[0]
53 self.nsd_id = self.nsd["_id"]
54 self.nsr = self.db.get_list("nsrs")[0]
55 self.nsr_id = self.nsr["_id"]
56 self.nsr_project = self.nsr["_admin"]["projects_read"][0]
57
58 self.vim = self.db.get_list("vim_accounts")[0]
59 self.vim_id = self.vim["_id"]
60
61 def test_create_instantiate(self):
62 session = {"force": False, "admin": False, "public": False, "project_id": [self.nsr_project], "method": "write"}
63 indata = {
64 "nsdId": self.nsd_id,
65 "nsInstanceId": self.nsr_id,
66 "nsName": "name",
67 "vimAccountId": self.vim_id,
68 "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
69 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}],
70 "vnf": [{"member-vnf-index": "1",
71 "vdu": [{"id": "dataVM", "interface": [{"name": "dataVM-eth0",
72 "ip-address": "10.11.12.13",
73 "floating-ip-required": True}]
74 }],
75 "internal-vld": [{"name": "internal", "vim-network-id": "vim-net-id"}]
76 }],
77 "lcmOperationType": "instantiate",
78
79 }
80 rollback = []
81 headers = {}
82
83 nslcmop_id, _ = self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers)
84
85 # check nslcmop is created at database
86 self.assertEqual(self.db.create.call_count, 1, "database create not called, or called more than once")
87 _call = self.db.create.call_args_list[0]
88 self.assertEqual(_call[0][0], "nslcmops", "must be create a nslcmops entry at database")
89
90 created_nslcmop = _call[0][1]
91 self.assertEqual(nslcmop_id, created_nslcmop["_id"], "mismatch between return id and database '_id'")
92 self.assertEqual(self.nsr_id, created_nslcmop["nsInstanceId"], "bad reference id from nslcmop to nsr")
93 self.assertTrue(created_nslcmop["_admin"].get("projects_read"),
94 "Database record must contain '_amdin.projects_read'")
95 self.assertIn("created", created_nslcmop["_admin"], "Database record must contain '_admin.created'")
96 self.assertTrue(created_nslcmop["lcmOperationType"] == "instantiate",
97 "Database record must contain 'lcmOperationType=instantiate'")
98
99 self.assertEqual(len(rollback), len(self.db.set_one.call_args_list) + 1,
100 "rollback mismatch with created/set items at database")
101
102 # test parameters with error
103 bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
104 test_set = (
105 ("nsr not found", {"nsInstanceId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
106 # TODO add "nsd"
107 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
108 ("bad member-vnf-index", {"vnf.0.member-vnf-index": "k"}, EngineException, HTTPStatus.BAD_REQUEST,
109 ("k",)),
110 )
111 for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
112 with self.assertRaises(expect_exc, msg=message) as e:
113 self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
114 if expect_code:
115 self.assertTrue(e.exception.http_code == expect_code)
116 if expect_text_list:
117 for expect_text in expect_text_list:
118 self.assertIn(expect_text, str(e.exception).lower(),
119 "Expected '{}' at exception text".format(expect_text))
120
121 def test_check_ns_operation_action(self):
122 nsrs = self.db.get_list("nsrs")[0]
123 session = {}
124
125 indata = {
126 "member_vnf_index": "1",
127 "vdu_id": None,
128 "primitive": "touch",
129 "primitive_params": {"filename": "file"}
130 }
131
132 self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
133 for k in indata:
134 indata_copy = indata.copy()
135 if k == "primitive_params":
136 continue
137 indata_copy[k] = "non_existing"
138 with self.assertRaises(EngineException) as exc_manager:
139 self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata_copy)
140 exc = exc_manager.exception
141 self.assertEqual(exc.http_code, HTTPStatus.BAD_REQUEST, "Engine exception bad http_code with {}".
142 format(indata_copy))
143
144
145 class TestNsrTopic(unittest.TestCase):
146
147 def setUp(self):
148 self.db = DbMemory()
149 self.fs = Mock(FsBase())
150 self.fs.get_params.return_value = {"./fake/folder"}
151 self.fs.file_open = mock_open()
152 self.msg = Mock(MsgBase())
153 # create class
154 self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
155 self.nsr_topic.check_quota = Mock(return_value=None) # skip quota
156
157 self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
158 self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
159 self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
160 self.db.create = Mock(return_value="created_id")
161 self.nsd = self.db.get_list("nsds")[0]
162 self.nsd_id = self.nsd["_id"]
163 self.nsd_project = self.nsd["_admin"]["projects_read"][0]
164
165 self.vim = self.db.get_list("vim_accounts")[0]
166 self.vim_id = self.vim["_id"]
167
168 def test_create(self):
169 session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
170 indata = {
171 "nsdId": self.nsd_id,
172 "nsName": "name",
173 "vimAccountId": self.vim_id,
174 "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
175 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}]
176 }
177 rollback = []
178 headers = {}
179
180 self.nsr_topic.new(rollback, session, indata=indata, kwargs=None, headers=headers)
181
182 # check vnfrs and nsrs created in whatever order
183 created_vnfrs = []
184 created_nsrs = []
185 nsr_id = None
186 for _call in self.db.create.call_args_list:
187 assert len(_call[0]) >= 2, "called db.create with few parameters"
188 created_item = _call[0][1]
189 if _call[0][0] == "vnfrs":
190 created_vnfrs.append(created_item)
191 self.assertIn("member-vnf-index-ref", created_item,
192 "Created item must contain member-vnf-index-ref section")
193 if nsr_id:
194 self.assertEqual(nsr_id, created_item["nsr-id-ref"], "bad reference id from vnfr to nsr")
195 else:
196 nsr_id = created_item["nsr-id-ref"]
197
198 elif _call[0][0] == "nsrs":
199 created_nsrs.append(created_item)
200 if nsr_id:
201 self.assertEqual(nsr_id, created_item["_id"], "bad reference id from vnfr to nsr")
202 else:
203 nsr_id = created_item["_id"]
204 else:
205 assert True, "created an unknown record {} at database".format(_call[0][0])
206
207 self.assertTrue(created_item["_admin"].get("projects_read"),
208 "Database record must contain '_amdin.projects_read'")
209 self.assertIn("created", created_item["_admin"], "Database record must contain '_admin.created'")
210 self.assertTrue(created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
211 "Database record must contain '_admin.nstate=NOT INSTANTIATE'")
212
213 self.assertEqual(len(created_vnfrs), len(self.nsd["constituent-vnfd"]),
214 "created a mismatch number of vnfr at database")
215 self.assertEqual(len(created_nsrs), 1, "Only one nsrs must be created at database")
216 self.assertEqual(len(rollback), len(created_vnfrs) + 1, "rollback mismatch with created items at database")
217
218 # test parameters with error
219 bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
220 test_set = (
221 # TODO add "nsd"
222 ("nsd not found", {"nsdId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
223 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
224 ("additional params not supply", {"additionalParamsForVnf.0.member-vnf-index": "k"}, EngineException,
225 HTTPStatus.BAD_REQUEST, None),
226 )
227 for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
228 with self.assertRaises(expect_exc, msg=message) as e:
229 self.nsr_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
230 if expect_code:
231 self.assertTrue(e.exception.http_code == expect_code)
232 if expect_text_list:
233 for expect_text in expect_text_list:
234 self.assertIn(expect_text, str(e.exception).lower(),
235 "Expected '{}' at exception text".format(expect_text))
236
237 def test_delete_ns(self):
238 self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
239 self.nsr = self.db.get_list("nsrs")[0]
240 self.nsr_id = self.nsr["_id"]
241 self.db_set_one = self.db.set_one
242 p_id = self.nsd_project
243 p_other = "other_p"
244
245 session = {"force": False, "admin": False, "public": None, "project_id": [p_id], "method": "delete"}
246 session2 = {"force": False, "admin": False, "public": None, "project_id": [p_other], "method": "delete"}
247 session_force = {"force": True, "admin": True, "public": None, "project_id": [], "method": "delete"}
248 with self.subTest(i=1, t='Normal Deletion'):
249 self.db.del_one = Mock()
250 self.db.set_one = Mock()
251 self.nsr_topic.delete(session, self.nsr_id)
252
253 db_args = self.db.del_one.call_args[0]
254 msg_args = self.msg.write.call_args[0]
255 self.assertEqual(msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic")
256 self.assertEqual(msg_args[1], "deleted", "Wrong message action")
257 self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
258 self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
259 self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
260 self.assertEqual(db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter")
261 self.db.set_one.assert_not_called()
262 fs_del_calls = self.fs.file_delete.call_args_list
263 self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
264 with self.subTest(i=2, t='No delete because referenced by other project'):
265 self.db_set_one("nsrs", {"_id": self.nsr_id}, update_dict=None, push={"_admin.projects_read": p_other,
266 "_admin.projects_write": p_other})
267 self.db.del_one.reset_mock()
268 self.db.set_one.reset_mock()
269 self.msg.write.reset_mock()
270 self.fs.file_delete.reset_mock()
271
272 self.nsr_topic.delete(session2, self.nsr_id)
273 self.db.del_one.assert_not_called()
274 self.msg.write.assert_not_called()
275 db_s1_args = self.db.set_one.call_args
276 self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic")
277 self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID")
278 self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
279 self.assertIn("_admin.projects_read." + p_other, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
280 self.assertIn("_admin.projects_write." + p_other, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
281 self.fs.file_delete.assert_not_called()
282 with self.subTest(i=4, t='Delete with force and admin'):
283 self.db.del_one.reset_mock()
284 self.db.set_one.reset_mock()
285 self.msg.write.reset_mock()
286 self.fs.file_delete.reset_mock()
287 self.nsr_topic.delete(session_force, self.nsr_id)
288
289 db_args = self.db.del_one.call_args[0]
290 msg_args = self.msg.write.call_args[0]
291 self.assertEqual(msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic")
292 self.assertEqual(msg_args[1], "deleted", "Wrong message action")
293 self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
294 self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
295 self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
296 self.db.set_one.assert_not_called()
297 fs_del_calls = self.fs.file_delete.call_args_list
298 self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
299 with self.subTest(i=3, t='Conflict on Delete - NS in INSTANTIATED state'):
300 self.db_set_one("nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"},
301 pull={"_admin.projects_read": p_other, "_admin.projects_write": p_other})
302 self.db.del_one.reset_mock()
303 self.db.set_one.reset_mock()
304 self.msg.write.reset_mock()
305 self.fs.file_delete.reset_mock()
306
307 with self.assertRaises(EngineException, msg="Accepted NSR with nsState INSTANTIATED") as e:
308 self.nsr_topic.delete(session, self.nsr_id)
309 self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
310 self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text")
311 # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
312
313 with self.subTest(i=4, t='Non-existent NS'):
314 self.db.del_one.reset_mock()
315 self.db.set_one.reset_mock()
316 self.msg.write.reset_mock()
317 self.fs.file_delete.reset_mock()
318 excp_msg = "Not found"
319 with self.assertRaises(DbException, msg="Accepted non-existent NSD ID") as e:
320 self.nsr_topic.delete(session2, "other_id")
321 self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
322 self.assertIn(excp_msg, str(e.exception), "Wrong exception text")
323 self.assertIn("other_id", str(e.exception), "Wrong exception text")
324 return