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 get_normatives(self):
  53         return self.con.get("/sdc2/rest/v1/screen", with_buffer=True)
  54
  55     def post_file(self, path, multi_part_form_data, buffer=None):
  56         return self.con.post_file(path, multi_part_form_data, buffer)
  57
  58     def put_file(self, path, multi_part_form_data, buffer=None):
  59         return self.con.put_file(path, multi_part_form_data, buffer)
  60
  61     def get_response_from_buffer(self):
  62         value = self.con.buffer.getvalue()
  63         self.con.buffer.truncate(0)
  64         self.con.buffer.seek(0)
  65
  66         response = value.decode(self.CHARTSET).split(self.BODY_SEPARATOR)
  67         return response[1] if len(response) == 2 else response[0]
  68
  69
  70 class CurlConnector:
  71     CONTENT_TYPE_HEADER = "Content-Type: application/json"
  72     ACCEPT_HEADER = "Accept: application/json; charset=UTF-8"
  73
  74     def __init__(self, url, user_id_header, header, buffer=None, protocol="http", debug=False):
  75         self.__debug = debug
  76         self.__protocol = protocol
  77         self.c = self.__build_default_curl()
  78
  79         self.user_header = "USER_ID: " + user_id_header
  80         self.url = url
  81
  82         if not buffer:
  83             self.buffer = BytesIO()
  84
  85         if header is None:
  86             self.basicauth_header = ""
  87         else:
  88             self.basicauth_header = "Authorization: Basic " + header
  89
  90     def get(self, path, buffer=None, with_buffer=False):
  91         try:
  92             self.c.setopt(pycurl.URL, self.url + path)
  93             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
  94                                               CurlConnector.CONTENT_TYPE_HEADER,
  95                                               CurlConnector.ACCEPT_HEADER,
  96                                               self.basicauth_header])
  97
  98             if with_buffer:
  99                 write = self.buffer.write if not buffer else buffer.write
 100                 self.c.setopt(pycurl.WRITEFUNCTION, write)
 101
 102             self.c.perform()
 103             return self.c.getinfo(pycurl.RESPONSE_CODE)
 104         except pycurl.error:
 105             return 111
 106
 107     def post(self, path, data):
 108         try:
 109             self.c.setopt(pycurl.URL, self.url + path)
 110             self.c.setopt(pycurl.POST, 1)
 111
 112             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
 113                                               CurlConnector.CONTENT_TYPE_HEADER,
 114                                               CurlConnector.ACCEPT_HEADER,
 115                                               self.basicauth_header])
 116
 117             self.c.setopt(pycurl.POSTFIELDS, data)
 118
 119             self.c.perform()
 120             self.c.setopt(pycurl.POST, 0)
 121
 122             return self.c.getinfo(pycurl.RESPONSE_CODE)
 123         except pycurl.error:
 124             return 111
 125
 126     def post_file(self, path, post_body, buffer=None):
 127         try:
 128             self.c.setopt(pycurl.URL, self.url + path)
 129             self.c.setopt(pycurl.POST, 1)
 130             self.c.setopt(pycurl.HTTPHEADER, [self.user_header, self.basicauth_header])
 131
 132             self.c.setopt(pycurl.HTTPPOST, post_body)
 133
 134             write = self.buffer.write if not buffer else buffer.write
 135             self.c.setopt(pycurl.WRITEFUNCTION, write)
 136
 137             self.c.perform()
 138             self.c.setopt(pycurl.POST, 0)
 139             return self.c.getinfo(pycurl.RESPONSE_CODE)
 140         except pycurl.error as ex:
 141             print(ex)
 142             return 111
 143
 144     def put_file(self, path, post_body, response_write_buffer=None):
 145         curl = self.__build_default_curl()
 146         curl.setopt(pycurl.URL, self.url + path)
 147         curl.setopt(pycurl.HTTPHEADER, [self.user_header, self.basicauth_header])
 148         curl.setopt(pycurl.CUSTOMREQUEST, "PUT")
 149
 150         curl.setopt(pycurl.HTTPPOST, post_body)
 151
 152         write = self.buffer.write if not response_write_buffer else response_write_buffer.write
 153         curl.setopt(pycurl.WRITEFUNCTION, write)
 154
 155         curl.perform()
 156         response_code = curl.getinfo(pycurl.RESPONSE_CODE)
 157         curl.close()
 158         return response_code
 159
 160     def __build_default_curl(self):
 161         curl = pycurl.Curl()
 162         if not self.__debug:
 163             # disable printing not necessary logs in the terminal
 164             curl.setopt(pycurl.WRITEFUNCTION, lambda x: None)
 165         else:
 166             curl.setopt(pycurl.VERBOSE, 1)
 167
 168         if self.__protocol == 'https':
 169             curl.setopt(pycurl.SSL_VERIFYPEER, 0)
 170             curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 171         curl.setopt(pycurl.HEADER, True)
 172         return curl
 173
 174     def __del__(self):
 175         self.c.close()