diff options
Diffstat (limited to 'robotframework-onap/ONAPLibrary/RequestsHelper.py')
| -rw-r--r-- | robotframework-onap/ONAPLibrary/RequestsHelper.py | 69 |
1 files changed, 42 insertions, 27 deletions
diff --git a/robotframework-onap/ONAPLibrary/RequestsHelper.py b/robotframework-onap/ONAPLibrary/RequestsHelper.py index 0fd0115..2dba36e 100644 --- a/robotframework-onap/ONAPLibrary/RequestsHelper.py +++ b/robotframework-onap/ONAPLibrary/RequestsHelper.py @@ -17,6 +17,7 @@ from RequestsLibrary import RequestsLibrary from robot.api import logger import hashlib from ONAPLibrary.Base64Keywords import Base64Keywords +from ONAPLibrary.HTTPKeywords import HTTPKeywords class RequestsHelper(object): @@ -27,34 +28,23 @@ class RequestsHelper(object): self.uuid = UUIDKeywords() self.application_id = "robot-ete" self.requests = RequestsLibrary() + self.http = HTTPKeywords() - def create_headers(self, sdc_user_id=None, accept="application/json", content_type="application/json", md5=None): - """Create the headers that are used by onap""" - uuid = self.uuid.generate_uuid4() - headers = { - "Accept": accept, - "Content-Type": content_type, - "X-TransactionId": self.application_id + "-" + uuid, - "X-FromAppId": self.application_id - } - if sdc_user_id is not None: - headers["USER_ID"] = sdc_user_id - if md5 is not None: - headers["Content-MD5"] = md5 - return headers - - def get_request(self, alias, endpoint, data_path, sdc_user=None, accept="application/json", auth=None): + def get_request(self, alias, endpoint, data_path, sdc_user=None, accept="application/json", auth=None, + client_certs=None): """Runs a get request""" + self.http.disable_warnings() logger.info("Creating session" + endpoint) - self.requests.create_session(alias, endpoint, auth=auth) - headers = self.create_headers(sdc_user_id=sdc_user, accept=accept) + self._create_session(alias, endpoint, auth=auth, client_certs=client_certs) + headers = self._create_headers(sdc_user_id=sdc_user, accept=accept) resp = self.requests.get_request(alias, data_path, headers=headers) logger.info("Received response from [" + alias + "]: " + resp.text) return resp def post_request(self, alias, endpoint, data_path, data, sdc_user=None, files=None, accept="application/json", - content_type="application/json", auth=None): + content_type="application/json", auth=None, client_certs=None): """Runs a post request""" + self.http.disable_warnings() logger.info("Creating session" + endpoint) if data is not None: md5 = hashlib.md5() @@ -62,26 +52,51 @@ class RequestsHelper(object): md5checksum = Base64Keywords().base64_encode(md5.hexdigest()) else: md5checksum = None - self.requests.create_session(alias, endpoint, auth=auth) - headers = self.create_headers(sdc_user_id=sdc_user, accept=accept, content_type=content_type, md5=md5checksum) + self._create_session(alias, endpoint, auth=auth, client_certs=client_certs) + headers = self._create_headers(sdc_user_id=sdc_user, accept=accept, content_type=content_type, md5=md5checksum) resp = self.requests.post_request(alias, data_path, files=files, data=data, headers=headers) logger.info("Received response from [" + alias + "]: " + resp.text) return resp - def put_request(self, alias, endpoint, data_path, data, sdc_user=None, accept="application/json", auth=None): + def put_request(self, alias, endpoint, data_path, data, sdc_user=None, accept="application/json", + auth=None, client_certs=None): """Runs a put request""" + self.http.disable_warnings() logger.info("Creating session" + endpoint) - self.requests.create_session(alias, endpoint, auth=auth) - headers = self.create_headers(sdc_user_id=sdc_user, accept=accept) + self._create_session(alias, endpoint, auth=auth, client_certs=client_certs) + headers = self._create_headers(sdc_user_id=sdc_user, accept=accept) resp = self.requests.put_request(alias, data_path, data=data, headers=headers) logger.info("Received response from [" + alias + "]: " + resp.text) return resp - def delete_request(self, alias, endpoint, data_path, data=None, sdc_user=None, accept="application/json", auth=None): + def delete_request(self, alias, endpoint, data_path, data=None, sdc_user=None, accept="application/json", + auth=None, client_certs=None): """Runs a delete request""" + self.http.disable_warnings() logger.info("Creating session" + endpoint) - self.requests.create_session(alias, endpoint, auth=auth) - headers = self.create_headers(sdc_user_id=sdc_user, accept=accept) + self._create_session(alias, endpoint, auth=auth, client_certs=client_certs) + headers = self._create_headers(sdc_user_id=sdc_user, accept=accept) resp = self.requests.delete_request(alias, data_path, data=data, headers=headers) logger.info("Received response from [" + alias + "]: " + resp.text) return resp + + def _create_session(self, alias, endpoint, auth=None, client_certs=None): + if client_certs is not None: + self.requests.create_client_cert_session(alias, endpoint, client_certs=client_certs) + else: + self.requests.create_session(alias, endpoint, auth=auth) + + def _create_headers(self, sdc_user_id=None, accept="application/json", content_type="application/json", md5=None): + """Create the headers that are used by onap""" + uuid = self.uuid.generate_uuid4() + headers = { + "Accept": accept, + "Content-Type": content_type, + "X-TransactionId": self.application_id + "-" + uuid, + "X-FromAppId": self.application_id + } + if sdc_user_id is not None: + headers["USER_ID"] = sdc_user_id + if md5 is not None: + headers["Content-MD5"] = md5 + return headers |