X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=lcm%2Flcm%2Fnf%2Ftests%2Ftest_operate_vnf.py;h=17408c2eb43fe8cdbab1c1c8d3a4e128a7e82a90;hb=46ad7c172411214c5432ed93fda4271288077447;hp=2071472cfd00f6b15ea70a067737dd3d17ac6fab;hpb=0b8b410f563039cfc8879631bcb2587974de7875;p=vfc%2Fgvnfm%2Fvnflcm.git diff --git a/lcm/lcm/nf/tests/test_operate_vnf.py b/lcm/lcm/nf/tests/test_operate_vnf.py index 2071472c..17408c2e 100644 --- a/lcm/lcm/nf/tests/test_operate_vnf.py +++ b/lcm/lcm/nf/tests/test_operate_vnf.py @@ -13,15 +13,20 @@ # limitations under the License. import json - +import uuid import mock + from django.test import TestCase, Client from rest_framework import status from lcm.nf.biz.operate_vnf import OperateVnf -from lcm.pub.database.models import NfInstModel, JobStatusModel, VmInstModel +from lcm.pub.database.models import NfInstModel +from lcm.pub.database.models import JobStatusModel +from lcm.pub.database.models import VmInstModel +from lcm.pub.database.models import SubscriptionModel from lcm.pub.utils import restcall from lcm.pub.utils.jobutil import JobUtil +from lcm.pub.utils.notificationsutil import NotificationsUtil from lcm.pub.utils.timeutil import now_time from lcm.pub.vimapi import api @@ -34,70 +39,156 @@ class TestNFOperate(TestCase): VmInstModel.objects.all().delete() def assert_job_result(self, job_id, job_progress, job_detail): - jobs = JobStatusModel.objects.filter(jobid=job_id, - progress=job_progress, - descp=job_detail) + jobs = JobStatusModel.objects.filter( + jobid=job_id, + progress=job_progress, + descp=job_detail + ) self.assertEqual(1, len(jobs)) def test_operate_vnf_not_found(self): req_data = { "changeStateTo": "STARTED" } - response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json') - self.failUnlessEqual(status.HTTP_404_NOT_FOUND, response.status_code) + response = self.client.post( + "/api/vnflcm/v1/vnf_instances/12/operate", + data=req_data, + format='json' + ) + self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code) def test_operate_vnf_conflict(self): req_data = { "changeStateTo": "STARTED" } - NfInstModel(nfinstid='12', nf_name='VNF1', status='NOT_INSTANTIATED').save() - response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json') - self.failUnlessEqual(status.HTTP_409_CONFLICT, response.status_code) - NfInstModel(nfinstid='12', nf_name='VNF1', status='NOT_INSTANTIATED').delete() + NfInstModel( + nfinstid='12', + nf_name='VNF1', + status='NOT_INSTANTIATED' + ).save() + response = self.client.post( + "/api/vnflcm/v1/vnf_instances/12/operate", + data=req_data, + format='json' + ) + self.assertEqual(status.HTTP_409_CONFLICT, response.status_code) + NfInstModel( + nfinstid='12', + nf_name='VNF1', + status='NOT_INSTANTIATED' + ).delete() + + def test_operate_vnf_badreq(self): + NfInstModel( + nfinstid='345', + nf_name='VNF1', + nf_desc="VNF DESC", + vnfdid="1", + netype="XGW", + vendor="ZTE", + vnfSoftwareVersion="V1", + version="V1", + package_id="2", + status='INSTANTIATED' + ).save() + url = "/api/vnflcm/v1/vnf_instances/345/operate" + response = self.client.post( + url, + data={}, + format='json' + ) + NfInstModel.objects.filter(nfinstid='345').delete() + self.assertEqual( + status.HTTP_400_BAD_REQUEST, + response.status_code + ) @mock.patch.object(OperateVnf, 'run') def test_operate_vnf_success(self, mock_run): req_data = { "changeStateTo": "STARTED" } - NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').save() - response = self.client.post("/api/vnflcm/v1/vnf_instances/12/operate", data=req_data, format='json') + NfInstModel( + nfinstid='12', + nf_name='VNF1', + status='INSTANTIATED' + ).save() + response = self.client.post( + "/api/vnflcm/v1/vnf_instances/12/operate", + data=req_data, + format='json' + ) mock_run.re.return_value = None - self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code) - NfInstModel(nfinstid='12', nf_name='VNF1', status='INSTANTIATED').delete() + self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code) + NfInstModel( + nfinstid='12', + nf_name='VNF1', + status='INSTANTIATED' + ).delete() @mock.patch.object(restcall, 'call_req') @mock.patch.object(api, 'call') - def test_operate_vnf_success_start(self, mock_call, mock_call_req): - NfInstModel.objects.create(nfinstid='1111', - nf_name='2222', - vnfminstid='1', - package_id='todo', - version='', - vendor='', - netype='', - vnfd_model='', - status='INSTANTIATED', - nf_desc='', - vnfdid='', - vnfSoftwareVersion='', - vnfConfigurableProperties='todo', - localizationLanguage='EN_US', - create_time=now_time()) - - VmInstModel.objects.create(vmid="1", - vimid="1", - resouceid="11", - insttype=0, - instid="1111", - vmname="test_01", - is_predefined=1, - operationalstate=1) + @mock.patch.object(NotificationsUtil, 'post_notification') + def test_operate_vnf_success_start(self, mock_post_notification, mock_call, mock_call_req): + NfInstModel.objects.create( + nfinstid='1111', + nf_name='2222', + vnfminstid='1', + package_id='todo', + version='', + vendor='', + netype='', + vnfd_model='', + status='INSTANTIATED', + nf_desc='', + vnfdid='', + vnfSoftwareVersion='', + vnfConfigurableProperties='todo', + localizationLanguage='EN_US', + create_time=now_time() + ) + + VmInstModel.objects.create( + vmid="1", + vimid="1", + resourceid="11", + insttype=0, + instid="1111", + vmname="test_01", + is_predefined=1, + operationalstate=1 + ) + + SubscriptionModel.objects.create( + subscription_id=str(uuid.uuid4()), + callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification', + auth_info=json.JSONEncoder().encode({ + 'authType': ['BASIC'], + 'paramsBasic': { + 'userName': 'username', + 'password': 'password' + } + }), + notification_types=str([ + 'VnfLcmOperationOccurrenceNotification', + 'VnfIdentifierCreationNotification', + 'VnfIdentifierDeletionNotification' + ]), + operation_types=str(['OPERATE']), + operation_states=str(['COMPLETED']), + vnf_instance_filter=json.JSONEncoder().encode({ + 'vnfdIds': [], + 'vnfProductsFromProviders': [], + 'vnfInstanceIds': ['1111'], + 'vnfInstanceNames': [], + }) + ) + t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200'] - t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200'] t3_action_vm_start_result = [0, json.JSONEncoder().encode(''), '202'] - mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_start_result] + mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_start_result] mock_call.return_value = None + mock_post_notification.return_value = None req_data = { "changeStateTo": "STARTED" } @@ -105,42 +196,71 @@ class TestNFOperate(TestCase): self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id) JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY") OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run() - vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11") - self.assertEqual("ACTIVE", vm[0].operationalstate) self.assert_job_result(self.job_id, 100, "Operate Vnf success.") @mock.patch.object(restcall, 'call_req') @mock.patch.object(api, 'call') - def test_operate_vnf_success_stop(self, mock_call, mock_call_req): - NfInstModel.objects.create(nfinstid='1111', - nf_name='2222', - vnfminstid='1', - package_id='todo', - version='', - vendor='', - netype='', - vnfd_model='', - status='INSTANTIATED', - nf_desc='', - vnfdid='', - vnfSoftwareVersion='', - vnfConfigurableProperties='todo', - localizationLanguage='EN_US', - create_time=now_time()) - - VmInstModel.objects.create(vmid="1", - vimid="1", - resouceid="11", - insttype=0, - instid="1111", - vmname="test_01", - is_predefined=1, - operationalstate=1) + @mock.patch.object(NotificationsUtil, 'post_notification') + def test_operate_vnf_success_stop(self, mock_post_notification, mock_call, mock_call_req): + NfInstModel.objects.create( + nfinstid='1111', + nf_name='2222', + vnfminstid='1', + package_id='todo', + version='', + vendor='', + netype='', + vnfd_model='', + status='INSTANTIATED', + nf_desc='', + vnfdid='', + vnfSoftwareVersion='', + vnfConfigurableProperties='todo', + localizationLanguage='EN_US', + create_time=now_time() + ) + + VmInstModel.objects.create( + vmid="1", + vimid="1", + resourceid="11", + insttype=0, + instid="1111", + vmname="test_01", + is_predefined=1, + operationalstate=1 + ) + + SubscriptionModel.objects.create( + subscription_id=str(uuid.uuid4()), + callback_uri='api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification', + auth_info=json.JSONEncoder().encode({ + 'authType': ['BASIC'], + 'paramsBasic': { + 'userName': 'username', + 'password': 'password' + } + }), + notification_types=str([ + 'VnfLcmOperationOccurrenceNotification', + 'VnfIdentifierCreationNotification', + 'VnfIdentifierDeletionNotification' + ]), + operation_types=str(['OPERATE']), + operation_states=str(['COMPLETED']), + vnf_instance_filter=json.JSONEncoder().encode({ + 'vnfdIds': [], + 'vnfProductsFromProviders': [], + 'vnfInstanceIds': ['1111'], + 'vnfInstanceNames': [], + }) + ) + t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200'] - t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200'] t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202'] - mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result] + mock_call_req.side_effect = [t1_apply_grant_result, t3_action_vm_stop_result] mock_call.return_value = None + mock_post_notification.return_value = None req_data = { "changeStateTo": "STOPPED" } @@ -148,41 +268,47 @@ class TestNFOperate(TestCase): self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id) JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY") OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run() - vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11") - self.assertEqual("INACTIVE", vm[0].operationalstate) self.assert_job_result(self.job_id, 100, "Operate Vnf success.") @mock.patch.object(restcall, 'call_req') @mock.patch.object(api, 'call') def test_operate_vnf_success_stop_graceful(self, mock_call, mock_call_req): - NfInstModel.objects.create(nfinstid='1111', - nf_name='2222', - vnfminstid='1', - package_id='todo', - version='', - vendor='', - netype='', - vnfd_model='', - status='INSTANTIATED', - nf_desc='', - vnfdid='', - vnfSoftwareVersion='', - vnfConfigurableProperties='todo', - localizationLanguage='EN_US', - create_time=now_time()) - - VmInstModel.objects.create(vmid="1", - vimid="1", - resouceid="11", - insttype=0, - instid="1111", - vmname="test_01", - is_predefined=1, - operationalstate=1) + NfInstModel.objects.create( + nfinstid='1111', + nf_name='2222', + vnfminstid='1', + package_id='todo', + version='', + vendor='', + netype='', + vnfd_model='', + status='INSTANTIATED', + nf_desc='', + vnfdid='', + vnfSoftwareVersion='', + vnfConfigurableProperties='todo', + localizationLanguage='EN_US', + create_time=now_time() + ) + + VmInstModel.objects.create( + vmid="1", + vimid="1", + resourceid="11", + insttype=0, + instid="1111", + vmname="test_01", + is_predefined=1, + operationalstate=1 + ) t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200'] t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200'] t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202'] - mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result] + mock_call_req.side_effect = [ + t1_apply_grant_result, + t2_lcm_notify_result, + t3_action_vm_stop_result + ] mock_call.return_value = None req_data = { "changeStateTo": "STOPPED", @@ -193,41 +319,47 @@ class TestNFOperate(TestCase): self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id) JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY") OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run() - vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11") - self.assertEqual("INACTIVE", vm[0].operationalstate) self.assert_job_result(self.job_id, 100, "Operate Vnf success.") @mock.patch.object(restcall, 'call_req') @mock.patch.object(api, 'call') def test_operate_vnf_success_stop_forceful(self, mock_call, mock_call_req): - NfInstModel.objects.create(nfinstid='1111', - nf_name='2222', - vnfminstid='1', - package_id='todo', - version='', - vendor='', - netype='', - vnfd_model='', - status='INSTANTIATED', - nf_desc='', - vnfdid='', - vnfSoftwareVersion='', - vnfConfigurableProperties='todo', - localizationLanguage='EN_US', - create_time=now_time()) - - VmInstModel.objects.create(vmid="1", - vimid="1", - resouceid="11", - insttype=0, - instid="1111", - vmname="test_01", - is_predefined=1, - operationalstate=1) + NfInstModel.objects.create( + nfinstid='1111', + nf_name='2222', + vnfminstid='1', + package_id='todo', + version='', + vendor='', + netype='', + vnfd_model='', + status='INSTANTIATED', + nf_desc='', + vnfdid='', + vnfSoftwareVersion='', + vnfConfigurableProperties='todo', + localizationLanguage='EN_US', + create_time=now_time() + ) + + VmInstModel.objects.create( + vmid="1", + vimid="1", + resourceid="11", + insttype=0, + instid="1111", + vmname="test_01", + is_predefined=1, + operationalstate=1 + ) t1_apply_grant_result = [0, json.JSONEncoder().encode(''), '200'] t2_lcm_notify_result = [0, json.JSONEncoder().encode(''), '200'] t3_action_vm_stop_result = [0, json.JSONEncoder().encode(''), '202'] - mock_call_req.side_effect = [t1_apply_grant_result, t2_lcm_notify_result, t3_action_vm_stop_result] + mock_call_req.side_effect = [ + t1_apply_grant_result, + t2_lcm_notify_result, + t3_action_vm_stop_result + ] mock_call.return_value = None req_data = { "changeStateTo": "STOPPED", @@ -237,6 +369,4 @@ class TestNFOperate(TestCase): self.job_id = JobUtil.create_job('NF', 'OPERATE', self.nf_inst_id) JobUtil.add_job_status(self.job_id, 0, "OPERATE_VNF_READY") OperateVnf(req_data, nf_inst_id=self.nf_inst_id, job_id=self.job_id).run() - vm = VmInstModel.objects.filter(vmid="1", vimid="1", resouceid="11") - self.assertEqual("INACTIVE", vm[0].operationalstate) self.assert_job_result(self.job_id, 100, "Operate Vnf success.")