X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=lcm%2Fns%2Ftests%2Ftest_sol_ns_instantiate_api.py;h=f13b7c4161a9e02abc96bdc4af656459d2429b46;hb=0e1fd96cd9ad2b1697a3a47bf534eca329fb9a93;hp=37e9eda6cfb577f8fd931c66460af25569fdc53e;hpb=e485f64104f71b4b08c385883f3ba3b27305dfaa;p=vfc%2Fnfvo%2Flcm.git diff --git a/lcm/ns/tests/test_sol_ns_instantiate_api.py b/lcm/ns/tests/test_sol_ns_instantiate_api.py index 37e9eda6..f13b7c41 100644 --- a/lcm/ns/tests/test_sol_ns_instantiate_api.py +++ b/lcm/ns/tests/test_sol_ns_instantiate_api.py @@ -14,6 +14,7 @@ import json import mock +import uuid from mock import MagicMock from django.test import TestCase from rest_framework import status @@ -23,204 +24,10 @@ 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 +from lcm.ns.tests import NSD_MODEL_DICT, NSD_MODEL_WITH_PNF_DICT, VNFM_LIST_IN_AAI_DICT, VNFM_IN_AAI_DICT, VCPE_NS_MODEL_DICT, SOL_INSTANTIATE_NS_DICT, SOL_INSTANTIATE_NS_VCPE_DICT, SOL_INSTANTIATE_NS_WITH_PNF_DICT -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" - } -}) +nsd_model = json.dumps(NSD_MODEL_DICT) +nsd = json.dumps(NSD_MODEL_WITH_PNF_DICT) vnfminfo = {"vnfmId": "1"} @@ -230,67 +37,12 @@ class TestInstantiateNsApi(TestCase): 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.req_data = SOL_INSTANTIATE_NS_DICT 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() + self.vnfms = json.dumps(VNFM_LIST_IN_AAI_DICT) + self.vnfm = json.dumps(VNFM_IN_AAI_DICT) + self.nsInstanceId = str(uuid.uuid4()) + NSInstModel(id=self.nsInstanceId, nspackage_id="7", nsd_id="2", status="active").save() def tearDown(self): pass @@ -301,24 +53,25 @@ class TestInstantiateNsApi(TestCase): 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') + response = self.client.post(self.url % self.nsInstanceId, data=self.req_data, format='json') self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code) self.assertIsNotNone(response['Location']) + response = self.client.get(response['Location'], format='json') + self.assertEqual(response.status_code, status.HTTP_200_OK) @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_do_biz.return_value = {'occ_id': "1", 'data': {}} + response = self.client.post(self.url % self.nsInstanceId, data=self.req_data, format='json') + self.assertEqual(status.HTTP_202_ACCEPTED, response.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') + resp = self.client.post(self.url % self.nsInstanceId, 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') @@ -328,41 +81,37 @@ class TestInstantiateNsApi(TestCase): @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() + nsInstanceId = str(uuid.uuid4()) + NSInstModel(id=nsInstanceId, 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') + response = self.client.post(self.url % nsInstanceId, data=SOL_INSTANTIATE_NS_WITH_PNF_DICT, format='json') self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) self.assertIsNotNone(response['Location']) + response = self.client.get(response['Location'], format='json') + self.assertEqual(response.status_code, status.HTTP_200_OK) 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) + response = self.client.put(self.url % self.nsInstanceId, data=self.req_data, format='json') + self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.patch(self.url % self.nsInstanceId, data=self.req_data, format='json') + self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.delete(self.url % self.nsInstanceId, data=self.req_data, format='json') + self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + response = self.client.get(self.url % self.nsInstanceId, data=self.req_data, format='json') + self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code) + + @mock.patch.object(restcall, 'call_req') + @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=json.dumps({"model": json.dumps(VCPE_NS_MODEL_DICT)}))) + @mock.patch.object(BuildInWorkflowThread, 'run') + def test_ns_instantiate_vcpe(self, mock_run, mock_call_req): + config.WORKFLOW_OPTION = "buildin" + mock_call_req.side_effect = [ + [0, self.vnfms, '200'], + [0, self.vnfm, '200'] + ] + response = self.client.post(self.url % self.nsInstanceId, data=SOL_INSTANTIATE_NS_VCPE_DICT, format='json') + self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code) + self.assertIsNotNone(response['Location']) + response = self.client.get(response['Location'], format='json') + self.assertEqual(response.status_code, status.HTTP_200_OK)