# Copyright 2018 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 copy import json import mock import os from django.test import TestCase from rest_framework import status from rest_framework.test import APIClient from genericparser.packages.biz.ns_descriptor import NsDescriptor from genericparser.packages.const import PKG_STATUS from genericparser.packages.tests.const import nsd_data from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH from genericparser.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel from genericparser.pub.utils import toscaparsers class TestNsDescriptor(TestCase): def setUp(self): self.client = APIClient() self.user_defined_data = { 'key1': 'value1', 'key2': 'value2', 'key3': 'value3', } self.expected_nsd_info = { 'id': None, 'nsdId': None, 'nsdName': None, 'nsdVersion': None, 'nsdDesigner': None, 'nsdInvariantId': None, 'vnfPkgIds': [], 'pnfdInfoIds': [], 'nestedNsdInfoIds': [], 'nsdOnboardingState': 'CREATED', 'onboardingFailureDetails': None, 'nsdOperationalState': 'DISABLED', 'nsdUsageState': 'NOT_IN_USE', 'userDefinedData': self.user_defined_data, '_links': None } self.nsdModel = { "pnfs": [{"properties": {"id": "m6000_s"}}], "vnfs": [{"properties": {"id": "123"}}] } def tearDown(self): pass def test_nsd_create_normal(self): reqest_data = {'userDefinedData': self.user_defined_data} expected_reponse_data = { 'nsdOnboardingState': 'CREATED', 'nsdOperationalState': 'DISABLED', 'nsdUsageState': 'NOT_IN_USE', 'userDefinedData': self.user_defined_data, '_links': None } response = self.client.post( '/api/nsd/v1/ns_descriptors', data=reqest_data, format='json' ) response.data.pop('id') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(expected_reponse_data, response.data) def test_query_multiple_nsds_normal(self): expected_reponse_data = [ copy.deepcopy(self.expected_nsd_info), copy.deepcopy(self.expected_nsd_info) ] expected_reponse_data[0]['id'] = '0' expected_reponse_data[0]['nsdId'] = '0' expected_reponse_data[1]['id'] = '1' expected_reponse_data[1]['nsdId'] = '1' user_defined_data = json.JSONEncoder().encode(self.user_defined_data) for i in range(2): NSPackageModel( nsPackageId=str(i), onboardingState='CREATED', operationalState='DISABLED', usageState='NOT_IN_USE', userDefinedData=user_defined_data, nsdId=str(i) ).save() response = self.client.get('/api/nsd/v1/ns_descriptors', format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(expected_reponse_data, response.data) expected_reponse_data = [ copy.deepcopy(self.expected_nsd_info) ] expected_reponse_data[0]['id'] = '1' expected_reponse_data[0]['nsdId'] = '1' response = self.client.get('/api/nsd/v1/ns_descriptors?nsdId=1', format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(expected_reponse_data, response.data) def test_query_single_nsd_normal(self): expected_reponse_data = copy.deepcopy(self.expected_nsd_info) expected_reponse_data['id'] = '22' user_defined_data = json.JSONEncoder().encode(self.user_defined_data) NSPackageModel( nsPackageId='22', onboardingState='CREATED', operationalState='DISABLED', usageState='NOT_IN_USE', userDefinedData=user_defined_data, nsdModel=json.JSONEncoder().encode(self.nsdModel) ).save() response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(expected_reponse_data, response.data) def test_query_single_when_ns_not_exist(self): response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json') self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_delete_single_nsd_normal(self): user_defined_data = json.JSONEncoder().encode(self.user_defined_data) NSPackageModel( nsPackageId='21', operationalState='DISABLED', usageState='NOT_IN_USE', userDefinedData=user_defined_data, nsdModel='test' ).save() response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json') self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertEqual(None, response.data) def test_delete_when_ns_not_exist(self): response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json') self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) @mock.patch.object(toscaparsers, 'parse_nsd') def test_nsd_content_upload_normal(self, mock_parse_nsd): user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data) mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data) VnfPackageModel( vnfPackageId="111", vnfdId="vcpe_vfw_zte_1_0" ).save() PnfPackageModel( pnfPackageId="112", pnfdId="m6000_s" ).save() NSPackageModel( nsPackageId='22', operationalState='DISABLED', usageState='NOT_IN_USE', userDefinedData=user_defined_data_json, ).save() with open('nsd_content.txt', 'wb') as fp: fp.write('test') with open('nsd_content.txt', 'rb') as fp: resp = self.client.put( "/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp}, ) file_content = '' with open(os.path.join(GENERICPARSER_ROOT_PATH, '22/nsd_content.txt')) as fp: data = fp.read() file_content = '%s%s' % (file_content, data) ns_pkg = NSPackageModel.objects.filter(nsPackageId="22") self.assertEqual("VCPE_NS", ns_pkg[0].nsdId) self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState) self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) self.assertEqual(None, resp.data) self.assertEqual(file_content, 'test') os.remove('nsd_content.txt') def test_nsd_content_upload_failure(self): with open('nsd_content.txt', 'wb') as fp: fp.write('test') with open('nsd_content.txt', 'rb') as fp: response = self.client.put( "/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp}, ) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) def test_nsd_content_download_normal(self): with open('nsd_content.txt', 'wb') as fp: fp.writelines('test1') fp.writelines('test2') NSPackageModel.objects.create( nsPackageId='23', onboardingState='ONBOARDED', localFilePath='nsd_content.txt' ) response = self.client.get( "/api/nsd/v1/ns_descriptors/23/nsd_content", format='json' ) file_content = "" for data in response.streaming_content: file_content = '%s%s' % (file_content, data) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual('test1test2', file_content) os.remove('nsd_content.txt') def test_nsd_content_download_when_ns_not_exist(self): response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json') self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_nsd_content_download_failed(self): NSPackageModel.objects.create( nsPackageId='23', onboardingState='CREATED', localFilePath='nsd_content.txt' ) response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) def test_nsd_content_partial_download_normal(self): with open('nsd_content.txt', 'wb') as fp: fp.writelines('test1') fp.writelines('test2') NSPackageModel( nsPackageId='23', onboardingState='ONBOARDED', localFilePath='nsd_content.txt' ).save() response = self.client.get( "/api/nsd/v1/ns_descriptors/23/nsd_content", HTTP_RANGE='5-10', format='json' ) partial_file_content = '' for data in response.streaming_content: partial_file_content = '%s%s' % (partial_file_content, data) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual('test2', partial_file_content) os.remove('nsd_content.txt') @mock.patch.object(NsDescriptor, 'create') def test_nsd_create_when_catch_exception(self, mock_create): reqest_data = {'userDefinedData': self.user_defined_data} mock_create.side_effect = TypeError("integer type") response = self.client.post('/api/nsd/v1/ns_descriptors', data=reqest_data, format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) @mock.patch.object(NsDescriptor, 'query_single') def test_query_single_when_catch_exception(self, mock_query_single): mock_query_single.side_effect = TypeError("integer type") response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) @mock.patch.object(NsDescriptor, 'query_multiple') def test_query_multiple_when_catch_exception(self, mock_query_multipe): mock_query_multipe.side_effect = TypeError("integer type") response = self.client.get('/api/nsd/v1/ns_descriptors', format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) @mock.patch.object(NsDescriptor, 'delete_single') def test_delete_when_catch_exception(self, mock_delete_single): mock_delete_single.side_effect = TypeError("integer type") response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) @mock.patch.object(NsDescriptor, 'upload') def test_upload_when_catch_exception(self, mock_upload): mock_upload.side_effect = TypeError("integer type") with open('nsd_content.txt', 'wb') as fp: fp.write('test') with open('nsd_content.txt', 'rb') as fp: response = self.client.put("/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp}) self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) os.remove('nsd_content.txt') @mock.patch.object(NsDescriptor, 'download') def test_download_when_catch_exception(self, mock_download): mock_download.side_effect = TypeError("integer type") response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)