genericparser/packages/tests/test_ns_descriptor.py

   1 # Copyright 2018 ZTE Corporation.
   2 #
   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
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import copy
  16 import json
  17 import mock
  18 import os
  19
  20 from django.test import TestCase
  21 from rest_framework import status
  22 from rest_framework.test import APIClient
  23 from genericparser.packages.biz.ns_descriptor import NsDescriptor
  24 from genericparser.packages.const import PKG_STATUS
  25 from genericparser.packages.tests.const import nsd_data
  26 from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH
  27 from genericparser.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
  28 from genericparser.pub.utils import toscaparsers
  29
  30
  31 class TestNsDescriptor(TestCase):
  32     def setUp(self):
  33         self.client = APIClient()
  34         self.user_defined_data = {
  35             'key1': 'value1',
  36             'key2': 'value2',
  37             'key3': 'value3',
  38         }
  39         self.expected_nsd_info = {
  40             'id': None,
  41             'nsdId': None,
  42             'nsdName': None,
  43             'nsdVersion': None,
  44             'nsdDesigner': None,
  45             'nsdInvariantId': None,
  46             'vnfPkgIds': [],
  47             'pnfdInfoIds': [],
  48             'nestedNsdInfoIds': [],
  49             'nsdOnboardingState': 'CREATED',
  50             'onboardingFailureDetails': None,
  51             'nsdOperationalState': 'DISABLED',
  52             'nsdUsageState': 'NOT_IN_USE',
  53             'userDefinedData': self.user_defined_data,
  54             '_links': None
  55         }
  56         self.nsdModel = {
  57             "pnfs": [{"properties": {"id": "m6000_s"}}],
  58             "vnfs": [{"properties": {"id": "123"}}]
  59         }
  60
  61     def tearDown(self):
  62         pass
  63
  64     def test_nsd_create_normal(self):
  65         reqest_data = {'userDefinedData': self.user_defined_data}
  66         expected_reponse_data = {
  67             'nsdOnboardingState': 'CREATED',
  68             'nsdOperationalState': 'DISABLED',
  69             'nsdUsageState': 'NOT_IN_USE',
  70             'userDefinedData': self.user_defined_data,
  71             '_links': None
  72         }
  73
  74         response = self.client.post(
  75             '/api/nsd/v1/ns_descriptors',
  76             data=reqest_data,
  77             format='json'
  78         )
  79         response.data.pop('id')
  80         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
  81         self.assertEqual(expected_reponse_data, response.data)
  82
  83     def test_query_multiple_nsds_normal(self):
  84         expected_reponse_data = [
  85             copy.deepcopy(self.expected_nsd_info),
  86             copy.deepcopy(self.expected_nsd_info)
  87         ]
  88         expected_reponse_data[0]['id'] = '0'
  89         expected_reponse_data[0]['nsdId'] = '0'
  90         expected_reponse_data[1]['id'] = '1'
  91         expected_reponse_data[1]['nsdId'] = '1'
  92
  93         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
  94         for i in range(2):
  95             NSPackageModel(
  96                 nsPackageId=str(i),
  97                 onboardingState='CREATED',
  98                 operationalState='DISABLED',
  99                 usageState='NOT_IN_USE',
 100                 userDefinedData=user_defined_data,
 101                 nsdId=str(i)
 102             ).save()
 103
 104         response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
 105         self.assertEqual(response.status_code, status.HTTP_200_OK)
 106         self.assertEqual(expected_reponse_data, response.data)
 107
 108         expected_reponse_data = [
 109             copy.deepcopy(self.expected_nsd_info)
 110         ]
 111         expected_reponse_data[0]['id'] = '1'
 112         expected_reponse_data[0]['nsdId'] = '1'
 113         response = self.client.get('/api/nsd/v1/ns_descriptors?nsdId=1', format='json')
 114         self.assertEqual(response.status_code, status.HTTP_200_OK)
 115         self.assertEqual(expected_reponse_data, response.data)
 116
 117     def test_query_single_nsd_normal(self):
 118         expected_reponse_data = copy.deepcopy(self.expected_nsd_info)
 119         expected_reponse_data['id'] = '22'
 120
 121         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 122         NSPackageModel(
 123             nsPackageId='22',
 124             onboardingState='CREATED',
 125             operationalState='DISABLED',
 126             usageState='NOT_IN_USE',
 127             userDefinedData=user_defined_data,
 128             nsdModel=json.JSONEncoder().encode(self.nsdModel)
 129         ).save()
 130
 131         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 132         self.assertEqual(response.status_code, status.HTTP_200_OK)
 133         self.assertEqual(expected_reponse_data, response.data)
 134
 135     def test_query_single_when_ns_not_exist(self):
 136         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 137         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 138
 139     def test_delete_single_nsd_normal(self):
 140         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 141         NSPackageModel(
 142             nsPackageId='21',
 143             operationalState='DISABLED',
 144             usageState='NOT_IN_USE',
 145             userDefinedData=user_defined_data,
 146             nsdModel='test'
 147         ).save()
 148
 149         response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 150         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 151         self.assertEqual(None, response.data)
 152
 153     def test_delete_when_ns_not_exist(self):
 154         response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 155         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 156
 157     @mock.patch.object(toscaparsers, 'parse_nsd')
 158     def test_nsd_content_upload_normal(self, mock_parse_nsd):
 159         user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
 160         mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
 161         VnfPackageModel(
 162             vnfPackageId="111",
 163             vnfdId="vcpe_vfw_zte_1_0"
 164         ).save()
 165
 166         PnfPackageModel(
 167             pnfPackageId="112",
 168             pnfdId="m6000_s"
 169         ).save()
 170
 171         NSPackageModel(
 172             nsPackageId='22',
 173             operationalState='DISABLED',
 174             usageState='NOT_IN_USE',
 175             userDefinedData=user_defined_data_json,
 176         ).save()
 177
 178         with open('nsd_content.txt', 'wb') as fp:
 179             fp.write('test')
 180         with open('nsd_content.txt', 'rb') as fp:
 181             resp = self.client.put(
 182                 "/api/nsd/v1/ns_descriptors/22/nsd_content",
 183                 {'file': fp},
 184             )
 185         file_content = ''
 186         with open(os.path.join(GENERICPARSER_ROOT_PATH, '22/nsd_content.txt')) as fp:
 187             data = fp.read()
 188             file_content = '%s%s' % (file_content, data)
 189         ns_pkg = NSPackageModel.objects.filter(nsPackageId="22")
 190         self.assertEqual("VCPE_NS", ns_pkg[0].nsdId)
 191         self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
 192         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 193         self.assertEqual(None, resp.data)
 194         self.assertEqual(file_content, 'test')
 195         os.remove('nsd_content.txt')
 196
 197     def test_nsd_content_upload_failure(self):
 198         with open('nsd_content.txt', 'wb') as fp:
 199             fp.write('test')
 200         with open('nsd_content.txt', 'rb') as fp:
 201             response = self.client.put(
 202                 "/api/nsd/v1/ns_descriptors/22/nsd_content",
 203                 {'file': fp},
 204             )
 205         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 206
 207     def test_nsd_content_download_normal(self):
 208         with open('nsd_content.txt', 'wb') as fp:
 209             fp.writelines('test1')
 210             fp.writelines('test2')
 211         NSPackageModel.objects.create(
 212             nsPackageId='23',
 213             onboardingState='ONBOARDED',
 214             localFilePath='nsd_content.txt'
 215         )
 216         response = self.client.get(
 217             "/api/nsd/v1/ns_descriptors/23/nsd_content", format='json'
 218         )
 219         file_content = ""
 220         for data in response.streaming_content:
 221             file_content = '%s%s' % (file_content, data)
 222         self.assertEqual(response.status_code, status.HTTP_200_OK)
 223         self.assertEqual('test1test2', file_content)
 224         os.remove('nsd_content.txt')
 225
 226     def test_nsd_content_download_when_ns_not_exist(self):
 227         response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
 228         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 229
 230     def test_nsd_content_download_failed(self):
 231         NSPackageModel.objects.create(
 232             nsPackageId='23',
 233             onboardingState='CREATED',
 234             localFilePath='nsd_content.txt'
 235         )
 236         response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
 237         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 238
 239     def test_nsd_content_partial_download_normal(self):
 240         with open('nsd_content.txt', 'wb') as fp:
 241             fp.writelines('test1')
 242             fp.writelines('test2')
 243         NSPackageModel(
 244             nsPackageId='23',
 245             onboardingState='ONBOARDED',
 246             localFilePath='nsd_content.txt'
 247         ).save()
 248
 249         response = self.client.get(
 250             "/api/nsd/v1/ns_descriptors/23/nsd_content",
 251             HTTP_RANGE='5-10',
 252             format='json'
 253         )
 254         partial_file_content = ''
 255         for data in response.streaming_content:
 256             partial_file_content = '%s%s' % (partial_file_content, data)
 257         self.assertEqual(response.status_code, status.HTTP_200_OK)
 258         self.assertEqual('test2', partial_file_content)
 259         os.remove('nsd_content.txt')
 260
 261     @mock.patch.object(NsDescriptor, 'create')
 262     def test_nsd_create_when_catch_exception(self, mock_create):
 263         reqest_data = {'userDefinedData': self.user_defined_data}
 264         mock_create.side_effect = TypeError("integer type")
 265         response = self.client.post('/api/nsd/v1/ns_descriptors', data=reqest_data, format='json')
 266         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 267
 268     @mock.patch.object(NsDescriptor, 'query_single')
 269     def test_query_single_when_catch_exception(self, mock_query_single):
 270         mock_query_single.side_effect = TypeError("integer type")
 271         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 272         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 273
 274     @mock.patch.object(NsDescriptor, 'query_multiple')
 275     def test_query_multiple_when_catch_exception(self, mock_query_multipe):
 276         mock_query_multipe.side_effect = TypeError("integer type")
 277         response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
 278         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 279
 280     @mock.patch.object(NsDescriptor, 'delete_single')
 281     def test_delete_when_catch_exception(self, mock_delete_single):
 282         mock_delete_single.side_effect = TypeError("integer type")
 283         response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 284         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 285
 286     @mock.patch.object(NsDescriptor, 'upload')
 287     def test_upload_when_catch_exception(self, mock_upload):
 288         mock_upload.side_effect = TypeError("integer type")
 289         with open('nsd_content.txt', 'wb') as fp:
 290             fp.write('test')
 291         with open('nsd_content.txt', 'rb') as fp:
 292             response = self.client.put("/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp})
 293         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 294         os.remove('nsd_content.txt')
 295
 296     @mock.patch.object(NsDescriptor, 'download')
 297     def test_download_when_catch_exception(self, mock_download):
 298         mock_download.side_effect = TypeError("integer type")
 299         response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
 300         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)