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
6 # http://www.apache.org/licenses/LICENSE-2.0
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
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
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
,\
29 from copy
import deepcopy
33 class TestNsLcmOpTopic(unittest
.TestCase
):
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())
42 self
.nslcmop_topic
= NsLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
43 self
.nslcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
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]
58 self
.vim
= self
.db
.get_list("vim_accounts")[0]
59 self
.vim_id
= self
.vim
["_id"]
61 def test_create_instantiate(self
):
62 session
= {"force": False, "admin": False, "public": False, "project_id": [self
.nsr_project
], "method": "write"}
65 "nsInstanceId": self
.nsr_id
,
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}]
75 "internal-vld": [{"name": "internal", "vim-network-id": "vim-net-id"}]
77 "lcmOperationType": "instantiate",
83 nslcmop_id
, _
= self
.nslcmop_topic
.new(rollback
, session
, indata
=deepcopy(indata
), kwargs
=None, headers
=headers
)
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")
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'")
99 self
.assertEqual(len(rollback
), len(self
.db
.set_one
.call_args_list
) + 1,
100 "rollback mismatch with created/set items at database")
102 # test parameters with error
103 bad_id
= "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
105 ("nsr not found", {"nsInstanceId": bad_id
}, DbException
, HTTPStatus
.NOT_FOUND
, ("not found", bad_id
)),
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
,
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
)
115 self
.assertTrue(e
.exception
.http_code
== expect_code
)
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
))
121 def test_check_ns_operation_action(self
):
122 nsrs
= self
.db
.get_list("nsrs")[0]
126 "member_vnf_index": "1",
128 "primitive": "touch",
129 "primitive_params": {"filename": "file"}
132 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "action", indata
)
134 indata_copy
= indata
.copy()
135 if k
== "primitive_params":
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 {}".
145 class TestNsrTopic(unittest
.TestCase
):
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())
154 self
.nsr_topic
= NsrTopic(self
.db
, self
.fs
, self
.msg
, None)
155 self
.nsr_topic
.check_quota
= Mock(return_value
=None) # skip quota
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]
165 self
.vim
= self
.db
.get_list("vim_accounts")[0]
166 self
.vim_id
= self
.vim
["_id"]
168 def test_create(self
):
169 session
= {"force": False, "admin": False, "public": False, "project_id": [self
.nsd_project
], "method": "write"}
171 "nsdId": self
.nsd_id
,
173 "vimAccountId": self
.vim_id
,
174 "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
175 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}]
180 self
.nsr_topic
.new(rollback
, session
, indata
=indata
, kwargs
=None, headers
=headers
)
182 # check vnfrs and nsrs created in whatever order
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")
194 self
.assertEqual(nsr_id
, created_item
["nsr-id-ref"], "bad reference id from vnfr to nsr")
196 nsr_id
= created_item
["nsr-id-ref"]
198 elif _call
[0][0] == "nsrs":
199 created_nsrs
.append(created_item
)
201 self
.assertEqual(nsr_id
, created_item
["_id"], "bad reference id from vnfr to nsr")
203 nsr_id
= created_item
["_id"]
205 assert True, "created an unknown record {} at database".format(_call
[0][0])
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'")
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")
218 # test parameters with error
219 bad_id
= "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
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),
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
)
231 self
.assertTrue(e
.exception
.http_code
== expect_code
)
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
))
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
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
)
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()
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
.assertEqual(db_s1_args
[1]["pull_list"],
280 {"_admin.projects_read": [p_other
], "_admin.projects_write": [p_other
]},
281 "Wrong DB pull_list dictionary")
282 self
.fs
.file_delete
.assert_not_called()
283 with self
.subTest(i
=4, t
='Delete with force and admin'):
284 self
.db
.del_one
.reset_mock()
285 self
.db
.set_one
.reset_mock()
286 self
.msg
.write
.reset_mock()
287 self
.fs
.file_delete
.reset_mock()
288 self
.nsr_topic
.delete(session_force
, self
.nsr_id
)
290 db_args
= self
.db
.del_one
.call_args
[0]
291 msg_args
= self
.msg
.write
.call_args
[0]
292 self
.assertEqual(msg_args
[0], self
.nsr_topic
.topic_msg
, "Wrong message topic")
293 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
294 self
.assertEqual(msg_args
[2], {"_id": self
.nsr_id
}, "Wrong message content")
295 self
.assertEqual(db_args
[0], self
.nsr_topic
.topic
, "Wrong DB topic")
296 self
.assertEqual(db_args
[1]["_id"], self
.nsr_id
, "Wrong DB ID")
297 self
.db
.set_one
.assert_not_called()
298 fs_del_calls
= self
.fs
.file_delete
.call_args_list
299 self
.assertEqual(fs_del_calls
[0][0][0], self
.nsr_id
, "Wrong FS file id")
300 with self
.subTest(i
=3, t
='Conflict on Delete - NS in INSTANTIATED state'):
301 self
.db_set_one("nsrs", {"_id": self
.nsr_id
}, {"_admin.nsState": "INSTANTIATED"},
302 pull
={"_admin.projects_read": p_other
, "_admin.projects_write": p_other
})
303 self
.db
.del_one
.reset_mock()
304 self
.db
.set_one
.reset_mock()
305 self
.msg
.write
.reset_mock()
306 self
.fs
.file_delete
.reset_mock()
308 with self
.assertRaises(EngineException
, msg
="Accepted NSR with nsState INSTANTIATED") as e
:
309 self
.nsr_topic
.delete(session
, self
.nsr_id
)
310 self
.assertEqual(e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code")
311 self
.assertIn("INSTANTIATED", str(e
.exception
), "Wrong exception text")
312 # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
314 with self
.subTest(i
=4, t
='Non-existent NS'):
315 self
.db
.del_one
.reset_mock()
316 self
.db
.set_one
.reset_mock()
317 self
.msg
.write
.reset_mock()
318 self
.fs
.file_delete
.reset_mock()
319 excp_msg
= "Not found"
320 with self
.assertRaises(DbException
, msg
="Accepted non-existent NSD ID") as e
:
321 self
.nsr_topic
.delete(session2
, "other_id")
322 self
.assertEqual(e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code")
323 self
.assertIn(excp_msg
, str(e
.exception
), "Wrong exception text")
324 self
.assertIn("other_id", str(e
.exception
), "Wrong exception text")