diff options
author | k.kedron <k.kedron@partner.samsung.com> | 2020-03-05 13:12:27 +0100 |
---|---|---|
committer | Ofir Sonsino <ofir.sonsino@intl.att.com> | 2020-06-24 06:01:11 +0000 |
commit | cb30f3a566b73c78c3ea666acfcd3f288098684a (patch) | |
tree | 38cd1acb83ddde2b927485611d130bbeec3c1a4b /catalog-be/src/main/resources/scripts/sdcBePy/common/sdcBeProxy.py | |
parent | bdbfc2e460ccb561c3e174260b2908b974996d4f (diff) |
Refactoring the sdc-BE-init python scripts
Deeper refactoring of python script:
- create the python package with init script
- support for python 3.x
- reuse code
- new design
- support for .json conf file
- update the docker chef script
Issue-ID: SDC-2784
Signed-off-by: Krystian Kedron <k.kedron@partner.samsung.com>
Change-Id: I02169eb7d0e3e90851ba1811536d1712c3b4145f
Diffstat (limited to 'catalog-be/src/main/resources/scripts/sdcBePy/common/sdcBeProxy.py')
-rwxr-xr-x | catalog-be/src/main/resources/scripts/sdcBePy/common/sdcBeProxy.py | 117 |
1 files changed, 117 insertions, 0 deletions
diff --git a/catalog-be/src/main/resources/scripts/sdcBePy/common/sdcBeProxy.py b/catalog-be/src/main/resources/scripts/sdcBePy/common/sdcBeProxy.py new file mode 100755 index 0000000000..48261704f7 --- /dev/null +++ b/catalog-be/src/main/resources/scripts/sdcBePy/common/sdcBeProxy.py @@ -0,0 +1,117 @@ +import json +from io import BytesIO + +import pycurl + +from sdcBePy.common.helpers import check_arguments_not_none + + +def get_url(ip, port, protocol): + return "%s://%s:%s" % (protocol, ip, port) + + +class SdcBeProxy: + + def __init__(self, be_ip, be_port, scheme, user_id="jh0003", + debug=False, connector=None): + if not check_arguments_not_none(be_ip, be_port, scheme, user_id): + raise AttributeError("The be_host, be_port, scheme or admin_user are missing") + url = get_url(be_ip, be_port, scheme) + self.con = connector if connector \ + else CurlConnector(url, user_id, scheme=scheme, debug=debug) + + def check_backend(self): + return self.con.get('/sdc2/rest/v1/user/jh0003') + + def check_user(self, user_name): + return self.con.get("/sdc2/rest/v1/user/" + user_name) + + def create_user(self, first_name, last_name, user_id, email, role): + return self.con.post('/sdc2/rest/v1/user', json.dumps({ + 'firstName': first_name, + 'lastName': last_name, + 'userId': user_id, + 'email': email, + 'role': role + })) + + def post_file(self, path, multi_part_form_data): + return self.con.post_file(path, multi_part_form_data) + + +class CurlConnector: + CONTENT_TYPE_HEADER = "Content-Type: application/json" + ACCEPT_HEADER = "Accept: application/json; charset=UTF-8" + + def __init__(self, url, user_id_header, buffer=None, scheme="http", debug=False): + self.c = pycurl.Curl() + self.c.setopt(pycurl.HEADER, True) + + self.user_header = "USER_ID: " + user_id_header + + if not debug: + # disable printing not necessary logs in the terminal + self.c.setopt(pycurl.WRITEFUNCTION, lambda x: None) + else: + self.c.setopt(pycurl.VERBOSE, 1) + + if not buffer: + self.buffer = BytesIO() + + self.url = url + self._check_schema(scheme) + + def get(self, path): + try: + self.c.setopt(pycurl.URL, self.url + path) + self.c.setopt(pycurl.HTTPHEADER, [self.user_header, + CurlConnector.CONTENT_TYPE_HEADER, + CurlConnector.ACCEPT_HEADER]) + + self.c.perform() + return self.c.getinfo(pycurl.RESPONSE_CODE) + except pycurl.error: + return 111 + + def post(self, path, data): + try: + self.c.setopt(pycurl.URL, self.url + path) + self.c.setopt(pycurl.POST, 1) + self.c.setopt(pycurl.HTTPHEADER, [self.user_header, + CurlConnector.CONTENT_TYPE_HEADER, + CurlConnector.ACCEPT_HEADER]) + + self.c.setopt(pycurl.POSTFIELDS, data) + + self.c.perform() + self.c.setopt(pycurl.POST, 0) + + return self.c.getinfo(pycurl.RESPONSE_CODE) + except pycurl.error: + return 111 + + def post_file(self, path, post_body, buffer=None): + try: + self.c.setopt(pycurl.URL, self.url + path) + self.c.setopt(pycurl.POST, 1) + self.c.setopt(pycurl.HTTPHEADER, [self.user_header]) + + self.c.setopt(pycurl.HTTPPOST, post_body) + + write = self.buffer.write if not buffer else buffer.write + self.c.setopt(pycurl.WRITEFUNCTION, write) + + self.c.perform() + self.c.setopt(pycurl.POST, 0) + + return self.c.getinfo(pycurl.RESPONSE_CODE) + except pycurl.error: + return 111 + + def _check_schema(self, scheme): + if scheme == 'https': + self.c.setopt(pycurl.SSL_VERIFYPEER, 0) + self.c.setopt(pycurl.SSL_VERIFYHOST, 0) + + def __del__(self): + self.c.close() |