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