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