catalog-be/src/main/resources/scripts/sdcBePy/tosca/models/model_import_manager.py

   1 # ============LICENSE_START=======================================================
   2 #  Copyright (C) 2021 Nordix Foundation
   3 #  ===============================================================================
   4 #  Licensed under the Apache License, Version 2.0 (the "License");
   5 #  you may not use this file except in compliance with the License.
   6 #  You may obtain a copy of the License at
   7 #
   8 #        http://www.apache.org/licenses/LICENSE-2.0
   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 #  SPDX-License-Identifier: Apache-2.0
  16 #  ============LICENSE_END========================================================
  17
  18 import json
  19 import os
  20 import zipfile
  21 from sdcBePy.tosca.models.normativeTypeCandidate import NormativeTypeCandidate
  22 from pathlib import Path
  23
  24
  25 class ModelImportManager:
  26     """Manages the model imports directory"""
  27
  28     INIT_FOLDER_NAME = 'init'
  29     UPGRADE_FOLDER_NAME = 'upgrade'
  30     IMPORTS_FOLDER_NAME = 'imports'
  31     ACTION_UPGRADE = 'upgrade'
  32     ACTION_INIT = 'init'
  33     NODE_TYPES = 'node-types/'
  34
  35     def __init__(self, model_imports_path, model_client):
  36         self.__model_base_path = model_imports_path
  37         self.__model_init_path = self.__model_base_path / self.INIT_FOLDER_NAME
  38         self.__model_upgrade_path = self.__model_base_path / self.UPGRADE_FOLDER_NAME
  39         self.__model_client = model_client
  40
  41     def create_models(self):
  42         for model_folder_name in self.__get_model_init_list():
  43             model_imports_zip_path = self.__zip_model_imports(model_folder_name, self.ACTION_INIT)
  44             model_payload_dict = self.__read_model_payload(model_folder_name, self.ACTION_INIT)
  45             self.__model_client.create_model(model_payload_dict, model_imports_zip_path)
  46             tosca_path = self.__get_model_tosca_path(self.ACTION_INIT, model_folder_name)
  47             self.__model_client.import_model_elements(model_payload_dict, tosca_path)
  48             if os.path.isdir(tosca_path + self.NODE_TYPES):
  49                 self.__model_client.import_model_types(model_payload_dict, self.__get_model_normative_type_candidate(tosca_path), False)
  50             self.__model_client.import_model_elements(model_payload_dict, tosca_path, True)
  51
  52     def update_models(self):
  53         for model_folder_name in self.__get_model_upgrade_list():
  54             model_imports_zip_path = self.__zip_model_imports(model_folder_name, self.ACTION_UPGRADE)
  55             model_payload_dict = self.__read_model_payload(model_folder_name, self.ACTION_UPGRADE)
  56             self.__model_client.update_model_imports(model_payload_dict, model_imports_zip_path)
  57             tosca_path = self.__get_model_tosca_path(self.ACTION_UPGRADE, model_folder_name)
  58             if os.path.isdir(tosca_path + self.NODE_TYPES):
  59                 self.__model_client.import_model_types(model_payload_dict, self.__get_model_normative_type_candidate(tosca_path), True)
  60
  61     def __get_model_init_list(self):
  62         return self.__get_model_list(self.__model_init_path)
  63
  64     def __get_model_upgrade_list(self):
  65         return self.__get_model_list(self.__model_upgrade_path)
  66
  67     @staticmethod
  68     def __get_model_list(path):
  69         model_list = []
  70         for (dirpath, dirnames, filenames) in os.walk(path):
  71             model_list.extend(dirnames)
  72             break
  73         return model_list
  74
  75     def __get_model_normative_type_candidate(self, tosca_path):
  76         path = tosca_path + self.NODE_TYPES
  77         return [NormativeTypeCandidate(path, self.__read_model_type_json(path))]
  78
  79     def __zip_model_imports(self, model, action_type) -> Path:
  80         base_path = self.__get_base_action_path(action_type)
  81         model_path = base_path / model
  82         model_imports_path = base_path / model / self.IMPORTS_FOLDER_NAME
  83         zip_file_path = model_path / "{}.zip".format(model)
  84         zip_file = zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED)
  85         for root, dirs, files in os.walk(model_imports_path):
  86             for file in files:
  87                 zip_file.write(os.path.join(root, file), os.path.relpath(os.path.join(root, file), model_imports_path))
  88         zip_file.close()
  89         return zip_file_path
  90
  91     def __read_model_payload_as_string(self, model, action_type) -> str:
  92         base_path = self.__get_base_action_path(action_type)
  93         model_payload_path = base_path / model / "payload.json"
  94         json_file = open(model_payload_path)
  95         json_data = json.load(json_file, strict=False)
  96         return json.dumps(json_data)
  97
  98     def __read_model_type_json(self, tosca_path):
  99         path = tosca_path + "types.json"
 100         if not os.path.isfile(path):
 101             return []
 102         json_file = open(path)
 103         return json.load(json_file)
 104
 105     def __read_model_payload(self, model, action_type) -> dict:
 106         base_path = self.__get_base_action_path(action_type)
 107         model_payload_path = base_path / model / "payload.json"
 108         json_file = open(model_payload_path)
 109         return json.load(json_file, strict=False)
 110
 111     def __get_base_action_path(self, action_type) -> Path:
 112         return self.__model_init_path if action_type == self.INIT_FOLDER_NAME else self.__model_upgrade_path
 113
 114     def __get_model_tosca_path(self, action, model):
 115         return str(self.__get_base_action_path(action) / model) + "/tosca/"