catalog/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
  16 import json
  17 import os
  18
  19 from django.test import TestCase
  20 from rest_framework import status
  21 from rest_framework.test import APIClient
  22 from catalog.pub.database.models import NSPackageModel
  23 from catalog.pub.config.config import CATALOG_ROOT_PATH
  24
  25
  26 class TestNsDescriptor(TestCase):
  27     def setUp(self):
  28         self.client = APIClient()
  29         self.user_defined_data = {
  30             'key1': 'value1',
  31             'key2': 'value2',
  32             'key3': 'value3',
  33         }
  34
  35     def tearDown(self):
  36         pass
  37
  38     def test_nsd_create_normal(self):
  39         reqest_data = {
  40             'userDefinedData': self.user_defined_data
  41         }
  42         expected_reponse_data = {
  43             'nsdOnboardingState': 'CREATED',
  44             'nsdOperationalState': 'DISABLED',
  45             'nsdUsageState': 'NOT_IN_USE',
  46             'userDefinedData': self.user_defined_data,
  47             '_links': None
  48         }
  49         response = self.client.post(
  50             '/api/nsd/v1/ns_descriptors',
  51             data=reqest_data,
  52             format='json'
  53         )
  54         response.data.pop('id')
  55         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
  56         self.assertEqual(expected_reponse_data, response.data)
  57
  58     def test_query_multiple_nsds_normal(self):
  59         expected_reponse_data = [
  60             {
  61                 'id': '0',
  62                 'nsdId': None,
  63                 'nsdName': None,
  64                 'nsdVersion': None,
  65                 'nsdDesigner': None,
  66                 'nsdInvariantId': None,
  67                 'vnfPkgIds': [],
  68                 'pnfdInfoIds': [],
  69                 'nestedNsdInfoIds': [],
  70                 'nsdOnboardingState': 'CREATED',
  71                 'onboardingFailureDetails': None,
  72                 'nsdOperationalState': 'DISABLED',
  73                 'nsdUsageState': 'NOT_IN_USE',
  74                 'userDefinedData': self.user_defined_data,
  75                 '_links': None
  76             },
  77             {
  78                 'id': '1',
  79                 'nsdId': None,
  80                 'nsdName': None,
  81                 'nsdVersion': None,
  82                 'nsdDesigner': None,
  83                 'nsdInvariantId': None,
  84                 'vnfPkgIds': [],
  85                 'pnfdInfoIds': [],
  86                 'nestedNsdInfoIds': [],
  87                 'nsdOnboardingState': 'CREATED',
  88                 'onboardingFailureDetails': None,
  89                 'nsdOperationalState': 'DISABLED',
  90                 'nsdUsageState': 'NOT_IN_USE',
  91                 'userDefinedData': self.user_defined_data,
  92                 '_links': None
  93             }
  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             ).save()
 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     def test_query_single_nsd_normal(self):
 109         expected_reponse_data = {
 110             'id': '22',
 111             'nsdId': None,
 112             'nsdName': None,
 113             'nsdVersion': None,
 114             'nsdDesigner': None,
 115             'nsdInvariantId': None,
 116             'vnfPkgIds': [],
 117             'pnfdInfoIds': [],
 118             'nestedNsdInfoIds': [],
 119             'nsdOnboardingState': 'CREATED',
 120             'onboardingFailureDetails': None,
 121             'nsdOperationalState': 'DISABLED',
 122             'nsdUsageState': 'NOT_IN_USE',
 123             'userDefinedData': self.user_defined_data,
 124             '_links': None
 125         }
 126         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 127         NSPackageModel(
 128             nsPackageId='22',
 129             onboardingState='CREATED',
 130             operationalState='DISABLED',
 131             usageState='NOT_IN_USE',
 132             userDefinedData=user_defined_data
 133         ).save()
 134
 135         response = self.client.get('/api/nsd/v1/ns_descriptors/22', format='json')
 136         self.assertEqual(response.status_code, status.HTTP_200_OK)
 137         self.assertEqual(expected_reponse_data, response.data)
 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         resp = self.client.delete("/api/nsd/v1/ns_descriptors/21", format='json')
 149         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 150         self.assertEqual({}, resp.data)
 151
 152     def test_nsd_content_upload_normal(self):
 153         user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
 154         NSPackageModel(
 155             nsPackageId='22',
 156             operationalState='DISABLED',
 157             usageState='NOT_IN_USE',
 158             userDefinedData=user_defined_data_json,
 159         ).save()
 160         with open('nsd_content.txt', 'wb') as fp:
 161             fp.write('test')
 162         with open('nsd_content.txt', 'rb') as fp:
 163             resp = self.client.put(
 164                 "/api/nsd/v1/ns_descriptors/22/nsd_content",
 165                 {'file': fp},
 166             )
 167         file_content = ''
 168         with open(os.path.join(CATALOG_ROOT_PATH, '22/nsd_content.txt')) as fp:
 169             data = fp.read()
 170             file_content = '%s%s' % (file_content, data)
 171         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 172         self.assertEqual({}, resp.data)
 173         self.assertEqual(file_content, 'test')
 174         os.remove('nsd_content.txt')
 175
 176     def test_nsd_content_upload_failure(self):
 177         pass
 178
 179     def test_nsd_content_download_normal(self):
 180         pass
 181
 182     def test_nsd_content_partial_download_normal(self):
 183         with open('nsd_content.txt', 'wb') as fp:
 184             fp.writelines('test1')
 185             fp.writelines('test2')
 186         NSPackageModel(
 187             nsPackageId='23',
 188             onboardingState='ONBOARDED',
 189             localFilePath='nsd_content.txt'
 190         ).save()
 191         response = self.client.get(
 192             "/api/nsd/v1/ns_descriptors/23/nsd_content",
 193             RANGE='5-10',
 194             format='json'
 195         )
 196         partial_file_content = ''
 197         for data in response.streaming_content:
 198             partial_file_content = '%s%s' % (partial_file_content, data)
 199         self.assertEqual(response.status_code, status.HTTP_200_OK)
 200         self.assertEqual('test2', partial_file_content)
 201         os.remove('nsd_content.txt')