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
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
98 NSInstModel.objects.all().delete()
99 NfInstModel.objects.all().delete()
101 @mock.patch.object(TerminateVnfs, "run")
102 def test_terminate_vnf_url(self, mock_run):
104 "terminationType": "forceful",
105 "gracefulTerminationTimeout": "600"}
107 response = self.client.post("/api/nslcm/v1/ns/terminatevnf/%s" % self.nf_inst_id, data=req_data)
108 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
110 @mock.patch.object(time, "sleep")
111 @mock.patch.object(restcall, "call_req")
112 @mock.patch.object(SubscriptionDeletion, "send_subscription_deletion_request")
113 def test_terminate_vnf(self, mock_send_subscription_deletion_request, mock_call_req, mock_sleep):
114 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
117 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
120 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
121 [0, json.JSONEncoder().encode(vnfm_info), "200"],
122 "/api/ztevnfmdriver/v1/1/vnfs/111/terminate":
123 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"],
124 "/api/ztevnfmdriver/v1/1/jobs/" + job_id + "?responseId=0":
125 [0, json.JSONEncoder().encode(job_info), "200"],
126 "/api/resmgr/v1/vnf/1":
127 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"]
130 def side_effect(*args):
131 return mock_vals[args[4]]
132 mock_call_req.side_effect = side_effect
133 TerminateVnfs(self.data, self.nf_inst_id, job_id).run()
134 nfinst = NfInstModel.objects.filter(nfinstid=self.nf_inst_id)
136 self.assertEqual(1, 0)
138 self.assertEqual(1, 1)
139 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 100)
141 def test_terminate_vnf_when_vnf_is_dealing(self):
142 NfInstModel.objects.filter(nfinstid=self.nf_inst_id).update(status=VNF_STATUS.TERMINATING)
143 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
144 TerminateVnfs(self.data, self.nf_inst_id, job_id).run()
145 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.FAILED)
146 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 255)
148 @mock.patch.object(time, "sleep")
149 @mock.patch.object(restcall, "call_req")
150 @mock.patch.object(SubscriptionDeletion, "send_subscription_deletion_request")
151 def test_terminate_vnf_when_job_error(self, mock_send_subscription_deletion_request, mock_call_req, mock_sleep):
152 job_id = JobUtil.create_job(JOB_TYPE.VNF, JOB_ACTION.TERMINATE, self.nf_inst_id)
155 "responsedescriptor": {"status": JOB_MODEL_STATUS.ERROR}
158 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
159 [0, json.JSONEncoder().encode(vnfm_info), "200"],
160 "/api/ztevnfmdriver/v1/1/vnfs/111/terminate":
161 [0, json.JSONEncoder().encode({"jobId": job_id}), "200"],
162 "/api/ztevnfmdriver/v1/1/jobs/" + job_id + "?responseId=0":
163 [0, json.JSONEncoder().encode(job_info), "200"]
166 def side_effect(*args):
167 return mock_vals[args[4]]
169 mock_call_req.side_effect = side_effect
170 TerminateVnfs(self.data, self.nf_inst_id, job_id).run()
171 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.FAILED)
172 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, 255)
175 class TestScaleVnfViews(TestCase):
177 self.client = Client()
178 self.nf_inst_id = str(uuid.uuid4())
179 self.url = "/api/nslcm/v1/ns/ns_vnfs/%s/scaling" % self.nf_inst_id
184 "aspectId": "demo_aspect1",
186 "additionalParam": {}
189 NfInstModel.objects.create(nfinstid=self.nf_inst_id, vnfm_inst_id="vnfm_inst_id_001",
190 mnfinstid="m_nf_inst_id_001")
193 NfInstModel.objects.all().delete()
195 # def test_scale_vnf_view(self):
196 # response = self.client.post(self.url, self.data)
197 # self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
199 @mock.patch.object(time, "sleep")
200 @mock.patch.object(restcall, "call_req")
201 def test_scale_vnf_success(self, mock_call_req, mock_sleep):
202 scale_service = NFManualScaleService(self.nf_inst_id, self.data)
204 "jobId": scale_service.job_id,
205 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
208 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
209 [0, json.JSONEncoder().encode(vnfm_info), "200"],
210 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/scale":
211 [0, json.JSONEncoder().encode({"jobId": scale_service.job_id}), "200"],
212 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + scale_service.job_id + "?responseId=0":
213 [0, json.JSONEncoder().encode(job_info), "200"]
216 def side_effect(*args):
217 return mock_vals[args[4]]
218 mock_call_req.side_effect = side_effect
220 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
221 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
223 jobs = JobModel.objects.get(jobid=scale_service.job_id)
224 self.assertEqual(JOB_PROGRESS.FINISHED, jobs.progress)
226 @mock.patch.object(time, "sleep")
227 @mock.patch.object(restcall, "call_req")
228 def test_scale_vnf_when_job_fail(self, mock_call_req, mock_sleep):
229 scale_service = NFManualScaleService(self.nf_inst_id, self.data)
231 "jobId": scale_service.job_id,
232 "responsedescriptor": {"status": JOB_MODEL_STATUS.ERROR}
235 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
236 [0, json.JSONEncoder().encode(vnfm_info), "200"],
237 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/scale":
238 [0, json.JSONEncoder().encode({"jobId": scale_service.job_id}), "200"],
239 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + scale_service.job_id + "?responseId=0":
240 [0, json.JSONEncoder().encode(job_info), "200"]
243 def side_effect(*args):
244 return mock_vals[args[4]]
246 mock_call_req.side_effect = side_effect
248 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
249 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
250 jobs = JobModel.objects.get(jobid=scale_service.job_id)
251 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
253 def test_scale_vnf_when_exception(self):
258 "aspectId": "demo_aspect1",
263 "aspectId": "demo_aspect2",
268 scale_service = NFManualScaleService(self.nf_inst_id, req_data)
270 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
271 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
273 jobs = JobModel.objects.get(jobid=scale_service.job_id)
274 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
276 def test_scale_vnf_when_nf_instance_does_not_exist(self):
281 "aspectId": "demo_aspect1",
285 scale_service = NFManualScaleService("nf_instance_does_not_exist", req_data)
288 jobs = JobModel.objects.get(jobid=scale_service.job_id)
289 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
291 def test_scale_vnf_when_scale_vnf_data_does_not_exist(self):
295 scale_service = NFManualScaleService(self.nf_inst_id, req_data)
297 nsIns = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
298 self.assertEqual(nsIns.status, VNF_STATUS.ACTIVE)
300 jobs = JobModel.objects.get(jobid=scale_service.job_id)
301 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
304 class TestHealVnfViews(TestCase):
306 self.client = Client()
307 self.ns_inst_id = str(uuid.uuid4())
308 self.nf_inst_id = str(uuid.uuid4())
317 "additionalParams": {
323 NSInstModel(id=self.ns_inst_id, name="ns_name").save()
324 NfInstModel.objects.create(nfinstid=self.nf_inst_id, status=VNF_STATUS.NULL, vnfm_inst_id="vnfm_inst_id_001",
325 mnfinstid="m_nf_inst_id_001")
326 NfInstModel.objects.create(nfinstid="non_vud_id", status=VNF_STATUS.NULL, vnfm_inst_id="vnfm_inst_id_001",
327 mnfinstid="m_nf_inst_id_001")
328 VNFCInstModel.objects.create(nfinstid=self.nf_inst_id, vmid="vm_id_001", vduid="vdu_id_001")
329 VmInstModel.objects.create(resouceid="vm_id_001", vmname="vm_name_001")
332 NSInstModel.objects.all().delete()
333 NfInstModel.objects.all().delete()
335 @mock.patch.object(time, "sleep")
336 @mock.patch.object(restcall, "call_req")
337 def test_heal_vnf_success(self, mock_call_req, mock_sleep):
338 heal_service = NFHealService(self.nf_inst_id, self.data)
340 "/test/bins/1?timeout=15000":
341 [0, json.JSONEncoder().encode(['{"powering-off": "", "instance_id": "vm_id_001", '
342 '"display_name": ""}']), "200"],
343 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
344 [0, json.JSONEncoder().encode(vnfm_info), "200"],
345 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/heal":
346 [0, json.JSONEncoder().encode({"jobId": heal_service.job_id}), "200"],
347 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + heal_service.job_id + "?responseId=0":
348 [0, json.JSONEncoder().encode({
349 "jobId": heal_service.job_id,
350 "responsedescriptor": {
351 "status": JOB_MODEL_STATUS.FINISHED,
352 "responsehistorylist": [{
354 "status": JOB_MODEL_STATUS.PROCESSING,
356 "statusdescription": "creating",
363 def side_effect(*args):
364 return mock_vals[args[4]]
366 mock_call_req.side_effect = side_effect
368 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.ACTIVE)
369 jobs = JobModel.objects.get(jobid=heal_service.job_id)
370 self.assertEqual(JOB_PROGRESS.FINISHED, jobs.progress)
372 def test_heal_vnf_when_non_existing_vnf(self, ):
373 heal_service = NFHealService("on_existing_vnf", self.data)
375 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.NULL)
376 jobs = JobModel.objects.get(jobid=heal_service.job_id)
377 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
379 def test_heal_vnf_when_additional_params_non_exist(self):
380 data = {"action": "vmReset"}
381 heal_service = NFHealService(self.nf_inst_id, data)
383 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.NULL)
384 jobs = JobModel.objects.get(jobid=heal_service.job_id)
385 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
387 def test_heal_vnf_when_non_vud_id(self, ):
388 heal_service = NFHealService("non_vud_id", self.data)
390 self.assertEqual(NfInstModel.objects.get(nfinstid="non_vud_id").status, VNF_STATUS.NULL)
391 jobs = JobModel.objects.get(jobid=heal_service.job_id)
392 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
394 @mock.patch.object(restcall, "call_req")
395 def test_heal_vnf_when_no_vnfm_job_id(self, mock_call_req):
396 heal_service = NFHealService(self.nf_inst_id, self.data)
398 "/test/bins/1?timeout=15000":
399 [0, json.JSONEncoder().encode(['{"powering-off": "", "instance_id": "vm_id_001", '
400 '"display_name": ""}']), "200"],
401 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
402 [0, json.JSONEncoder().encode(vnfm_info), "200"],
403 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/heal":
404 [0, json.JSONEncoder().encode({}), "200"]
407 def side_effect(*args):
408 return mock_vals[args[4]]
410 mock_call_req.side_effect = side_effect
412 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.ACTIVE)
413 jobs = JobModel.objects.get(jobid=heal_service.job_id)
414 self.assertEqual(JOB_PROGRESS.FINISHED, jobs.progress)
416 @mock.patch.object(time, "sleep")
417 @mock.patch.object(restcall, "call_req")
418 def test_heal_vnf_when_job_bot_finish(self, mock_call_req, mock_sleep):
419 heal_service = NFHealService(self.nf_inst_id, self.data)
421 "/test/bins/1?timeout=15000":
422 [0, json.JSONEncoder().encode(['{"powering-off": "", "instance_id": "vm_id_001", '
423 '"display_name": ""}']), "200"],
424 "/external-system/esr-vnfm-list/esr-vnfm/vnfm_inst_id_001?depth=all":
425 [0, json.JSONEncoder().encode(vnfm_info), "200"],
426 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/vnfs/m_nf_inst_id_001/heal":
427 [0, json.JSONEncoder().encode({"jobId": heal_service.job_id}), "200"],
428 "/api/ztevnfmdriver/v1/vnfm_inst_id_001/jobs/" + heal_service.job_id + "?responseId=0":
429 [0, json.JSONEncoder().encode({
430 "jobId": heal_service.job_id,
431 "responsedescriptor": {
432 "status": JOB_MODEL_STATUS.ERROR,
433 "responsehistorylist": [{
435 "status": JOB_MODEL_STATUS.PROCESSING,
437 "statusdescription": "creating",
444 def side_effect(*args):
445 return mock_vals[args[4]]
447 mock_call_req.side_effect = side_effect
449 self.assertEqual(NfInstModel.objects.get(nfinstid=self.nf_inst_id).status, VNF_STATUS.HEALING)
450 jobs = JobModel.objects.get(jobid=heal_service.job_id)
451 self.assertEqual(JOB_PROGRESS.ERROR, jobs.progress)
454 class TestGetVnfmInfoViews(TestCase):
456 self.client = Client()
457 self.vnfm_id = str(uuid.uuid4())
462 @mock.patch.object(restcall, "call_req")
463 def test_get_vnfm_info(self, mock_call_req):
465 "vnfm-id": "example-vnfm-id-val-62576",
466 "vim-id": "example-vim-id-val-35114",
467 "certificate-url": "example-certificate-url-val-90242",
468 "esr-system-info-list": {
471 "esr-system-info-id": "example-esr-system-info-id-val-78484",
472 "system-name": "example-system-name-val-23790",
473 "type": "example-type-val-52596",
474 "vendor": "example-vendor-val-47399",
475 "version": "example-version-val-42051",
476 "service-url": "example-service-url-val-10731",
477 "user-name": "example-user-name-val-65946",
478 "password": "example-password-val-22505",
479 "system-type": "example-system-type-val-27221",
480 "protocal": "example-protocal-val-54632",
481 "ssl-cacert": "example-ssl-cacert-val-45965",
482 "ssl-insecure": True,
483 "ip-address": "example-ip-address-val-19212",
484 "port": "example-port-val-57641",
485 "cloud-domain": "example-cloud-domain-val-26296",
486 "default-tenant": "example-default-tenant-val-87724"
491 r1 = [0, json.JSONEncoder().encode(vnfm_info_aai), "200"]
492 mock_call_req.side_effect = [r1]
493 esr_system_info = ignore_case_get(ignore_case_get(vnfm_info_aai, "esr-system-info-list"), "esr-system-info")
495 "vnfmId": vnfm_info_aai["vnfm-id"],
496 "name": vnfm_info_aai["vnfm-id"],
497 "type": ignore_case_get(esr_system_info[0], "type"),
498 "vimId": vnfm_info_aai["vim-id"],
499 "vendor": ignore_case_get(esr_system_info[0], "vendor"),
500 "version": ignore_case_get(esr_system_info[0], "version"),
501 "description": "vnfm",
502 "certificateUrl": vnfm_info_aai["certificate-url"],
503 "url": ignore_case_get(esr_system_info[0], "service-url"),
504 "userName": ignore_case_get(esr_system_info[0], "user-name"),
505 "password": ignore_case_get(esr_system_info[0], "password"),
509 response = self.client.get("/api/nslcm/v1/vnfms/%s" % self.vnfm_id)
510 self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
511 context = json.loads(response.content)
512 self.assertEqual(expect_data, context)
515 class TestGetVimInfoViews(TestCase):
517 self.client = Client()
518 self.vim_id = {"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}
523 @mock.patch.object(restcall, "call_req")
524 def test_get_vim_info(self, mock_call_req):
525 r1 = [0, json.JSONEncoder().encode(vim_info), "200"]
526 mock_call_req.side_effect = [r1]
527 esr_system_info = ignore_case_get(ignore_case_get(vim_info, "esr-system-info-list"), "esr-system-info")
529 "vimId": self.vim_id,
531 "url": ignore_case_get(esr_system_info[0], "service-url"),
532 "userName": ignore_case_get(esr_system_info[0], "user-name"),
533 "password": ignore_case_get(esr_system_info[0], "password"),
534 # "tenant": ignore_case_get(tenants[0], "tenant-id"),
535 "tenant": ignore_case_get(esr_system_info[0], "default-tenant"),
536 "vendor": ignore_case_get(esr_system_info[0], "vendor"),
537 "version": ignore_case_get(esr_system_info[0], "version"),
538 "description": "vim",
540 "type": ignore_case_get(esr_system_info[0], "type"),
544 # response = self.client.get("/api/nslcm/v1/vims/%s" % self.vim_id)
545 response = self.client.get("/api/nslcm/v1/vims/%s/%s" % (self.vim_id["cloud_owner"], self.vim_id["cloud_regionid"]))
546 self.assertEqual(status.HTTP_200_OK, response.status_code)
547 context = json.loads(response.content)
548 self.assertEqual(expect_data["url"], context["url"])
551 class TestPlaceVnfViews(TestCase):
553 self.vnf_inst_id = "1234"
555 self.client = Client()
556 OOFDataModel.objects.all().delete()
557 OOFDataModel.objects.create(
559 transaction_id="1234",
560 request_status="init",
561 request_module_name=self.vnf_id,
562 service_resource_id=self.vnf_inst_id,
570 OOFDataModel.objects.all().delete()
572 @mock.patch.object(restcall, "call_req")
573 def test_place_vnf(self, mock_call_req):
576 "flavorName": "HPA.flavor.1",
580 PlaceVnfs(vnf_place_request).extract()
581 db_info = OOFDataModel.objects.filter(request_id=vnf_place_request.get("requestId"), transaction_id=vnf_place_request.get("transactionId"))
582 self.assertEqual(db_info[0].request_status, "completed")
583 self.assertEqual(db_info[0].vim_id, "CloudOwner1_DLLSTX1A")
584 self.assertEqual(db_info[0].cloud_owner, "CloudOwner1")
585 self.assertEqual(db_info[0].cloud_region_id, "DLLSTX1A")
586 self.assertEqual(db_info[0].vdu_info, json.dumps(vdu_info_json))
588 def test_place_vnf_with_invalid_response(self):
591 "transactionId": "1234",
592 "statusMessage": "xx",
593 "requestStatus": "pending",
595 "placementSolutions": [
598 "resourceModuleName": self.vnf_id,
599 "serviceResourceId": self.vnf_inst_id,
601 "identifierType": "serviceInstanceId",
605 "cloudOwner": "CloudOwner1 "
611 "licenseSolutions": [
613 "resourceModuleName": "string",
614 "serviceResourceId": "string",
615 "entitlementPoolUUID": [
618 "licenseKeyGroupUUID": [
621 "entitlementPoolInvariantUUID": [
624 "licenseKeyGroupInvariantUUID": [
631 PlaceVnfs(resp).extract()
632 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
633 self.assertEqual(db_info[0].request_status, "pending")
634 self.assertEqual(db_info[0].vim_id, "none")
635 self.assertEqual(db_info[0].cloud_owner, "none")
636 self.assertEqual(db_info[0].cloud_region_id, "none")
637 self.assertEqual(db_info[0].vdu_info, "none")
639 def test_place_vnf_with_no_assignment_info(self):
642 "transactionId": "1234",
643 "statusMessage": "xx",
644 "requestStatus": "completed",
646 "placementSolutions": [
649 "resourceModuleName": self.vnf_id,
650 "serviceResourceId": self.vnf_inst_id,
652 "identifierType": "serviceInstanceId",
656 "cloudOwner": "CloudOwner1 "
661 "licenseSolutions": [
663 "resourceModuleName": "string",
664 "serviceResourceId": "string",
665 "entitlementPoolUUID": [
668 "licenseKeyGroupUUID": [
671 "entitlementPoolInvariantUUID": [
674 "licenseKeyGroupInvariantUUID": [
681 PlaceVnfs(resp).extract()
682 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
683 self.assertEqual(db_info[0].request_status, "completed")
684 self.assertEqual(db_info[0].vim_id, "none")
685 self.assertEqual(db_info[0].cloud_owner, "none")
686 self.assertEqual(db_info[0].cloud_region_id, "none")
687 self.assertEqual(db_info[0].vdu_info, "none")
689 def test_place_vnf_no_directives(self):
692 "transactionId": "1234",
693 "statusMessage": "xx",
694 "requestStatus": "completed",
696 "placementSolutions": [
699 "resourceModuleName": self.vnf_id,
700 "serviceResourceId": self.vnf_inst_id,
702 "identifierType": "serviceInstanceId",
706 "cloudOwner": "CloudOwner1 "
709 {"key": "locationId",
718 "resourceModuleName": "string",
719 "serviceResourceId": "string",
720 "entitlementPoolUUID": [
723 "licenseKeyGroupUUID": [
726 "entitlementPoolInvariantUUID": [
729 "licenseKeyGroupInvariantUUID": [
736 PlaceVnfs(resp).extract()
737 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
738 self.assertEqual(db_info[0].request_status, "completed")
739 self.assertEqual(db_info[0].vim_id, "none")
740 self.assertEqual(db_info[0].cloud_owner, "none")
741 self.assertEqual(db_info[0].cloud_region_id, "none")
742 self.assertEqual(db_info[0].vdu_info, "none")
744 def test_place_vnf_with_no_solution(self):
747 "transactionId": "1234",
748 "statusMessage": "xx",
749 "requestStatus": "completed",
751 "placementSolutions": [],
752 "licenseSoutions": []
755 PlaceVnfs(resp).extract()
756 db_info = OOFDataModel.objects.filter(request_id=resp.get("requestId"), transaction_id=resp.get("transactionId"))
757 self.assertEqual(db_info[0].request_status, "completed")
758 self.assertEqual(db_info[0].vim_id, "none")
759 self.assertEqual(db_info[0].cloud_owner, "none")
760 self.assertEqual(db_info[0].cloud_region_id, "none")
761 self.assertEqual(db_info[0].vdu_info, "none")
764 class TestGrantVnfsViews(TestCase):
766 self.vnf_inst_id = str(uuid.uuid4())
768 "vnfInstanceId": self.vnf_inst_id,
769 "lifecycleOperation": LIFE_CYCLE_OPERATION.INSTANTIATE,
770 "addResource": [{"type": RESOURCE_CHANGE_TYPE.VDU, "vdu": "vdu_grant_vnf_add_resources"}],
772 "vnfmid": "vnfm_inst_id_001",
773 "vimid": '{"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}'
776 self.client = Client()
777 self.url = "/api/nslcm/v1/ns/grantvnf"
778 NfInstModel(mnfinstid=self.vnf_inst_id, nfinstid="vnf_inst_id_001", package_id="package_id_001",
779 vnfm_inst_id="vnfm_inst_id_001").save()
782 OOFDataModel.objects.all().delete()
783 NfInstModel.objects.all().delete()
785 # @mock.patch.object(restcall, "call_req")
786 # def test_nf_grant_view(self, mock_call_req):
788 # "/api/catalog/v1/vnfpackages/package_id_001":
789 # [0, json.JSONEncoder().encode(nf_package_info), "200"],
790 # "/api/resmgr/v1/resource/grant":
791 # [1, json.JSONEncoder().encode({}), "200"],
792 # "/cloud-infrastructure/cloud-regions/cloud-region/VCPE/RegionOne?depth=all":
793 # [0, json.JSONEncoder().encode(vim_info), "201"],
796 # def side_effect(*args):
797 # return mock_vals[args[4]]
799 # mock_call_req.side_effect = side_effect
801 # "vnfInstanceId": self.vnf_inst_id,
802 # "lifecycleOperation": LIFE_CYCLE_OPERATION.INSTANTIATE
804 # response = self.client.post(self.url, data=data)
805 # self.assertEqual(response.status_code, status.HTTP_201_CREATED)
807 @mock.patch.object(restcall, "call_req")
808 def test_nf_grant_view_when_add_resource(self, mock_call_req):
810 "/api/catalog/v1/vnfpackages/package_id_001":
811 [0, json.JSONEncoder().encode(nf_package_info), "200"],
812 "/api/resmgr/v1/resource/grant":
813 [1, json.JSONEncoder().encode({}), "200"],
814 "/cloud-infrastructure/cloud-regions/cloud-region/VCPE/RegionOne?depth=all":
815 [0, json.JSONEncoder().encode(vim_info), "201"],
818 def side_effect(*args):
819 return mock_vals[args[4]]
820 mock_call_req.side_effect = side_effect
821 resp = GrantVnfs(json.dumps(self.data), "").send_grant_vnf_to_resMgr()
822 return_success = {"vim": {"accessInfo": {"tenant": "admin"},
823 "vimId": "example-cloud-owner-val-97336_example-cloud-region-id-val-35532"}}
824 self.assertEqual(resp, return_success)
826 @mock.patch.object(restcall, "call_req")
827 def test_nf_grant_view_when_remove_resource(self, mock_call_req):
829 "/api/catalog/v1/vnfpackages/package_id_001":
830 [0, json.JSONEncoder().encode(nf_package_info), "200"],
831 "/api/resmgr/v1/resource/grant":
832 [1, json.JSONEncoder().encode({}), "200"],
833 "/cloud-infrastructure/cloud-regions/cloud-region/VCPE/RegionOne?depth=all":
834 [0, json.JSONEncoder().encode(vim_info), "201"],
837 def side_effect(*args):
838 return mock_vals[args[4]]
840 mock_call_req.side_effect = side_effect
841 self.data.pop("addResource")
842 self.data["removeResource"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
843 resp = GrantVnfs(json.dumps(self.data), "").send_grant_vnf_to_resMgr()
844 return_success = {"vim": {"accessInfo": {"tenant": "admin"},
845 "vimId": "example-cloud-owner-val-97336_example-cloud-region-id-val-35532"}}
846 self.assertEqual(resp, return_success)
849 class TestGrantVnfViews(TestCase):
851 self.vnf_inst_id = str(uuid.uuid4())
853 "vnfInstanceId": self.vnf_inst_id,
854 "vnfLcmOpOccId": "vnf_lcm_op_occ_id",
855 "addResources": [{"vdu": "vdu_grant_vnf_add_resources"}],
856 "operation": "INSTANTIATE"
858 self.client = Client()
859 vdu_info_dict = [{"vduName": "vg", "flavorName": "flavor_1", "flavorId": "flavor_id_001", "directive": []}]
860 OOFDataModel(request_id="request_id_001", transaction_id="transaction_id_001", request_status="done",
861 request_module_name="vg", service_resource_id=self.vnf_inst_id, vim_id="cloudOwner_casa",
862 cloud_owner="cloudOwner", cloud_region_id="casa", vdu_info=json.dumps(vdu_info_dict)).save()
863 NfInstModel(mnfinstid=self.vnf_inst_id, nfinstid="vnf_inst_id_001", package_id="package_id_001",
864 vnfm_inst_id="vnfm_id_001").save()
867 OOFDataModel.objects.all().delete()
868 NfInstModel.objects.all().delete()
870 @mock.patch.object(resmgr, "grant_vnf")
871 def test_vnf_grant_view(self, mock_grant):
872 resmgr_grant_resp = {
874 "vimId": "cloudOwner_casa",
880 mock_grant.return_value = resmgr_grant_resp
881 self.data.pop("addResources")
882 response = self.client.post("/api/nslcm/v2/grants", data=self.data)
883 self.assertEqual(response.status_code, status.HTTP_201_CREATED)
884 self.assertEqual(response.data["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
885 self.assertEqual(response.data["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
886 self.assertEqual(response.data["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
888 @mock.patch.object(restcall, "call_req")
889 @mock.patch.object(resmgr, "grant_vnf")
890 def test_exec_grant_when_add_resources_success(self, mock_grant, mock_call_req):
892 "/api/catalog/v1/vnfpackages/package_id_001":
893 [0, json.JSONEncoder().encode(nf_package_info), "200"],
896 def side_effect(*args):
897 return mock_vals[args[4]]
899 mock_call_req.side_effect = side_effect
900 resmgr_grant_resp = {
902 "vimId": "cloudOwner_casa",
908 mock_grant.return_value = resmgr_grant_resp
909 resp = GrantVnf(json.dumps(self.data)).exec_grant()
910 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
911 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
912 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
914 def test_exec_grant_when_add_resources_but_no_vnfinst(self):
915 self.data["vnfInstanceId"] = "no_vnfinst"
918 resp = GrantVnf(json.dumps(self.data)).exec_grant()
919 except NSLCMException as e:
920 self.assertEqual(type(e), NSLCMException)
922 self.assertEqual(resp, None)
924 @mock.patch.object(time, "sleep")
925 @mock.patch.object(restcall, "call_req")
926 @mock.patch.object(resmgr, "grant_vnf")
927 def test_exec_grant_when_add_resources_but_no_off(self, mock_grant, mock_call_req, mock_sleep):
928 NfInstModel(mnfinstid="add_resources_but_no_off", nfinstid="vnf_inst_id_002",
929 package_id="package_id_002").save()
930 mock_sleep.return_value = None
932 "/api/catalog/v1/vnfpackages/package_id_002":
933 [0, json.JSONEncoder().encode(nf_package_info), "200"],
936 def side_effect(*args):
937 return mock_vals[args[4]]
939 mock_call_req.side_effect = side_effect
940 resmgr_grant_resp = {
942 "vimId": "cloudOwner_casa",
948 mock_grant.return_value = resmgr_grant_resp
949 self.data["vnfInstanceId"] = "add_resources_but_no_off"
950 resp = GrantVnf(json.dumps(self.data)).exec_grant()
951 self.assertEqual(resp["vnfInstanceId"], "add_resources_but_no_off")
952 self.assertEqual(resp["vnfLcmOpOccId"], "vnf_lcm_op_occ_id")
954 "id": "cloudOwner_casa",
955 "vimId": "cloudOwner_casa",
957 "interfaceInfo": None,
958 "accessInfo": {"tenant": "tenantA"},
961 self.assertEqual(resp["vimConnections"], vimConnections)
963 @mock.patch.object(resmgr, "grant_vnf")
964 def test_exec_grant_when_resource_template_in_add_resources(self, mock_grant):
965 resmgr_grant_resp = {
967 "vimId": "cloudOwner_casa",
973 mock_grant.return_value = resmgr_grant_resp
974 self.data["addResources"] = [{"vdu": "vdu_grant_vnf_add_resources"}, "resourceTemplate"]
975 resp = GrantVnf(json.dumps(self.data)).exec_grant()
976 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
977 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
978 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
980 @mock.patch.object(restcall, "call_req")
981 @mock.patch.object(resmgr, "grant_vnf")
982 def test_exec_grant_when_remove_resources_success(self, mock_grant, mock_call_req):
984 "/api/catalog/v1/vnfpackages/package_id_001":
985 [0, json.JSONEncoder().encode(nf_package_info), "200"],
988 def side_effect(*args):
989 return mock_vals[args[4]]
991 mock_call_req.side_effect = side_effect
992 resmgr_grant_resp = {
994 "vimId": "cloudOwner_casa",
1000 mock_grant.return_value = resmgr_grant_resp
1001 self.data.pop("addResources")
1002 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
1003 self.data["additionalparams"] = {"vnfmid": "vnfm_id_001"}
1004 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1005 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
1006 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
1007 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
1009 def test_exec_grant_when_remove_resources_no_vnfinst(self):
1010 self.data.pop("addResources")
1011 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
1012 self.data["additionalparams"] = {"vnfmid": "vnfm_id_002"}
1015 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1016 except NSLCMException as e:
1017 self.assertEqual(type(e), NSLCMException)
1019 self.assertEqual(resp, None)
1021 @mock.patch.object(time, "sleep")
1022 @mock.patch.object(restcall, "call_req")
1023 @mock.patch.object(resmgr, "grant_vnf")
1024 def test_exec_grant_when_remove_resources_but_no_off(self, mock_grant, mock_call_req, mock_sleep):
1025 NfInstModel(mnfinstid="remove_resources_but_no_off", nfinstid="vnf_inst_id_002", package_id="package_id_002",
1026 vnfm_inst_id="vnfm_id_002").save()
1027 mock_sleep.return_value = None
1029 "/api/catalog/v1/vnfpackages/package_id_002":
1030 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1033 def side_effect(*args):
1034 return mock_vals[args[4]]
1036 mock_call_req.side_effect = side_effect
1037 resmgr_grant_resp = {
1039 "vimId": "cloudOwner_casa",
1045 mock_grant.return_value = resmgr_grant_resp
1046 self.data["vnfInstanceId"] = "remove_resources_but_no_off"
1047 self.data.pop("addResources")
1048 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}]
1049 self.data["additionalparams"] = {"vnfmid": "vnfm_id_002"}
1050 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1051 self.assertEqual(resp["vnfInstanceId"], "remove_resources_but_no_off")
1052 self.assertEqual(resp["vnfLcmOpOccId"], "vnf_lcm_op_occ_id")
1054 "id": "cloudOwner_casa",
1055 "vimId": "cloudOwner_casa",
1057 "interfaceInfo": None,
1058 "accessInfo": {"tenant": "tenantA"},
1061 self.assertEqual(resp["vimConnections"], vimConnections)
1063 @mock.patch.object(resmgr, "grant_vnf")
1064 def test_exec_grant_when_resource_template_in_remove_resources(self, mock_grant):
1065 resmgr_grant_resp = {
1067 "vimId": "cloudOwner_casa",
1073 mock_grant.return_value = resmgr_grant_resp
1074 self.data.pop("addResources")
1075 self.data["removeResources"] = [{"vdu": "vdu_grant_vnf_remove_resources"}, "resourceTemplate"]
1076 resp = GrantVnf(json.dumps(self.data)).exec_grant()
1077 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimConnectionId"], "cloudOwner_casa")
1078 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["resourceProviderId"], "vg")
1079 self.assertEqual(resp["vimAssets"]["computeResourceFlavours"][0]["vimFlavourId"], "flavor_id_001")
1082 class TestCreateVnfViews(TestCase):
1084 self.ns_inst_id = str(uuid.uuid4())
1085 self.job_id = str(uuid.uuid4())
1088 "nsInstanceId": self.ns_inst_id,
1089 # "additionalParamForNs": {"inputs": json.dumps({})},
1090 "additionalParamForVnf": [
1092 "vnfprofileid": "VBras",
1093 "additionalparam": {
1094 "inputs": json.dumps({
1098 "vnfminstanceid": "1",
1099 # "vimId": "zte_test",
1100 "vimId": '{"cloud_owner": "VCPE", "cloud_regionid": "RegionOne"}'
1105 self.client = Client()
1106 NSInstModel(id=self.ns_inst_id, name="ns", nspackage_id="1", nsd_id="nsd_id", description="description",
1107 status="instantiating", nsd_model=json.dumps(nsd_model_dict), create_time=now_time(),
1108 lastuptime=now_time()).save()
1109 VLInstModel(vldid="ext_mnet_network", ownertype=OWNER_TYPE.NS, ownerid=self.ns_inst_id,
1113 NfInstModel.objects.all().delete()
1114 JobModel.objects.all().delete()
1116 @mock.patch.object(CreateVnfs, "run")
1117 def test_create_vnf_view(self, mock_run):
1118 response = self.client.post("/api/nslcm/v1/ns/vnfs", data=self.data)
1119 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
1120 context = json.loads(response.content)
1121 self.assertTrue(NfInstModel.objects.filter(nfinstid=context["vnfInstId"]).exists())
1123 @mock.patch.object(time, "sleep")
1124 @mock.patch.object(restcall, "call_req")
1125 def test_create_vnf_thread_sucess(self, mock_call_req, mock_sleep):
1126 mock_sleep.return_value = None
1127 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1129 "/api/catalog/v1/vnfpackages/zte_vbras":
1130 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1131 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
1132 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1133 "/api/ztevnfmdriver/v1/1/vnfs":
1134 [0, json.JSONEncoder().encode({"jobId": self.job_id, "vnfInstanceId": 3}), "200"],
1135 "/api/oof/v1/placement":
1136 [0, json.JSONEncoder().encode({}), "202"],
1137 "/api/resmgr/v1/vnf":
1138 [0, json.JSONEncoder().encode({}), "200"],
1139 "/api/ztevnfmdriver/v1/1/jobs/" + self.job_id + "?responseId=0":
1140 [0, json.JSONEncoder().encode({"jobid": self.job_id,
1141 "responsedescriptor": {"progress": "100",
1142 "status": JOB_MODEL_STATUS.FINISHED,
1144 "statusdescription": "creating",
1146 "responsehistorylist": [
1148 "status": JOB_MODEL_STATUS.PROCESSING,
1150 "statusdescription": "creating",
1151 "errorcode": "0"}]}}), "200"],
1152 "api/gvnfmdriver/v1/1/subscriptions":
1153 [0, json.JSONEncoder().encode({}), "200"],
1154 "/api/resmgr/v1/vnfinfo":
1155 [0, json.JSONEncoder().encode(subscription_response_data), "200"],
1157 # "/network/generic-vnfs/generic-vnf/%s" % nf_inst_id:
1158 # [0, json.JSONEncoder().encode({}), "201"],
1159 # "/cloud-infrastructure/cloud-regions/cloud-region/zte/test?depth=all":
1160 # [0, json.JSONEncoder().encode(vim_info), "201"],
1161 # "/cloud-infrastructure/cloud-regions/cloud-region/zte/test/tenants/tenant/admin/vservers/vserver/1":
1162 # [0, json.JSONEncoder().encode({}), "201"],
1166 def side_effect(*args):
1167 return mock_vals[args[4]]
1168 mock_call_req.side_effect = side_effect
1170 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1171 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1172 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1173 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1175 CreateVnfs(data, nf_inst_id, job_id).run()
1176 self.assertEqual(NfInstModel.objects.get(nfinstid=nf_inst_id).status, VNF_STATUS.ACTIVE)
1177 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, JOB_PROGRESS.FINISHED)
1179 def test_create_vnf_thread_when_the_name_of_vnf_instance_already_exists(self):
1180 NfInstModel(nf_name="").save()
1181 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1183 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1184 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1185 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1186 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1188 CreateVnfs(data, nf_inst_id, job_id).run()
1189 self.assertEqual(NfInstModel.objects.get(nfinstid=nf_inst_id).status, VNF_STATUS.FAILED)
1190 self.assertEqual(JobModel.objects.get(jobid=job_id).progress, JOB_PROGRESS.ERROR)
1192 @mock.patch.object(time, "sleep")
1193 @mock.patch.object(restcall, "call_req")
1194 def test_create_vnf_thread_when_data_has_vnfd_id(self, mock_call_req, mock_sleep):
1195 mock_sleep.return_value = None
1196 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1198 "/api/catalog/v1/vnfpackages/data_has_vnfd_id":
1199 [0, json.JSONEncoder().encode(nf_package_info), "200"],
1200 "/external-system/esr-vnfm-list/esr-vnfm/1?depth=all":
1201 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1202 "/api/ztevnfmdriver/v1/1/vnfs":
1203 [0, json.JSONEncoder().encode({"jobId": self.job_id, "vnfInstanceId": 3}), "200"],
1204 "/api/oof/v1/placement":
1205 [0, json.JSONEncoder().encode({}), "202"],
1206 "/api/resmgr/v1/vnf":
1207 [0, json.JSONEncoder().encode({}), "200"],
1208 "/api/ztevnfmdriver/v1/1/jobs/" + self.job_id + "?responseId=0":
1209 [0, json.JSONEncoder().encode({"jobid": self.job_id,
1210 "responsedescriptor": {"progress": "100",
1211 "status": JOB_MODEL_STATUS.FINISHED,
1213 "statusdescription": "creating",
1215 "responsehistorylist": [
1217 "status": JOB_MODEL_STATUS.PROCESSING,
1219 "statusdescription": "creating",
1220 "errorcode": "0"}]}}), "200"],
1221 "api/gvnfmdriver/v1/1/subscriptions":
1222 [0, json.JSONEncoder().encode({}), "200"],
1223 "/api/resmgr/v1/vnfinfo":
1224 [0, json.JSONEncoder().encode(subscription_response_data), "200"]
1227 def side_effect(*args):
1228 return mock_vals[args[4]]
1230 mock_call_req.side_effect = side_effect
1231 self.data["additionalParamForVnf"][0]["additionalparam"]["vnfdId"] = "data_has_vnfd_id"
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 @mock.patch.object(restcall, "call_req")
1243 @mock.patch.object(CreateVnfs, "build_homing_request")
1244 def test_send_homing_request(self, mock_build_req, mock_call_req):
1245 nf_inst_id, job_id = create_vnfs.prepare_create_params()
1246 OOFDataModel.objects.all().delete()
1248 "requestId": "1234",
1249 "transactionId": "1234",
1250 "requestStatus": "accepted"
1252 mock_build_req.return_value = {
1254 "transactionId": "1234",
1255 "requestId": "1234",
1256 "callbackUrl": "xx",
1258 "requestType": "create",
1260 "optimizers": ["placement"],
1264 "placementDemands": [
1266 "resourceModuleName": "vG",
1267 "serviceResourceId": "1234",
1268 "resourceModelInfo": {
1269 "modelInvariantId": "1234",
1270 "modelVersionId": "1234"
1276 "serviceInstanceId": "1234",
1277 "serviceName": "1234",
1279 "modelInvariantId": "5678",
1280 "modelVersionId": "7890"
1284 mock_call_req.return_value = [0, json.JSONEncoder().encode(resp), "202"]
1286 "ns_instance_id": ignore_case_get(self.data, "nsInstanceId"),
1287 "additional_param_for_ns": ignore_case_get(self.data, "additionalParamForNs"),
1288 "additional_param_for_vnf": ignore_case_get(self.data, "additionalParamForVnf"),
1289 "vnf_index": ignore_case_get(self.data, "vnfIndex")
1291 CreateVnfs(data, nf_inst_id, job_id).send_homing_request_to_OOF()
1292 ret = OOFDataModel.objects.filter(request_id="1234", transaction_id="1234")
1293 self.assertIsNotNone(ret)
1296 class TestUpdateVnfsViews(TestCase):
1298 self.client = Client()
1300 "terminationType": "forceful",
1301 "gracefulTerminationTimeout": "600",
1302 "additionalParams": ""
1304 self.nf_inst_id = "test_update_vnf"
1305 self.m_nf_inst_id = "test_update_vnf_m_nf_inst_id"
1306 self.vnfm_inst_id = "test_update_vnf_vnfm_inst_id"
1307 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
1308 NfInstModel.objects.all().delete()
1309 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
1310 vnfm_inst_id=self.vnfm_inst_id,
1311 status=VNF_STATUS.NULL,
1312 mnfinstid=self.m_nf_inst_id,
1313 vnfd_model=self.vnfd_model
1317 NfInstModel.objects.all().delete()
1319 @mock.patch.object(time, "sleep")
1320 @mock.patch.object(restcall, "call_req")
1321 def test_update_vnf_thread(self, mock_call_req, mock_sleep):
1322 vnf_update_service = NFOperateService(self.nf_inst_id, self.data)
1324 "jobId": vnf_update_service.job_id,
1325 "responsedescriptor": {"status": JOB_MODEL_STATUS.FINISHED}
1328 "/external-system/esr-vnfm-list/esr-vnfm/test_update_vnf_vnfm_inst_id?depth=all":
1329 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1330 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/vnfs/test_update_vnf_m_nf_inst_id/operate":
1331 [0, json.JSONEncoder().encode({"jobId": vnf_update_service.job_id}), "200"],
1332 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/jobs/" + vnf_update_service.job_id + "?responseId=0":
1333 [0, json.JSONEncoder().encode(job_info), "200"],
1336 def side_effect(*args):
1337 return mock_vals[args[4]]
1338 mock_call_req.side_effect = side_effect
1339 vnf_update_service.run()
1340 nfinst = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
1341 self.assertEqual(nfinst.status, VNF_STATUS.ACTIVE)
1342 self.assertEqual(JobModel.objects.get(jobid=vnf_update_service.job_id).progress, JOB_PROGRESS.FINISHED)
1344 def test_update_vnf_thread_when_no_nf(self):
1345 NfInstModel.objects.all().delete()
1346 vnf_update_service = NFOperateService(self.nf_inst_id, self.data)
1347 vnf_update_service.run()
1348 self.assertEqual(JobModel.objects.get(jobid=vnf_update_service.job_id).progress, JOB_PROGRESS.ERROR)
1350 @mock.patch.object(time, "sleep")
1351 @mock.patch.object(restcall, "call_req")
1352 def test_update_vnf_thread_when_nf_update_failed(self, mock_call_req, mock_sleep):
1353 vnf_update_service = NFOperateService(self.nf_inst_id, self.data)
1355 "jobId": vnf_update_service.job_id,
1356 "responsedescriptor": {"status": JOB_MODEL_STATUS.ERROR}
1359 "/external-system/esr-vnfm-list/esr-vnfm/test_update_vnf_vnfm_inst_id?depth=all":
1360 [0, json.JSONEncoder().encode(vnfm_info), "200"],
1361 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/vnfs/test_update_vnf_m_nf_inst_id/operate":
1362 [0, json.JSONEncoder().encode({"jobId": vnf_update_service.job_id}), "200"],
1363 "/api/ztevnfmdriver/v1/test_update_vnf_vnfm_inst_id/jobs/" + vnf_update_service.job_id + "?responseId=0":
1364 [0, json.JSONEncoder().encode(job_info), "200"],
1367 def side_effect(*args):
1368 return mock_vals[args[4]]
1370 mock_call_req.side_effect = side_effect
1371 vnf_update_service.run()
1372 nfinst = NfInstModel.objects.get(nfinstid=self.nf_inst_id)
1373 self.assertEqual(nfinst.status, VNF_STATUS.UPDATING)
1374 self.assertEqual(JobModel.objects.get(jobid=vnf_update_service.job_id).progress, JOB_PROGRESS.ERROR)
1377 class TestVerifyVnfsViews(TestCase):
1379 self.client = Client()
1380 self.url = "/api/nslcm/v1/vnfonboarding"
1381 self.package_id = "test_verify_vnfs_package_id"
1382 self.nf_inst_id = "test_verify_vnfs"
1383 self.m_nf_inst_id = "test_verify_vnfs_m_nf_inst_id"
1384 self.vnfm_inst_id = "test_verify_vnfs_vnfm_inst_id"
1385 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
1387 "PackageID": self.package_id,
1389 self.job_id = JobUtil.create_job(JOB_TYPE.VNF, "verify_vnfs", self.nf_inst_id)
1390 NfInstModel.objects.all().delete()
1391 NfInstModel.objects.create(package_id=self.package_id,
1392 nfinstid=self.nf_inst_id,
1393 vnfm_inst_id=self.vnfm_inst_id,
1394 status=VNF_STATUS.NULL,
1395 mnfinstid=self.m_nf_inst_id,
1396 vnfd_model=self.vnfd_model
1400 NfInstModel.objects.all().delete()
1402 @mock.patch.object(VerifyVnfs, "run")
1403 def test_verify_vnfs_view(self, mock_run):
1404 response = self.client.post(self.url, data=self.data)
1405 self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
1407 def test_verify_vnfs_view_when_data_is_not_valid(self):
1408 response = self.client.post(self.url, data={})
1409 self.assertEqual(status.HTTP_409_CONFLICT, response.status_code)
1411 @mock.patch.object(time, "sleep")
1412 @mock.patch.object(restcall, "call_req")
1413 def test_verify_vnfs_thread(self, mock_call_req, mock_sleep):
1416 "responseDescriptor": {
1417 "status": JOB_MODEL_STATUS.FINISHED,
1420 "statusDescription": ""
1425 "responseDescriptor": {
1426 "status": JOB_MODEL_STATUS.FINISHED,
1429 "statusDescription": ""
1434 "responseDescriptor": {
1435 "status": JOB_MODEL_STATUS.FINISHED,
1438 "statusDescription": ""
1442 "/api/nslcm/v1/vnfpackage":
1443 [0, json.JSONEncoder().encode(job_info_1), "200"],
1444 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1445 [0, json.JSONEncoder().encode(job_info_1), "200"],
1446 "/api/nslcm/v1/ns/ns_vnfs":
1447 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1448 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1449 [0, json.JSONEncoder().encode(job_info_2), "200"],
1450 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1451 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1452 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1453 [0, json.JSONEncoder().encode(job_info_task), "200"],
1454 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1455 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1458 def side_effect(*args):
1459 return mock_vals[args[4]]
1460 mock_call_req.side_effect = side_effect
1461 VerifyVnfs(self.data, self.job_id).run()
1462 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.FINISHED)
1464 @mock.patch.object(time, "sleep")
1465 @mock.patch.object(restcall, "call_req")
1466 def test_verify_vnfs_thread_when_failed_to_call_vnf_onboarding(self, mock_call_req, mock_sleep):
1468 "/api/nslcm/v1/vnfpackage":
1469 [1, json.JSONEncoder().encode({}), "200"]
1472 def side_effect(*args):
1473 return mock_vals[args[4]]
1475 mock_call_req.side_effect = side_effect
1476 VerifyVnfs(self.data, self.job_id).run()
1477 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1479 @mock.patch.object(time, "sleep")
1480 @mock.patch.object(restcall, "call_req")
1481 def test_verify_vnfs_thread_when_do_on_boarding_failed_to_query_job(self, mock_call_req, mock_sleep):
1484 "responseDescriptor": {
1485 "status": JOB_MODEL_STATUS.FINISHED,
1488 "statusDescription": ""
1492 "/api/nslcm/v1/vnfpackage":
1493 [0, json.JSONEncoder().encode(job_info_1), "200"],
1494 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1495 [1, json.JSONEncoder().encode(job_info_1), "200"],
1498 def side_effect(*args):
1499 return mock_vals[args[4]]
1501 mock_call_req.side_effect = side_effect
1502 VerifyVnfs(self.data, self.job_id).run()
1503 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1505 @mock.patch.object(time, "sleep")
1506 @mock.patch.object(restcall, "call_req")
1507 def test_verify_vnfs_thread_when_do_on_boarding_job_does_not_exist(self, mock_call_req, mock_sleep):
1512 "/api/nslcm/v1/vnfpackage":
1513 [0, json.JSONEncoder().encode(job_info_1), "200"],
1514 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1515 [0, json.JSONEncoder().encode(job_info_1), "200"],
1518 def side_effect(*args):
1519 return mock_vals[args[4]]
1521 mock_call_req.side_effect = side_effect
1522 VerifyVnfs(self.data, self.job_id).run()
1523 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1525 @mock.patch.object(time, "sleep")
1526 @mock.patch.object(restcall, "call_req")
1527 def test_verify_vnfs_thread_when_do_on_boarding_job_process_error(self, mock_call_req, mock_sleep):
1530 "responseDescriptor": {
1531 "status": JOB_MODEL_STATUS.FINISHED,
1532 "progress": JOB_PROGRESS.ERROR,
1534 "statusDescription": "already onBoarded"
1539 "responseDescriptor": {
1540 "status": JOB_MODEL_STATUS.FINISHED,
1543 "statusDescription": ""
1548 "responseDescriptor": {
1549 "status": JOB_MODEL_STATUS.FINISHED,
1552 "statusDescription": ""
1556 "/api/nslcm/v1/vnfpackage":
1557 [0, json.JSONEncoder().encode(job_info_1), "200"],
1558 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1559 [0, json.JSONEncoder().encode(job_info_1), "200"],
1560 "/api/nslcm/v1/ns/ns_vnfs":
1561 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1562 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1563 [0, json.JSONEncoder().encode(job_info_2), "200"],
1564 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1565 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1566 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1567 [0, json.JSONEncoder().encode(job_info_task), "200"],
1568 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1569 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1572 def side_effect(*args):
1573 return mock_vals[args[4]]
1575 mock_call_req.side_effect = side_effect
1576 VerifyVnfs(self.data, self.job_id).run()
1577 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.FINISHED)
1579 @mock.patch.object(time, "sleep")
1580 @mock.patch.object(restcall, "call_req")
1581 def test_verify_vnfs_thread_when_failed_to_call_inst_vnf(self, mock_call_req, mock_sleep):
1584 "responseDescriptor": {
1585 "status": JOB_MODEL_STATUS.FINISHED,
1588 "statusDescription": ""
1592 "/api/nslcm/v1/vnfpackage":
1593 [0, json.JSONEncoder().encode(job_info_1), "200"],
1594 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1595 [0, json.JSONEncoder().encode(job_info_1), "200"],
1596 "/api/nslcm/v1/ns/ns_vnfs":
1597 [1, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1600 def side_effect(*args):
1601 return mock_vals[args[4]]
1603 mock_call_req.side_effect = side_effect
1604 VerifyVnfs(self.data, self.job_id).run()
1605 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1607 @mock.patch.object(time, "sleep")
1608 @mock.patch.object(restcall, "call_req")
1609 def test_verify_vnfs_thread_when_do_term_vnf_job_process_error(self, mock_call_req, mock_sleep):
1612 "responseDescriptor": {
1613 "status": JOB_MODEL_STATUS.FINISHED,
1616 "statusDescription": ""
1621 "responseDescriptor": {
1622 "status": JOB_MODEL_STATUS.FINISHED,
1623 "progress": JOB_PROGRESS.ERROR,
1625 "statusDescription": ""
1629 "/api/nslcm/v1/vnfpackage":
1630 [0, json.JSONEncoder().encode(job_info_1), "200"],
1631 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1632 [0, json.JSONEncoder().encode(job_info_1), "200"],
1633 "/api/nslcm/v1/ns/ns_vnfs":
1634 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1635 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1636 [0, json.JSONEncoder().encode(job_info_2), "200"],
1639 def side_effect(*args):
1640 return mock_vals[args[4]]
1642 mock_call_req.side_effect = side_effect
1643 VerifyVnfs(self.data, self.job_id).run()
1644 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1646 @mock.patch.object(time, "sleep")
1647 @mock.patch.object(restcall, "call_req")
1648 def test_verify_vnfs_thread_when_failed_to_call_func_test(self, mock_call_req, mock_sleep):
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 [1, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"]
1680 def side_effect(*args):
1681 return mock_vals[args[4]]
1683 mock_call_req.side_effect = side_effect
1684 VerifyVnfs(self.data, self.job_id).run()
1685 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1687 @mock.patch.object(time, "sleep")
1688 @mock.patch.object(restcall, "call_req")
1689 def test_verify_vnfs_thread_when_do_func_test_failed_query_job(self, mock_call_req, mock_sleep):
1692 "responseDescriptor": {
1693 "status": JOB_MODEL_STATUS.FINISHED,
1696 "statusDescription": ""
1701 "responseDescriptor": {
1702 "status": JOB_MODEL_STATUS.FINISHED,
1705 "statusDescription": ""
1710 "responseDescriptor": {
1711 "status": JOB_MODEL_STATUS.FINISHED,
1714 "statusDescription": ""
1718 "/api/nslcm/v1/vnfpackage":
1719 [0, json.JSONEncoder().encode(job_info_1), "200"],
1720 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1721 [0, json.JSONEncoder().encode(job_info_1), "200"],
1722 "/api/nslcm/v1/ns/ns_vnfs":
1723 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1724 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1725 [0, json.JSONEncoder().encode(job_info_2), "200"],
1726 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1727 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1728 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1729 [1, json.JSONEncoder().encode(job_info_task), "200"],
1732 def side_effect(*args):
1733 return mock_vals[args[4]]
1735 mock_call_req.side_effect = side_effect
1736 VerifyVnfs(self.data, self.job_id).run()
1737 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1739 @mock.patch.object(time, "sleep")
1740 @mock.patch.object(restcall, "call_req")
1741 def test_verify_vnfs_thread_when_do_func_test_job_does_not_exist(self, mock_call_req, mock_sleep):
1744 "responseDescriptor": {
1745 "status": JOB_MODEL_STATUS.FINISHED,
1748 "statusDescription": ""
1753 "responseDescriptor": {
1754 "status": JOB_MODEL_STATUS.FINISHED,
1757 "statusDescription": ""
1764 "/api/nslcm/v1/vnfpackage":
1765 [0, json.JSONEncoder().encode(job_info_1), "200"],
1766 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1767 [0, json.JSONEncoder().encode(job_info_1), "200"],
1768 "/api/nslcm/v1/ns/ns_vnfs":
1769 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1770 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1771 [0, json.JSONEncoder().encode(job_info_2), "200"],
1772 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1773 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1774 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1775 [0, json.JSONEncoder().encode(job_info_task), "200"],
1776 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1777 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1780 def side_effect(*args):
1781 return mock_vals[args[4]]
1783 mock_call_req.side_effect = side_effect
1784 VerifyVnfs(self.data, self.job_id).run()
1785 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1787 @mock.patch.object(time, "sleep")
1788 @mock.patch.object(restcall, "call_req")
1789 def test_verify_vnfs_thread_when_do_func_test_job_process_error(self, mock_call_req, mock_sleep):
1792 "responseDescriptor": {
1793 "status": JOB_MODEL_STATUS.FINISHED,
1796 "statusDescription": ""
1801 "responseDescriptor": {
1802 "status": JOB_MODEL_STATUS.FINISHED,
1805 "statusDescription": ""
1810 "responseDescriptor": {
1811 "status": JOB_MODEL_STATUS.ERROR,
1812 "progress": JOB_PROGRESS.ERROR,
1814 "statusDescription": "already onBoarded"
1818 "/api/nslcm/v1/vnfpackage":
1819 [0, json.JSONEncoder().encode(job_info_1), "200"],
1820 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1821 [0, json.JSONEncoder().encode(job_info_1), "200"],
1822 "/api/nslcm/v1/ns/ns_vnfs":
1823 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1824 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1825 [0, json.JSONEncoder().encode(job_info_2), "200"],
1826 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1827 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1828 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1829 [0, json.JSONEncoder().encode(job_info_task), "200"],
1830 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1831 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1834 def side_effect(*args):
1835 return mock_vals[args[4]]
1837 mock_call_req.side_effect = side_effect
1838 VerifyVnfs(self.data, self.job_id).run()
1839 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.FINISHED)
1841 @mock.patch.object(time, "sleep")
1842 @mock.patch.object(restcall, "call_req")
1843 def test_verify_vnfs_thread_when_do_func_test_failed_to_get_func_test_result(self, mock_call_req, mock_sleep):
1846 "responseDescriptor": {
1847 "status": JOB_MODEL_STATUS.FINISHED,
1850 "statusDescription": ""
1855 "responseDescriptor": {
1856 "status": JOB_MODEL_STATUS.FINISHED,
1859 "statusDescription": ""
1864 "responseDescriptor": {
1865 "status": JOB_MODEL_STATUS.FINISHED,
1868 "statusDescription": "already onBoarded"
1872 "/api/nslcm/v1/vnfpackage":
1873 [0, json.JSONEncoder().encode(job_info_1), "200"],
1874 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_1", 0):
1875 [0, json.JSONEncoder().encode(job_info_1), "200"],
1876 "/api/nslcm/v1/ns/ns_vnfs":
1877 [0, json.JSONEncoder().encode({"jobId": "test_2", "vnfInstId": ""}), "200"],
1878 "/api/nslcm/v1/jobs/%s?responseId=%s" % ("test_2", 0):
1879 [0, json.JSONEncoder().encode(job_info_2), "200"],
1880 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks":
1881 [0, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1882 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s?responseId=%s" % ("task_id", 0):
1883 [0, json.JSONEncoder().encode(job_info_task), "200"],
1884 "/openapi/vnfsdk/v1/functest/taskmanager/testtasks/%s/result" % "task_id":
1885 [1, json.JSONEncoder().encode({"TaskID": "task_id"}), "200"],
1888 def side_effect(*args):
1889 return mock_vals[args[4]]
1891 mock_call_req.side_effect = side_effect
1892 VerifyVnfs(self.data, self.job_id).run()
1893 self.assertEqual(JobModel.objects.get(jobid=self.job_id).progress, JOB_PROGRESS.ERROR)
1896 class TestLcmNotifyViews(TestCase):
1898 self.client = Client()
1901 "operation": "Instantiate",
1904 "affectedVnfc": [{"vnfcInstanceId": "vnfc_instance_id",
1906 "changeType": VNFC_CHANGE_TYPE.ADDED,
1911 "affectedVl": [{"vlInstanceId": "vl_instance_id",
1913 "changeType": VNFC_CHANGE_TYPE.ADDED,
1914 "networkResource": {
1915 "resourceType": NETWORK_RESOURCE_TYPE.NETWORK,
1916 "resourceId": "resource_id",
1917 "resourceName": "resource_name"
1920 "affectedCp": [{"changeType": VNFC_CHANGE_TYPE.ADDED,
1921 "virtualLinkInstanceId": "virtual_link_instance_id",
1922 "cpInstanceId": "cp_instance_id",
1925 "ownerId": "owner_id",
1928 "resourceId": "resource_id",
1929 "resourceName": "resource_name",
1931 "ipAddress": "ip_address",
1932 "macAddress": "mac_address",
1936 "affectedVirtualStorage": [{}]
1938 self.nf_inst_id = "test_lcm_notify"
1939 self.m_nf_inst_id = "test_lcm_notify_m_nf_inst_id"
1940 self.vnfm_inst_id = "test_lcm_notify_vnfm_inst_id"
1941 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
1942 self.url = "/api/nslcm/v1/ns/%s/vnfs/%s/Notify" % (self.m_nf_inst_id, self.vnfm_inst_id)
1943 NfInstModel.objects.all().delete()
1944 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
1945 vnfm_inst_id=self.vnfm_inst_id,
1946 status=VNF_STATUS.NULL,
1947 mnfinstid=self.m_nf_inst_id,
1948 vnfd_model=self.vnfd_model
1952 NfInstModel.objects.all().delete()
1953 VNFCInstModel.objects.all().delete()
1954 VmInstModel.objects.all().delete()
1955 VLInstModel.objects.all().delete()
1956 PortInstModel.objects.all().delete()
1957 CPInstModel.objects.all().delete()
1959 def test_lcm_notify_view_when_change_type_is_added(self):
1960 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, self.data).do_biz()
1961 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id", vmid="vm_id",
1962 nfinstid=self.nf_inst_id)
1963 self.assertIsInstance(vnfc_inst, VNFCInstModel)
1964 vm_inst = VmInstModel.objects.get(vmid="vm_id", vimid="vim_id", resouceid="vm_id", insttype=INST_TYPE.VNF,
1965 instid=self.nf_inst_id, vmname="vm_name", hostid='1')
1966 self.assertIsInstance(vm_inst, VmInstModel)
1967 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="vld_id", vlinstancename="resource_name",
1968 ownertype=0, ownerid=self.nf_inst_id, relatednetworkid="resource_id",
1970 self.assertIsInstance(vl_inst, VLInstModel)
1971 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_id",
1972 resourceid="resource_id", name="resource_name", instid="inst_id",
1973 cpinstanceid="cp_instance_id", bandwidth='unknown',
1974 operationalstate='active', ipaddress="ip_address",
1975 macaddress='mac_address', floatipaddress='unknown',
1976 serviceipaddress='unknown', typevirtualnic='unknown',
1977 sfcencapsulation='gre', direction='unknown', tenant="tenant")
1978 self.assertIsInstance(port_inst, PortInstModel)
1979 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
1980 ownerid=self.nf_inst_id, relatedtype=2, status='active')
1981 self.assertIsInstance(cp_inst, CPInstModel)
1983 def test_lcm_notify_view__when_change_type_is_added_when_nf_not_exists(self):
1984 NfInstModel.objects.all().delete()
1987 "operation": "Instantiate",
1992 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
1993 self.assertEqual(1, 0)
1995 self.assertEqual(1, 1)
1997 def test_lcm_notify_view_when_change_type_is_removeed(self):
1998 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
1999 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2000 CPInstModel.objects.create(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0, ownerid=self.nf_inst_id,
2001 relatedtype=2, relatedport="related_port", status="active")
2004 "operation": "Instantiate",
2007 "affectedVnfc": [{"vnfcInstanceId": "vnfc_instance_id",
2009 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2014 "affectedVl": [{"vlInstanceId": "vl_instance_id",
2016 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2017 "networkResource": {
2018 "resourceType": NETWORK_RESOURCE_TYPE.NETWORK,
2019 "resourceId": "resource_id",
2020 "resourceName": "resource_name"
2023 "affectedCp": [{"changeType": VNFC_CHANGE_TYPE.REMOVED,
2024 "virtualLinkInstanceId": "virtual_link_instance_id",
2025 "cpInstanceId": "cp_instance_id",
2028 "ownerId": "owner_id",
2031 "resourceId": "resource_id",
2032 "resourceName": "resource_name",
2034 "ipAddress": "ip_address",
2035 "macAddress": "mac_address",
2039 "affectedVirtualStorage": [{}]
2041 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2042 vnfc_inst = VNFCInstModel.objects.filter(vnfcinstanceid="vnfc_instance_id")
2043 self.assertEqual(len(vnfc_inst), 0)
2044 vl_inst = VLInstModel.objects.filter(vlinstanceid="vl_instance_id")
2045 self.assertEqual(len(vl_inst), 0)
2046 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_id",
2047 resourceid="resource_id", name="resource_name", instid="inst_id",
2048 cpinstanceid="cp_instance_id", bandwidth='unknown',
2049 operationalstate='active', ipaddress="ip_address",
2050 macaddress='mac_address', floatipaddress='unknown',
2051 serviceipaddress='unknown', typevirtualnic='unknown',
2052 sfcencapsulation='gre', direction='unknown', tenant="tenant")
2053 self.assertIsInstance(port_inst, PortInstModel)
2054 cp_inst = CPInstModel.objects.filter(cpinstanceid="cp_instance_id")
2055 self.assertEqual(len(cp_inst), 0)
2057 def test_lcm_notify_view_when_change_type_is_modified(self):
2058 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2059 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2060 CPInstModel.objects.create(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0, ownerid=self.nf_inst_id,
2061 relatedtype=2, relatedport="related_port")
2064 "operation": "Instantiate",
2067 "affectedVnfc": [{"vnfcInstanceId": "vnfc_instance_id",
2069 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2074 "affectedVl": [{"vlInstanceId": "vl_instance_id",
2076 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2077 "networkResource": {
2078 "resourceType": NETWORK_RESOURCE_TYPE.NETWORK,
2079 "resourceId": "resource_id",
2080 "resourceName": "resource_name"
2083 "affectedCp": [{"changeType": VNFC_CHANGE_TYPE.MODIFIED,
2084 "virtualLinkInstanceId": "virtual_link_instance_id",
2085 "cpInstanceId": "cp_instance_id",
2088 "ownerId": "owner_id",
2091 "resourceId": "resource_id",
2092 "resourceName": "resource_name",
2094 "ipAddress": "ip_address",
2095 "macAddress": "mac_address",
2099 "affectedVirtualStorage": [{}]
2101 NotifyLcm(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2102 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id", vmid="vm_id",
2103 nfinstid=self.nf_inst_id)
2104 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2105 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="vld_id", vlinstancename="resource_name",
2106 ownertype=0, ownerid=self.nf_inst_id, relatednetworkid="resource_id",
2108 self.assertIsInstance(vl_inst, VLInstModel)
2109 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_id",
2110 resourceid="resource_id", name="resource_name", instid="inst_id",
2111 cpinstanceid="cp_instance_id", bandwidth='unknown',
2112 operationalstate='active', ipaddress="ip_address",
2113 macaddress='mac_address', floatipaddress='unknown',
2114 serviceipaddress='unknown', typevirtualnic='unknown',
2115 sfcencapsulation='gre', direction='unknown', tenant="tenant")
2116 self.assertIsInstance(port_inst, PortInstModel)
2117 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2118 ownerid=self.nf_inst_id, relatedtype=2)
2119 self.assertIsInstance(cp_inst, CPInstModel)
2122 class TestVnfNotifyView(TestCase):
2124 self.client = Client()
2125 self.nf_inst_id = "test_vnf_notify"
2126 self.m_nf_inst_id = "test_vnf_notify_m_nf_inst_id"
2127 self.vnfm_inst_id = "test_vnf_notify_vnfm_inst_id"
2128 self.vnfd_model = {"metadata": {"vnfdId": "1"}}
2129 self.url = "/api/nslcm/v2/ns/%s/vnfs/%s/Notify" % (self.vnfm_inst_id, self.m_nf_inst_id)
2132 "notificationType": "VnfLcmOperationOccurrenceNotification",
2133 "subscriptionId": "1111",
2134 "timeStamp": "1111",
2135 "notificationStatus": "START",
2136 "operationState": "STARTING",
2137 "vnfInstanceId": self.nf_inst_id,
2138 "operation": "INSTANTIATE",
2139 "isAutomaticInvocation": "1111",
2140 "vnfLcmOpOccId": "1111",
2141 "affectedVnfcs": [{"id": "vnfc_instance_id",
2143 "changeType": VNFC_CHANGE_TYPE.ADDED,
2144 "computeResource": {
2145 "vimConnectionId": "vim_connection_id",
2146 "resourceId": "resource_id"
2149 "affectedVirtualLinks": [{"id": "vl_instance_id",
2150 "virtualLinkDescId": "virtual_link_desc_id",
2151 "changeType": VNFC_CHANGE_TYPE.ADDED,
2152 "networkResource": {
2153 "vimLevelResourceType": "network",
2154 "resourceId": "resource_id"
2156 "changedExtConnectivity": [{"id": "virtual_link_instance_id",
2157 "extLinkPorts": [{"cpInstanceId": "cp_instance_id",
2160 "vimConnectionId": "vim_connection_id",
2161 "resourceId": "resource_id"
2166 NfInstModel.objects.all().delete()
2167 VNFCInstModel.objects.all().delete()
2168 VmInstModel.objects.all().delete()
2169 NfInstModel.objects.create(nfinstid=self.nf_inst_id,
2170 vnfm_inst_id=self.vnfm_inst_id,
2171 status=VNF_STATUS.NULL,
2172 mnfinstid=self.m_nf_inst_id,
2173 vnfd_model=self.vnfd_model
2177 NfInstModel.objects.all().delete()
2178 VNFCInstModel.objects.all().delete()
2179 VmInstModel.objects.all().delete()
2180 VLInstModel.objects.all().delete()
2181 PortInstModel.objects.all().delete()
2182 CPInstModel.objects.all().delete()
2184 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_added(self):
2185 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, self.data).do_biz()
2186 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id",
2187 nfinstid=self.nf_inst_id, vmid="resource_id")
2188 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2189 vm_inst = VmInstModel.objects.get(vmid="resource_id", vimid="vim_connection_id", resouceid="resource_id",
2190 insttype=INST_TYPE.VNF, instid=self.nf_inst_id, vmname="resource_id",
2192 self.assertIsInstance(vm_inst, VmInstModel)
2193 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="virtual_link_desc_id",
2194 vlinstancename="resource_id", ownertype=0, ownerid=self.nf_inst_id,
2195 relatednetworkid="resource_id", vltype=0)
2196 self.assertIsInstance(vl_inst, VLInstModel)
2197 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_connection_id",
2198 resourceid="resource_id", name="resource_id", instid="unknown",
2199 cpinstanceid="cp_instance_id", bandwidth='unknown',
2200 operationalstate='active', ipaddress="unkown", macaddress='unknown',
2201 floatipaddress='unknown', serviceipaddress='unknown',
2202 typevirtualnic='unknown', sfcencapsulation='gre', direction='unknown',
2204 self.assertIsInstance(port_inst, PortInstModel)
2205 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2206 ownerid=self.nf_inst_id, relatedtype=2, status='active')
2207 self.assertIsInstance(cp_inst, CPInstModel)
2209 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_added_when_nf_not_exists(self):
2212 "notificationType": "VnfLcmOperationOccurrenceNotification",
2213 "subscriptionId": "1111",
2214 "timeStamp": "1111",
2215 "notificationStatus": "START",
2216 "operationState": "STARTING",
2217 "vnfInstanceId": "nf_not_exists",
2218 "operation": "INSTANTIATE",
2219 "isAutomaticInvocation": "1111",
2220 "vnfLcmOpOccId": "1111"
2223 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2224 self.assertEqual(1, 0)
2226 self.assertEqual(1, 1)
2228 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_removed(self):
2229 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2230 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2233 "notificationType": "VnfLcmOperationOccurrenceNotification",
2234 "subscriptionId": "1111",
2235 "timeStamp": "1111",
2236 "notificationStatus": "START",
2237 "operationState": "STARTING",
2238 "vnfInstanceId": self.nf_inst_id,
2239 "operation": "INSTANTIATE",
2240 "isAutomaticInvocation": "1111",
2241 "vnfLcmOpOccId": "1111",
2242 "affectedVnfcs": [{"id": "vnfc_instance_id",
2244 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2245 "computeResource": {
2246 "vimConnectionId": "vim_connection_id",
2247 "resourceId": "resource_id"
2249 "affectedVirtualLinks": [{"id": "vl_instance_id",
2250 "virtualLinkDescId": "virtual_link_desc_id",
2251 "changeType": VNFC_CHANGE_TYPE.REMOVED,
2252 "networkResource": {
2253 "vimLevelResourceType": "network",
2254 "resourceId": "resource_id"
2257 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2258 vnfc_inst = VNFCInstModel.objects.filter(vnfcinstanceid="vnfc_instance_id")
2259 self.assertEqual(len(vnfc_inst), 0)
2260 vl_inst = VLInstModel.objects.filter(vlinstanceid="vl_instance_id")
2261 self.assertEqual(len(vl_inst), 0)
2263 def test_handle_vnf_lcm_ooc_notification_when_change_type_is_modified(self):
2264 VNFCInstModel.objects.create(vnfcinstanceid="vnfc_instance_id")
2265 VLInstModel.objects.create(vlinstanceid="vl_instance_id", ownertype=0)
2268 "notificationType": "VnfLcmOperationOccurrenceNotification",
2269 "subscriptionId": "1111",
2270 "timeStamp": "1111",
2271 "notificationStatus": "START",
2272 "operationState": "STARTING",
2273 "vnfInstanceId": self.nf_inst_id,
2274 "operation": "INSTANTIATE",
2275 "isAutomaticInvocation": "1111",
2276 "vnfLcmOpOccId": "1111",
2277 "affectedVnfcs": [{"id": "vnfc_instance_id",
2279 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2280 "computeResource": {
2281 "vimConnectionId": "vim_connection_id",
2282 "resourceId": "resource_id"
2284 "affectedVirtualLinks": [{"id": "vl_instance_id",
2285 "virtualLinkDescId": "virtual_link_desc_id",
2286 "changeType": VNFC_CHANGE_TYPE.MODIFIED,
2287 "networkResource": {
2288 "vimLevelResourceType": "network",
2289 "resourceId": "resource_id"
2291 "changedExtConnectivity": [{"id": "virtual_link_instance_id",
2292 "extLinkPorts": [{"cpInstanceId": "cp_instance_id",
2295 "vimConnectionId": "vim_connection_id",
2296 "resourceId": "resource_id"
2301 HandleVnfLcmOocNotification(self.vnfm_inst_id, self.m_nf_inst_id, data).do_biz()
2302 vnfc_inst = VNFCInstModel.objects.get(vnfcinstanceid="vnfc_instance_id", vduid="vdu_id",
2303 nfinstid=self.nf_inst_id, vmid="resource_id")
2304 self.assertIsInstance(vnfc_inst, VNFCInstModel)
2305 vl_inst = VLInstModel.objects.get(vlinstanceid="vl_instance_id", vldid="virtual_link_desc_id",
2306 vlinstancename="resource_id", ownertype=0, ownerid=self.nf_inst_id,
2307 relatednetworkid="resource_id", vltype=0)
2308 self.assertIsInstance(vl_inst, VLInstModel)
2309 port_inst = PortInstModel.objects.get(networkid='unknown', subnetworkid='unknown', vimid="vim_connection_id",
2310 resourceid="resource_id", name="resource_id", instid="unknown",
2311 cpinstanceid="cp_instance_id", bandwidth='unknown',
2312 operationalstate='active', ipaddress="unkown", macaddress='unknown',
2313 floatipaddress='unknown', serviceipaddress='unknown',
2314 typevirtualnic='unknown', sfcencapsulation='gre', direction='unknown',
2316 self.assertIsInstance(port_inst, PortInstModel)
2317 cp_inst = CPInstModel.objects.get(cpinstanceid="cp_instance_id", cpdid="cpd_id", ownertype=0,
2318 ownerid=self.nf_inst_id, relatedtype=2, status='active')
2319 self.assertIsInstance(cp_inst, CPInstModel)
2321 def test_handle_vnf_identifier_creation_notification(self):
2323 vnf_instance_id = "vnf_instance_id"
2325 "timeStamp": "20190809",
2327 HandleVnfIdentifierCreationNotification(vnfm_id, vnf_instance_id, data).do_biz()
2328 nf_inst = NfInstModel.objects.get(mnfinstid=vnf_instance_id, vnfm_inst_id=vnfm_id, create_time="20190809")
2329 self.assertIsInstance(nf_inst, NfInstModel)
2331 def test_handle_vnf_identifier_deletion_notification(self):
2332 nf_inst_id = "nf_inst_id"
2334 vnf_instance_id = "vnf_instance_id"
2335 NfInstModel.objects.create(nfinstid=nf_inst_id,
2336 vnfm_inst_id=vnfm_id,
2337 status=VNF_STATUS.NULL,
2338 mnfinstid=vnf_instance_id,
2339 vnfd_model=self.vnfd_model
2342 "timeStamp": "20190809",
2344 HandleVnfIdentifierDeletionNotification(vnfm_id, vnf_instance_id, data).do_biz()
2345 nf_inst = NfInstModel.objects.filter(mnfinstid=vnf_instance_id, vnfm_inst_id=vnfm_id)
2346 self.assertEqual(len(nf_inst), 0)
2348 def test_handle_vnf_identifier_deletion_notification_when_nf_not_exists(self):
2349 NfInstModel.objects.all().delete()
2350 vnfm_id = "nf_not_exists"
2351 vnf_instance_id = "nf_not_exists"
2353 "timeStamp": "20190809",
2356 HandleVnfIdentifierDeletionNotification(vnfm_id, vnf_instance_id, data).do_biz()
2357 self.assertEqual(1, 0)
2359 self.assertEqual(1, 1)