diff options
Diffstat (limited to 'robotframework-onap')
| -rw-r--r-- | robotframework-onap/ONAPLibrary/BaseAAIKeywords.py | 14 | ||||
| -rw-r--r-- | robotframework-onap/ONAPLibrary/RequestsHelper.py | 69 | ||||
| -rw-r--r-- | robotframework-onap/ONAPLibrary/VariableKeywords.py | 22 |
3 files changed, 63 insertions, 42 deletions
diff --git a/robotframework-onap/ONAPLibrary/BaseAAIKeywords.py b/robotframework-onap/ONAPLibrary/BaseAAIKeywords.py index a512203..222c54b 100644 --- a/robotframework-onap/ONAPLibrary/BaseAAIKeywords.py +++ b/robotframework-onap/ONAPLibrary/BaseAAIKeywords.py @@ -18,7 +18,6 @@ from robot.libraries.BuiltIn import BuiltIn import time from ONAPLibrary.RequestsHelper import RequestsHelper -from ONAPLibrary.HTTPKeywords import HTTPKeywords from ONAPLibrary.VariableKeywords import VariableKeywords @@ -30,36 +29,31 @@ class BaseAAIKeywords(object): super(BaseAAIKeywords, self).__init__() self.reqs = RequestsHelper() self.builtin = BuiltIn() - self.http = HTTPKeywords() self.vars = VariableKeywords() - aai_ip_addr = self.vars.get_globally_injected_parameters()['GLOBAL_INJECTED_AAI_IP_ADDR'] - aai_server_protocol = self.vars.get_global_parameters()['GLOBAL_AAI_SERVER_PROTOCOL'] - aai_server_port = self.vars.get_global_parameters()['GLOBAL_AAI_SERVER_PORT'] + aai_ip_addr = self.vars.get_globally_injected_parameters().get('GLOBAL_INJECTED_AAI_IP_ADDR', '') + aai_server_protocol = self.vars.get_global_parameters().get('GLOBAL_AAI_SERVER_PROTOCOL', '') + aai_server_port = self.vars.get_global_parameters().get('GLOBAL_AAI_SERVER_PORT', '') self.aai_endpoint = aai_server_protocol + '://' + aai_ip_addr + ':' + aai_server_port @keyword def run_get_request(self, endpoint, data_path, accept="application/json", auth=None): """Runs an AAI get request""" - self.http.disable_warnings() return self.reqs.get_request("aai", endpoint, data_path, sdc_user=None, accept=accept, auth=auth) @keyword def run_post_request(self, endpoint, data_path, data, accept="application/json", auth=None): """Runs an AAI post request""" - self.http.disable_warnings() return self.reqs.post_request("aai", endpoint, data_path, data, sdc_user=None, files=None, accept=accept, auth=auth) @keyword def run_put_request(self, endpoint, data_path, data, accept="application/json", auth=None): """Runs an AAI post request""" - self.http.disable_warnings() return self.reqs.put_request("aai", endpoint, data_path, data, sdc_user=None, accept=accept, auth=auth) @keyword def run_delete_request(self, endpoint, data_path, resource_version, accept="application/json", auth=None): """Runs an AAI delete request""" - self.http.disable_warnings() return self.reqs.delete_request("aai", endpoint, data_path + '?resource-version=' + resource_version, data=None, sdc_user=None, accept=accept, auth=auth) @@ -67,6 +61,7 @@ class BaseAAIKeywords(object): def wait_for_node_to_exist(self, search_node_type, key, uuid, auth=None): logger.info('Waiting for AAI traversal to complete...') for i in range(30): + logger.trace("running iteration " + str(i)) time.sleep(1) result = self.find_node(search_node_type, key, uuid, auth=auth) if result: @@ -81,7 +76,6 @@ class BaseAAIKeywords(object): def find_node(self, search_node_type, key, node_uuid, auth=None): data_path = '/aai/v11/search/nodes-query?search-node-type={0}&filter={1}:EQUALS:{2}'.format( search_node_type, key, node_uuid) - self.http.disable_warnings() resp = self.reqs.get_request("aai", self.aai_endpoint, data_path, accept="application/json", auth=auth) response = resp.json() return 'result-data' in response 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 diff --git a/robotframework-onap/ONAPLibrary/VariableKeywords.py b/robotframework-onap/ONAPLibrary/VariableKeywords.py index 6338aa2..81dcb9f 100644 --- a/robotframework-onap/ONAPLibrary/VariableKeywords.py +++ b/robotframework-onap/ONAPLibrary/VariableKeywords.py @@ -12,8 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -from robot.libraries.BuiltIn import BuiltIn +from robot.libraries.BuiltIn import BuiltIn, RobotNotRunningError from robot.api.deco import keyword +import os class VariableKeywords(object): @@ -25,18 +26,29 @@ class VariableKeywords(object): @keyword def get_globally_injected_parameters(self): - dictionary = self.builtin.get_variables(no_decoration=True) - return self._filter_variables_by_key_prefix(dictionary, "GLOBAL_INJECTED_") + return self._filter_variables_by_key_prefix(self._retrieve_robot_variables(), "GLOBAL_INJECTED_") @keyword def get_global_parameters(self): - dictionary = self.builtin.get_variables(no_decoration=True) - global_variables = self._filter_variables_by_key_prefix(dictionary, "GLOBAL_") + global_variables = self._filter_variables_by_key_prefix(self._retrieve_robot_variables(), "GLOBAL_") # strip out global injected (get those above) for key in self.get_globally_injected_parameters(): del global_variables[key] return global_variables + def _retrieve_robot_variables(self): + """ try to get the parameters from the robot keyword, but if it is ran out of robot context, + allow an env to be used instead """ + dictionary = dict() + try: + dictionary = self.builtin.get_variables(no_decoration=True) + except RobotNotRunningError: + try: + dictionary = os.environ['GLOBAL_ROBOT_VARIABLES'] + except KeyError: + pass + return dictionary + @staticmethod def _filter_variables_by_key_prefix(dictionary, partial): matches = dict() |