1 # Copyright 2016-2017 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.
19 from django.test import TestCase, Client
20 from rest_framework import status
22 from lcm.pub.database.models import VLInstModel, NfInstModel, JobModel, NSInstModel, VmInstModel, \
23 OOFDataModel, VNFCInstModel, PortInstModel, CPInstModel, SubscriptionModel
24 from lcm.pub.exceptions import NSLCMException
25 from lcm.pub.utils import restcall
26 from lcm.jobs.enum import JOB_MODEL_STATUS, JOB_TYPE, JOB_ACTION, JOB_PROGRESS
27 from lcm.pub.utils.jobutil import JobUtil
28 from lcm.pub.utils.timeutil import now_time
29 from lcm.pub.utils.values import ignore_case_get
30 from lcm.ns_vnfs.biz.grant_vnf import GrantVnf
31 from lcm.ns_vnfs.biz.heal_vnfs import NFHealService
32 from lcm.ns_vnfs.biz.scale_vnfs import NFManualScaleService
33 from lcm.ns_vnfs.biz.subscribe import SubscriptionDeletion
34 from lcm.ns_vnfs.biz.terminate_nfs import TerminateVnfs
35 from lcm.ns_vnfs.enum import VNF_STATUS, LIFE_CYCLE_OPERATION, RESOURCE_CHANGE_TYPE, VNFC_CHANGE_TYPE, INST_TYPE, \
37 from lcm.ns_vnfs.biz.place_vnfs import PlaceVnfs
38 from lcm.pub.msapi import resmgr
39 from lcm.ns_vnfs.tests.test_data import vnfm_info, vim_info, vnf_place_request
40 from lcm.ns_vnfs.tests.test_data import nf_package_info, nsd_model_dict, subscription_response_data
41 from lcm.ns_vnfs.biz.create_vnfs import CreateVnfs
42 from lcm.ns_vnfs.biz import create_vnfs
43 from lcm.ns_vnfs.biz.grant_vnfs import GrantVnfs
44 from lcm.ns_vnfs.biz.update_vnfs import NFOperateService
45 from lcm.ns_vnfs.biz.verify_vnfs import VerifyVnfs
46 from lcm.ns.enum import OWNER_TYPE
47 from lcm.ns_vnfs.biz.handle_notification import HandleVnfLcmOocNotification, HandleVnfIdentifierCreationNotification, \
48 HandleVnfIdentifierDeletionNotification
49 from lcm.ns_vnfs.biz.notify_lcm import NotifyLcm
52 class TestGetVnfViews(TestCase):
54 self.client = Client()
55 self.nf_inst_id = str(uuid.uuid4())
56 NfInstModel(nfinstid=self.nf_inst_id, nf_name="vnf1", vnfm_inst_id="1", vnf_id="vnf_id1",
57 status=VNF_STATUS.ACTIVE, create_time=now_time(), lastuptime=now_time()).save()
60 NfInstModel.objects.all().delete()
62 def test_get_vnf(self):
63 response = self.client.get("/api/nslcm/v1/ns/vnfs/%s" % self.nf_inst_id)
64 self.assertEqual(status.HTTP_200_OK, response.status_code)
65 context = json.loads(response.content)
66 self.assertEqual(self.nf_inst_id, context["vnfInstId"])
69 class TestTerminateVnfViews(TestCase):
71 self.client = Client()
73 "terminationType": "forceful",
74 "gracefulTerminationTimeout": "600"
76 self.ns_inst_id = str(uuid.uuid4())
78 self.vim_id = str(uuid.uuid4())
79 self.job_id = str(uuid.uuid4())
81 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
82 NSInstModel.objects.all().delete()
83 NfInstModel.objects.all().delete()
84 VmInstModel.objects.all().delete()
85 NSInstModel(id=self.ns_inst_id, name="ns_name").save()
86 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
89 mnfinstid=self.nf_uuid,
90 vnfd_model=self.vnfd_model
92 VmInstModel.objects.create(vmid="1",
93 vimid='{"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}',
94 instid=self.nf_inst_id
96 SubscriptionModel(vnf_instance_filter=self.nf_inst_id, callback_uri="", links="").save()
99 NSInstModel.objects.all().delete()
100 NfInstModel.objects.all().delete()
101 VmInstModel.objects.all().delete()
102 SubscriptionModel.objects.all().delete()
104 @mock.patch.object(TerminateVnfs, "run")
105 def test_terminate_vnf_url(self, mock_run):
107 "terminationType": "forceful",
108 "gracefulTerminationTimeout": "600"}
110 response = self.client.post("/api/nslcm/v1/ns/terminatevnf/%s" % self.nf_inst_id, data=req_data)
111 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
113 @mock.patch.object(time, "sleep")
114 @mock.patch.object(restcall, "call_req")
115 def test_terminate_vnf(self, mock_call_req, mock_sleep):
116 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
119 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
122 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
123 [0, json.JSONEncoder().encode(vnfm_info), "200"],
124 "/api/ztevnfmdriver/v1/1/vnfs/111/terminate":
125 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"],
126 "/api/ztevnfmdriver/v1/1/jobs/" + job_id + "?responseId=0":
127 [0, json.JSONEncoder().encode(job_info), "200"],
128 "/api/resmgr/v1/vnf/1":
129 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"],
130 "api/gvnfmdriver/v1/1/subscriptions/":
131 [0, json.JSONEncoder().encode({}), "200"]
134 def side_effect(*args):
135 return mock_vals[args[4]]
136 mock_call_req.side_effect = side_effect
137 TerminateVnfs(self.data, self.nf_inst_id, job_id).run()
138 nfinst = NfInstModel.objects.filter(nfinstid=self.nf_inst_id)
140 self.assertEqual(1, 0)
142 self.assertEqual(1, 1)
143 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 100)
145 @mock.patch.object(time, "sleep")
146 @mock.patch.object(restcall, "call_req")
147 @mock.patch.object(SubscriptionDeletion, "send_subscription_deletion_request")
148 def test_terminate_vnf_when_no_vnf_uuid(self, mock_send_subscription_deletion_request, mock_call_req, mock_sleep):
149 nf_inst_id = "test_terminate_vnf_when_no_vnf_uuid"
150 NSInstModel(id=nf_inst_id, name="ns_name_2").save()
151 NfInstModel.objects.create(nfinstid=nf_inst_id,
154 vnfd_model=self.vnfd_model
156 VmInstModel.objects.create(vmid="2",
157 vimid='{"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}',
160 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, nf_inst_id)
163 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
166 "/external-system/esr-vnfm-list/esr-vnfm/2?depth=all":
167 [0, json.JSONEncoder().encode(vnfm_info), "200"],
168 "/api/ztevnfmdriver/v1/2/vnfs/None/terminate":
169 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"],
170 "/api/ztevnfmdriver/v1/2/jobs/" + job_id + "?responseId=0":
171 [0, json.JSONEncoder().encode(job_info), "200"],
172 "/api/resmgr/v1/vnf/%s" % nf_inst_id:
173 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"]
176 def side_effect(*args):
177 return mock_vals[args[4]]
179 mock_call_req.side_effect = side_effect
180 TerminateVnfs(self.data, nf_inst_id, job_id).run()
181 nfinst = NfInstModel.objects.filter(nfinstid=nf_inst_id)
183 self.assertEqual(1, 0)
185 self.assertEqual(1, 1)
186 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 100)
188 @mock.patch.object(time, "sleep")
189 @mock.patch.object(restcall, "call_req")
190 @mock.patch.object(SubscriptionDeletion, "send_subscription_deletion_request")
191 def test_terminate_vnf_when_nf_not_exists(self, mock_send_subscription_deletion_request, mock_call_req, mock_sleep):
192 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
193 TerminateVnfs(self.data, "nf_not_exists", job_id).run()
194 nfinst = NfInstModel.objects.filter(nfinstid="nf_not_exists")
196 self.assertEqual(1, 0)
198 self.assertEqual(1, 1)
199 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 255)
201 def test_terminate_vnf_when_vnf_is_dealing(self):
202 NfInstModel.objects.filter(nfinstid=self.nf_inst_id).update(status=VNF_STATUS.TERMINATING)
203 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
204 TerminateVnfs(self.data, self.nf_inst_id, job_id).run()
205 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.FAILED)
206 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 255)
208 @mock.patch.object(time, "sleep")
209 @mock.patch.object(restcall, "call_req")
210 @mock.patch.object(SubscriptionDeletion, "send_subscription_deletion_request")
211 def test_terminate_vnf_when_job_error(self, mock_send_subscription_deletion_request, mock_call_req, mock_sleep):
212 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
215 "responsedescriptor": {"status": JOB_MODEL_STATUS.ERROR}
218 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
219 [0, json.JSONEncoder().encode(vnfm_info), "200"],
220 "/api/ztevnfmdriver/v1/1/vnfs/111/terminate":
221 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"],
222 "/api/ztevnfmdriver/v1/1/jobs/" + job_id + "?responseId=0":
223 [0, json.JSONEncoder().encode(job_info), "200"]
226 def side_effect(*args):
227 return mock_vals[args[4]]
229 mock_call_req.side_effect = side_effect
230 TerminateVnfs(self.data, self.nf_inst_id, job_id).run()
231 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.FAILED)
232 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 255)
235 class TestScaleVnfViews(TestCase):
237 self.client = Client()
238 self.nf_inst_id = str(uuid.uuid4())
239 self.url = "/api/nslcm/v1/ns/ns_vnfs/%s/scaling" % self.nf_inst_id
244 "aspectId": "demo_aspect1",
246 "additionalParam": {}
249 NfInstModel.objects.create(nfinstid=self.nf_inst_id, vnfm_inst_id="vnfm_inst_id_001",
250 mnfinstid="m_nf_inst_id_001")
253 NfInstModel.objects.all().delete()
255 # def test_scale_vnf_view(self):
256 # response = self.client.post(self.url, self.data)
257 # self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
259 @mock.patch.object(time, "sleep")
260 @mock.patch.object(restcall, "call_req")
261 def test_scale_vnf_success(self, mock_call_req, mock_sleep):
262 scale_service = NFManualScaleService(self.nf_inst_id, self.data)
264 "jobId": scale_service.job_id,
265 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
268 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
269 [0, json.JSONEncoder().encode(vnfm_info), "200"],
270 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/scale":
271 [0, json.JSONEncoder().encode({"jobId": scale_service.job_id}), "200"],
272 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + scale_service.job_id + "?responseId=0":
273 [0, json.JSONEncoder().encode(job_info), "200"]
276 def side_effect(*args):
277 return mock_vals[args[4]]
278 mock_call_req.side_effect = side_effect
280 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
281 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
283 jobs = JobModel.objects.get(jobid=scale_service.job_id)
284 self.assertEqual(JOB_PROGRESS.FINISHED, jobs.progress)
286 @mock.patch.object(time, "sleep")
287 @mock.patch.object(restcall, "call_req")
288 def test_scale_vnf_when_job_fail(self, mock_call_req, mock_sleep):
289 scale_service = NFManualScaleService(self.nf_inst_id, self.data)
291 "jobId": scale_service.job_id,
292 "responsedescriptor": {"status": JOB_MODEL_STATUS.ERROR}
295 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
296 [0, json.JSONEncoder().encode(vnfm_info), "200"],
297 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/scale":
298 [0, json.JSONEncoder().encode({"jobId": scale_service.job_id}), "200"],
299 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + scale_service.job_id + "?responseId=0":
300 [0, json.JSONEncoder().encode(job_info), "200"]
303 def side_effect(*args):
304 return mock_vals[args[4]]
306 mock_call_req.side_effect = side_effect
308 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
309 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
310 jobs = JobModel.objects.get(jobid=scale_service.job_id)
311 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
313 def test_scale_vnf_when_exception(self):
318 "aspectId": "demo_aspect1",
323 "aspectId": "demo_aspect2",
328 scale_service = NFManualScaleService(self.nf_inst_id, req_data)
330 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
331 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
333 jobs = JobModel.objects.get(jobid=scale_service.job_id)
334 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
336 def test_scale_vnf_when_nf_instance_does_not_exist(self):
341 "aspectId": "demo_aspect1",
345 scale_service = NFManualScaleService("nf_instance_does_not_exist", req_data)
348 jobs = JobModel.objects.get(jobid=scale_service.job_id)
349 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
351 def test_scale_vnf_when_scale_vnf_data_does_not_exist(self):
355 scale_service = NFManualScaleService(self.nf_inst_id, req_data)
357 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
358 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
360 jobs = JobModel.objects.get(jobid=scale_service.job_id)
361 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
364 class TestHealVnfViews(TestCase):
366 self.client = Client()
367 self.ns_inst_id = str(uuid.uuid4())
368 self.nf_inst_id = str(uuid.uuid4())
377 "additionalParams": {
383 NSInstModel(id=self.ns_inst_id, name="ns_name").save()
384 NfInstModel.objects.create(nfinstid=self.nf_inst_id, status=VNF_STATUS.NULL, vnfm_inst_id="vnfm_inst_id_001",
385 mnfinstid="m_nf_inst_id_001")
386 NfInstModel.objects.create(nfinstid="non_vud_id", status=VNF_STATUS.NULL, vnfm_inst_id="vnfm_inst_id_001",
387 mnfinstid="m_nf_inst_id_001")
388 VNFCInstModel.objects.create(nfinstid=self.nf_inst_id, vmid="vm_id_001", vduid="vdu_id_001")
389 VmInstModel.objects.create(resouceid="vm_id_001", vmname="vm_name_001")
392 NSInstModel.objects.all().delete()
393 NfInstModel.objects.all().delete()
395 @mock.patch.object(time, "sleep")
396 @mock.patch.object(restcall, "call_req")
397 def test_heal_vnf_success(self, mock_call_req, mock_sleep):
398 heal_service = NFHealService(self.nf_inst_id, self.data)
400 "/test/bins/1?timeout=15000":
401 [0, json.JSONEncoder().encode(['{"powering-off": "", "instance_id": "vm_id_001", '
402 '"display_name": ""}']), "200"],
403 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
404 [0, json.JSONEncoder().encode(vnfm_info), "200"],
405 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/heal":
406 [0, json.JSONEncoder().encode({"jobId": heal_service.job_id}), "200"],
407 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + heal_service.job_id + "?responseId=0":
408 [0, json.JSONEncoder().encode({
409 "jobId": heal_service.job_id,
410 "responsedescriptor": {
411 "status": JOB_MODEL_STATUS.FINISHED,
412 "responsehistorylist": [{
414 "status": JOB_MODEL_STATUS.PROCESSING,
416 "statusdescription": "creating",
423 def side_effect(*args):
424 return mock_vals[args[4]]
426 mock_call_req.side_effect = side_effect
428 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.ACTIVE)
429 jobs = JobModel.objects.get(jobid=heal_service.job_id)
430 self.assertEqual(JOB_PROGRESS.FINISHED, jobs.progress)
432 def test_heal_vnf_when_non_existing_vnf(self, ):
433 heal_service = NFHealService("on_existing_vnf", self.data)
435 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.NULL)
436 jobs = JobModel.objects.get(jobid=heal_service.job_id)
437 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
439 def test_heal_vnf_when_additional_params_non_exist(self):
440 data = {"action": "vmReset"}
441 heal_service = NFHealService(self.nf_inst_id, data)
443 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.NULL)
444 jobs = JobModel.objects.get(jobid=heal_service.job_id)
445 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
447 def test_heal_vnf_when_non_vud_id(self, ):
448 heal_service = NFHealService("non_vud_id", self.data)
450 self.assertEqual(NfInstModel.objects.get(nfinstid="non_vud_id").status, VNF_STATUS.NULL)
451 jobs = JobModel.objects.get(jobid=heal_service.job_id)
452 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
454 @mock.patch.object(restcall, "call_req")
455 def test_heal_vnf_when_no_vnfm_job_id(self, mock_call_req):
456 heal_service = NFHealService(self.nf_inst_id, self.data)
458 "/test/bins/1?timeout=15000":
459 [0, json.JSONEncoder().encode(['{"powering-off": "", "instance_id": "vm_id_001", '
460 '"display_name": ""}']), "200"],
461 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
462 [0, json.JSONEncoder().encode(vnfm_info), "200"],
463 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/heal":
464 [0, json.JSONEncoder().encode({}), "200"]
467 def side_effect(*args):
468 return mock_vals[args[4]]
470 mock_call_req.side_effect = side_effect
472 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.ACTIVE)
473 jobs = JobModel.objects.get(jobid=heal_service.job_id)
474 self.assertEqual(JOB_PROGRESS.FINISHED, jobs.progress)
476 @mock.patch.object(time, "sleep")
477 @mock.patch.object(restcall, "call_req")
478 def test_heal_vnf_when_job_bot_finish(self, mock_call_req, mock_sleep):
479 heal_service = NFHealService(self.nf_inst_id, self.data)
481 "/test/bins/1?timeout=15000":
482 [0, json.JSONEncoder().encode(['{"powering-off": "", "instance_id": "vm_id_001", '
483 '"display_name": ""}']), "200"],
484 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
485 [0, json.JSONEncoder().encode(vnfm_info), "200"],
486 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/heal":
487 [0, json.JSONEncoder().encode({"jobId": heal_service.job_id}), "200"],
488 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + heal_service.job_id + "?responseId=0":
489 [0, json.JSONEncoder().encode({
490 "jobId": heal_service.job_id,
491 "responsedescriptor": {
492 "status": JOB_MODEL_STATUS.ERROR,
493 "responsehistorylist": [{
495 "status": JOB_MODEL_STATUS.PROCESSING,
497 "statusdescription": "creating",
504 def side_effect(*args):
505 return mock_vals[args[4]]
507 mock_call_req.side_effect = side_effect
509 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.HEALING)
510 jobs = JobModel.objects.get(jobid=heal_service.job_id)
511 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
514 class TestGetVnfmInfoViews(TestCase):
516 self.client = Client()
517 self.vnfm_id = str(uuid.uuid4())
522 @mock.patch.object(restcall, "call_req")
523 def test_get_vnfm_info(self, mock_call_req):
525 "vnfm-id": "example-vnfm-id-val-62576",
526 "vim-id": "example-vim-id-val-35114",
527 "certificate-url": "example-certificate-url-val-90242",
528 "esr-system-info-list": {
531 "esr-system-info-id": "example-esr-system-info-id-val-78484",
532 "system-name": "example-system-name-val-23790",
533 "type": "example-type-val-52596",
534 "vendor": "example-vendor-val-47399",
535 "version": "example-version-val-42051",
536 "service-url": "example-service-url-val-10731",
537 "user-name": "example-user-name-val-65946",
538 "password": "example-password-val-22505",
539 "system-type": "example-system-type-val-27221",
540 "protocal": "example-protocal-val-54632",
541 "ssl-cacert": "example-ssl-cacert-val-45965",
542 "ssl-insecure": True,
543 "ip-address": "example-ip-address-val-19212",
544 "port": "example-port-val-57641",
545 "cloud-domain": "example-cloud-domain-val-26296",
546 "default-tenant": "example-default-tenant-val-87724"
551 r1 = [0, json.JSONEncoder().encode(vnfm_info_aai), "200"]
552 mock_call_req.side_effect = [r1]
553 esr_system_info = ignore_case_get(ignore_case_get(vnfm_info_aai, "esr-system-info-list"), "esr-system-info")
555 "vnfmId": vnfm_info_aai["vnfm-id"],
556 "name": vnfm_info_aai["vnfm-id"],
557 "type": ignore_case_get(esr_system_info[0], "type"),
558 "vimId": vnfm_info_aai["vim-id"],
559 "vendor": ignore_case_get(esr_system_info[0], "vendor"),
560 "version": ignore_case_get(esr_system_info[0], "version"),
561 "description": "vnfm",
562 "certificateUrl": vnfm_info_aai["certificate-url"],
563 "url": ignore_case_get(esr_system_info[0], "service-url"),
564 "userName": ignore_case_get(esr_system_info[0], "user-name"),
565 "password": ignore_case_get(esr_system_info[0], "password"),
569 response = self.client.get("/api/nslcm/v1/vnfms/%s" % self.vnfm_id)
570 self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
571 context = json.loads(response.content)
572 self.assertEqual(expect_data, context)
575 class TestGetVimInfoViews(TestCase):
577 self.client = Client()
578 self.vim_id = {"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}
583 @mock.patch.object(restcall, "call_req")
584 def test_get_vim_info(self, mock_call_req):
585 r1 = [0, json.JSONEncoder().encode(vim_info), "200"]
586 mock_call_req.side_effect = [r1]
587 esr_system_info = ignore_case_get(ignore_case_get(vim_info, "esr-system-info-list"), "esr-system-info")
589 "vimId": self.vim_id,
591 "url": ignore_case_get(esr_system_info[0], "service-url"),
592 "userName": ignore_case_get(esr_system_info[0], "user-name"),
593 "password": ignore_case_get(esr_system_info[0], "password"),
594 # "tenant": ignore_case_get(tenants[0], "tenant-id"),
595 "tenant": ignore_case_get(esr_system_info[0], "default-tenant"),
596 "vendor": ignore_case_get(esr_system_info[0], "vendor"),
597 "version": ignore_case_get(esr_system_info[0], "version"),
598 "description": "vim",
600 "type": ignore_case_get(esr_system_info[0], "type"),
604 # response = self.client.get("/api/nslcm/v1/vims/%s" % self.vim_id)
605 response = self.client.get("/api/nslcm/v1/vims/%s/%s" % (self.vim_id["cloud_owner"], self.vim_id["cloud_regionid"]))
606 self.assertEqual(status.HTTP_200_OK, response.status_code)
607 context = json.loads(response.content)
608 self.assertEqual(expect_data["url"], context["url"])
611 class TestPlaceVnfViews(TestCase):
613 self.vnf_inst_id = "1234"
615 self.client = Client()
616 self.url = "/api/nslcm/v1/ns/placevnf"
617 self.data = vnf_place_request
618 OOFDataModel.objects.all().delete()
619 OOFDataModel.objects.create(
621 transaction_id="1234",
622 request_status="init",
623 request_module_name=self.vnf_id,
624 service_resource_id=self.vnf_inst_id,
632 OOFDataModel.objects.all().delete()
634 @mock.patch.object(restcall, "call_req")
635 def test_place_vnf(self, mock_call_req):
638 "flavorName": "HPA.flavor.1",
642 # response = self.client.post(self.url, data=self.data)
643 PlaceVnfs(vnf_place_request).extract()
644 db_info = OOFDataModel.objects.filter(request_id=vnf_place_request.get("requestId"), transaction_id=vnf_place_request.get("transactionId"))
645 self.assertEqual(db_info[0].request_status, "completed")
646 self.assertEqual(db_info[0].vim_id, "CloudOwner1_DLLSTX1A")
647 self.assertEqual(db_info[0].cloud_owner, "CloudOwner1")
648 self.assertEqual(db_info[0].cloud_region_id, "DLLSTX1A")
649 self.assertEqual(db_info[0].vdu_info, json.dumps(vdu_info_json))
651 def test_place_vnf_with_invalid_response(self):
654 "transactionId": "1234",
655 "statusMessage": "xx",
656 "requestStatus": "pending",
658 "placementSolutions": [
661 "resourceModuleName": self.vnf_id,
662 "serviceResourceId": self.vnf_inst_id,
664 "identifierType": "serviceInstanceId",
668 "cloudOwner": "CloudOwner1 "
674 "licenseSolutions": [
676 "resourceModuleName": "string",
677 "serviceResourceId": "string",
678 "entitlementPoolUUID": [
681 "licenseKeyGroupUUID": [
684 "entitlementPoolInvariantUUID": [
687 "licenseKeyGroupInvariantUUID": [
694 PlaceVnfs(resp).extract()
695 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
696 self.assertEqual(db_info[0].request_status, "pending")
697 self.assertEqual(db_info[0].vim_id, "none")
698 self.assertEqual(db_info[0].cloud_owner, "none")
699 self.assertEqual(db_info[0].cloud_region_id, "none")
700 self.assertEqual(db_info[0].vdu_info, "none")
702 def test_place_vnf_with_no_assignment_info(self):
705 "transactionId": "1234",
706 "statusMessage": "xx",
707 "requestStatus": "completed",
709 "placementSolutions": [
712 "resourceModuleName": self.vnf_id,
713 "serviceResourceId": self.vnf_inst_id,
715 "identifierType": "serviceInstanceId",
719 "cloudOwner": "CloudOwner1 "
724 "licenseSolutions": [
726 "resourceModuleName": "string",
727 "serviceResourceId": "string",
728 "entitlementPoolUUID": [
731 "licenseKeyGroupUUID": [
734 "entitlementPoolInvariantUUID": [
737 "licenseKeyGroupInvariantUUID": [
744 PlaceVnfs(resp).extract()
745 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
746 self.assertEqual(db_info[0].request_status, "completed")
747 self.assertEqual(db_info[0].vim_id, "none")
748 self.assertEqual(db_info[0].cloud_owner, "none")
749 self.assertEqual(db_info[0].cloud_region_id, "none")
750 self.assertEqual(db_info[0].vdu_info, "none")
752 def test_place_vnf_no_directives(self):
755 "transactionId": "1234",
756 "statusMessage": "xx",
757 "requestStatus": "completed",
759 "placementSolutions": [
762 "resourceModuleName": self.vnf_id,
763 "serviceResourceId": self.vnf_inst_id,
765 "identifierType": "serviceInstanceId",
769 "cloudOwner": "CloudOwner1 "
772 {"key": "locationId",
781 "resourceModuleName": "string",
782 "serviceResourceId": "string",
783 "entitlementPoolUUID": [
786 "licenseKeyGroupUUID": [
789 "entitlementPoolInvariantUUID": [
792 "licenseKeyGroupInvariantUUID": [
799 PlaceVnfs(resp).extract()
800 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
801 self.assertEqual(db_info[0].request_status, "completed")
802 self.assertEqual(db_info[0].vim_id, "none")
803 self.assertEqual(db_info[0].cloud_owner, "none")
804 self.assertEqual(db_info[0].cloud_region_id, "none")
805 self.assertEqual(db_info[0].vdu_info, "none")
807 def test_place_vnf_with_no_solution(self):
810 "transactionId": "1234",
811 "statusMessage": "xx",
812 "requestStatus": "completed",
814 "placementSolutions": [],
815 "licenseSoutions": []
818 PlaceVnfs(resp).extract()
819 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
820 self.assertEqual(db_info[0].request_status, "completed")
821 self.assertEqual(db_info[0].vim_id, "none")
822 self.assertEqual(db_info[0].cloud_owner, "none")
823 self.assertEqual(db_info[0].cloud_region_id, "none")
824 self.assertEqual(db_info[0].vdu_info, "none")
827 class TestGrantVnfsViews(TestCase):
829 self.vnf_inst_id = str(uuid.uuid4())
831 "vnfInstanceId": self.vnf_inst_id,
832 "lifecycleOperation": LIFE_CYCLE_OPERATION.INSTANTIATE,
833 "addResource": [{"type": RESOURCE_CHANGE_TYPE.VDU, "vdu": "vdu_grant_vnf_add_resources"}],
835 "vnfmid": "vnfm_inst_id_001",
836 "vimid": '{"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}'
839 self.client = Client()
840 self.url = "/api/nslcm/v1/ns/grantvnf"
841 NfInstModel(mnfinstid=self.vnf_inst_id, nfinstid="vnf_inst_id_001", package_id="package_id_001",
842 vnfm_inst_id="vnfm_inst_id_001").save()
845 OOFDataModel.objects.all().delete()
846 NfInstModel.objects.all().delete()
848 # @mock.patch.object(restcall, "call_req")
849 # def test_nf_grant_view(self, mock_call_req):
851 # "/api/catalog/v1/vnfpackages/package_id_001":
852 # [0, json.JSONEncoder().encode(nf_package_info), "200"],
853 # "/api/resmgr/v1/resource/grant":
854 # [1, json.JSONEncoder().encode({}), "200"],
855 # "/cloud-infrastructure/cloud-regions/cloud-region/VCPE/RegionOne?depth=all":
856 # [0, json.JSONEncoder().encode(vim_info), "201"],
859 # def side_effect(*args):
860 # return mock_vals[args[4]]
862 # mock_call_req.side_effect = side_effect
864 # "vnfInstanceId": self.vnf_inst_id,
865 # "lifecycleOperation": LIFE_CYCLE_OPERATION.INSTANTIATE
867 # response = self.client.post(self.url, data=data)
868 # self.assertEqual(response.status_code, status.HTTP_201_CREATED)
870 @mock.patch.object(restcall, "call_req")
871 def test_nf_grant_view_when_add_resource(self, mock_call_req):
873 "/api/catalog/v1/vnfpackages/package_id_001":
874 [0, json.JSONEncoder().encode(nf_package_info), "200"],
875 "/api/resmgr/v1/resource/grant":
876 [1, json.JSONEncoder().encode({}), "200"],
877 "/cloud-infrastructure/cloud-regions/cloud-region/VCPE/RegionOne?depth=all":
878 [0, json.JSONEncoder().encode(vim_info), "201"],
881 def side_effect(*args):
882 return mock_vals[args[4]]
883 mock_call_req.side_effect = side_effect
884 resp = GrantVnfs(json.dumps(self.data), "").send_grant_vnf_to_resMgr()
885 return_success = {"vim": {"accessInfo": {"tenant": "admin"},
886 "vimId": "example-cloud-owner-val-97336_example-cloud-region-id-val-35532"}}
887 self.assertEqual(resp, return_success)
889 @mock.patch.object(restcall, "call_req")
890 def test_nf_grant_view_when_remove_resource(self, mock_call_req):
892 "/api/catalog/v1/vnfpackages/package_id_001":
893 [0, json.JSONEncoder().encode(nf_package_info), "200"],
894 "/api/resmgr/v1/resource/grant":
895 [1, json.JSONEncoder().encode({}), "200"],
896 "/cloud-infrastructure/cloud-regions/cloud-region/VCPE/RegionOne?depth=all":
897 [0, json.JSONEncoder().encode(vim_info), "201"],
900 def side_effect(*args):
901 return mock_vals[args[4]]
903 mock_call_req.side_effect = side_effect
904 self.data.pop("addResource")
905 self.data["removeResource"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
906 resp = GrantVnfs(json.dumps(self.data), "").send_grant_vnf_to_resMgr()
907 return_success = {"vim": {"accessInfo": {"tenant": "admin"},
908 "vimId": "example-cloud-owner-val-97336_example-cloud-region-id-val-35532"}}
909 self.assertEqual(resp, return_success)
912 class TestGrantVnfViews(TestCase):
914 self.vnf_inst_id = str(uuid.uuid4())
916 "vnfInstanceId": self.vnf_inst_id,
917 "vnfLcmOpOccId": "vnf_lcm_op_occ_id",
918 "addResources": [{"vdu": "vdu_grant_vnf_add_resources"}],
919 "operation": "INSTANTIATE"
921 self.client = Client()
922 vdu_info_dict = [{"vduName": "vg", "flavorName": "flavor_1", "flavorId": "flavor_id_001", "directive": []}]
923 OOFDataModel(request_id="request_id_001", transaction_id="transaction_id_001", request_status="done",
924 request_module_name="vg", service_resource_id=self.vnf_inst_id, vim_id="cloudOwner_casa",
925 cloud_owner="cloudOwner", cloud_region_id="casa", vdu_info=json.dumps(vdu_info_dict)).save()
926 NfInstModel(mnfinstid=self.vnf_inst_id, nfinstid="vnf_inst_id_001", package_id="package_id_001",
927 vnfm_inst_id="vnfm_id_001").save()
930 OOFDataModel.objects.all().delete()
931 NfInstModel.objects.all().delete()
933 @mock.patch.object(resmgr, "grant_vnf")
934 def test_vnf_grant_view(self, mock_grant):
935 resmgr_grant_resp = {
937 "vimId": "cloudOwner_casa",
943 mock_grant.return_value = resmgr_grant_resp
944 self.data.pop("addResources")
945 response = self.client.post("/api/nslcm/v2/grants", data=self.data)
946 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
947 self.assertEqual(response.data["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
948 self.assertEqual(response.data["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
949 self.assertEqual(response.data["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
951 @mock.patch.object(restcall, "call_req")
952 @mock.patch.object(resmgr, "grant_vnf")
953 def test_exec_grant_when_add_resources_success(self, mock_grant, mock_call_req):
955 "/api/catalog/v1/vnfpackages/package_id_001":
956 [0, json.JSONEncoder().encode(nf_package_info), "200"],
959 def side_effect(*args):
960 return mock_vals[args[4]]
962 mock_call_req.side_effect = side_effect
963 resmgr_grant_resp = {
965 "vimId": "cloudOwner_casa",
971 mock_grant.return_value = resmgr_grant_resp
972 resp = GrantVnf(json.dumps(self.data)).exec_grant()
973 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
974 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
975 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
977 def test_exec_grant_when_add_resources_but_no_vnfinst(self):
978 self.data["vnfInstanceId"] = "no_vnfinst"
981 resp = GrantVnf(json.dumps(self.data)).exec_grant()
982 except NSLCMException as e:
983 self.assertEqual(type(e), NSLCMException)
985 self.assertEqual(resp, None)
987 @mock.patch.object(time, "sleep")
988 @mock.patch.object(restcall, "call_req")
989 @mock.patch.object(resmgr, "grant_vnf")
990 def test_exec_grant_when_add_resources_but_no_off(self, mock_grant, mock_call_req, mock_sleep):
991 NfInstModel(mnfinstid="add_resources_but_no_off", nfinstid="vnf_inst_id_002",
992 package_id="package_id_002").save()
993 mock_sleep.return_value = None
995 "/api/catalog/v1/vnfpackages/package_id_002":
996 [0, json.JSONEncoder().encode(nf_package_info), "200"],
999 def side_effect(*args):
1000 return mock_vals[args[4]]
1002 mock_call_req.side_effect = side_effect
1003 resmgr_grant_resp = {
1005 "vimId": "cloudOwner_casa",
1011 mock_grant.return_value = resmgr_grant_resp
1012 self.data["vnfInstanceId"] = "add_resources_but_no_off"
1013 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1014 self.assertEqual(resp["vnfInstanceId"], "add_resources_but_no_off")
1015 self.assertEqual(resp["vnfLcmOpOccId"], "vnf_lcm_op_occ_id")
1017 "id": "cloudOwner_casa",
1018 "vimId": "cloudOwner_casa",
1020 "interfaceInfo": None,
1021 "accessInfo": {"tenant": "tenantA"},
1024 self.assertEqual(resp["vimConnections"], vimConnections)
1026 @mock.patch.object(resmgr, "grant_vnf")
1027 def test_exec_grant_when_resource_template_in_add_resources(self, mock_grant):
1028 resmgr_grant_resp = {
1030 "vimId": "cloudOwner_casa",
1036 mock_grant.return_value = resmgr_grant_resp
1037 self.data["addResources"] = [{"vdu": "vdu_grant_vnf_add_resources"}, "resourceTemplate"]
1038 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1039 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
1040 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
1041 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
1043 @mock.patch.object(restcall, "call_req")
1044 @mock.patch.object(resmgr, "grant_vnf")
1045 def test_exec_grant_when_remove_resources_success(self, mock_grant, mock_call_req):
1047 "/api/catalog/v1/vnfpackages/package_id_001":
1048 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1051 def side_effect(*args):
1052 return mock_vals[args[4]]
1054 mock_call_req.side_effect = side_effect
1055 resmgr_grant_resp = {
1057 "vimId": "cloudOwner_casa",
1063 mock_grant.return_value = resmgr_grant_resp
1064 self.data.pop("addResources")
1065 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
1066 self.data["additionalparams"] = {"vnfmid": "vnfm_id_001"}
1067 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1068 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
1069 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
1070 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
1072 def test_exec_grant_when_remove_resources_no_vnfinst(self):
1073 self.data.pop("addResources")
1074 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
1075 self.data["additionalparams"] = {"vnfmid": "vnfm_id_002"}
1078 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1079 except NSLCMException as e:
1080 self.assertEqual(type(e), NSLCMException)
1082 self.assertEqual(resp, None)
1084 @mock.patch.object(time, "sleep")
1085 @mock.patch.object(restcall, "call_req")
1086 @mock.patch.object(resmgr, "grant_vnf")
1087 def test_exec_grant_when_remove_resources_but_no_off(self, mock_grant, mock_call_req, mock_sleep):
1088 NfInstModel(mnfinstid="remove_resources_but_no_off", nfinstid="vnf_inst_id_002", package_id="package_id_002",
1089 vnfm_inst_id="vnfm_id_002").save()
1090 mock_sleep.return_value = None
1092 "/api/catalog/v1/vnfpackages/package_id_002":
1093 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1096 def side_effect(*args):
1097 return mock_vals[args[4]]
1099 mock_call_req.side_effect = side_effect
1100 resmgr_grant_resp = {
1102 "vimId": "cloudOwner_casa",
1108 mock_grant.return_value = resmgr_grant_resp
1109 self.data["vnfInstanceId"] = "remove_resources_but_no_off"
1110 self.data.pop("addResources")
1111 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
1112 self.data["additionalparams"] = {"vnfmid": "vnfm_id_002"}
1113 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1114 self.assertEqual(resp["vnfInstanceId"], "remove_resources_but_no_off")
1115 self.assertEqual(resp["vnfLcmOpOccId"], "vnf_lcm_op_occ_id")
1117 "id": "cloudOwner_casa",
1118 "vimId": "cloudOwner_casa",
1120 "interfaceInfo": None,
1121 "accessInfo": {"tenant": "tenantA"},
1124 self.assertEqual(resp["vimConnections"], vimConnections)
1126 @mock.patch.object(resmgr, "grant_vnf")
1127 def test_exec_grant_when_resource_template_in_remove_resources(self, mock_grant):
1128 resmgr_grant_resp = {
1130 "vimId": "cloudOwner_casa",
1136 mock_grant.return_value = resmgr_grant_resp
1137 self.data.pop("addResources")
1138 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}, "resourceTemplate"]
1139 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1140 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
1141 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
1142 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
1145 class TestCreateVnfViews(TestCase):
1147 self.ns_inst_id = str(uuid.uuid4())
1148 self.job_id = str(uuid.uuid4())
1151 "nsInstanceId": self.ns_inst_id,
1152 # "additionalParamForNs": {"inputs": json.dumps({})},
1153 "additionalParamForVnf": [
1155 "vnfprofileid": "VBras",
1156 "additionalparam": {
1157 "inputs": json.dumps({
1161 "vnfminstanceid": "1",
1162 # "vimId": "zte_test",
1163 "vimId": '{"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}'
1168 self.client = Client()
1169 NSInstModel(id=self.ns_inst_id, name="ns", nspackage_id="1", nsd_id="nsd_id", description="description",
1170 status="instantiating", nsd_model=json.dumps(nsd_model_dict), create_time=now_time(),
1171 lastuptime=now_time()).save()
1172 VLInstModel(vldid="ext_mnet_network", ownertype=OWNER_TYPE.NS, ownerid=self.ns_inst_id,
1176 NfInstModel.objects.all().delete()
1177 JobModel.objects.all().delete()
1179 @mock.patch.object(CreateVnfs, "run")
1180 def test_create_vnf_view(self, mock_run):
1181 response = self.client.post("/api/nslcm/v1/ns/vnfs", data=self.data)
1182 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
1183 context = json.loads(response.content)
1184 self.assertTrue(NfInstModel.objects.filter(nfinstid=context["vnfInstId"]).exists())
1186 @mock.patch.object(time, "sleep")
1187 @mock.patch.object(restcall, "call_req")
1188 def test_create_vnf_thread_sucess(self, mock_call_req, mock_sleep):
1189 mock_sleep.return_value = None
1190 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1192 "/api/catalog/v1/vnfpackages/zte_vbras":
1193 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1194 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
1195 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1196 "/api/ztevnfmdriver/v1/1/vnfs":
1197 [0, json.JSONEncoder().encode({"jobId": self.job_id, "vnfInstanceId": 3}), "200"],
1198 "/api/oof/v1/placement":
1199 [0, json.JSONEncoder().encode({}), "202"],
1200 "/api/resmgr/v1/vnf":
1201 [0, json.JSONEncoder().encode({}), "200"],
1202 "/api/ztevnfmdriver/v1/1/jobs/" + self.job_id + "?responseId=0":
1203 [0, json.JSONEncoder().encode({"jobid": self.job_id,
1204 "responsedescriptor": {"progress": "100",
1205 "status": JOB_MODEL_STATUS.FINISHED,
1207 "statusdescription": "creating",
1209 "responsehistorylist": [
1211 "status": JOB_MODEL_STATUS.PROCESSING,
1213 "statusdescription": "creating",
1214 "errorcode": "0"}]}}), "200"],
1215 "api/gvnfmdriver/v1/1/subscriptions":
1216 [0, json.JSONEncoder().encode(subscription_response_data), "200"],
1217 "/api/resmgr/v1/vnfinfo":
1218 [0, json.JSONEncoder().encode(subscription_response_data), "200"],
1220 # "/network/generic-vnfs/generic-vnf/%s" % nf_inst_id:
1221 # [0, json.JSONEncoder().encode({}), "201"],
1222 # "/cloud-infrastructure/cloud-regions/cloud-region/zte/test?depth=all":
1223 # [0, json.JSONEncoder().encode(vim_info), "201"],
1224 # "/cloud-infrastructure/cloud-regions/cloud-region/zte/test/tenants/tenant/admin/vservers/vserver/1":
1225 # [0, json.JSONEncoder().encode({}), "201"],
1229 def side_effect(*args):
1230 return mock_vals[args[4]]
1231 mock_call_req.side_effect = side_effect
1233 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1234 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1235 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1236 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1238 CreateVnfs(data, nf_inst_id, job_id).run()
1239 self.assertEqual(NfInstModel.objects.get(nfinstid=nf_inst_id).status, VNF_STATUS.ACTIVE)
1240 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, JOB_PROGRESS.FINISHED)
1242 def test_create_vnf_thread_when_the_name_of_vnf_instance_already_exists(self):
1243 NfInstModel(nf_name="").save()
1244 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1246 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1247 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1248 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1249 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1251 CreateVnfs(data, nf_inst_id, job_id).run()
1252 self.assertEqual(NfInstModel.objects.get(nfinstid=nf_inst_id).status, VNF_STATUS.FAILED)
1253 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, JOB_PROGRESS.ERROR)
1255 @mock.patch.object(time, "sleep")
1256 @mock.patch.object(restcall, "call_req")
1257 def test_create_vnf_thread_when_data_has_vnfd_id(self, mock_call_req, mock_sleep):
1258 mock_sleep.return_value = None
1259 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1261 "/api/catalog/v1/vnfpackages/data_has_vnfd_id":
1262 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1263 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
1264 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1265 "/api/ztevnfmdriver/v1/1/vnfs":
1266 [0, json.JSONEncoder().encode({"jobId": self.job_id, "vnfInstanceId": 3}), "200"],
1267 "/api/oof/v1/placement":
1268 [0, json.JSONEncoder().encode({}), "202"],
1269 "/api/resmgr/v1/vnf":
1270 [0, json.JSONEncoder().encode({}), "200"],
1271 "/api/ztevnfmdriver/v1/1/jobs/" + self.job_id + "?responseId=0":
1272 [0, json.JSONEncoder().encode({"jobid": self.job_id,
1273 "responsedescriptor": {"progress": "100",
1274 "status": JOB_MODEL_STATUS.FINISHED,
1276 "statusdescription": "creating",
1278 "responsehistorylist": [
1280 "status": JOB_MODEL_STATUS.PROCESSING,
1282 "statusdescription": "creating",
1283 "errorcode": "0"}]}}), "200"],
1284 "api/gvnfmdriver/v1/1/subscriptions":
1285 [0, json.JSONEncoder().encode({}), "200"],
1286 "/api/resmgr/v1/vnfinfo":
1287 [0, json.JSONEncoder().encode(subscription_response_data), "200"]
1290 def side_effect(*args):
1291 return mock_vals[args[4]]
1293 mock_call_req.side_effect = side_effect
1294 self.data["additionalParamForVnf"][0]["additionalparam"]["vnfdId"] = "data_has_vnfd_id"
1296 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1297 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1298 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1299 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1301 CreateVnfs(data, nf_inst_id, job_id).run()
1302 self.assertEqual(NfInstModel.objects.get(nfinstid=nf_inst_id).status, VNF_STATUS.ACTIVE)
1303 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, JOB_PROGRESS.FINISHED)
1305 @mock.patch.object(restcall, "call_req")
1306 @mock.patch.object(CreateVnfs, "build_homing_request")
1307 def test_send_homing_request(self, mock_build_req, mock_call_req):
1308 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1309 OOFDataModel.objects.all().delete()
1311 "requestId": "1234",
1312 "transactionId": "1234",
1313 "requestStatus": "accepted"
1315 mock_build_req.return_value = {
1317 "transactionId": "1234",
1318 "requestId": "1234",
1319 "callbackUrl": "xx",
1321 "requestType": "create",
1323 "optimizers": ["placement"],
1327 "placementDemands": [
1329 "resourceModuleName": "vG",
1330 "serviceResourceId": "1234",
1331 "resourceModelInfo": {
1332 "modelInvariantId": "1234",
1333 "modelVersionId": "1234"
1339 "serviceInstanceId": "1234",
1340 "serviceName": "1234",
1342 "modelInvariantId": "5678",
1343 "modelVersionId": "7890"
1347 mock_call_req.return_value = [0, json.JSONEncoder().encode(resp), "202"]
1349 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1350 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1351 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1352 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1354 CreateVnfs(data, nf_inst_id, job_id).send_homing_request_to_OOF()
1355 ret = OOFDataModel.objects.filter(request_id="1234", transaction_id="1234")
1356 self.assertIsNotNone(ret)
1358 @mock.patch.object(time, "sleep")
1359 @mock.patch.object(restcall, "call_req")
1360 def test_create_vnf_thread_sucess_when_failed_to_subscribe_from_vnfm(self, mock_call_req, mock_sleep):
1361 mock_sleep.return_value = None
1362 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1364 "/api/catalog/v1/vnfpackages/zte_vbras":
1365 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1366 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
1367 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1368 "/api/ztevnfmdriver/v1/1/vnfs":
1369 [0, json.JSONEncoder().encode({"jobId": self.job_id, "vnfInstanceId": 3}), "200"],
1370 "/api/oof/v1/placement":
1371 [0, json.JSONEncoder().encode({}), "202"],
1372 "/api/resmgr/v1/vnf":
1373 [0, json.JSONEncoder().encode({}), "200"],
1374 "/api/ztevnfmdriver/v1/1/jobs/" + self.job_id + "?responseId=0":
1375 [0, json.JSONEncoder().encode({"jobid": self.job_id,
1376 "responsedescriptor": {"progress": "100",
1377 "status": JOB_MODEL_STATUS.FINISHED,
1379 "statusdescription": "creating",
1381 "responsehistorylist": [
1383 "status": JOB_MODEL_STATUS.PROCESSING,
1385 "statusdescription": "creating",
1386 "errorcode": "0"}]}}), "200"],
1387 "api/gvnfmdriver/v1/1/subscriptions":
1388 [1, json.JSONEncoder().encode(subscription_response_data), "200"],
1389 "/api/resmgr/v1/vnfinfo":
1390 [0, json.JSONEncoder().encode(subscription_response_data), "200"],
1393 def side_effect(*args):
1394 return mock_vals[args[4]]
1396 mock_call_req.side_effect = side_effect
1398 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1399 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1400 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1401 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1403 CreateVnfs(data, nf_inst_id, job_id).run()
1404 self.assertEqual(NfInstModel.objects.get(nfinstid=nf_inst_id).status, VNF_STATUS.ACTIVE)
1405 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, JOB_PROGRESS.FINISHED)
1408 class TestUpdateVnfsViews(TestCase):
1410 self.client = Client()
1412 "terminationType": "forceful",
1413 "gracefulTerminationTimeout": "600",
1414 "additionalParams": ""
1416 self.nf_inst_id = "test_update_vnf"
1417 self.m_nf_inst_id = "test_update_vnf_m_nf_inst_id"
1418 self.vnfm_inst_id = "test_update_vnf_vnfm_inst_id"
1419 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
1420 NfInstModel.objects.all().delete()
1421 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
1422 vnfm_inst_id=self.vnfm_inst_id,
1423 status=VNF_STATUS.NULL,
1424 mnfinstid=self.m_nf_inst_id,
1425 vnfd_model=self.vnfd_model
1429 NfInstModel.objects.all().delete()
1431 @mock.patch.object(time, "sleep")
1432 @mock.patch.object(restcall, "call_req")
1433 def test_update_vnf_thread(self, mock_call_req, mock_sleep):
1434 vnf_update_service = NFOperateService(self.nf_inst_id, self.data)
1436 "jobId": vnf_update_service.job_id,
1437 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
1440 "/external-system/esr-vnfm-list/esr-vnfm/test_update_vnf_vnfm_inst_id?depth=all":
1441 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1442 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/vnfs/test_update_vnf_m_nf_inst_id/operate":
1443 [0, json.JSONEncoder().encode({"jobId": vnf_update_service.job_id}), "200"],
1444 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/jobs/" + vnf_update_service.job_id + "?responseId=0":
1445 [0, json.JSONEncoder().encode(job_info), "200"],
1448 def side_effect(*args):
1449 return mock_vals[args[4]]
1450 mock_call_req.side_effect = side_effect
1451 vnf_update_service.run()
1452 nfinst = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
1453 self.assertEqual(nfinst.status, VNF_STATUS.ACTIVE)
1454 self.assertEqual(JobModel.objects.get(jobid=vnf_update_service.job_id).progress, JOB_PROGRESS.FINISHED)
1456 def test_update_vnf_thread_when_no_nf(self):
1457 NfInstModel.objects.all().delete()
1458 vnf_update_service = NFOperateService(self.nf_inst_id, self.data)
1459 vnf_update_service.run()
1460 self.assertEqual(JobModel.objects.get(jobid=vnf_update_service.job_id).progress, JOB_PROGRESS.ERROR)
1462 @mock.patch.object(time, "sleep")
1463 @mock.patch.object(restcall, "call_req")
1464 def test_update_vnf_thread_when_nf_update_failed(self, mock_call_req, mock_sleep):
1465 vnf_update_service = NFOperateService(self.nf_inst_id, self.data)
1467 "jobId": vnf_update_service.job_id,
1468 "responsedescriptor": {"status": JOB_MODEL_STATUS.ERROR}
1471 "/external-system/esr-vnfm-list/esr-vnfm/test_update_vnf_vnfm_inst_id?depth=all":
1472 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1473 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/vnfs/test_update_vnf_m_nf_inst_id/operate":
1474 [0, json.JSONEncoder().encode({"jobId": vnf_update_service.job_id}), "200"],
1475 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/jobs/" + vnf_update_service.job_id + "?responseId=0":
1476 [0, json.JSONEncoder().encode(job_info), "200"],
1479 def side_effect(*args):
1480 return mock_vals[args[4]]
1482 mock_call_req.side_effect = side_effect
1483 vnf_update_service.run()
1484 nfinst = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
1485 self.assertEqual(nfinst.status, VNF_STATUS.UPDATING)
1486 self.assertEqual(JobModel.objects.get(jobid=vnf_update_service.job_id).progress, JOB_PROGRESS.ERROR)
1489 class TestVerifyVnfsViews(TestCase):
1491 self.client = Client()
1492 self.url = "/api/nslcm/v1/vnfonboarding"
1493 self.package_id = "test_verify_vnfs_package_id"
1494 self.nf_inst_id = "test_verify_vnfs"
1495 self.m_nf_inst_id = "test_verify_vnfs_m_nf_inst_id"
1496 self.vnfm_inst_id = "test_verify_vnfs_vnfm_inst_id"
1497 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
1499 "PackageID": self.package_id,
1501 self.job_id = JobUtil.create_job(JOB_TYPE.VNF, "verify_vnfs", self.nf_inst_id)
1502 NfInstModel.objects.all().delete()
1503 NfInstModel.objects.create(package_id=self.package_id,
1504 nfinstid=self.nf_inst_id,
1505 vnfm_inst_id=self.vnfm_inst_id,
1506 status=VNF_STATUS.NULL,
1507 mnfinstid=self.m_nf_inst_id,
1508 vnfd_model=self.vnfd_model
1512 NfInstModel.objects.all().delete()
1514 @mock.patch.object(VerifyVnfs, "run")
1515 def test_verify_vnfs_view(self, mock_run):
1516 response = self.client.post(self.url, data=self.data)
1517 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
1519 def test_verify_vnfs_view_when_data_is_not_valid(self):
1520 response = self.client.post(self.url, data={})
1521 self.assertEqual(status.HTTP_409_CONFLICT, response.status_code)
1523 @mock.patch.object(time, "sleep")
1524 @mock.patch.object(restcall, "call_req")
1525 def test_verify_vnfs_thread(self, mock_call_req, mock_sleep):
1528 "responseDescriptor": {
1529 "status": JOB_MODEL_STATUS.FINISHED,
1532 "statusDescription": ""
1537 "responseDescriptor": {
1538 "status": JOB_MODEL_STATUS.FINISHED,
1541 "statusDescription": ""
1546 "responseDescriptor": {
1547 "status": JOB_MODEL_STATUS.FINISHED,
1550 "statusDescription": ""
1554 "/api/nslcm/v1/vnfpackage":
1555 [0, json.JSONEncoder().encode(job_info_1), "200"],
1556 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1557 [0, json.JSONEncoder().encode(job_info_1), "200"],
1558 "/api/nslcm/v1/ns/ns_vnfs":
1559 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1560 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1561 [0, json.JSONEncoder().encode(job_info_2), "200"],
1562 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1563 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1564 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1565 [0, json.JSONEncoder().encode(job_info_task), "200"],
1566 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1567 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1570 def side_effect(*args):
1571 return mock_vals[args[4]]
1572 mock_call_req.side_effect = side_effect
1573 VerifyVnfs(self.data, self.job_id).run()
1574 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.FINISHED)
1576 @mock.patch.object(time, "sleep")
1577 @mock.patch.object(restcall, "call_req")
1578 def test_verify_vnfs_thread_when_failed_to_call_vnf_onboarding(self, mock_call_req, mock_sleep):
1580 "/api/nslcm/v1/vnfpackage":
1581 [1, json.JSONEncoder().encode({}), "200"]
1584 def side_effect(*args):
1585 return mock_vals[args[4]]
1587 mock_call_req.side_effect = side_effect
1588 VerifyVnfs(self.data, self.job_id).run()
1589 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1591 @mock.patch.object(time, "sleep")
1592 @mock.patch.object(restcall, "call_req")
1593 def test_verify_vnfs_thread_when_do_on_boarding_failed_to_query_job(self, mock_call_req, mock_sleep):
1596 "responseDescriptor": {
1597 "status": JOB_MODEL_STATUS.FINISHED,
1600 "statusDescription": ""
1604 "/api/nslcm/v1/vnfpackage":
1605 [0, json.JSONEncoder().encode(job_info_1), "200"],
1606 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1607 [1, json.JSONEncoder().encode(job_info_1), "200"],
1610 def side_effect(*args):
1611 return mock_vals[args[4]]
1613 mock_call_req.side_effect = side_effect
1614 VerifyVnfs(self.data, self.job_id).run()
1615 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1617 @mock.patch.object(time, "sleep")
1618 @mock.patch.object(restcall, "call_req")
1619 def test_verify_vnfs_thread_when_do_on_boarding_job_does_not_exist(self, mock_call_req, mock_sleep):
1624 "/api/nslcm/v1/vnfpackage":
1625 [0, json.JSONEncoder().encode(job_info_1), "200"],
1626 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1627 [0, json.JSONEncoder().encode(job_info_1), "200"],
1630 def side_effect(*args):
1631 return mock_vals[args[4]]
1633 mock_call_req.side_effect = side_effect
1634 VerifyVnfs(self.data, self.job_id).run()
1635 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1637 @mock.patch.object(time, "sleep")
1638 @mock.patch.object(restcall, "call_req")
1639 def test_verify_vnfs_thread_when_do_on_boarding_job_process_error(self, mock_call_req, mock_sleep):
1642 "responseDescriptor": {
1643 "status": JOB_MODEL_STATUS.FINISHED,
1644 "progress": JOB_PROGRESS.ERROR,
1646 "statusDescription": "already onBoarded"
1651 "responseDescriptor": {
1652 "status": JOB_MODEL_STATUS.FINISHED,
1655 "statusDescription": ""
1660 "responseDescriptor": {
1661 "status": JOB_MODEL_STATUS.FINISHED,
1664 "statusDescription": ""
1668 "/api/nslcm/v1/vnfpackage":
1669 [0, json.JSONEncoder().encode(job_info_1), "200"],
1670 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1671 [0, json.JSONEncoder().encode(job_info_1), "200"],
1672 "/api/nslcm/v1/ns/ns_vnfs":
1673 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1674 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1675 [0, json.JSONEncoder().encode(job_info_2), "200"],
1676 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1677 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1678 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1679 [0, json.JSONEncoder().encode(job_info_task), "200"],
1680 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1681 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1684 def side_effect(*args):
1685 return mock_vals[args[4]]
1687 mock_call_req.side_effect = side_effect
1688 VerifyVnfs(self.data, self.job_id).run()
1689 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.FINISHED)
1691 @mock.patch.object(time, "sleep")
1692 @mock.patch.object(restcall, "call_req")
1693 def test_verify_vnfs_thread_when_failed_to_call_inst_vnf(self, mock_call_req, mock_sleep):
1696 "responseDescriptor": {
1697 "status": JOB_MODEL_STATUS.FINISHED,
1700 "statusDescription": ""
1704 "/api/nslcm/v1/vnfpackage":
1705 [0, json.JSONEncoder().encode(job_info_1), "200"],
1706 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1707 [0, json.JSONEncoder().encode(job_info_1), "200"],
1708 "/api/nslcm/v1/ns/ns_vnfs":
1709 [1, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1712 def side_effect(*args):
1713 return mock_vals[args[4]]
1715 mock_call_req.side_effect = side_effect
1716 VerifyVnfs(self.data, self.job_id).run()
1717 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1719 @mock.patch.object(time, "sleep")
1720 @mock.patch.object(restcall, "call_req")
1721 def test_verify_vnfs_thread_when_do_term_vnf_job_process_error(self, mock_call_req, mock_sleep):
1724 "responseDescriptor": {
1725 "status": JOB_MODEL_STATUS.FINISHED,
1728 "statusDescription": ""
1733 "responseDescriptor": {
1734 "status": JOB_MODEL_STATUS.FINISHED,
1735 "progress": JOB_PROGRESS.ERROR,
1737 "statusDescription": ""
1741 "/api/nslcm/v1/vnfpackage":
1742 [0, json.JSONEncoder().encode(job_info_1), "200"],
1743 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1744 [0, json.JSONEncoder().encode(job_info_1), "200"],
1745 "/api/nslcm/v1/ns/ns_vnfs":
1746 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1747 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1748 [0, json.JSONEncoder().encode(job_info_2), "200"],
1751 def side_effect(*args):
1752 return mock_vals[args[4]]
1754 mock_call_req.side_effect = side_effect
1755 VerifyVnfs(self.data, self.job_id).run()
1756 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1758 @mock.patch.object(time, "sleep")
1759 @mock.patch.object(restcall, "call_req")
1760 def test_verify_vnfs_thread_when_failed_to_call_func_test(self, mock_call_req, mock_sleep):
1763 "responseDescriptor": {
1764 "status": JOB_MODEL_STATUS.FINISHED,
1767 "statusDescription": ""
1772 "responseDescriptor": {
1773 "status": JOB_MODEL_STATUS.FINISHED,
1776 "statusDescription": ""
1780 "/api/nslcm/v1/vnfpackage":
1781 [0, json.JSONEncoder().encode(job_info_1), "200"],
1782 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1783 [0, json.JSONEncoder().encode(job_info_1), "200"],
1784 "/api/nslcm/v1/ns/ns_vnfs":
1785 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1786 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1787 [0, json.JSONEncoder().encode(job_info_2), "200"],
1788 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1789 [1, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"]
1792 def side_effect(*args):
1793 return mock_vals[args[4]]
1795 mock_call_req.side_effect = side_effect
1796 VerifyVnfs(self.data, self.job_id).run()
1797 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1799 @mock.patch.object(time, "sleep")
1800 @mock.patch.object(restcall, "call_req")
1801 def test_verify_vnfs_thread_when_do_func_test_failed_query_job(self, mock_call_req, mock_sleep):
1804 "responseDescriptor": {
1805 "status": JOB_MODEL_STATUS.FINISHED,
1808 "statusDescription": ""
1813 "responseDescriptor": {
1814 "status": JOB_MODEL_STATUS.FINISHED,
1817 "statusDescription": ""
1822 "responseDescriptor": {
1823 "status": JOB_MODEL_STATUS.FINISHED,
1826 "statusDescription": ""
1830 "/api/nslcm/v1/vnfpackage":
1831 [0, json.JSONEncoder().encode(job_info_1), "200"],
1832 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1833 [0, json.JSONEncoder().encode(job_info_1), "200"],
1834 "/api/nslcm/v1/ns/ns_vnfs":
1835 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1836 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1837 [0, json.JSONEncoder().encode(job_info_2), "200"],
1838 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1839 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1840 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1841 [1, json.JSONEncoder().encode(job_info_task), "200"],
1844 def side_effect(*args):
1845 return mock_vals[args[4]]
1847 mock_call_req.side_effect = side_effect
1848 VerifyVnfs(self.data, self.job_id).run()
1849 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1851 @mock.patch.object(time, "sleep")
1852 @mock.patch.object(restcall, "call_req")
1853 def test_verify_vnfs_thread_when_do_func_test_job_does_not_exist(self, mock_call_req, mock_sleep):
1856 "responseDescriptor": {
1857 "status": JOB_MODEL_STATUS.FINISHED,
1860 "statusDescription": ""
1865 "responseDescriptor": {
1866 "status": JOB_MODEL_STATUS.FINISHED,
1869 "statusDescription": ""
1876 "/api/nslcm/v1/vnfpackage":
1877 [0, json.JSONEncoder().encode(job_info_1), "200"],
1878 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1879 [0, json.JSONEncoder().encode(job_info_1), "200"],
1880 "/api/nslcm/v1/ns/ns_vnfs":
1881 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1882 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1883 [0, json.JSONEncoder().encode(job_info_2), "200"],
1884 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1885 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1886 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1887 [0, json.JSONEncoder().encode(job_info_task), "200"],
1888 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1889 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1892 def side_effect(*args):
1893 return mock_vals[args[4]]
1895 mock_call_req.side_effect = side_effect
1896 VerifyVnfs(self.data, self.job_id).run()
1897 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1899 @mock.patch.object(time, "sleep")
1900 @mock.patch.object(restcall, "call_req")
1901 def test_verify_vnfs_thread_when_do_func_test_job_process_error(self, mock_call_req, mock_sleep):
1904 "responseDescriptor": {
1905 "status": JOB_MODEL_STATUS.FINISHED,
1908 "statusDescription": ""
1913 "responseDescriptor": {
1914 "status": JOB_MODEL_STATUS.FINISHED,
1917 "statusDescription": ""
1922 "responseDescriptor": {
1923 "status": JOB_MODEL_STATUS.ERROR,
1924 "progress": JOB_PROGRESS.ERROR,
1926 "statusDescription": "already onBoarded"
1930 "/api/nslcm/v1/vnfpackage":
1931 [0, json.JSONEncoder().encode(job_info_1), "200"],
1932 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1933 [0, json.JSONEncoder().encode(job_info_1), "200"],
1934 "/api/nslcm/v1/ns/ns_vnfs":
1935 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1936 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1937 [0, json.JSONEncoder().encode(job_info_2), "200"],
1938 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1939 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1940 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1941 [0, json.JSONEncoder().encode(job_info_task), "200"],
1942 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1943 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1946 def side_effect(*args):
1947 return mock_vals[args[4]]
1949 mock_call_req.side_effect = side_effect
1950 VerifyVnfs(self.data, self.job_id).run()
1951 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.FINISHED)
1953 @mock.patch.object(time, "sleep")
1954 @mock.patch.object(restcall, "call_req")
1955 def test_verify_vnfs_thread_when_do_func_test_failed_to_get_func_test_result(self, mock_call_req, mock_sleep):
1958 "responseDescriptor": {
1959 "status": JOB_MODEL_STATUS.FINISHED,
1962 "statusDescription": ""
1967 "responseDescriptor": {
1968 "status": JOB_MODEL_STATUS.FINISHED,
1971 "statusDescription": ""
1976 "responseDescriptor": {
1977 "status": JOB_MODEL_STATUS.FINISHED,
1980 "statusDescription": "already onBoarded"
1984 "/api/nslcm/v1/vnfpackage":
1985 [0, json.JSONEncoder().encode(job_info_1), "200"],
1986 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1987 [0, json.JSONEncoder().encode(job_info_1), "200"],
1988 "/api/nslcm/v1/ns/ns_vnfs":
1989 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1990 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1991 [0, json.JSONEncoder().encode(job_info_2), "200"],
1992 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1993 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1994 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1995 [0, json.JSONEncoder().encode(job_info_task), "200"],
1996 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1997 [1, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
2000 def side_effect(*args):
2001 return mock_vals[args[4]]
2003 mock_call_req.side_effect = side_effect
2004 VerifyVnfs(self.data, self.job_id).run()
2005 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
2008 class TestLcmNotifyViews(TestCase):
2010 self.client = Client()
2013 "operation": "Instantiate",
2016 "affectedVnfc": [{"vnfcInstanceId": "vnfc_instance_id",
2018 "changeType": VNFC_CHANGE_TYPE.ADDED,
2023 "affectedVl": [{"vlInstanceId": "vl_instance_id",
2025 "changeType": VNFC_CHANGE_TYPE.ADDED,
2026 "networkResource": {
2027 "resourceType": NETWORK_RESOURCE_TYPE.NETWORK,
2028 "resourceId": "resource_id",
2029 "resourceName": "resource_name"
2032 "affectedCp": [{"changeType": VNFC_CHANGE_TYPE.ADDED,
2033 "virtualLinkInstanceId": "virtual_link_instance_id",
2034 "cpInstanceId": "cp_instance_id",
2037 "ownerId": "owner_id",
2040 "resourceId": "resource_id",
2041 "resourceName": "resource_name",
2043 "ipAddress": "ip_address",
2044 "macAddress": "mac_address",
2048 "affectedVirtualStorage": [{}]
2050 self.nf_inst_id = "test_lcm_notify"
2051 self.m_nf_inst_id = "test_lcm_notify_m_nf_inst_id"
2052 self.vnfm_inst_id = "test_lcm_notify_vnfm_inst_id"
2053 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
2054 self.url = "/api/nslcm/v1/ns/%s/vnfs/%s/Notify" % (self.m_nf_inst_id, self.vnfm_inst_id)
2055 NfInstModel.objects.all().delete()
2056 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
2057 vnfm_inst_id=self.vnfm_inst_id,
2058 status=VNF_STATUS.NULL,
2059 mnfinstid=self.m_nf_inst_id,
2060 vnfd_model=self.vnfd_model
2064 NfInstModel.objects.all().delete()
2065 VNFCInstModel.objects.all().delete()
2066 VmInstModel.objects.all().delete()
2067 VLInstModel.objects.all().delete()
2068 PortInstModel.objects.all().delete()
2069 CPInstModel.objects.all().delete()
2071 def test_lcm_notify_view_when_change_type_is_added(self):
2072 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, self.data).do_biz()
2073 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id", vmid="vm_id",
2074 nfinstid=self.nf_inst_id)
2075 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2076 vm_inst = VmInstModel.objects.get(vmid="vm_id", vimid="vim_id", resouceid="vm_id", insttype=INST_TYPE.VNF,
2077 instid=self.nf_inst_id, vmname="vm_name", hostid='1')
2078 self.assertIsInstance(vm_inst, VmInstModel)
2079 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="vld_id", vlinstancename="resource_name",
2080 ownertype=0, ownerid=self.nf_inst_id, relatednetworkid="resource_id",
2082 self.assertIsInstance(vl_inst, VLInstModel)
2083 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_id",
2084 resourceid="resource_id", name="resource_name", instid="inst_id",
2085 cpinstanceid="cp_instance_id", bandwidth='unknown',
2086 operationalstate='active', ipaddress="ip_address",
2087 macaddress='mac_address', floatipaddress='unknown',
2088 serviceipaddress='unknown', typevirtualnic='unknown',
2089 sfcencapsulation='gre', direction='unknown', tenant="tenant")
2090 self.assertIsInstance(port_inst, PortInstModel)
2091 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2092 ownerid=self.nf_inst_id, relatedtype=2, status='active')
2093 self.assertIsInstance(cp_inst, CPInstModel)
2095 def test_lcm_notify_view__when_change_type_is_added_when_nf_not_exists(self):
2096 NfInstModel.objects.all().delete()
2099 "operation": "Instantiate",
2104 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2105 self.assertEqual(1, 0)
2107 self.assertEqual(1, 1)
2109 def test_lcm_notify_view_when_change_type_is_removeed(self):
2110 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2111 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2112 CPInstModel.objects.create(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0, ownerid=self.nf_inst_id,
2113 relatedtype=2, relatedport="related_port", status="active")
2116 "operation": "Instantiate",
2119 "affectedVnfc": [{"vnfcInstanceId": "vnfc_instance_id",
2121 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2126 "affectedVl": [{"vlInstanceId": "vl_instance_id",
2128 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2129 "networkResource": {
2130 "resourceType": NETWORK_RESOURCE_TYPE.NETWORK,
2131 "resourceId": "resource_id",
2132 "resourceName": "resource_name"
2135 "affectedCp": [{"changeType": VNFC_CHANGE_TYPE.REMOVED,
2136 "virtualLinkInstanceId": "virtual_link_instance_id",
2137 "cpInstanceId": "cp_instance_id",
2140 "ownerId": "owner_id",
2143 "resourceId": "resource_id",
2144 "resourceName": "resource_name",
2146 "ipAddress": "ip_address",
2147 "macAddress": "mac_address",
2151 "affectedVirtualStorage": [{}]
2153 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2154 vnfc_inst = VNFCInstModel.objects.filter(vnfcinstanceid="vnfc_instance_id")
2155 self.assertEqual(len(vnfc_inst), 0)
2156 vl_inst = VLInstModel.objects.filter(vlinstanceid="vl_instance_id")
2157 self.assertEqual(len(vl_inst), 0)
2158 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_id",
2159 resourceid="resource_id", name="resource_name", instid="inst_id",
2160 cpinstanceid="cp_instance_id", bandwidth='unknown',
2161 operationalstate='active', ipaddress="ip_address",
2162 macaddress='mac_address', floatipaddress='unknown',
2163 serviceipaddress='unknown', typevirtualnic='unknown',
2164 sfcencapsulation='gre', direction='unknown', tenant="tenant")
2165 self.assertIsInstance(port_inst, PortInstModel)
2166 cp_inst = CPInstModel.objects.filter(cpinstanceid="cp_instance_id")
2167 self.assertEqual(len(cp_inst), 0)
2169 def test_lcm_notify_view_when_change_type_is_modified(self):
2170 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2171 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2172 CPInstModel.objects.create(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0, ownerid=self.nf_inst_id,
2173 relatedtype=2, relatedport="related_port")
2176 "operation": "Instantiate",
2179 "affectedVnfc": [{"vnfcInstanceId": "vnfc_instance_id",
2181 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2186 "affectedVl": [{"vlInstanceId": "vl_instance_id",
2188 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2189 "networkResource": {
2190 "resourceType": NETWORK_RESOURCE_TYPE.NETWORK,
2191 "resourceId": "resource_id",
2192 "resourceName": "resource_name"
2195 "affectedCp": [{"changeType": VNFC_CHANGE_TYPE.MODIFIED,
2196 "virtualLinkInstanceId": "virtual_link_instance_id",
2197 "cpInstanceId": "cp_instance_id",
2200 "ownerId": "owner_id",
2203 "resourceId": "resource_id",
2204 "resourceName": "resource_name",
2206 "ipAddress": "ip_address",
2207 "macAddress": "mac_address",
2211 "affectedVirtualStorage": [{}]
2213 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2214 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id", vmid="vm_id",
2215 nfinstid=self.nf_inst_id)
2216 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2217 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="vld_id", vlinstancename="resource_name",
2218 ownertype=0, ownerid=self.nf_inst_id, relatednetworkid="resource_id",
2220 self.assertIsInstance(vl_inst, VLInstModel)
2221 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_id",
2222 resourceid="resource_id", name="resource_name", instid="inst_id",
2223 cpinstanceid="cp_instance_id", bandwidth='unknown',
2224 operationalstate='active', ipaddress="ip_address",
2225 macaddress='mac_address', floatipaddress='unknown',
2226 serviceipaddress='unknown', typevirtualnic='unknown',
2227 sfcencapsulation='gre', direction='unknown', tenant="tenant")
2228 self.assertIsInstance(port_inst, PortInstModel)
2229 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2230 ownerid=self.nf_inst_id, relatedtype=2)
2231 self.assertIsInstance(cp_inst, CPInstModel)
2234 class TestVnfNotifyView(TestCase):
2236 self.client = Client()
2237 self.nf_inst_id = "test_vnf_notify"
2238 self.m_nf_inst_id = "test_vnf_notify_m_nf_inst_id"
2239 self.vnfm_inst_id = "test_vnf_notify_vnfm_inst_id"
2240 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
2241 self.url = "/api/nslcm/v2/ns/%s/vnfs/%s/Notify" % (self.vnfm_inst_id, self.m_nf_inst_id)
2244 "notificationType": "VnfLcmOperationOccurrenceNotification",
2245 "subscriptionId": "1111",
2246 "timeStamp": "1111",
2247 "notificationStatus": "START",
2248 "operationState": "STARTING",
2249 "vnfInstanceId": self.nf_inst_id,
2250 "operation": "INSTANTIATE",
2251 "isAutomaticInvocation": "1111",
2252 "vnfLcmOpOccId": "1111",
2253 "affectedVnfcs": [{"id": "vnfc_instance_id",
2255 "changeType": VNFC_CHANGE_TYPE.ADDED,
2256 "computeResource": {
2257 "vimConnectionId": "vim_connection_id",
2258 "resourceId": "resource_id"
2261 "affectedVirtualLinks": [{"id": "vl_instance_id",
2262 "virtualLinkDescId": "virtual_link_desc_id",
2263 "changeType": VNFC_CHANGE_TYPE.ADDED,
2264 "networkResource": {
2265 "vimLevelResourceType": "network",
2266 "resourceId": "resource_id"
2268 "changedExtConnectivity": [{"id": "virtual_link_instance_id",
2269 "extLinkPorts": [{"cpInstanceId": "cp_instance_id",
2272 "vimConnectionId": "vim_connection_id",
2273 "resourceId": "resource_id",
2274 "resourceProviderId": "resourceProviderId",
2276 "ipAddress": "ipAddress",
2277 "macAddress": "macAddress",
2279 "networkId": "networkId",
2280 "subnetId": "subnetId"
2285 NfInstModel.objects.all().delete()
2286 VNFCInstModel.objects.all().delete()
2287 VmInstModel.objects.all().delete()
2288 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
2289 vnfm_inst_id=self.vnfm_inst_id,
2290 status=VNF_STATUS.NULL,
2291 mnfinstid=self.m_nf_inst_id,
2292 vnfd_model=self.vnfd_model
2296 NfInstModel.objects.all().delete()
2297 VNFCInstModel.objects.all().delete()
2298 VmInstModel.objects.all().delete()
2299 VLInstModel.objects.all().delete()
2300 PortInstModel.objects.all().delete()
2301 CPInstModel.objects.all().delete()
2303 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_added(self):
2304 # response = self.client.post(self.url, data=self.data)
2305 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, self.data).do_biz()
2306 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id",
2307 nfinstid=self.nf_inst_id, vmid="resource_id")
2308 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2309 vm_inst = VmInstModel.objects.get(vmid="resource_id", vimid="vim_connection_id", resouceid="resource_id",
2310 insttype=INST_TYPE.VNF, instid=self.nf_inst_id, vmname="resource_id",
2312 self.assertIsInstance(vm_inst, VmInstModel)
2313 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="virtual_link_desc_id",
2314 vlinstancename="resource_id", ownertype=0, ownerid=self.nf_inst_id,
2315 relatednetworkid="resource_id", vltype=0)
2316 self.assertIsInstance(vl_inst, VLInstModel)
2317 port_inst = PortInstModel.objects.get(networkid='networkId', subnetworkid='subnetId', vimid="vim_connection_id",
2318 resourceid="resource_id", name="resourceProviderId", instid="instId",
2319 cpinstanceid="cp_instance_id", bandwidth='unknown',
2320 operationalstate='active', ipaddress="ipAddress", macaddress='macAddress',
2321 floatipaddress='unknown', serviceipaddress='unknown',
2322 typevirtualnic='unknown', sfcencapsulation='gre', direction='unknown',
2324 self.assertIsInstance(port_inst, PortInstModel)
2325 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2326 ownerid=self.nf_inst_id, relatedtype=2, status='active')
2327 self.assertIsInstance(cp_inst, CPInstModel)
2329 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_added_when_nf_not_exists(self):
2332 "notificationType": "VnfLcmOperationOccurrenceNotification",
2333 "subscriptionId": "1111",
2334 "timeStamp": "1111",
2335 "notificationStatus": "START",
2336 "operationState": "STARTING",
2337 "vnfInstanceId": "nf_not_exists",
2338 "operation": "INSTANTIATE",
2339 "isAutomaticInvocation": "1111",
2340 "vnfLcmOpOccId": "1111"
2343 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2344 self.assertEqual(1, 0)
2346 self.assertEqual(1, 1)
2348 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_removed(self):
2349 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2350 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2353 "notificationType": "VnfLcmOperationOccurrenceNotification",
2354 "subscriptionId": "1111",
2355 "timeStamp": "1111",
2356 "notificationStatus": "START",
2357 "operationState": "STARTING",
2358 "vnfInstanceId": self.nf_inst_id,
2359 "operation": "INSTANTIATE",
2360 "isAutomaticInvocation": "1111",
2361 "vnfLcmOpOccId": "1111",
2362 "affectedVnfcs": [{"id": "vnfc_instance_id",
2364 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2365 "computeResource": {
2366 "vimConnectionId": "vim_connection_id",
2367 "resourceId": "resource_id"
2369 "affectedVirtualLinks": [{"id": "vl_instance_id",
2370 "virtualLinkDescId": "virtual_link_desc_id",
2371 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2372 "networkResource": {
2373 "vimLevelResourceType": "network",
2374 "resourceId": "resource_id"
2377 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2378 vnfc_inst = VNFCInstModel.objects.filter(vnfcinstanceid="vnfc_instance_id")
2379 self.assertEqual(len(vnfc_inst), 0)
2380 vl_inst = VLInstModel.objects.filter(vlinstanceid="vl_instance_id")
2381 self.assertEqual(len(vl_inst), 0)
2383 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_modified(self):
2384 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2385 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2388 "notificationType": "VnfLcmOperationOccurrenceNotification",
2389 "subscriptionId": "1111",
2390 "timeStamp": "1111",
2391 "notificationStatus": "START",
2392 "operationState": "STARTING",
2393 "vnfInstanceId": self.nf_inst_id,
2394 "operation": "INSTANTIATE",
2395 "isAutomaticInvocation": "1111",
2396 "vnfLcmOpOccId": "1111",
2397 "affectedVnfcs": [{"id": "vnfc_instance_id",
2399 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2400 "computeResource": {
2401 "vimConnectionId": "vim_connection_id",
2402 "resourceId": "resource_id"
2404 "affectedVirtualLinks": [{"id": "vl_instance_id",
2405 "virtualLinkDescId": "virtual_link_desc_id",
2406 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2407 "networkResource": {
2408 "vimLevelResourceType": "network",
2409 "resourceId": "resource_id"
2411 "changedExtConnectivity": [{"id": "virtual_link_instance_id",
2412 "extLinkPorts": [{"cpInstanceId": "cp_instance_id",
2415 "vimConnectionId": "vim_connection_id",
2416 "resourceId": "resource_id",
2417 "resourceProviderId": "resourceProviderId",
2419 "ipAddress": "ipAddress",
2420 "macAddress": "macAddress",
2422 "networkId": "networkId",
2423 "subnetId": "subnetId"
2428 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2429 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id",
2430 nfinstid=self.nf_inst_id, vmid="resource_id")
2431 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2432 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="virtual_link_desc_id",
2433 vlinstancename="resource_id", ownertype=0, ownerid=self.nf_inst_id,
2434 relatednetworkid="resource_id", vltype=0)
2435 self.assertIsInstance(vl_inst, VLInstModel)
2436 port_inst = PortInstModel.objects.get(networkid='networkId', subnetworkid='subnetId', vimid="vim_connection_id",
2437 resourceid="resource_id", name="resourceProviderId", instid="instId",
2438 cpinstanceid="cp_instance_id", bandwidth='unknown',
2439 operationalstate='active', ipaddress="ipAddress", macaddress='macAddress',
2440 floatipaddress='unknown', serviceipaddress='unknown',
2441 typevirtualnic='unknown', sfcencapsulation='gre', direction='unknown',
2443 self.assertIsInstance(port_inst, PortInstModel)
2444 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2445 ownerid=self.nf_inst_id, relatedtype=2, status='active')
2446 self.assertIsInstance(cp_inst, CPInstModel)
2448 def test_handle_vnf_identifier_creation_notification(self):
2450 vnf_instance_id = "vnf_instance_id"
2452 "timeStamp": "20190809",
2454 HandleVnfIdentifierCreationNotification(vnfm_id, vnf_instance_id, data).do_biz()
2455 nf_inst = NfInstModel.objects.get(mnfinstid=vnf_instance_id, vnfm_inst_id=vnfm_id, create_time="20190809")
2456 self.assertIsInstance(nf_inst, NfInstModel)
2458 def test_handle_vnf_identifier_deletion_notification(self):
2459 nf_inst_id = "nf_inst_id"
2461 vnf_instance_id = "vnf_instance_id"
2462 NfInstModel.objects.create(nfinstid=nf_inst_id,
2463 vnfm_inst_id=vnfm_id,
2464 status=VNF_STATUS.NULL,
2465 mnfinstid=vnf_instance_id,
2466 vnfd_model=self.vnfd_model
2469 "timeStamp": "20190809",
2471 HandleVnfIdentifierDeletionNotification(vnfm_id, vnf_instance_id, data).do_biz()
2472 nf_inst = NfInstModel.objects.filter(mnfinstid=vnf_instance_id, vnfm_inst_id=vnfm_id)
2473 self.assertEqual(len(nf_inst), 0)
2475 def test_handle_vnf_identifier_deletion_notification_when_nf_not_exists(self):
2476 NfInstModel.objects.all().delete()
2477 vnfm_id = "nf_not_exists"
2478 vnf_instance_id = "nf_not_exists"
2480 "timeStamp": "20190809",
2483 HandleVnfIdentifierDeletionNotification(vnfm_id, vnf_instance_id, data).do_biz()
2484 self.assertEqual(1, 0)
2486 self.assertEqual(1, 1)