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 post_file(self, path, multi_part_form_data):
  39         return self.con.post_file(path, multi_part_form_data)
  40
  41
  42 class CurlConnector:
  43     CONTENT_TYPE_HEADER = "Content-Type: application/json"
  44     ACCEPT_HEADER = "Accept: application/json; charset=UTF-8"
  45
  46     def __init__(self, url, user_id_header, buffer=None, scheme="http", debug=False):
  47         self.c = pycurl.Curl()
  48         self.c.setopt(pycurl.HEADER, True)
  49
  50         self.user_header = "USER_ID: " + user_id_header
  51
  52         if not debug:
  53             # disable printing not necessary logs in the terminal
  54             self.c.setopt(pycurl.WRITEFUNCTION, lambda x: None)
  55         else:
  56             self.c.setopt(pycurl.VERBOSE, 1)
  57
  58         if not buffer:
  59             self.buffer = BytesIO()
  60
  61         self.url = url
  62         self._check_schema(scheme)
  63
  64     def get(self, path):
  65         try:
  66             self.c.setopt(pycurl.URL, self.url + path)
  67             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
  68                                               CurlConnector.CONTENT_TYPE_HEADER,
  69                                               CurlConnector.ACCEPT_HEADER])
  70
  71             self.c.perform()
  72             return self.c.getinfo(pycurl.RESPONSE_CODE)
  73         except pycurl.error:
  74             return 111
  75
  76     def post(self, path, data):
  77         try:
  78             self.c.setopt(pycurl.URL, self.url + path)
  79             self.c.setopt(pycurl.POST, 1)
  80             self.c.setopt(pycurl.HTTPHEADER, [self.user_header,
  81                                               CurlConnector.CONTENT_TYPE_HEADER,
  82                                               CurlConnector.ACCEPT_HEADER])
  83
  84             self.c.setopt(pycurl.POSTFIELDS, data)
  85
  86             self.c.perform()
  87             self.c.setopt(pycurl.POST, 0)
  88
  89             return self.c.getinfo(pycurl.RESPONSE_CODE)
  90         except pycurl.error:
  91             return 111
  92
  93     def post_file(self, path, post_body, buffer=None):
  94         try:
  95             self.c.setopt(pycurl.URL, self.url + path)
  96             self.c.setopt(pycurl.POST, 1)
  97             self.c.setopt(pycurl.HTTPHEADER, [self.user_header])
  98
  99             self.c.setopt(pycurl.HTTPPOST, post_body)
 100
 101             write = self.buffer.write if not buffer else buffer.write
 102             self.c.setopt(pycurl.WRITEFUNCTION, write)
 103
 104             self.c.perform()
 105             self.c.setopt(pycurl.POST, 0)
 106
 107             return self.c.getinfo(pycurl.RESPONSE_CODE)
 108         except pycurl.error:
 109             return 111
 110
 111     def _check_schema(self, scheme):
 112         if scheme == 'https':
 113             self.c.setopt(pycurl.SSL_VERIFYPEER, 0)
 114             self.c.setopt(pycurl.SSL_VERIFYHOST, 0)
 115
 116     def __del__(self):
 117         self.c.close()