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 os
  19 import shutil
  20
  21 import mock
  22 from django.test import TestCase
  23 from rest_framework import status
  24 from rest_framework.test import APIClient
  25
  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/parser/v1/parserpnfd", req_data, format='json')
 285         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 286         self.assertEqual({"model": '{"c": "d"}'}, resp.data)