--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import robot.utils
+from robot.api.deco import keyword
+from robot.libraries.BuiltIn import BuiltIn
+import json
+
+from ONAPLibrary.Utilities import Utilities
+
+
+class BaseOpenstackKeywords(object):
+ """SO is an ONAP testing library for Robot Framework that provides functionality for interacting with the serivce
+ orchestrator. """
+
+ def __init__(self):
+ super(BaseOpenstackKeywords, self).__init__()
+ self._cache = robot.utils.ConnectionCache('No connections created')
+ self.application_id = "robot-ete"
+ self.uuid = Utilities()
+ self.builtin = BuiltIn()
+
+ @keyword
+ def save_openstack_auth(self, alias, response, token, version='v2.0'):
+ """Save Openstack Auth takes in an openstack auth response and saves it to allow easy retrival of token
+ and service catalog"""
+ self.builtin.log('Creating connection: %s' % alias, 'DEBUG')
+ json_response = json.loads(response)
+ json_response['auth_token'] = token
+ json_response['keystone_api_version'] = version
+ self._cache.register(json_response, alias=alias)
+
+ @keyword
+ def get_openstack_token(self, alias):
+ """Get Openstack auth token from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, str):
+ json_response = json.loads(response)
+ else:
+ json_response = response
+ if json_response['keystone_api_version'] == 'v2.0':
+ return json_response['access']['token']['id']
+ else:
+ return json_response['auth_token']
+
+ @keyword
+ def get_openstack_catalog(self, alias):
+ """Get Openstack service catalog from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, str):
+ json_response = json.loads(response)
+ else:
+ json_response = response
+ if json_response['keystone_api_version'] == 'v2.0':
+ return json_response['access']['serviceCatalog']
+ else:
+ return json_response['token']['catalog']
+
+ @keyword
+ def get_current_openstack_tenant(self, alias):
+ """Get Openstack tenant from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, str):
+ json_response = json.loads(response)
+ else:
+ json_response = response
+ if json_response['keystone_api_version'] == 'v2.0':
+ return json_response['access']['token']['tenant']
+ else:
+ return json_response['token']['project']
+
+ @keyword
+ def get_current_openstack_tenant_id(self, alias):
+ """Get Openstack tenant id from the current alias"""
+ tenant = self.get_current_openstack_tenant(alias)
+ return tenant['id']
+
+ @keyword
+ def get_openstack_regions(self, alias):
+ """Get all Openstack regions from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, str):
+ json_response = json.loads(response)
+ else:
+ json_response = response
+ regions = []
+ if json_response['keystone_api_version'] == 'v2.0':
+ resp = json_response['access']['serviceCatalog']
+ else:
+ resp = json_response['token']['catalog']
+ for catalogEntry in resp:
+ list_of_endpoints = catalogEntry['endpoints']
+ for endpoint in list_of_endpoints:
+ if 'region' in endpoint:
+ if endpoint['region'] not in regions:
+ regions.append(endpoint['region'])
+ return regions
+
+ @keyword
+ def get_openstack_service_url(self, alias, servicetype, region=None, tenant_id=None):
+ """Get Openstack service catalog from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, str):
+ json_response = json.loads(response)
+ else:
+ json_response = response
+ endpoint = None
+ if json_response['keystone_api_version'] == 'v2.0':
+ resp = json_response['access']['serviceCatalog']
+ else:
+ resp = json_response['token']['catalog']
+ for catalogEntry in resp:
+ if self.__determine_match(catalogEntry['type'], servicetype):
+ list_of_endpoints = catalogEntry['endpoints']
+ # filter out non matching regions if provided
+ list_of_endpoints[:] = [x for x in list_of_endpoints if self.__determine_match(x['region'], region)]
+ # filter out non matching tenants if provided
+ # Only provide tenant id when authorizing without qualifying with tenant id
+ # WindRiver does not return the tenantId on the endpoint in this case.
+ if tenant_id is not None:
+ list_of_endpoints[:] = [y for y in list_of_endpoints if
+ self.__determine_match(y['tenantId'], tenant_id)]
+ if json_response['keystone_api_version'] == 'v3':
+ list_of_endpoints[:] = [z for z in list_of_endpoints if
+ self.__determine_match(z['interface'], 'public')]
+ if len(list_of_endpoints) > 0:
+ if json_response['keystone_api_version'] == 'v2.0':
+ endpoint = list_of_endpoints[0]['publicURL']
+ else:
+ endpoint = list_of_endpoints[0]['url']
+ if endpoint is None:
+ self.builtin.should_not_be_empty("", "Service Endpoint Url should not be empty")
+ return endpoint
+
+ @staticmethod
+ def __determine_match(list_item, item):
+ if item is None:
+ return True
+ elif list_item == item:
+ return True
+ else:
+ return False
--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from robot.api.deco import keyword
+from robot.libraries.BuiltIn import BuiltIn
+import json
+import yaml
+import io
+import copy
+from hashlib import md5
+from paramiko import RSAKey
+from paramiko.ssh_exception import PasswordRequiredException
+
+from ONAPLibrary.Utilities import Utilities
+
+
+class HEATKeywords(object):
+ """Utilities useful for constructing OpenStack HEAT requests. """
+
+ def __init__(self):
+ super(HEATKeywords, self).__init__()
+ self.application_id = "robot-ete"
+ self.uuid = Utilities()
+ self.builtin = BuiltIn()
+
+ @keyword
+ def get_yaml(self, template_file):
+ """Template Yaml To Json reads a YAML Heat template file returns a JSON string that can be used included
+ in an Openstack Add Stack Request"""
+ if isinstance(template_file, str) or isinstance(template_file, unicode):
+ fin = open(template_file, 'r')
+ yamlobj = yaml.load(fin)
+ return yamlobj
+ return None
+
+ @keyword
+ def template_yaml_to_json(self, template_file):
+ """Template Yaml To Json reads a YAML Heat template file returns a JSON string that can be used included
+ in an Openstack Add Stack Request"""
+ contents = None
+ if isinstance(template_file, str) or isinstance(template_file, unicode):
+ fin = open(template_file, 'r')
+ yamlobj = yaml.load(fin)
+ fin.close()
+ if 'heat_template_version' in yamlobj:
+ datetime = yamlobj['heat_template_version']
+ yamlobj['heat_template_version'] = str(datetime)
+ fout = io.BytesIO()
+ json.dump(yamlobj, fout)
+ contents = fout.getvalue()
+ fout.close()
+ return contents
+
+ @keyword
+ def env_yaml_to_json(self, template_file):
+ """Env Yaml To JSon reads a YAML Heat env file and returns a JSON string that can be used included
+ in an Openstack Add Stack Request"""
+ if isinstance(template_file, str) or isinstance(template_file, unicode):
+ fin = open(template_file, 'r')
+ yamlobj = yaml.load(fin)
+ fin.close()
+ if 'parameters' in yamlobj:
+ fout = io.BytesIO()
+ json.dump(yamlobj['parameters'], fout)
+ contents = fout.getvalue()
+ fout.close()
+ return contents
+ return None
+
+ @keyword
+ def stack_info_parse(self, stack_info):
+ """ returns a flattened version of the Openstack Find Stack results """
+ d = {}
+ if isinstance(stack_info, dict):
+ s = stack_info['stack']
+ p = s['parameters']
+ d = copy.deepcopy(p)
+ d['id'] = s['id']
+ d['name'] = s['stack_name']
+ d['stack_status'] = s['stack_status']
+ return d
+
+ @keyword
+ def match_fingerprint(self, pvt_file, pw, fingerprint):
+ try:
+ ssh_key = RSAKey.from_private_key_file(pvt_file, pw)
+ keybytes = md5(ssh_key.asbytes()).hexdigest()
+ printable_fingerprint = ':'.join(a + b for a, b in zip(keybytes[::2], keybytes[1::2]))
+ return printable_fingerprint == fingerprint.__str__()
+ except PasswordRequiredException:
+ return False
+
+ @keyword
+ def match_private_key_file_to_keypair(self, files, keypair):
+ for keyfile in files:
+ if self.match_fingerprint(keyfile, None, keypair['keypair']['fingerprint']):
+ return keyfile
+ return None
+
+ @keyword
+ def get_openstack_server_ip(self, server, network_name="public", ipversion=4):
+ ipaddr = None
+ try:
+ versions = server['addresses'][network_name]
+ for version in versions:
+ if version['version'] == ipversion:
+ ipaddr = version['addr']
+ break
+ except ValueError:
+ return ipaddr
+ return ipaddr
Code Review / testsuite / python-testing-utils.git / commitdiff