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