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