summaryrefslogtreecommitdiffstats
path: root/robotframework-onap/ONAPLibrary
diff options
context:
space:
mode:
Diffstat (limited to 'robotframework-onap/ONAPLibrary')
-rw-r--r--robotframework-onap/ONAPLibrary/BaseAAIKeywords.py14
-rw-r--r--robotframework-onap/ONAPLibrary/RequestsHelper.py69
-rw-r--r--robotframework-onap/ONAPLibrary/VariableKeywords.py22
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()