genericparser/packages/tests/test_pnf_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 copy
  17 import json
  18 import mock
  19 import os
  20 import shutil
  21
  22 from django.test import TestCase
  23 from rest_framework import status
  24 from rest_framework.test import APIClient
  25 from genericparser.packages.biz.pnf_descriptor import PnfDescriptor
  26 from genericparser.packages.const import PKG_STATUS
  27 from genericparser.packages.tests.const import pnfd_data
  28 from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH
  29 from genericparser.pub.database.models import PnfPackageModel, NSPackageModel
  30 from genericparser.pub.utils import toscaparsers
  31
  32
  33 class TestPnfDescriptor(TestCase):
  34     def setUp(self):
  35         self.client = APIClient()
  36         self.user_defined_data = {
  37             'key1': 'value1',
  38             'key2': 'value2',
  39             'key3': 'value3',
  40         }
  41         self.expected_pnfd_info = {
  42             'id': None,
  43             'pnfdId': None,
  44             'pnfdName': None,
  45             'pnfdVersion': None,
  46             'pnfdProvider': None,
  47             'pnfdInvariantId': None,
  48             'pnfdOnboardingState': 'CREATED',
  49             'onboardingFailureDetails': None,
  50             'pnfdUsageState': 'NOT_IN_USE',
  51             'userDefinedData': self.user_defined_data,
  52             '_links': None
  53         }
  54         self.nsdModel = {
  55             "pnfs": [{"properties": {"id": "m6000_s"}}]
  56         }
  57
  58     def tearDown(self):
  59         file_path = os.path.join(GENERICPARSER_ROOT_PATH, "22")
  60         if os.path.exists(file_path):
  61             shutil.rmtree(file_path)
  62
  63     def test_pnfd_create_normal(self):
  64         request_data = {'userDefinedData': self.user_defined_data}
  65         expected_reponse_data = {
  66             'pnfdOnboardingState': 'CREATED',
  67             'pnfdUsageState': 'NOT_IN_USE',
  68             'userDefinedData': self.user_defined_data,
  69             '_links': None
  70         }
  71
  72         response = self.client.post(
  73             '/api/nsd/v1/pnf_descriptors',
  74             data=request_data,
  75             format='json'
  76         )
  77         response.data.pop('id')
  78         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
  79         # self.assertEqual(expected_reponse_data, response.data)
  80         for key, value in expected_reponse_data.items():
  81             self.assertEqual(response.data[key], value)
  82
  83     def test_query_multiple_pnfds_normal(self):
  84         expected_reponse_data = [
  85             copy.deepcopy(self.expected_pnfd_info),
  86             copy.deepcopy(self.expected_pnfd_info)
  87         ]
  88         expected_reponse_data[0]['id'] = '0'
  89         expected_reponse_data[1]['id'] = '1'
  90
  91         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
  92         for i in range(2):
  93             PnfPackageModel(
  94                 pnfPackageId=str(i),
  95                 onboardingState='CREATED',
  96                 usageState='NOT_IN_USE',
  97                 userDefinedData=user_defined_data
  98             ).save()
  99         response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
 100         self.assertEqual(response.status_code, status.HTTP_200_OK)
 101         self.assertEqual(expected_reponse_data, response.data)
 102
 103     def test_query_single_pnfd_normal(self):
 104         expected_reponse_data = copy.deepcopy(self.expected_pnfd_info)
 105         expected_reponse_data['id'] = '22'
 106
 107         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 108         PnfPackageModel(
 109             pnfPackageId='22',
 110             onboardingState='CREATED',
 111             usageState='NOT_IN_USE',
 112             userDefinedData=user_defined_data
 113         ).save()
 114
 115         response = self.client.get('/api/nsd/v1/pnf_descriptors/22', 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_pnfd_failed(self):
 120         response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
 121         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 122
 123     def test_delete_single_pnfd_normal(self):
 124         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 125         PnfPackageModel(
 126             pnfPackageId='22',
 127             usageState=PKG_STATUS.NOT_IN_USE,
 128             userDefinedData=user_defined_data,
 129             pnfdModel='test'
 130         ).save()
 131         NSPackageModel.objects.create(
 132             nsPackageId="111",
 133             nsdModel=json.JSONEncoder().encode(self.nsdModel)
 134         )
 135         resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
 136         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 137         self.assertEqual(None, resp.data)
 138
 139     def test_delete_single_pnfd_when_not_exist(self):
 140         resp = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
 141         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 142         self.assertEqual(None, resp.data)
 143
 144     @mock.patch.object(toscaparsers, "parse_pnfd")
 145     def test_pnfd_content_upload_normal(self, mock_parse_pnfd):
 146         user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
 147         PnfPackageModel(
 148             pnfPackageId='22',
 149             usageState=PKG_STATUS.NOT_IN_USE,
 150             userDefinedData=user_defined_data_json,
 151         ).save()
 152         mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
 153         with open('pnfd_content.txt', 'wb') as fp:
 154             fp.write('test')
 155
 156         with open('pnfd_content.txt', 'rb') as fp:
 157             resp = self.client.put(
 158                 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
 159                 {'file': fp},
 160             )
 161         pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId="22")
 162         self.assertEqual(pnf_pkg[0].pnfdId, "zte-1.0")
 163         self.assertEqual(pnf_pkg[0].onboardingState, PKG_STATUS.ONBOARDED)
 164         self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
 165         self.assertEqual(None, resp.data)
 166         os.remove('pnfd_content.txt')
 167
 168     def test_pnfd_content_upload_when_pnf_not_exist(self):
 169         with open('pnfd_content.txt', 'wb') as fp:
 170             fp.write('test')
 171
 172         with open('pnfd_content.txt', 'rb') as fp:
 173             resp = self.client.put(
 174                 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
 175                 {'file': fp},
 176             )
 177         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 178
 179     @mock.patch.object(toscaparsers, "parse_pnfd")
 180     def test_pnfd_content_upload_when_pnfd_exist(self, mock_parse_pnfd):
 181         with open('pnfd_content.txt', 'wb') as fp:
 182             fp.write('test')
 183         PnfPackageModel(
 184             pnfPackageId='22',
 185             usageState=PKG_STATUS.NOT_IN_USE,
 186             pnfdId="zte-1.1"
 187         ).save()
 188         PnfPackageModel(
 189             pnfPackageId='23',
 190             usageState=PKG_STATUS.NOT_IN_USE,
 191             pnfdId="zte-1.0"
 192         ).save()
 193         mock_parse_pnfd.return_value = json.JSONEncoder().encode(pnfd_data)
 194         with open('pnfd_content.txt', 'rb') as fp:
 195             resp = self.client.put(
 196                 "/api/nsd/v1/pnf_descriptors/22/pnfd_content",
 197                 {'file': fp},
 198             )
 199         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 200
 201     def test_pnfd_download_normal(self):
 202         with open('pnfd_content.txt', 'wb') as fp:
 203             fp.writelines('test1')
 204             fp.writelines('test2')
 205         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 206         PnfPackageModel(
 207             pnfPackageId='22',
 208             usageState=PKG_STATUS.NOT_IN_USE,
 209             onboardingState=PKG_STATUS.ONBOARDED,
 210             userDefinedData=user_defined_data,
 211             localFilePath="pnfd_content.txt",
 212             pnfdModel='test'
 213         ).save()
 214         resp = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
 215         file_content = ""
 216         for data in resp.streaming_content:
 217             file_content = '%s%s' % (file_content, data)
 218         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 219         self.assertEqual('test1test2', file_content)
 220         os.remove('pnfd_content.txt')
 221
 222     def test_pnfd_download_failed(self):
 223         response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
 224         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 225
 226     def test_pnfd_download_when_not_on_boarded(self):
 227         with open('pnfd_content.txt', 'wb') as fp:
 228             fp.writelines('test1')
 229             fp.writelines('test2')
 230         user_defined_data = json.JSONEncoder().encode(self.user_defined_data)
 231         PnfPackageModel(
 232             pnfPackageId='22',
 233             usageState=PKG_STATUS.NOT_IN_USE,
 234             onboardingState=PKG_STATUS.CREATED,
 235             userDefinedData=user_defined_data,
 236             localFilePath="pnfd_content.txt",
 237             pnfdModel='test'
 238         ).save()
 239         response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
 240         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 241         os.remove('pnfd_content.txt')
 242
 243     @mock.patch.object(PnfDescriptor, "create")
 244     def test_pnfd_create_when_catch_exception(self, mock_create):
 245         request_data = {'userDefinedData': self.user_defined_data}
 246         mock_create.side_effect = TypeError('integer type')
 247         response = self.client.post('/api/nsd/v1/pnf_descriptors', data=request_data, format='json')
 248         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 249
 250     @mock.patch.object(PnfDescriptor, "delete_single")
 251     def test_delete_single_when_catch_exception(self, mock_delete_single):
 252         mock_delete_single.side_effect = TypeError("integer type")
 253         response = self.client.delete("/api/nsd/v1/pnf_descriptors/22", format='json')
 254         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 255
 256     @mock.patch.object(PnfDescriptor, "query_single")
 257     def test_query_single_when_catch_exception(self, mock_query_single):
 258         mock_query_single.side_effect = TypeError("integer type")
 259         response = self.client.get('/api/nsd/v1/pnf_descriptors/22', format='json')
 260         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 261
 262     @mock.patch.object(PnfDescriptor, "query_multiple")
 263     def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 264         mock_query_muitiple.side_effect = TypeError("integer type")
 265         response = self.client.get('/api/nsd/v1/pnf_descriptors', format='json')
 266         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 267
 268     @mock.patch.object(PnfDescriptor, "upload")
 269     def test_upload_when_catch_exception(self, mock_upload):
 270         mock_upload.side_effect = TypeError("integer type")
 271         response = self.client.put("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
 272         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 273
 274     @mock.patch.object(PnfDescriptor, "download")
 275     def test_download_when_catch_exception(self, mock_download):
 276         mock_download.side_effect = TypeError("integer type")
 277         response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
 278         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 279
 280     @mock.patch.object(toscaparsers, 'parse_pnfd')
 281     def test_pnfd_parse_normal(self, mock_parse_pnfd):
 282         PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
 283         mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
 284         req_data = {"csarId": "8", "inputs": []}
 285         resp = self.client.post("/api/parser/v1/parserpnfd", req_data, format='json')
 286         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 287         self.assertEqual({"model": '{"c": "d"}'}, resp.data)