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     def __init__(self, be_ip, be_port, scheme, user_id="jh0003",
  16                  debug=False, connector=None):
  17         if not check_arguments_not_none(be_ip, be_port, scheme, user_id):
  18             raise AttributeError("The be_host, be_port, scheme or admin_user are missing")
  19         url = get_url(be_ip, be_port, scheme)
  20         self.con = connector if connector \
  21             else CurlConnector(url, user_id, scheme=scheme, debug=debug)
  22
  23     def check_backend(self):
  24         return self.con.get('/sdc2/rest/v1/user/jh0003')
  25
  26     def check_user(self, user_name):
  27         return self.con.get("/sdc2/rest/v1/user/" + user_name)
  28
  29     def create_user(self, first_name, last_name, user_id, email, role):
  30         return self.con.post('/sdc2/rest/v1/user', json.dumps({
  31             'firstName': first_name,
  32             'lastName': last_name,
  33             'userId': user_id,
  34             'email': email,
  35             'role': role
  36         }))
  37
  38     def check_consumer(self, consumer_name):
  39         return self.con.get("/sdc2/rest/v1/consumers/" + consumer_name)
  40
  41     def create_consumer(self, consumer_name, slat, password):
  42         return self.con.post("/sdc2/rest/v1/consumers/", json.dumps({
  43             'consumerName': consumer_name,
  44             'consumerSalt': slat,
  45             'consumerPassword': password
  46         }))
  47
  48     def post_file(self, path, multi_part_form_data):
  49         return self.con.post_file(path, multi_part_form_data)
  50
  51
  52 class CurlConnector:
  53     CONTENT_TYPE_HEADER = "Content-Type: application/json"
  54     ACCEPT_HEADER = "Accept: application/json; charset=UTF-8"
  55
  56     def __init__(self, url, user_id_header, buffer=None, scheme="http", debug=False):
  57         self.c = pycurl.Curl()
  58         self.c.setopt(pycurl.HEADER, True)
  59
  60         self.user_header = "USER_ID: " + user_id_header
  61
  62         if not debug:
  63             # disable printing not necessary logs in the terminal
  64             self.c.setopt(pycurl.WRITEFUNCTION, lambda x: None)
  65         else:
  66             self.c.setopt(pycurl.VERBOSE, 1)
  67
  68         if not buffer:
  69             self.buffer = BytesIO()
  70
  71         self.url = url
  72         self._check_schema(scheme)
  73
  74     def get(self, path):
  75         try:
  76             self.c.setopt(pycurl.URL, self.url + path)
  77             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
  78                                               CurlConnector.CONTENT_TYPE_HEADER,
  79                                               CurlConnector.ACCEPT_HEADER])
  80
  81             self.c.perform()
  82             return self.c.getinfo(pycurl.RESPONSE_CODE)
  83         except pycurl.error:
  84             return 111
  85
  86     def post(self, path, data):
  87         try:
  88             self.c.setopt(pycurl.URL, self.url + path)
  89             self.c.setopt(pycurl.POST, 1)
  90             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
  91                                               CurlConnector.CONTENT_TYPE_HEADER,
  92                                               CurlConnector.ACCEPT_HEADER])
  93
  94             self.c.setopt(pycurl.POSTFIELDS, data)
  95
  96             self.c.perform()
  97             self.c.setopt(pycurl.POST, 0)
  98
  99             return self.c.getinfo(pycurl.RESPONSE_CODE)
 100         except pycurl.error:
 101             return 111
 102
 103     def post_file(self, path, post_body, buffer=None):
 104         try:
 105             self.c.setopt(pycurl.URL, self.url + path)
 106             self.c.setopt(pycurl.POST, 1)
 107             self.c.setopt(pycurl.HTTPHEADER, [self.user_header])
 108
 109             self.c.setopt(pycurl.HTTPPOST, post_body)
 110
 111             write = self.buffer.write if not buffer else buffer.write
 112             self.c.setopt(pycurl.WRITEFUNCTION, write)
 113
 114             self.c.perform()
 115             self.c.setopt(pycurl.POST, 0)
 116
 117             return self.c.getinfo(pycurl.RESPONSE_CODE)
 118         except pycurl.error:
 119             return 111
 120
 121     def _check_schema(self, scheme):
 122         if scheme == 'https':
 123             self.c.setopt(pycurl.SSL_VERIFYPEER, 0)
 124             self.c.setopt(pycurl.SSL_VERIFYHOST, 0)
 125
 126     def __del__(self):
 127         self.c.close()