robotframework-onap/ONAPLibrary/RequestsHelper.py

   1 # Copyright 2019 AT&T Intellectual Property. All rights reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 from ONAPLibrary.UUIDKeywords import UUIDKeywords
  16 from RequestsLibrary import RequestsLibrary
  17 from robot.api import logger
  18 import hashlib
  19 from ONAPLibrary.Base64Keywords import Base64Keywords
  20
  21
  22 class RequestsHelper(object):
  23     """ non keyword methods related to Requests library """
  24
  25     def __init__(self):
  26         super(RequestsHelper, self).__init__()
  27         self.uuid = UUIDKeywords()
  28         self.application_id = "robot-ete"
  29         self.requests = RequestsLibrary()
  30
  31     def create_headers(self, sdc_user_id=None, accept="application/json", content_type="application/json", md5=None):
  32         """Create the headers that are used by onap"""
  33         uuid = self.uuid.generate_uuid4()
  34         headers = {
  35             "Accept": accept,
  36             "Content-Type": content_type,
  37             "X-TransactionId": self.application_id + "-" + uuid,
  38             "X-FromAppId": self.application_id
  39         }
  40         if not sdc_user_id:
  41             headers["USER_ID"] = sdc_user_id
  42         if not md5:
  43             headers["Content-MD5"] = md5
  44         return headers
  45
  46     def get_request(self, alias, endpoint, data_path, sdc_user=None, accept="application/json", auth=None):
  47         """Runs a get request"""
  48         logger.info("Creating session" + endpoint)
  49         self.requests.create_session(alias, endpoint, auth=auth)
  50         headers = self.create_headers(sdc_user_id=sdc_user, accept=accept)
  51         resp = self.requests.get_request(alias, data_path, headers=headers)
  52         logger.info("Received response from [" + alias + "]: " + resp.text)
  53         return resp
  54
  55     def post_request(self, alias, endpoint, data_path, data, sdc_user=None, files=None, accept="application/json",
  56                      content_type="application/json", auth=None):
  57         """Runs a post request"""
  58         logger.info("Creating session" + endpoint)
  59         md5 = hashlib.md5()
  60         md5.update(data)
  61         md5checksum = Base64Keywords().base64_encode(md5.hexdigest())
  62         self.requests.create_session(alias,  endpoint, auth=auth)
  63         headers = self.create_headers(sdc_user_id=sdc_user, accept=accept, content_type=content_type, md5=md5checksum)
  64         resp = self.requests.post_request(alias,  data_path, files=files, data=data, headers=headers)
  65         logger.info("Received response from [" + alias + "]: " + resp.text)
  66         return resp
  67
  68     def put_request(self, alias, endpoint, data_path, data, sdc_user=None, accept="application/json", auth=None):
  69         """Runs a put request"""
  70         logger.info("Creating session" + endpoint)
  71         self.requests.create_session(alias,  endpoint, auth=auth)
  72         headers = self.create_headers(sdc_user_id=sdc_user, accept=accept)
  73         resp = self.requests.put_request(alias,  data_path, data=data, headers=headers)
  74         logger.info("Received response from [" + alias + "]: " + resp.text)
  75         return resp
  76
  77     def delete_request(self, alias, endpoint, data_path, data=None, sdc_user=None, accept="application/json", auth=None):
  78         """Runs a delete request"""
  79         logger.info("Creating session" + endpoint)
  80         self.requests.create_session(alias, endpoint, auth=auth)
  81         headers = self.create_headers(sdc_user_id=sdc_user, accept=accept)
  82         resp = self.requests.delete_request(alias, data_path, data=data, headers=headers)
  83         logger.info("Received response from [" + alias + "]: " + resp.text)
  84         return resp