From 85a2c699512811122940d6896a0f40f3749527c8 Mon Sep 17 00:00:00 2001 From: maopengzhang Date: Wed, 27 Mar 2019 14:02:35 +0800 Subject: add NS heal test usecase add NS heal test usecase Change-Id: I8a011d6b1acdc0df7473acd35507b4a4edfd4a26 Issue-ID: VFC-1215 Signed-off-by: maopengzhang --- lcm/ns/biz/ns_terminate.py | 1 + lcm/ns/tests/test_sol_ns_heal_api.py | 173 +++++++++++++ lcm/ns/tests/test_sol_ns_instantiate.py | 356 --------------------------- lcm/ns/tests/test_sol_ns_instantiate_api.py | 368 ++++++++++++++++++++++++++++ lcm/ns/tests/test_sol_ns_terminate.py | 63 ----- lcm/ns/tests/test_sol_ns_terminate_api.py | 74 ++++++ 6 files changed, 616 insertions(+), 419 deletions(-) create mode 100644 lcm/ns/tests/test_sol_ns_heal_api.py delete mode 100644 lcm/ns/tests/test_sol_ns_instantiate.py create mode 100644 lcm/ns/tests/test_sol_ns_instantiate_api.py delete mode 100644 lcm/ns/tests/test_sol_ns_terminate.py create mode 100644 lcm/ns/tests/test_sol_ns_terminate_api.py diff --git a/lcm/ns/biz/ns_terminate.py b/lcm/ns/biz/ns_terminate.py index 4bc9d430..701e65c6 100644 --- a/lcm/ns/biz/ns_terminate.py +++ b/lcm/ns/biz/ns_terminate.py @@ -45,6 +45,7 @@ class TerminateNsService(threading.Thread): try: if not NSInstModel.objects.filter(id=self.ns_inst_id): JobUtil.add_job_status(self.job_id, 100, "Need not terminate.", '') + NsLcmOpOcc.update(self.occ_id, "COMPLETED") return JobUtil.add_job_status(self.job_id, 10, "Starting terminate...", '') diff --git a/lcm/ns/tests/test_sol_ns_heal_api.py b/lcm/ns/tests/test_sol_ns_heal_api.py new file mode 100644 index 00000000..8ff4ced5 --- /dev/null +++ b/lcm/ns/tests/test_sol_ns_heal_api.py @@ -0,0 +1,173 @@ +# Copyright 2019 ZTE Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json + +import mock +from django.test import Client +from django.test import TestCase +from rest_framework import status + +from lcm.ns.biz.ns_heal import NSHealService +from lcm.pub.database.models import NSInstModel, NfInstModel +from lcm.pub.exceptions import NSLCMException +from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE + + +class TestHealNsApi(TestCase): + def setUp(self): + self.url = "/api/nslcm/v1/ns_instances/%s/heal" + self.ns_inst_id = '1' + self.nf_inst_id = '1' + self.nf_uuid = '1-1-1' + + self.job_id = JobUtil.create_job("NS", JOB_TYPE.HEAL_VNF, self.ns_inst_id) + + self.client = Client() + + model = json.dumps({ + "metadata": { + "vnfdId": "1", + "vnfdName": "PGW001", + "vnfProvider": "zte", + "vnfdVersion": "V00001", + "vnfVersion": "V5.10.20", + "productType": "CN", + "vnfType": "PGW", + "description": "PGW VNFD description", + "isShared": True, + "vnfExtendType": "driver" + } + }) + NSInstModel.objects.filter().delete() + NfInstModel.objects.filter().delete() + NSInstModel(id=self.ns_inst_id, name="ns_name", status='null').save() + NfInstModel.objects.create(nfinstid=self.nf_inst_id, + nf_name='name_1', + vnf_id='1', + vnfm_inst_id='1', + ns_inst_id=self.ns_inst_id, + max_cpu='14', + max_ram='12296', + max_hd='101', + max_shd="20", + max_net=10, + status='null', + mnfinstid=self.nf_uuid, + package_id='pkg1', + vnfd_model=model) + + def tearDown(self): + pass + + @mock.patch.object(NSHealService, 'run') + def test_heal_vnf_url(self, mock_run): + + data = { + "healVnfData": [{ + "vnfInstanceId": self.nf_inst_id, + "cause": "vm is down", + "additionalParams": { + "action": "restartvm", + "actionvminfo": { + "vmid": "33", + "vduid": "", + "vmname": "xgw-smp11" + } + } + }] + } + + response = self.client.post(self.url % self.ns_inst_id, data=data) + self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.data) + self.assertIsNotNone(response.data) + self.assertIsNotNone(response['Location']) + + # add healNsData + + @mock.patch.object(NSHealService, 'run') + def test_heal_ns_url(self, mock_run): + + data = { + "healNsData": { + "vnfInstanceId": self.nf_inst_id, + "cause": "", + "additionalParams": { + "action": "vmreset", + "actionvminfo": { + "vmid": "33", + "vduid": "", + "vmname": "xgw-smp11" + } + } + } + } + + response = self.client.post(self.url % self.ns_inst_id, data=data) + self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.data) + self.assertIsNotNone(response['Location']) + + @mock.patch.object(NSHealService, "start") + def test_heal_vnf_non_existing_ns(self, mock_start): + mock_start.side_effect = NSLCMException("NS Not Found") + + ns_inst_id = "2" + + data = { + "healVnfData": [{ + "vnfInstanceId": self.nf_inst_id, + "cause": "vm is down", + "additionalParams": { + "action": "restartvm", + "actionvminfo": { + "vmid": "33", + "vmname": "xgw-smp11" + } + } + }] + } + + response = self.client.post(self.url % ns_inst_id, data=data) + self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) + + # add healNsData + @mock.patch.object(NSHealService, "start") + def test_heal_ns_heal_non_existing_ns(self, mock_start): + mock_start.side_effect = NSLCMException("NS Not Found") + + ns_inst_id = "2" + + data = { + "healNsData": { + "vnfInstanceId": self.nf_inst_id, + "cause": "", + "additionalParams": { + "action": "vmreset", + "actionvminfo": { + "vmid": "33", + "vduid": "", + "vmname": "xgw-smp11" + } + } + } + } + + response = self.client.post(self.url % ns_inst_id, data=data) + self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) + + @mock.patch.object(NSHealService, "start") + def test_heal_vnf_empty_post(self, mock_start): + mock_start.side_effect = NSLCMException("healVnfData parameter does not exist or value is incorrect.") + response = self.client.post(self.url % self.ns_inst_id, data={}) + self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) diff --git a/lcm/ns/tests/test_sol_ns_instantiate.py b/lcm/ns/tests/test_sol_ns_instantiate.py deleted file mode 100644 index 791bb1dd..00000000 --- a/lcm/ns/tests/test_sol_ns_instantiate.py +++ /dev/null @@ -1,356 +0,0 @@ -# Copyright 2019 ZTE Corporation. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import json -import mock -from mock import MagicMock -from django.test import TestCase -from rest_framework import status -from rest_framework.test import APIClient -from lcm.ns.biz.ns_instant import BuildInWorkflowThread -from lcm.ns.biz.ns_instant import InstantNSService -from lcm.pub.database.models import NSInstModel -from lcm.pub.utils import restcall -from lcm.pub.config import config - -nsd_model = json.dumps({ - "model": json.dumps({ - "vnfs": [{ - "vnf_id": "vnf1", - "properties": { - "id": "vnfd1", - "nf_type": "xgw" - }, - "dependencies": [{ - "vl_id": "5" - }] - }], - "vls": [{ - "vl_id": "5", - "properties": {} - }] - }) -}) - -nsd = json.dumps({ - "vnffgs": [ - ], - "inputs": { - }, - "pnfs": [ - { - "pnf_id": "du", - "networks": [ - ], - "description":"", - "properties":{ - "descriptor_id": "zte_ran_du_0001", - "descriptor_invariant_id": "1111", - "provider": "ZTE", - "version": "1.0", - "function_description": "RAN DU Function", - "name": "ZTE RAN DU" - } - } - ], - "ns_exposed": { - "external_cps": [ - ], - "forward_cps": [ - ] - }, - "graph": { - "cucp": [ - ], - "du": [ - ], - "vl_flat_net": [ - "cucp", - "cuup" - ], - "vl_ext_net": [ - "cucp", - "cuup" - ], - "cuup": [ - ] - }, - "basepath": "c:\\users\\10030173\\appdata\\local\\temp\\tmpvg5vto", - "vnfs": [ - { - "networks": [ - { - "key_name": "ran_ext_net", - "vl_id": "vl_ext_net" - }, - { - "key_name": "ran_flat_net", - "vl_id": "vl_flat_net" - } - ], - "dependencies": [ - { - "key_name": "ran_ext_net", - "vl_id": "vl_ext_net" - }, - { - "key_name": "ran_flat_net", - "vl_id": "vl_flat_net" - } - ], - "vnf_id": "cucp", - "description": "", - "properties": { - "descriptor_id": "zte_ran_cucp_0001", - "flavour_description": "default", - "software_version": "1.0.1", - "flavour_id": "1", - "descriptor_version": "1.0", - "provider": "ZTE", - "id": "zte_ran_cucp_0001", - "vnfm_info": [ - "GVNFM-Driver" - ], - "product_name":"ran" - } - }, - { - "networks": [ - { - "key_name": "ran_ext_net", - "vl_id": "vl_ext_net" - }, - { - "key_name": "ran_flat_net", - "vl_id": "vl_flat_net" - } - ], - "dependencies": [ - { - "key_name": "ran_ext_net", - "vl_id": "vl_ext_net" - }, - { - "key_name": "ran_flat_net", - "vl_id": "vl_flat_net" - } - ], - "vnf_id": "cuup", - "description": "", - "properties": { - "descriptor_id": "zte_ran_cuup_0001", - "flavour_description": "default", - "software_version": "1.0.1", - "flavour_id": "1", - "descriptor_version": "1.0", - "provider": "ZTE", - "id": "zte_ran_cuup_0001", - "vnfm_info": [ - "GVNFM-Driver" - ], - "product_name":"ran" - } - } - ], - "fps": [ - ], - "vls": [ - { - "vl_id": "vl_ext_net", - "description": "", - "properties": { - "connectivity_type": { - "layer_protocol": "ipv4" - }, - "vl_profile": { - "cidr": "10.0.0.0/24", - "max_bit_rate_requirements": { - "root": 10000000, - "leaf": 10000000 - }, - "networkName": "ran_ext_net", - "min_bit_rate_requirements": { - "root": 10000000, - "leaf": 10000000 - }, - "dhcpEnabled": False - }, - "version": "1.0.1" - } - }, - { - "vl_id": "vl_flat_net", - "description": "", - "properties": { - "connectivity_type": { - "layer_protocol": "ipv4" - }, - "vl_profile": { - "cidr": "10.1.0.0/24", - "max_bit_rate_requirements": { - "root": 10000000, - "leaf": 10000000 - }, - "networkName": "ran_flat_net", - "min_bit_rate_requirements": { - "root": 10000000, - "leaf": 10000000 - }, - "dhcpEnabled": False - }, - "version": "1.0.1" - } - } - ], - "nested_ns": [ - ], - "metadata": { - "template_name": "RAN-NS", - "template_version": "1.0", - "template_author": "ZTE" - } -}) -vnfminfo = {"vnfmId": "1"} - - -class TestInstantiateNs(TestCase): - - def setUp(self): - self.client = APIClient() - NSInstModel.objects.filter().delete() - self.url = "/api/nslcm/v1/ns_instances/%s/instantiate" - self.req_data = { - "additionalParamForNs": { - "location": "2" - }, - "nsFlavourId": 'default', - "localizationLanguage": [{ - "vnfProfileId": "vnfd1", - "locationConstraints": { - "countryCode": "countryCode", - "vimId": "3", - "civicAddressElement": [ - {"caType": "type1", - "caValue": 1 - } - ] - } - }] - } - self.nsd_model = nsd_model - self.updated_nsd_model = { - "vnfs": [{ - "dependencies": [{ - "vl_id": "5" - }], - "vnf_id": "vnf1", - "properties": { - "nf_type": "xgw", - "id": "vnfd1" - } - }], - "vls": [{ - "vl_id": "5", - "properties": { - "location_info": { - "vimid": "3" - } - } - }] - } - self.vnfms = json.dumps({ - "esr-vnfm": [{ - "vnfm-id": "4" - }] - }) - self.vnfm = json.dumps({ - "type": "xgw", - "vim-id": "3", - "vnfm-id": "4", - "certificate-url": "http://127.0.0.0/ztevnfm/v1/auth", - "esr-system-info-list": { - "esr-system-info": [{ - "type": "xgw", - "vendor": "zte", - "version": "1.0", - "service-url": "http://127.0.0.0/ztevnfm/v1", - "user-name": "admin", - "password": "admin123" - }] - } - }) - NSInstModel(id="2", nspackage_id="7", nsd_id="2", status="active").save() - - def tearDown(self): - pass - - @mock.patch.object(restcall, 'call_req') - @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model)) - @mock.patch.object(BuildInWorkflowThread, 'run') - def test_ns_instantiate_when_succeed_to_enter_workflow(self, mock_run, mock_call_req): - config.WORKFLOW_OPTION = "buildin" - mock_call_req.side_effect = [ - [0, self.nsd_model, '200'], - [0, self.vnfms, '200'], - [0, self.vnfm, '200'] - ] - resp = self.client.post(self.url % '2', data=self.req_data, format='json') - self.assertEqual(status.HTTP_202_ACCEPTED, resp.status_code) - - @mock.patch.object(InstantNSService, 'do_biz') - def test_ns_instantiate_normal(self, mock_do_biz): - mock_do_biz.return_value = {'occ_id': "1"} - resp = self.client.post(self.url % '2', data=self.req_data, format='json') - self.failUnlessEqual(status.HTTP_202_ACCEPTED, resp.status_code) - - @mock.patch.object(restcall, 'call_req') - def test_ns_instantiate_when_fail_to_parse_nsd(self, mock_call_req): - mock_call_req.return_value = [1, "Failed to parse nsd", '500'] - resp = self.client.post(self.url % '2', data=self.req_data, format='json') - self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) - - @mock.patch('lcm.ns.biz.ns_instantiate_flow.post_deal') - @mock.patch.object(restcall, 'call_req') - @mock.patch('lcm.ns.biz.ns_instantiate_flow.update_job') - @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd)) - @mock.patch('lcm.pub.msapi.extsys.select_vnfm', MagicMock(return_value=vnfminfo)) - def test_ns_instantiate_with_pnf(self, mock_updata_job, mock_call_req, mock_post_deal): - config.WORKFLOW_OPTION = "grapflow" - NSInstModel(id="1", name="test_ns", nspackage_id="1", status="created").save() - ret = [0, json.JSONEncoder().encode({'jobId': "1", "responseDescriptor": {"progress": 100}}), '200'] - mock_call_req.side_effect = [ret for i in range(1, 20)] - data = { - "nsFlavourId": 'default', - "additionalParamForNs": { - "sdnControllerId": "2" - }, - "additionalParamsForVnf": [ - { - "vnfProfileId": "zte_ran_cucp_0001", - "additionalParams": {"vimId": "3"} - }, - { - "vnfProfileId": "zte_ran_cuup_0001", - "additionalParams": {"vimId": "3"} - } - ], - "addpnfData": [{ - "pnfId": 1, - "pnfName": "test_pnf", - "pnfdId": "zte_ran_du_0001", - "pnfProfileId": "du" - }] - } - response = self.client.post(self.url % '1', data=data, format='json') - self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) diff --git a/lcm/ns/tests/test_sol_ns_instantiate_api.py b/lcm/ns/tests/test_sol_ns_instantiate_api.py new file mode 100644 index 00000000..37e9eda6 --- /dev/null +++ b/lcm/ns/tests/test_sol_ns_instantiate_api.py @@ -0,0 +1,368 @@ +# Copyright 2019 ZTE Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +import mock +from mock import MagicMock +from django.test import TestCase +from rest_framework import status +from rest_framework.test import APIClient +from lcm.ns.biz.ns_instant import BuildInWorkflowThread +from lcm.ns.biz.ns_instant import InstantNSService +from lcm.pub.database.models import NSInstModel +from lcm.pub.utils import restcall +from lcm.pub.config import config + +nsd_model = json.dumps({ + "model": json.dumps({ + "vnfs": [{ + "vnf_id": "vnf1", + "properties": { + "id": "vnfd1", + "nf_type": "xgw" + }, + "dependencies": [{ + "vl_id": "5" + }] + }], + "vls": [{ + "vl_id": "5", + "properties": {} + }] + }) +}) + +nsd = json.dumps({ + "vnffgs": [ + ], + "inputs": { + }, + "pnfs": [ + { + "pnf_id": "du", + "networks": [ + ], + "description":"", + "properties":{ + "descriptor_id": "zte_ran_du_0001", + "descriptor_invariant_id": "1111", + "provider": "ZTE", + "version": "1.0", + "function_description": "RAN DU Function", + "name": "ZTE RAN DU" + } + } + ], + "ns_exposed": { + "external_cps": [ + ], + "forward_cps": [ + ] + }, + "graph": { + "cucp": [ + ], + "du": [ + ], + "vl_flat_net": [ + "cucp", + "cuup" + ], + "vl_ext_net": [ + "cucp", + "cuup" + ], + "cuup": [ + ] + }, + "basepath": "c:\\users\\10030173\\appdata\\local\\temp\\tmpvg5vto", + "vnfs": [ + { + "networks": [ + { + "key_name": "ran_ext_net", + "vl_id": "vl_ext_net" + }, + { + "key_name": "ran_flat_net", + "vl_id": "vl_flat_net" + } + ], + "dependencies": [ + { + "key_name": "ran_ext_net", + "vl_id": "vl_ext_net" + }, + { + "key_name": "ran_flat_net", + "vl_id": "vl_flat_net" + } + ], + "vnf_id": "cucp", + "description": "", + "properties": { + "descriptor_id": "zte_ran_cucp_0001", + "flavour_description": "default", + "software_version": "1.0.1", + "flavour_id": "1", + "descriptor_version": "1.0", + "provider": "ZTE", + "id": "zte_ran_cucp_0001", + "vnfm_info": [ + "GVNFM-Driver" + ], + "product_name":"ran" + } + }, + { + "networks": [ + { + "key_name": "ran_ext_net", + "vl_id": "vl_ext_net" + }, + { + "key_name": "ran_flat_net", + "vl_id": "vl_flat_net" + } + ], + "dependencies": [ + { + "key_name": "ran_ext_net", + "vl_id": "vl_ext_net" + }, + { + "key_name": "ran_flat_net", + "vl_id": "vl_flat_net" + } + ], + "vnf_id": "cuup", + "description": "", + "properties": { + "descriptor_id": "zte_ran_cuup_0001", + "flavour_description": "default", + "software_version": "1.0.1", + "flavour_id": "1", + "descriptor_version": "1.0", + "provider": "ZTE", + "id": "zte_ran_cuup_0001", + "vnfm_info": [ + "GVNFM-Driver" + ], + "product_name":"ran" + } + } + ], + "fps": [ + ], + "vls": [ + { + "vl_id": "vl_ext_net", + "description": "", + "properties": { + "connectivity_type": { + "layer_protocol": "ipv4" + }, + "vl_profile": { + "cidr": "10.0.0.0/24", + "max_bit_rate_requirements": { + "root": 10000000, + "leaf": 10000000 + }, + "networkName": "ran_ext_net", + "min_bit_rate_requirements": { + "root": 10000000, + "leaf": 10000000 + }, + "dhcpEnabled": False + }, + "version": "1.0.1" + } + }, + { + "vl_id": "vl_flat_net", + "description": "", + "properties": { + "connectivity_type": { + "layer_protocol": "ipv4" + }, + "vl_profile": { + "cidr": "10.1.0.0/24", + "max_bit_rate_requirements": { + "root": 10000000, + "leaf": 10000000 + }, + "networkName": "ran_flat_net", + "min_bit_rate_requirements": { + "root": 10000000, + "leaf": 10000000 + }, + "dhcpEnabled": False + }, + "version": "1.0.1" + } + } + ], + "nested_ns": [ + ], + "metadata": { + "template_name": "RAN-NS", + "template_version": "1.0", + "template_author": "ZTE" + } +}) +vnfminfo = {"vnfmId": "1"} + + +class TestInstantiateNsApi(TestCase): + + def setUp(self): + self.client = APIClient() + NSInstModel.objects.filter().delete() + self.url = "/api/nslcm/v1/ns_instances/%s/instantiate" + self.req_data = { + "additionalParamForNs": { + "location": "2" + }, + "nsFlavourId": 'default', + "localizationLanguage": [{ + "vnfProfileId": "vnfd1", + "locationConstraints": { + "countryCode": "countryCode", + "vimId": "3", + "civicAddressElement": [ + {"caType": "type1", + "caValue": 1 + } + ] + } + }] + } + self.nsd_model = nsd_model + self.updated_nsd_model = { + "vnfs": [{ + "dependencies": [{ + "vl_id": "5" + }], + "vnf_id": "vnf1", + "properties": { + "nf_type": "xgw", + "id": "vnfd1" + } + }], + "vls": [{ + "vl_id": "5", + "properties": { + "location_info": { + "vimid": "3" + } + } + }] + } + self.vnfms = json.dumps({ + "esr-vnfm": [{ + "vnfm-id": "4" + }] + }) + self.vnfm = json.dumps({ + "type": "xgw", + "vim-id": "3", + "vnfm-id": "4", + "certificate-url": "http://127.0.0.0/ztevnfm/v1/auth", + "esr-system-info-list": { + "esr-system-info": [{ + "type": "xgw", + "vendor": "zte", + "version": "1.0", + "service-url": "http://127.0.0.0/ztevnfm/v1", + "user-name": "admin", + "password": "admin123" + }] + } + }) + NSInstModel(id="2", nspackage_id="7", nsd_id="2", status="active").save() + + def tearDown(self): + pass + + @mock.patch.object(restcall, 'call_req') + @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model)) + @mock.patch.object(BuildInWorkflowThread, 'run') + def test_ns_instantiate_when_succeed_to_enter_workflow(self, mock_run, mock_call_req): + config.WORKFLOW_OPTION = "buildin" + mock_call_req.side_effect = [ + [0, self.nsd_model, '200'], + [0, self.vnfms, '200'], + [0, self.vnfm, '200'] + ] + response = self.client.post(self.url % '2', data=self.req_data, format='json') + self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code) + self.assertIsNotNone(response['Location']) + + @mock.patch.object(InstantNSService, 'do_biz') + def test_ns_instantiate_normal(self, mock_do_biz): + mock_do_biz.return_value = {'occ_id': "1"} + resp = self.client.post(self.url % '2', data=self.req_data, format='json') + self.failUnlessEqual(status.HTTP_202_ACCEPTED, resp.status_code) + + @mock.patch.object(restcall, 'call_req') + def test_ns_instantiate_when_fail_to_parse_nsd(self, mock_call_req): + mock_call_req.return_value = [1, "Failed to parse nsd", '500'] + resp = self.client.post(self.url % '2', data=self.req_data, format='json') + self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) + + @mock.patch('lcm.ns.biz.ns_instantiate_flow.post_deal') + @mock.patch.object(restcall, 'call_req') + @mock.patch('lcm.ns.biz.ns_instantiate_flow.update_job') + @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd)) + @mock.patch('lcm.pub.msapi.extsys.select_vnfm', MagicMock(return_value=vnfminfo)) + def test_ns_instantiate_with_pnf(self, mock_updata_job, mock_call_req, mock_post_deal): + config.WORKFLOW_OPTION = "grapflow" + NSInstModel(id="1", name="test_ns", nspackage_id="1", status="created").save() + ret = [0, json.JSONEncoder().encode({'jobId': "1", "responseDescriptor": {"progress": 100}}), '200'] + mock_call_req.side_effect = [ret for i in range(1, 20)] + data = { + "nsFlavourId": 'default', + "additionalParamForNs": { + "sdnControllerId": "2" + }, + "additionalParamsForVnf": [ + { + "vnfProfileId": "zte_ran_cucp_0001", + "additionalParams": {"vimId": "3"} + }, + { + "vnfProfileId": "zte_ran_cuup_0001", + "additionalParams": {"vimId": "3"} + } + ], + "addpnfData": [{ + "pnfId": 1, + "pnfName": "test_pnf", + "pnfdId": "zte_ran_du_0001", + "pnfProfileId": "du" + }] + } + response = self.client.post(self.url % '1', data=data, format='json') + self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) + self.assertIsNotNone(response['Location']) + + def test_method_not_allowed(self): + response = self.client.put(self.url % '1', data=self.req_data, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.patch(self.url % '1', data=self.req_data, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.delete(self.url % '1', data=self.req_data, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.get(self.url % '1', data=self.req_data, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) diff --git a/lcm/ns/tests/test_sol_ns_terminate.py b/lcm/ns/tests/test_sol_ns_terminate.py deleted file mode 100644 index 7d6c52bb..00000000 --- a/lcm/ns/tests/test_sol_ns_terminate.py +++ /dev/null @@ -1,63 +0,0 @@ -# Copyright 2019 ZTE Corporation. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -import uuid - -import mock -from django.test import TestCase, Client -from rest_framework import status - -from lcm.ns.biz.ns_terminate import TerminateNsService -from lcm.pub.database.models import NfInstModel, NSInstModel - - -class TestTerminateNsViews(TestCase): - def setUp(self): - self.client = Client() - self.url = "/api/nslcm/v1/ns_instances/%s/terminate" - self.ns_inst_id = '1' - self.nf_inst_id = '1' - self.vnffg_id = str(uuid.uuid4()) - self.vim_id = str(uuid.uuid4()) - self.job_id = str(uuid.uuid4()) - self.nf_uuid = '1-1-1' - self.tenant = "tenantname" - model = {"metadata": { - "vnfdId": "1", - "vnfdName": "PGW001", - "vnfProvider": "zte", - "vnfdVersion": "V00001", - "vnfVersion": "V5.10.20", - "productType": "CN", - "vnfType": "PGW", - "description": "PGW VNFD description", - "isShared": True, - "vnfExtendType": "driver" - }} - NSInstModel(id=self.ns_inst_id, name="ns_name", status='null').save() - NfInstModel.objects.create(nfinstid=self.nf_inst_id, nf_name='name_1', vnf_id='1', - vnfm_inst_id='1', ns_inst_id='1-1-1,2-2-2', - max_cpu='14', max_ram='12296', max_hd='101', max_shd="20", max_net=10, - status='null', mnfinstid=self.nf_uuid, package_id='pkg1', - vnfd_model=str(model)) - - def tearDown(self): - NSInstModel.objects.all().delete() - NfInstModel.objects.all().delete() - - @mock.patch.object(TerminateNsService, 'run') - def test_terminate_vnf_url(self, mock_run): - mock_run.re.return_value = "1" - req_data = {} - response = self.client.post(self.url % self.ns_inst_id, data=req_data) - self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code) diff --git a/lcm/ns/tests/test_sol_ns_terminate_api.py b/lcm/ns/tests/test_sol_ns_terminate_api.py new file mode 100644 index 00000000..0027312a --- /dev/null +++ b/lcm/ns/tests/test_sol_ns_terminate_api.py @@ -0,0 +1,74 @@ +# Copyright 2019 ZTE Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import uuid + +import mock +from django.test import TestCase, Client +from rest_framework import status + +from lcm.ns.biz.ns_terminate import TerminateNsService +from lcm.pub.database.models import NfInstModel, NSInstModel + + +class TestTerminateNsApi(TestCase): + def setUp(self): + self.client = Client() + self.url = "/api/nslcm/v1/ns_instances/%s/terminate" + self.ns_inst_id = '1' + self.nf_inst_id = '1' + self.vnffg_id = str(uuid.uuid4()) + self.vim_id = str(uuid.uuid4()) + self.job_id = str(uuid.uuid4()) + self.nf_uuid = '1-1-1' + self.tenant = "tenantname" + model = {"metadata": { + "vnfdId": "1", + "vnfdName": "PGW001", + "vnfProvider": "zte", + "vnfdVersion": "V00001", + "vnfVersion": "V5.10.20", + "productType": "CN", + "vnfType": "PGW", + "description": "PGW VNFD description", + "isShared": True, + "vnfExtendType": "driver" + }} + NSInstModel(id=self.ns_inst_id, name="ns_name", status='null').save() + NfInstModel.objects.create(nfinstid=self.nf_inst_id, nf_name='name_1', vnf_id='1', + vnfm_inst_id='1', ns_inst_id='1-1-1,2-2-2', + max_cpu='14', max_ram='12296', max_hd='101', max_shd="20", max_net=10, + status='null', mnfinstid=self.nf_uuid, package_id='pkg1', + vnfd_model=str(model)) + + def tearDown(self): + NSInstModel.objects.all().delete() + NfInstModel.objects.all().delete() + + @mock.patch.object(TerminateNsService, 'run') + def test_terminate_vnf(self, mock_run): + mock_run.re.return_value = "1" + req_data = {} + response = self.client.post(self.url % self.ns_inst_id, data=req_data) + self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code) + self.assertIsNotNone(response['Location']) + + def test_method_not_allowed(self): + response = self.client.put(self.url % '1', data={}, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.patch(self.url % '1', data={}, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.delete(self.url % '1', data={}, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.get(self.url % '1', data={}, format='json') + self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) -- cgit 1.2.3-korg