genericparser/packages/tests/test_service_descriptor.py

   1 # Copyright (c) 2019, CMCC Technologies. Co., Ltd.
   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 import json
  15 import logging
  16
  17 from django.test import TestCase
  18 from mock import mock
  19
  20 from genericparser.packages.biz.service_descriptor import ServiceDescriptor
  21 from genericparser.packages.const import PKG_STATUS
  22 from genericparser.pub.database.models import ServicePackageModel, VnfPackageModel, PnfPackageModel
  23 from genericparser.pub.exceptions import PackageNotFoundException
  24 from genericparser.pub.utils import toscaparsers
  25 from .const import sd_data
  26
  27 logger = logging.getLogger(__name__)
  28
  29
  30 class TestServiceDescription(TestCase):
  31
  32     def setUp(self):
  33         self.user_defined_data = {
  34             'key1': 'value1',
  35             'key2': 'value2',
  36             'key3': 'value3',
  37         }
  38         self.data = {
  39             'userDefinedData': self.user_defined_data,
  40         }
  41         self.sd_data = sd_data
  42         ServicePackageModel.objects.filter().delete()
  43
  44     def tearDown(self):
  45         pass
  46
  47     def test_create(self):
  48         result_data = ServiceDescriptor().create(self.data)
  49         self.assertIsNotNone(result_data['id'])
  50         service_package = ServicePackageModel.objects.filter(servicePackageId=result_data['id'])[0]
  51         self.assertIsNotNone(service_package)
  52         self.assertEqual(PKG_STATUS.DISABLED, service_package.operationalState)
  53         self.assertEqual(PKG_STATUS.CREATED, service_package.onboardingState)
  54         self.assertEqual(PKG_STATUS.NOT_IN_USE, service_package.usageState)
  55
  56     def test_create_with_csarid(self):
  57         csar_id = '0b667470-e6b3-4ee8-8f08-186317a04dc2'
  58         result_data = ServiceDescriptor().create(self.data, csar_id)
  59         self.assertEqual(csar_id, result_data['id'])
  60         service_package = ServicePackageModel.objects.filter(servicePackageId=csar_id)[0]
  61         self.assertIsNotNone(service_package)
  62         self.assertEqual(PKG_STATUS.DISABLED, service_package.operationalState)
  63         self.assertEqual(PKG_STATUS.CREATED, service_package.onboardingState)
  64         self.assertEqual(PKG_STATUS.NOT_IN_USE, service_package.usageState)
  65
  66     @mock.patch.object(toscaparsers, 'parse_sd')
  67     def test_parse_serviced_and_save(self, mock_parse_sd):
  68         mock_parse_sd.return_value = json.JSONEncoder().encode(self.sd_data)
  69         servcie_desc = ServiceDescriptor()
  70         csar_id = '0b667470-e6b3-4ee8-8f08-186317a04dc2'
  71         servcie_desc.create(self.data, csar_id)
  72         VnfPackageModel(vnfPackageId="1", vnfdId="cd557883-ac4b-462d-aa01-421b5fa606b1").save()
  73         PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
  74         local_file_name = "/test.csar"
  75         servcie_desc.parse_serviced_and_save(csar_id, local_file_name)
  76
  77         service_package = ServicePackageModel.objects.filter(servicePackageId=csar_id)[0]
  78         self.assertIsNotNone(service_package)
  79
  80     def test_delete_single(self):
  81         servcie_desc = ServiceDescriptor()
  82         csar_id = '0b667470-e6b3-4ee8-8f08-186317a04dc2'
  83         servcie_desc.create(self.data, csar_id)
  84
  85         servcie_desc.delete_single(csar_id)
  86         self.assertTrue(len(ServicePackageModel.objects.filter(servicePackageId=csar_id)) == 0)
  87         self.assertFalse(ServicePackageModel.objects.filter(servicePackageId=csar_id).exists())
  88
  89     def test_delete_single_not_exists(self):
  90         csar_id = "8000"
  91         try:
  92             ServiceDescriptor().delete_single(csar_id)
  93         except Exception as e:
  94             self.assertTrue(isinstance(e, PackageNotFoundException))
  95             self.assertEqual("Service package[8000] not Found.", e.message)