1 # Copyright 2016 ZTE Corporation.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from rest_framework import status
17 from django.test import TestCase
18 from django.test import Client
20 from lcm.pub.utils import restcall
21 from lcm.pub.database.models import NSDModel, NSInstModel, NfPackageModel
24 class TestNsPackage(TestCase):
26 self.client = Client()
27 NSDModel.objects.filter().delete()
28 NSInstModel.objects.filter().delete()
29 NfPackageModel.objects.filter().delete()
38 "csarProvider": "ZTE",
40 "invariant_id": "VBRAS_NS_NO_SFC",
41 "id": "VBRAS_NS_ZTE_1.0",
42 "description": "VBRAS_ZTE_NS"
46 "id": "VBras_yfye7lsgi73p8j4p2a6vbguzd",
47 "type_name": "tosca.nodes.nfv.ext.zte.VNF.VBras",
48 "template_name": "VBras",
51 "type_name": "string",
55 "type_name": "string",
59 "type_name": "string",
63 "type_name": "string",
67 "type_name": "string",
71 "type_name": "string",
72 "value": "zte_vbras_1.0"
84 "csarProvider": "ZTE",
86 "invariant_id": "VBRAS_NS_NO_SFC",
87 "id": "VBRAS_NS_ZTE_1.0",
88 "description": "VBRAS_ZTE_NS"
93 "type_name": "tosca.nodes.nfv.ext.zte.VNF.VBras",
94 "default_instances": 1,
98 "type_name": "string",
102 "type_name": "string",
106 "type_name": "string",
110 "type_name": "string",
114 "type_name": "string",
118 "type_name": "string",
119 "value": "zte_vbras_1.0"
122 "requirement_templates": [
124 "name": "lb_mnet_vl_cp",
125 "target_node_template_name": "ext_mnet_net",
126 "target_capability_name": "virtual_linkable"
138 def set_nsd_metadata(self, key, val):
139 self.nsd_raw_data["rawData"]["instance"]["metadata"][key] = val
141 def set_nsd_vnf_id(self, val):
142 self.nsd_raw_data["rawData"]["instance"]["nodes"][0]["properties"]["id"]["value"] = val
144 @mock.patch.object(restcall, 'call_req')
145 def test_ns_pkg_on_boarding_when_on_boarded(self, mock_call_req):
146 mock_call_req.return_value = [0, json.JSONEncoder().encode({"onBoardState": "onBoarded"}), '200']
147 resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "1"}, format='json')
148 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
149 self.assertEqual("failed", resp.data["status"])
150 self.assertEqual("CSAR(1) already onBoarded.", resp.data["statusDescription"])
152 @mock.patch.object(restcall, 'call_req')
153 def test_ns_pkg_on_boarding_when_nsd_already_exists(self, mock_call_req):
154 self.set_nsd_metadata(key="id", val="2")
156 "/api/catalog/v1/csars/2":
157 [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'],
158 "/api/catalog/v1/servicetemplates/queryingrawdata":
159 [0, json.JSONEncoder().encode(self.nsd_raw_data), '200']}
161 def side_effect(*args):
162 return mock_vals[args[4]]
163 mock_call_req.side_effect = side_effect
165 NSDModel(id="1", nsd_id="2").save()
166 resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "2"}, format='json')
167 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
168 self.assertEqual("failed", resp.data["status"])
169 self.assertEqual("NSD(2) already exists.", resp.data["statusDescription"])
171 @mock.patch.object(restcall, 'call_req')
172 def test_ns_pkg_on_boarding_when_vnf_pkg_not_on_boarded(self, mock_call_req):
173 self.set_nsd_metadata(key="id", val="2")
174 self.set_nsd_vnf_id(val="3")
176 "/api/catalog/v1/csars/3":
177 [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'],
178 "/api/catalog/v1/servicetemplates/queryingrawdata":
179 [0, json.JSONEncoder().encode(self.nsd_raw_data), '200']}
181 def side_effect(*args):
182 return mock_vals[args[4]]
183 mock_call_req.side_effect = side_effect
185 resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "3"}, format='json')
186 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
187 self.assertEqual("failed", resp.data["status"])
188 self.assertEqual("VNF package(3) is not onBoarded.", resp.data["statusDescription"])
190 @mock.patch.object(restcall, 'call_req')
191 def test_ns_pkg_on_boarding_when_vnf_pkg_not_on_boarded_on_catalog(self, mock_call_req):
192 self.set_nsd_metadata(key="id", val="2")
193 self.set_nsd_vnf_id(val="6")
195 "/api/catalog/v1/csars/4":
196 [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'],
197 "/api/catalog/v1/servicetemplates/queryingrawdata":
198 [0, json.JSONEncoder().encode(self.nsd_raw_data), '200'],
199 "/api/catalog/v1/csars/5":
200 [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'], }
202 def side_effect(*args):
203 return mock_vals[args[4]]
204 mock_call_req.side_effect = side_effect
206 NfPackageModel(uuid="1", nfpackageid="5", vnfdid="6").save()
207 resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "4"}, format='json')
208 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
209 self.assertEqual("failed", resp.data["status"])
210 self.assertEqual("VNF package(5) is not onBoarded on catalog.", resp.data["statusDescription"])
212 @mock.patch.object(restcall, 'call_req')
213 def test_ns_pkg_on_boarding_when_on_board_success(self, mock_call_req):
214 self.set_nsd_metadata(key="id", val="2")
215 self.set_nsd_metadata(key="name", val="3")
216 self.set_nsd_metadata(key="vendor", val="4")
217 self.set_nsd_metadata(key="description", val="5")
218 self.set_nsd_metadata(key="version", val="6")
219 self.set_nsd_vnf_id(val="7")
221 "/api/catalog/v1/csars/5":
222 [0, json.JSONEncoder().encode({
223 "onBoardState": "non-onBoarded",
224 "createTime": "2016-05-15 12:30:34",
225 "modifyTime": "2016-05-15 12:30:34"}), '200'],
226 "/api/catalog/v1/servicetemplates/queryingrawdata":
227 [0, json.JSONEncoder().encode(self.nsd_raw_data), '200'],
228 "/api/catalog/v1/csars/6":
229 [0, json.JSONEncoder().encode({"onBoardState": "onBoarded"}), '200'],
230 "/api/catalog/v1/csars/5?operationalState=Enabled": [0, '{}', 200],
231 "/api/catalog/v1/csars/5?onBoardState=onBoarded": [0, "OK", '200']}
233 def side_effect(*args):
234 return mock_vals[args[4]]
236 mock_call_req.side_effect = side_effect
238 NfPackageModel(uuid="1", nfpackageid="6", vnfdid="7").save()
239 resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "5"}, format='json')
240 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
241 self.assertEqual("success", resp.data["status"])
242 self.assertEqual("CSAR(5) onBoarded successfully.", resp.data["statusDescription"])
243 nsds = NSDModel.objects.filter(id="5")
244 self.assertEqual(1, len(nsds))
245 self.assertEqual("2", nsds[0].nsd_id)
247 ###############################################################################################################
248 @mock.patch.object(restcall, 'call_req')
249 def test_delete_csar_when_id_not_exist(self, mock_call_req):
250 mock_call_req.return_value = [0, "", '204']
251 resp = self.client.delete("/api/nslcm/v0/nspackage/6")
252 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
253 self.assertEqual("success", resp.data["status"])
254 self.assertEqual("Delete CSAR(6) successfully.", resp.data["statusDescription"])
256 @mock.patch.object(restcall, 'call_req')
257 def test_delete_csar_when_ref_by_ns_inst(self, mock_call_req):
258 mock_call_req.return_value = [0, "OK", '200']
260 NSDModel(id="7", nsd_id="2").save()
261 NSInstModel(id="1", nspackage_id="7").save()
263 resp = self.client.delete("/api/nslcm/v0/nspackage/7")
264 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
265 self.assertEqual("success", resp.data["status"])
266 self.assertEqual("Set deletionPending to True of CSAR(7) successfully.",
267 resp.data["statusDescription"])
269 @mock.patch.object(restcall, 'call_req')
270 def test_delete_csar_when_delete_success(self, mock_call_req):
271 mock_call_req.return_value = [0, "OK", '204']
273 NSDModel(id="8", nsd_id="2").save()
275 resp = self.client.delete("/api/nslcm/v0/nspackage/8")
276 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
277 self.assertEqual("success", resp.data["status"])
278 self.assertEqual("Delete CSAR(8) successfully.", resp.data["statusDescription"])
280 ###############################################################################################################
281 def test_delete_pending_csar_when_id_not_exist(self):
282 resp = self.client.delete("/api/nslcm/v0/nspackage/9/deletionpending")
283 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
284 self.assertEqual("success", resp.data["status"])
285 self.assertEqual("Delete pending CSAR(9) successfully.", resp.data["statusDescription"])
287 @mock.patch.object(restcall, 'call_req')
288 def test_delete_pending_csar_when_pending_is_false(self, mock_call_req):
289 mock_call_req.return_value = [0, '{"deletionPending": "false"}', '200']
290 NSDModel(id="10", nsd_id="2").save()
291 resp = self.client.delete("/api/nslcm/v0/nspackage/10/deletionpending")
292 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
293 self.assertEqual("failed", resp.data["status"])
294 self.assertEqual("CSAR(10) need not to be deleted.", resp.data["statusDescription"])
296 @mock.patch.object(restcall, 'call_req')
297 def test_delete_pending_csar_when_refed_by_ns(self, mock_call_req):
298 mock_call_req.return_value = [0, '{"deletionPending": "true"}', '200']
299 NSDModel(id="11", nsd_id="2").save()
300 NSInstModel(id="1", nspackage_id="11").save()
301 resp = self.client.delete("/api/nslcm/v0/nspackage/11/deletionpending")
302 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
303 self.assertEqual("failed", resp.data["status"])
304 self.assertEqual("CSAR(11) is in using, cannot be deleted.", resp.data["statusDescription"])
306 @mock.patch.object(restcall, 'call_req')
307 def test_delete_pending_csar_when_delete_success(self, mock_call_req):
308 mock_call_req.side_effect = [
309 [0, '{"deletionPending": "true"}', '200'],
311 NSDModel(id="12", nsd_id="2").save()
312 resp = self.client.delete("/api/nslcm/v0/nspackage/12/deletionpending")
313 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
314 self.assertEqual("success", resp.data["status"])
315 self.assertEqual("Delete CSAR(12) successfully.", resp.data["statusDescription"])
317 ###############################################################################################################
318 @mock.patch.object(restcall, 'call_req')
319 def test_get_csar_successfully(self, mock_call_req):
320 mock_call_req.return_value = [0, json.JSONEncoder().encode({
324 "operationalState": "4",
328 "deletionPending": "8",
336 NSDModel(id="13", nsd_id="2", vendor="3", version="4").save()
337 NSInstModel(id="1", nspackage_id="13", name="11").save()
338 NSInstModel(id="2", nspackage_id="13", name="22").save()
340 resp = self.client.get("/api/nslcm/v0/nspackage/13")
341 self.assertEqual(resp.status_code, status.HTTP_200_OK)
342 expect_data = {"nsInstanceInfo": [{"nsInstanceId": "1", "nsInstanceName": "11"},
343 {"nsInstanceId": "2", "nsInstanceName": "22"}], "csarId": "13",
344 "packageInfo": {"nsdProvider": "3", "usageState": "5",
345 "onBoardState": "6", "name": "1", "format": "12",
346 "modifyTime": "11", "nsdId": "2", "nsdVersion": "4",
347 "deletionPending": "8", "version": "3", "downloadUri": "9",
348 "processState": "7", "provider": "2", "operationalState": "4",
349 "createTime": "10", "size": "13"}}
350 self.assertEqual(expect_data, resp.data)
352 ###############################################################################################################
353 def test_disable_csar_when_id_not_exist_table(self):
354 resp = self.client.put("/api/nslcm/v0/nspackage/14/disabled")
355 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
356 self.assertEqual("failed", resp.data["status"])
357 self.assertEqual("CSAR(14) does not exist.", resp.data["statusDescription"])
359 @mock.patch.object(restcall, 'call_req')
360 def test_disable_csar_when_csar_is_disabled(self, mock_call_req):
361 NSDModel(id="15", nsd_id="2").save()
362 mock_call_req.return_value = [0, json.JSONEncoder().encode({"operationalState": "Disabled"}), '200']
363 resp = self.client.put("/api/nslcm/v0/nspackage/15/disabled")
364 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
365 self.assertEqual("failed", resp.data["status"])
366 self.assertEqual("CSAR(15) already disabled.", resp.data["statusDescription"])
368 @mock.patch.object(restcall, 'call_req')
369 def test_disable_csar_successfully(self, mock_call_req):
370 NSDModel(id="16", nsd_id="2").save()
372 "/api/catalog/v1/csars/16":
373 [0, json.JSONEncoder().encode({"operationalState": "Enabled"}), '200'],
374 "/api/catalog/v1/csars/16?operationState=Disabled":
377 def side_effect(*args):
378 return mock_vals[args[4]]
379 mock_call_req.side_effect = side_effect
381 resp = self.client.put("/api/nslcm/v0/nspackage/16/disabled")
382 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
383 self.assertEqual("success", resp.data["status"])
384 self.assertEqual("Set operationState to Disabled of CSAR(16) successfully.", resp.data["statusDescription"])
386 ###############################################################################################################
387 def test_enable_csar_when_id_not_exist_table(self):
388 resp = self.client.put("/api/nslcm/v0/nspackage/17/enabled")
389 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
390 self.assertEqual("failed", resp.data["status"])
391 self.assertEqual("CSAR(17) does not exist.", resp.data["statusDescription"])
393 @mock.patch.object(restcall, 'call_req')
394 def test_enable_csar_when_csar_is_enabled(self, mock_call_req):
395 NSDModel(id="18", nsd_id="2").save()
396 mock_call_req.return_value = [0, json.JSONEncoder().encode({"operationalState": "Enabled"}), '200']
397 resp = self.client.put("/api/nslcm/v0/nspackage/18/enabled")
398 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
399 self.assertEqual("failed", resp.data["status"])
400 self.assertEqual("CSAR(18) already enabled.", resp.data["statusDescription"])
402 @mock.patch.object(restcall, 'call_req')
403 def test_enable_csar_successfully(self, mock_call_req):
404 NSDModel(id="19", nsd_id="2").save()
406 "/api/catalog/v1/csars/19":
407 [0, json.JSONEncoder().encode({"operationalState": "Disabled"}), '200'],
408 "/api/catalog/v1/csars/19?operationState=Enabled":
411 def side_effect(*args):
412 return mock_vals[args[4]]
413 mock_call_req.side_effect = side_effect
415 resp = self.client.put("/api/nslcm/v0/nspackage/19/enabled")
416 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
417 self.assertEqual("success", resp.data["status"])
418 self.assertEqual("Set operationState to Enabled of CSAR(19) successfully.", resp.data["statusDescription"])