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         for key, value in expected_reponse_data.items():
  83             self.assertEqual(response.data[key], value)
  84
  85     def test_query_multiple_nsds_normal(self):
  86         expected_reponse_data = [
  87             copy.deepcopy(self.expected_nsd_info),
  88             copy.deepcopy(self.expected_nsd_info)
  89         ]
  90         expected_reponse_data[0]['id'] = '0'
  91         expected_reponse_data[0]['nsdId'] = '0'
  92         expected_reponse_data[1]['id'] = '1'
  93         expected_reponse_data[1]['nsdId'] = '1'
  94
  95         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
  96         for i in range(2):
  97             NSPackageModel(
  98                 nsPackageId=str(i),
  99                 onboardingState='CREATED',
 100                 operationalState='DISABLED',
 101                 usageState='NOT_IN_USE',
 102                 userDefinedData=user_defined_data,
 103                 nsdId=str(i)
 104             ).save()
 105
 106         response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
 107         self.assertEqual(response.status_code, status.HTTP_200_OK)
 108         self.assertEqual(expected_reponse_data, response.data)
 109
 110         expected_reponse_data = [
 111             copy.deepcopy(self.expected_nsd_info)
 112         ]
 113         expected_reponse_data[0]['id'] = '1'
 114         expected_reponse_data[0]['nsdId'] = '1'
 115         response = self.client.get('/api/nsd/v1/ns_descriptors?nsdId=1', format='json')
 116         self.assertEqual(response.status_code, status.HTTP_200_OK)
 117         self.assertEqual(expected_reponse_data, response.data)
 118
 119     def test_query_single_nsd_normal(self):
 120         expected_reponse_data = copy.deepcopy(self.expected_nsd_info)
 121         expected_reponse_data['id'] = '22'
 122
 123         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 124         NSPackageModel(
 125             nsPackageId='22',
 126             onboardingState='CREATED',
 127             operationalState='DISABLED',
 128             usageState='NOT_IN_USE',
 129             userDefinedData=user_defined_data,
 130             nsdModel=json.JSONEncoder().encode(self.nsdModel)
 131         ).save()
 132
 133         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 134         self.assertEqual(response.status_code, status.HTTP_200_OK)
 135         self.assertEqual(expected_reponse_data, response.data)
 136
 137     def test_query_single_when_ns_not_exist(self):
 138         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 139         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 140
 141     def test_delete_single_nsd_normal(self):
 142         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 143         NSPackageModel(
 144             nsPackageId='21',
 145             operationalState='DISABLED',
 146             usageState='NOT_IN_USE',
 147             userDefinedData=user_defined_data,
 148             nsdModel='test'
 149         ).save()
 150
 151         response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 152         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 153         self.assertEqual(None, response.data)
 154
 155     def test_delete_when_ns_not_exist(self):
 156         response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 157         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 158
 159     @mock.patch.object(toscaparsers, 'parse_nsd')
 160     def test_nsd_content_upload_normal(self, mock_parse_nsd):
 161         user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
 162         mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
 163         VnfPackageModel(
 164             vnfPackageId="111",
 165             vnfdId="vcpe_vfw_zte_1_0"
 166         ).save()
 167
 168         PnfPackageModel(
 169             pnfPackageId="112",
 170             pnfdId="m6000_s"
 171         ).save()
 172
 173         NSPackageModel(
 174             nsPackageId='22',
 175             operationalState='DISABLED',
 176             usageState='NOT_IN_USE',
 177             userDefinedData=user_defined_data_json,
 178         ).save()
 179
 180         with open('nsd_content.txt', 'wb') as fp:
 181             fp.write('test')
 182         with open('nsd_content.txt', 'rb') as fp:
 183             resp = self.client.put(
 184                 "/api/nsd/v1/ns_descriptors/22/nsd_content",
 185                 {'file': fp},
 186             )
 187         file_content = ''
 188         with open(os.path.join(GENERICPARSER_ROOT_PATH, '22/nsd_content.txt')) as fp:
 189             data = fp.read()
 190             file_content = '%s%s' % (file_content, data)
 191         ns_pkg = NSPackageModel.objects.filter(nsPackageId="22")
 192         self.assertEqual("VCPE_NS", ns_pkg[0].nsdId)
 193         self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
 194         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 195         self.assertEqual(None, resp.data)
 196         self.assertEqual(file_content, 'test')
 197         os.remove('nsd_content.txt')
 198
 199     def test_nsd_content_upload_failure(self):
 200         with open('nsd_content.txt', 'wb') as fp:
 201             fp.write('test')
 202         with open('nsd_content.txt', 'rb') as fp:
 203             response = self.client.put(
 204                 "/api/nsd/v1/ns_descriptors/22/nsd_content",
 205                 {'file': fp},
 206             )
 207         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 208
 209     def test_nsd_content_download_normal(self):
 210         with open('nsd_content.txt', 'wb') as fp:
 211             fp.writelines('test1')
 212             fp.writelines('test2')
 213         NSPackageModel.objects.create(
 214             nsPackageId='23',
 215             onboardingState='ONBOARDED',
 216             localFilePath='nsd_content.txt'
 217         )
 218         response = self.client.get(
 219             "/api/nsd/v1/ns_descriptors/23/nsd_content", format='json'
 220         )
 221         file_content = ""
 222         for data in response.streaming_content:
 223             file_content = '%s%s' % (file_content, data)
 224         self.assertEqual(response.status_code, status.HTTP_200_OK)
 225         self.assertEqual('test1test2', file_content)
 226         os.remove('nsd_content.txt')
 227
 228     def test_nsd_content_download_when_ns_not_exist(self):
 229         response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
 230         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 231
 232     def test_nsd_content_download_failed(self):
 233         NSPackageModel.objects.create(
 234             nsPackageId='23',
 235             onboardingState='CREATED',
 236             localFilePath='nsd_content.txt'
 237         )
 238         response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
 239         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 240
 241     def test_nsd_content_partial_download_normal(self):
 242         with open('nsd_content.txt', 'wb') as fp:
 243             fp.writelines('test1')
 244             fp.writelines('test2')
 245         NSPackageModel(
 246             nsPackageId='23',
 247             onboardingState='ONBOARDED',
 248             localFilePath='nsd_content.txt'
 249         ).save()
 250
 251         response = self.client.get(
 252             "/api/nsd/v1/ns_descriptors/23/nsd_content",
 253             HTTP_RANGE='5-10',
 254             format='json'
 255         )
 256         partial_file_content = ''
 257         for data in response.streaming_content:
 258             partial_file_content = '%s%s' % (partial_file_content, data)
 259         self.assertEqual(response.status_code, status.HTTP_200_OK)
 260         self.assertEqual('test2', partial_file_content)
 261         os.remove('nsd_content.txt')
 262
 263     @mock.patch.object(NsDescriptor, 'create')
 264     def test_nsd_create_when_catch_exception(self, mock_create):
 265         reqest_data = {'userDefinedData': self.user_defined_data}
 266         mock_create.side_effect = TypeError("integer type")
 267         response = self.client.post('/api/nsd/v1/ns_descriptors', data=reqest_data, format='json')
 268         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 269
 270     @mock.patch.object(NsDescriptor, 'query_single')
 271     def test_query_single_when_catch_exception(self, mock_query_single):
 272         mock_query_single.side_effect = TypeError("integer type")
 273         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 274         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 275
 276     @mock.patch.object(NsDescriptor, 'query_multiple')
 277     def test_query_multiple_when_catch_exception(self, mock_query_multipe):
 278         mock_query_multipe.side_effect = TypeError("integer type")
 279         response = self.client.get('/api/nsd/v1/ns_descriptors', format='json')
 280         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 281
 282     @mock.patch.object(NsDescriptor, 'delete_single')
 283     def test_delete_when_catch_exception(self, mock_delete_single):
 284         mock_delete_single.side_effect = TypeError("integer type")
 285         response = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 286         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 287
 288     @mock.patch.object(NsDescriptor, 'upload')
 289     def test_upload_when_catch_exception(self, mock_upload):
 290         mock_upload.side_effect = TypeError("integer type")
 291         with open('nsd_content.txt', 'wb') as fp:
 292             fp.write('test')
 293         with open('nsd_content.txt', 'rb') as fp:
 294             response = self.client.put("/api/nsd/v1/ns_descriptors/22/nsd_content", {'file': fp})
 295         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 296         os.remove('nsd_content.txt')
 297
 298     @mock.patch.object(NsDescriptor, 'download')
 299     def test_download_when_catch_exception(self, mock_download):
 300         mock_download.side_effect = TypeError("integer type")
 301         response = self.client.get("/api/nsd/v1/ns_descriptors/23/nsd_content", format='json')
 302         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)