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