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

   1 import json
   2 import zipfile
   3
   4 import pycurl
   5
   6 from sdcBePy.common.logger import print_name_and_return_code, error_and_exit, log, debug
   7 from sdcBePy.common.sdcBeProxy import SdcBeProxy
   8
   9
  10 def process_and_create_normative_types(normative_type,
  11                                        scheme=None, be_host=None, be_port=None, admin_user=None,
  12                                        sdc_be_proxy=None,
  13                                        update_version=False,
  14                                        debug=False,
  15                                        exit_on_success=False):
  16     if sdc_be_proxy is None:
  17         sdc_be_proxy = SdcBeProxy(be_host, be_port, scheme, admin_user, debug=debug)
  18
  19     file_dir, normative_type_list = normative_type.get_parameters()
  20
  21     results = _create_normatives_type(file_dir, sdc_be_proxy, normative_type_list, update_version)
  22     print_and_check_results(results, update_version, exit_on_success)
  23
  24
  25 def print_and_check_results(results, update_version, exit_on_success=False):
  26     if results is not None:
  27         if len(results) == 0:
  28             return
  29         print("----------------------------------------")
  30         for result in results:
  31             print_name_and_return_code(result[0], result[1], with_line=False)
  32         print("----------------------------------------")
  33         check_results_and_exit(results, update_version, exit_on_success)
  34     else:
  35         error_and_exit(1, "results is none -> error occurred!!")
  36
  37
  38 def check_results_and_exit(results, update_version, exit_on_success):
  39     if not _results_ok(results, _get_response_code(update_version)):
  40         error_and_exit(1, "Failed to create the normatives types !!")
  41     else:
  42         if exit_on_success:
  43             error_and_exit(0, "All normatives types created successfully!!")
  44
  45
  46 def _create_normatives_type(file_dir, sdc_be_proxy, types, update_version):
  47     results = []
  48     response_codes = _get_response_code(update_version)
  49     for normativeType in types:
  50         result = _send_request(sdc_be_proxy, file_dir, normativeType, update_version)
  51         results.append(result)
  52         if result[1] is None or result[1] not in response_codes:
  53             print("Failed creating normative type " + normativeType + ". " + str(result[1]))
  54     return results
  55
  56
  57 def _send_request(sdc_be_proxy, file_dir, element_name, update_version):
  58     try:
  59         log("create normative type ", element_name)
  60         debug("userId", sdc_be_proxy.con.user_header)
  61         debug("fileDir", file_dir)
  62
  63         url = '/sdc2/rest/v1/catalog/upload/multipart'
  64         if update_version is not None:
  65             url += '?createNewVersion=' + _boolean_to_string(update_version)
  66
  67         send = _create_send_body(file_dir, element_name)
  68
  69         debug(send)
  70         httpRes = sdc_be_proxy.post_file(url, send)
  71         if httpRes is not None:
  72             debug("http response=", httpRes)
  73         debug(sdc_be_proxy.con.buffer.getvalue())
  74
  75         return element_name, httpRes, sdc_be_proxy.con.buffer.getvalue()
  76
  77     except Exception as inst:
  78         print("ERROR=" + str(inst))
  79         return element_name, None, None
  80
  81
  82 def _create_send_body(file_dir, element_name):
  83     yml_path = file_dir + element_name + "/" + element_name + ".yml"
  84     path = file_dir + element_name + "/" + element_name + ".zip"
  85
  86     zf = zipfile.ZipFile(path, "w")
  87     zf.write(yml_path, element_name + '.yml')
  88     zf.close()
  89
  90     debug(path)
  91     current_json_file = file_dir + element_name + "/" + element_name + ".json"
  92
  93     jsonFile = open(current_json_file)
  94
  95     debug("before load json")
  96     json_data = json.load(jsonFile, strict=False)
  97     debug(json_data)
  98
  99     jsonAsStr = json.dumps(json_data)
 100
 101     return [('resourceMetadata', jsonAsStr), ('resourceZip', (pycurl.FORM_FILE, path))]
 102
 103
 104 def _results_ok(results, response_codes):
 105     for result in results:
 106         if result[1] not in response_codes:
 107             return False
 108
 109     return True
 110
 111
 112 def _get_response_code(update_version):
 113     responseCodes = [200, 201]
 114     if update_version is False:
 115         responseCodes.append(409)
 116
 117     return responseCodes
 118
 119
 120 def _boolean_to_string(boolean_value):
 121     return "true" if boolean_value else "false"