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

   1 import json
   2 from io import BytesIO
   3
   4 import pycurl
   5
   6 from sdcBePy.common.helpers import check_arguments_not_none
   7
   8
   9 def get_url(ip, port, protocol):
  10     return "%s://%s:%s" % (protocol, ip, port)
  11
  12
  13 class SdcBeProxy:
  14
  15     BODY_SEPARATOR = "\r\n\r\n"
  16     CHARTSET = 'UTF-8'
  17
  18     def __init__(self, be_ip, be_port, header, scheme, user_id="jh0003",
  19                  debug=False, connector=None):
  20         if not check_arguments_not_none(be_ip, be_port, scheme, user_id):
  21             raise AttributeError("The be_host, be_port, scheme or admin_user are missing")
  22         url = get_url(be_ip, be_port, scheme)
  23         self.con = connector if connector \
  24             else CurlConnector(url, user_id, header, protocol=scheme, debug=debug)
  25
  26     def check_backend(self):
  27         return self.con.get('/sdc2/rest/v1/user/jh0003')
  28
  29     def check_user(self, user_name):
  30           return self.con.get("/sdc2/rest/v1/user" + user_name)
  31
  32     def create_user(self, first_name, last_name, user_id, email, role):
  33
  34         return self.con.post('/sdc2/rest/v1/user', json.dumps({
  35             'firstName': first_name,
  36             'lastName': last_name,
  37             'userId': user_id,
  38             'email': email,
  39             'role': role
  40         }))
  41
  42     def check_consumer(self, consumer_name):
  43         return self.con.get("/sdc2/rest/v1/consumers" + consumer_name)
  44
  45     def create_consumer(self, consumer_name, slat, password):
  46         return self.con.post("/sdc2/rest/v1/consumers", json.dumps({
  47             'consumerName': consumer_name,
  48             'consumerSalt': slat,
  49             'consumerPassword': password
  50         }))
  51
  52     def disable_locking(self, disable):
  53         return self.con.post("/sdc2/rest/v1/catalog/lock", disable)
  54
  55     def get_normatives(self):
  56         return self.con.get("/sdc2/rest/v1/screen", with_buffer=True)
  57
  58     def post_file(self, path, multi_part_form_data, buffer=None):
  59         return self.con.post_file(path, multi_part_form_data, buffer)
  60
  61     def put_file(self, path, multi_part_form_data, buffer=None):
  62         return self.con.put_file(path, multi_part_form_data, buffer)
  63
  64     def get_response_from_buffer(self):
  65         value = self.con.buffer.getvalue()
  66         self.con.buffer.truncate(0)
  67         self.con.buffer.seek(0)
  68
  69         response = value.decode(self.CHARTSET).split(self.BODY_SEPARATOR)
  70         return response[1] if len(response) == 2 else response[0]
  71
  72
  73 class CurlConnector:
  74     CONTENT_TYPE_HEADER = "Content-Type: application/json"
  75     ACCEPT_HEADER = "Accept: application/json; charset=UTF-8"
  76
  77     def __init__(self, url, user_id_header, header, buffer=None, protocol="http", debug=False):
  78         self.__debug = debug
  79         self.__protocol = protocol
  80         self.c = self.__build_default_curl()
  81
  82         self.user_header = "USER_ID: " + user_id_header
  83         self.url = url
  84
  85         if not buffer:
  86             self.buffer = BytesIO()
  87
  88         if header is None:
  89             self.basicauth_header = ""
  90         else:
  91             self.basicauth_header = "Authorization: Basic " + header
  92
  93     def get(self, path, buffer=None, with_buffer=False):
  94         try:
  95             self.c.setopt(pycurl.URL, self.url + path)
  96             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
  97                                               CurlConnector.CONTENT_TYPE_HEADER,
  98                                               CurlConnector.ACCEPT_HEADER,
  99                                               self.basicauth_header])
 100
 101             if with_buffer:
 102                 write = self.buffer.write if not buffer else buffer.write
 103                 self.c.setopt(pycurl.WRITEFUNCTION, write)
 104
 105             self.c.perform()
 106             return self.c.getinfo(pycurl.RESPONSE_CODE)
 107         except pycurl.error:
 108             return 111
 109
 110     def post(self, path, data):
 111         try:
 112             self.c.setopt(pycurl.URL, self.url + path)
 113             self.c.setopt(pycurl.POST, 1)
 114
 115             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
 116                                               CurlConnector.CONTENT_TYPE_HEADER,
 117                                               CurlConnector.ACCEPT_HEADER,
 118                                               self.basicauth_header])
 119
 120             self.c.setopt(pycurl.POSTFIELDS, data)
 121
 122             self.c.perform()
 123             self.c.setopt(pycurl.POST, 0)
 124
 125             return self.c.getinfo(pycurl.RESPONSE_CODE)
 126         except pycurl.error:
 127             return 111
 128
 129     def post_file(self, path, post_body, buffer=None):
 130         try:
 131             self.c.setopt(pycurl.URL, self.url + path)
 132             self.c.setopt(pycurl.POST, 1)
 133             self.c.setopt(pycurl.HTTPHEADER, [self.user_header, self.basicauth_header])
 134
 135             self.c.setopt(pycurl.HTTPPOST, post_body)
 136
 137             write = self.buffer.write if not buffer else buffer.write
 138             self.c.setopt(pycurl.WRITEFUNCTION, write)
 139
 140             self.c.perform()
 141             self.c.setopt(pycurl.POST, 0)
 142             return self.c.getinfo(pycurl.RESPONSE_CODE)
 143         except pycurl.error as ex:
 144             print(ex)
 145             return 111
 146
 147     def put_file(self, path, post_body, response_write_buffer=None):
 148         curl = self.__build_default_curl()
 149         curl.setopt(pycurl.URL, self.url + path)
 150         curl.setopt(pycurl.HTTPHEADER, [self.user_header, self.basicauth_header])
 151         curl.setopt(pycurl.CUSTOMREQUEST, "PUT")
 152
 153         curl.setopt(pycurl.HTTPPOST, post_body)
 154
 155         write = self.buffer.write if not response_write_buffer else response_write_buffer.write
 156         curl.setopt(pycurl.WRITEFUNCTION, write)
 157
 158         curl.perform()
 159         response_code = curl.getinfo(pycurl.RESPONSE_CODE)
 160         curl.close()
 161         return response_code
 162
 163     def __build_default_curl(self):
 164         curl = pycurl.Curl()
 165         if not self.__debug:
 166             # disable printing not necessary logs in the terminal
 167             curl.setopt(pycurl.WRITEFUNCTION, lambda x: None)
 168         else:
 169             curl.setopt(pycurl.VERBOSE, 1)
 170
 171         if self.__protocol == 'https':
 172             curl.setopt(pycurl.SSL_VERIFYPEER, 0)
 173             curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 174         curl.setopt(pycurl.HEADER, True)
 175         return curl
 176
 177     def __del__(self):
 178         self.c.close()