conductor/conductor/data/plugins/inventory_provider/utils/csar.py

   1
   2
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    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, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14 import os
  15 import os.path
  16 import requests
  17 from toscaparser.common.exception import ExceptionCollector
  18 from toscaparser.common.exception import ValidationError
  19 from toscaparser.prereq.csar import CSAR
  20 from toscaparser.utils.gettextutils import _
  21 from toscaparser.utils.urlutils import UrlUtils
  22 from toscaparser.utils import yamlparser
  23 import zipfile
  24
  25
  26 try:  # Python 2.x
  27     from BytesIO import BytesIO
  28 except ImportError:  # Python 3.x
  29     from io import BytesIO
  30
  31 TOSCA_META = 'TOSCA-Metadata/TOSCA.meta'
  32 YAML_LOADER = yamlparser.load_yaml
  33
  34
  35 class SDCCSAR(CSAR):
  36     def __init__(self, csar_file, model_name, a_file=True):
  37         super(SDCCSAR, self).__init__(csar_file, a_file=True)
  38         self.model_name = model_name
  39
  40     def validate(self):
  41         """Validate the provided CSAR file."""
  42         self.is_validated = True
  43         # validate that the file or URL exists
  44         missing_err_msg = (_('"%s" does not exist.') % self.path)
  45         if self.a_file:
  46             if not os.path.isfile(self.path):
  47                 ExceptionCollector.appendException(
  48                     ValidationError(message=missing_err_msg))
  49                 return False
  50             else:
  51                 self.csar = self.path
  52         else:  # a URL
  53             if not UrlUtils.validate_url(self.path):
  54                 ExceptionCollector.appendException(
  55                     ValidationError(message=missing_err_msg))
  56                 return False
  57             else:
  58                 response = requests.get(self.path)
  59                 self.csar = BytesIO(response.content)
  60
  61         # validate that it is a valid zip file
  62         if not zipfile.is_zipfile(self.csar):
  63             err_msg = (_('"%s" is not a valid zip file.') % self.path)
  64             ExceptionCollector.appendException(
  65                 ValidationError(message=err_msg))
  66             return False
  67
  68         # validate that it contains the metadata file in the correct location
  69         self.zfile = zipfile.ZipFile(self.csar, 'r')
  70         filelist = self.zfile.namelist()
  71         if TOSCA_META in filelist:
  72             self.is_tosca_metadata = True
  73             # validate that 'Entry-Definitions' property exists in TOSCA.meta
  74             is_validated = self._validate_tosca_meta(filelist)
  75         else:
  76             self.is_tosca_metadata = False
  77             is_validated = self._validate_root_level_yaml(filelist)
  78
  79         if is_validated:
  80             main_tpl = self._read_template_yaml(self.main_template_file_name)
  81             nst_properies_res = self.get_nst_properties(main_tpl)
  82             print("nst properties", nst_properies_res)
  83         return nst_properies_res
  84
  85     def get_nst_properties(self, main_tpl):
  86         importsarr = main_tpl.get('imports')
  87         for imports in importsarr:
  88             for key in imports:
  89                 if "service-{}-interface".format(self.model_name) in key:
  90                     val = imports[key]
  91         filename = val.get("file")
  92         datanew = self._read_template_yaml("Definitions/" + filename)
  93         node_types = datanew.get("node_types")
  94         for key in list(node_types):
  95             if "org.openecomp" in key:
  96                 nodedata = node_types[key]
  97         nst_properties = nodedata.get("properties")
  98         return nst_properties