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