catalog-be/src/main/resources/scripts/sdcBePy/common/normative/toscaElements.py

   1 import json
   2 import zipfile
   3
   4 import pycurl
   5
   6 from sdcBePy.common.logger import debug, log, print_name_and_return_code, print_and_exit
   7 from sdcBePy.common.sdcBeProxy import SdcBeProxy
   8 from sdcBePy.common.errors import ResourceCreationError
   9
  10
  11 def process_and_create_normative_element(normative_element,
  12                                          scheme=None, be_host=None, be_port=None, header=None, admin_user=None,
  13                                          tls_cert=None, tls_key=None, tls_key_pw=None, ca_cert=None, sdc_be_proxy=None,
  14                                          model=None, debug=False,
  15                                          exit_on_success=False):
  16     if sdc_be_proxy is None:
  17         sdc_be_proxy = SdcBeProxy(be_host, be_port, header, scheme, tls_cert, tls_key, tls_key_pw, ca_cert, admin_user, debug=debug)
  18
  19     file_dir, url_suffix, element_name, element_from_name, with_metadata = normative_element.get_parameters()
  20     _create_normative_element(sdc_be_proxy,
  21                               file_dir,
  22                               url_suffix,
  23                               element_name,
  24                               element_from_name,
  25                               model,
  26                               with_metadata,
  27                               exit_on_success)
  28
  29
  30 def _create_normative_element(sdc_be_proxy, file_dir,
  31                               url_suffix, element_name, element_form_name, model, with_metadata=False,
  32                               exit_on_success=False):
  33     result = _send_request(sdc_be_proxy,
  34                            file_dir,
  35                            url_suffix,
  36                            element_name,
  37                            element_form_name,
  38                            model,
  39                            with_metadata)
  40     print_and_check_result(result, exit_on_success)
  41
  42
  43 def _send_request(sdc_be_proxy, file_dir, url_suffix, element_name,
  44                   element_form_name, model,
  45                   with_metadata=False):
  46     try:
  47         log("create normative element ", element_name)
  48
  49         type_file_name = file_dir + element_name
  50         multi_part_form_data = _create_multipart_form_data(element_form_name, type_file_name, with_metadata,
  51                                                            element_name, model)
  52
  53         debug("http request url =", url_suffix)
  54         http_res = sdc_be_proxy.post_file(url_suffix, multi_part_form_data)
  55         if http_res is not None:
  56             debug("http response =", http_res)
  57
  58         response = sdc_be_proxy.get_response_from_buffer()
  59         debug("response buffer", response)
  60         # c.close()
  61         return element_name, http_res, response
  62
  63     except Exception as inst:
  64         print("ERROR=" + str(inst))
  65         return element_name, None, None
  66
  67
  68 def _create_multipart_form_data(element_form_name, type_file_name, with_metadata, element_name, model):
  69     tosca_type_zip_part = _create_zip_file_multi_part(element_form_name, type_file_name, element_name)
  70     multi_part_form_data = [tosca_type_zip_part]
  71     if with_metadata:
  72         metadata_type_part = _create_metadata_multipart(type_file_name)
  73         multi_part_form_data.append(metadata_type_part)
  74     if model is not None:
  75         model_data = ("model", model)
  76         multi_part_form_data.append(model_data)
  77     debug(multi_part_form_data)
  78     return multi_part_form_data
  79
  80
  81 def _create_metadata_multipart(type_file_name):
  82     metadata = _create_json_metadata_str(type_file_name)
  83     return "toscaTypeMetadata", metadata
  84
  85 def _create_zip_file_multi_part(element_form_name, type_file_name, element_name):
  86     zf = zipfile.ZipFile(type_file_name + ".zip", "w")
  87     zf.write(type_file_name + '.yml', element_name + '.yml')
  88     zf.close()
  89
  90     tosca_type_zip_path = type_file_name + ".zip"
  91     tosca_type_zip_part = (element_form_name, (pycurl.FORM_FILE, tosca_type_zip_path))
  92     return tosca_type_zip_part
  93
  94
  95 def _create_json_metadata_str(file_name):
  96     type_metadata_json_file = file_name + ".json"
  97     debug(type_metadata_json_file)
  98     json_file = open(type_metadata_json_file, encoding='utf-8')
  99
 100     debug("before load json")
 101     json_data = json.load(json_file, strict=False)
 102     debug(json_data)
 103
 104     return json.dumps(json_data)
 105
 106
 107 def print_and_check_result(result, exit_on_success):
 108     if result is not None:
 109         print_name_and_return_code(result[0], result[1])
 110         if result[1] is None or result[1] not in [200, 201, 409]:
 111             raise ResourceCreationError("Failed to create the normatives elements!!", 1)
 112         else:
 113             if exit_on_success is True:
 114                 print_and_exit(0, "All normatives elements created successfully!!")
 115     else:
 116         raise ResourceCreationError("Results is None!", 1)