+++ /dev/null
-import json
-import yaml
-import io
-import copy
-from hashlib import md5
-from paramiko import RSAKey
-from paramiko.ssh_exception import PasswordRequiredException
-
-
-class HEATUtils:
- """ Utilities useful for constructing OpenStack HEAT requests """
-
- 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
-
- 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
-
- 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
-
- 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
-
- def match_fingerprint(self, pvt_file, pw, fingerprint):
- try:
- sshKey = RSAKey.from_private_key_file(pvt_file, pw)
- keybytes = md5(sshKey.asbytes()).hexdigest()
- printableFingerprint = ':'.join(a + b for a, b in zip(keybytes[::2], keybytes[1::2]))
- return printableFingerprint == fingerprint.__str__()
- except PasswordRequiredException:
- return False
-
- 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
-
- 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
\ No newline at end of file
+++ /dev/null
-from robot.libraries.BuiltIn import BuiltIn
-import robot.utils
-import json
-
-class OpenstackLibrary:
- """OpenstackLibrary manages the connection state and service catalog of an openstack instance."""
-
- ROBOT_LIBRARY_SCOPE = 'Global'
-
-
- def __init__(self):
- self._cache = robot.utils.ConnectionCache('No connections created')
- self.builtin = BuiltIn()
-
- 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')
- jsonResponse = json.loads(response);
- jsonResponse['auth_token'] = token
- jsonResponse['keystone_api_version'] = version
- self._cache.register(jsonResponse, alias=alias)
-
- def get_openstack_token(self, alias):
- """Get Openstack auth token from the current alias"""
- response = self._cache.switch(alias)
- if isinstance(response, str):
- jsonResponse = json.loads(response);
- else:
- jsonResponse = response;
- if jsonResponse['keystone_api_version'] == 'v2.0':
- return jsonResponse['access']['token']['id']
- else:
- return jsonResponse['auth_token']
-
- def get_openstack_catalog(self, alias):
- """Get Openstack service catalog from the current alias"""
- response = self._cache.switch(alias)
- if isinstance(response, str):
- jsonResponse = json.loads(response);
- else:
- jsonResponse = response;
- if jsonResponse['keystone_api_version'] == 'v2.0':
- return jsonResponse['access']['serviceCatalog']
- else:
- return jsonResponse['token']['catalog']
-
-
- def get_current_openstack_tenant(self, alias):
- """Get Openstack tenant from the current alias"""
- response = self._cache.switch(alias)
- if isinstance(response, str):
- jsonResponse = json.loads(response);
- else:
- jsonResponse = response;
- if jsonResponse['keystone_api_version'] == 'v2.0':
- return jsonResponse['access']['token']['tenant']
- else:
- return jsonResponse['token']['project']
-
- 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']
-
- def get_openstack_regions(self, alias):
- """Get all Openstack regions from the current alias"""
- response = self._cache.switch(alias)
- if isinstance(response, str):
- jsonResponse = json.loads(response);
- else:
- jsonResponse = response;
- regions = [];
- if jsonResponse['keystone_api_version'] == 'v2.0':
- resp = jsonResponse['access']['serviceCatalog']
- else:
- resp = jsonResponse['token']['catalog']
- for catalogEntry in resp:
- listOfEndpoints = catalogEntry['endpoints'];
- for endpoint in listOfEndpoints:
- if 'region'in endpoint:
- if endpoint['region'] not in regions:
- regions.append(endpoint['region'])
- return regions;
-
- 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):
- jsonResponse = json.loads(response);
- else:
- jsonResponse = response;
- endPoint = None;
- if jsonResponse['keystone_api_version'] == 'v2.0':
- resp = jsonResponse['access']['serviceCatalog']
- else:
- resp = jsonResponse['token']['catalog']
- for catalogEntry in resp:
- if self.__determine_match(catalogEntry['type'], servicetype):
- listOfEndpoints = catalogEntry['endpoints'];
- # filter out non matching regions if provided
- listOfEndpoints[:] = [x for x in listOfEndpoints 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:
- listOfEndpoints[:] = [y for y in listOfEndpoints if self.__determine_match(y['tenantId'], tenant_id)];
- if jsonResponse['keystone_api_version'] == 'v3':
- listOfEndpoints[:] = [z for z in listOfEndpoints if self.__determine_match(z['interface'], 'public')];
- if len(listOfEndpoints) > 0:
- if jsonResponse['keystone_api_version'] == 'v2.0':
- endPoint = listOfEndpoints[0]['publicURL'];
- else:
- endPoint = listOfEndpoints[0]['url'];
- if endPoint == None:
- self.builtin.should_not_be_empty("", "Service Endpoint Url should not be empty")
- return endPoint;
-
- def __determine_match(self, listItem, item):
- if item is None:
- return True;
- elif listItem == item:
- return True;
- else:
- return False;
\ No newline at end of file
Code Review / testsuite / python-testing-utils.git / commitdiff