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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
import json
from io import BytesIO
import pycurl
from sdcBePy.common.helpers import check_arguments_not_none
def get_url(ip, port, protocol):
return "%s://%s:%s" % (protocol, ip, port)
class SdcBeProxy:
BODY_SEPARATOR = "\r\n\r\n"
CHARTSET = 'UTF-8'
def __init__(self, be_ip, be_port, header, scheme, user_id="jh0003",
debug=False, connector=None):
if not check_arguments_not_none(be_ip, be_port, scheme, user_id):
raise AttributeError("The be_host, be_port, scheme or admin_user are missing")
url = get_url(be_ip, be_port, scheme)
self.con = connector if connector \
else CurlConnector(url, user_id, header, scheme=scheme, debug=debug)
def check_backend(self):
return self.con.get('/sdc2/rest/v1/user/jh0003')
def check_user(self, user_name):
return self.con.get("/sdc2/rest/v1/user" + user_name)
def create_user(self, first_name, last_name, user_id, email, role):
return self.con.post('/sdc2/rest/v1/user', json.dumps({
'firstName': first_name,
'lastName': last_name,
'userId': user_id,
'email': email,
'role': role
}))
def check_consumer(self, consumer_name):
return self.con.get("/sdc2/rest/v1/consumers" + consumer_name)
def create_consumer(self, consumer_name, slat, password):
return self.con.post("/sdc2/rest/v1/consumers", json.dumps({
'consumerName': consumer_name,
'consumerSalt': slat,
'consumerPassword': password
}))
def get_normatives(self):
return self.con.get("/sdc2/rest/v1/screen", with_buffer=True)
def post_file(self, path, multi_part_form_data):
return self.con.post_file(path, multi_part_form_data)
def get_response_from_buffer(self):
value = self.con.buffer.getvalue()
self.con.buffer.truncate(0)
self.con.buffer.seek(0)
response = value.decode(self.CHARTSET).split(self.BODY_SEPARATOR)
return response[1] if len(response) == 2 else response[0]
class CurlConnector:
CONTENT_TYPE_HEADER = "Content-Type: application/json"
ACCEPT_HEADER = "Accept: application/json; charset=UTF-8"
def __init__(self, url, user_id_header, header, buffer=None, scheme="http", debug=False):
self.c = pycurl.Curl()
self.c.setopt(pycurl.HEADER, True)
self.user_header = "USER_ID: " + user_id_header
if not debug:
# disable printing not necessary logs in the terminal
self.c.setopt(pycurl.WRITEFUNCTION, lambda x: None)
else:
self.c.setopt(pycurl.VERBOSE, 1)
if not buffer:
self.buffer = BytesIO()
if header is None:
self.basicauth_header = ""
else:
self.basicauth_header = "Authorization: Basic " + header
self.url = url
self._check_schema(scheme)
def get(self, path, buffer=None, with_buffer=False):
try:
self.c.setopt(pycurl.URL, self.url + path)
self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
CurlConnector.CONTENT_TYPE_HEADER,
CurlConnector.ACCEPT_HEADER,
self.basicauth_header])
if with_buffer:
write = self.buffer.write if not buffer else buffer.write
self.c.setopt(pycurl.WRITEFUNCTION, write)
self.c.perform()
return self.c.getinfo(pycurl.RESPONSE_CODE)
except pycurl.error:
return 111
def post(self, path, data):
try:
self.c.setopt(pycurl.URL, self.url + path)
self.c.setopt(pycurl.POST, 1)
self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
CurlConnector.CONTENT_TYPE_HEADER,
CurlConnector.ACCEPT_HEADER,
self.basicauth_header])
self.c.setopt(pycurl.POSTFIELDS, data)
self.c.perform()
self.c.setopt(pycurl.POST, 0)
return self.c.getinfo(pycurl.RESPONSE_CODE)
except pycurl.error:
return 111
def post_file(self, path, post_body, buffer=None):
try:
self.c.setopt(pycurl.URL, self.url + path)
self.c.setopt(pycurl.POST, 1)
self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
self.basicauth_header])
self.c.setopt(pycurl.HTTPPOST, post_body)
write = self.buffer.write if not buffer else buffer.write
self.c.setopt(pycurl.WRITEFUNCTION, write)
self.c.perform()
self.c.setopt(pycurl.POST, 0)
return self.c.getinfo(pycurl.RESPONSE_CODE)
except pycurl.error:
return 111
def _check_schema(self, scheme):
if scheme == 'https':
self.c.setopt(pycurl.SSL_VERIFYPEER, 0)
self.c.setopt(pycurl.SSL_VERIFYHOST, 0)
def __del__(self):
self.c.close()
|
