import logging
import os.path
-import re
import sys
import time
from novaclient import client as novaclient
from keystoneclient import client as keystoneclient
from neutronclient.neutron import client as neutronclient
-from functest.utils.constants import CONST
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
logger = logging.getLogger(__name__)
return creds
-def source_credentials(rc_file):
- with open(rc_file, "r") as f:
- for line in f:
- var = (line.rstrip('"\n').replace('export ', '').split("=")
- if re.search(r'(.*)=(.*)', line) else None)
- # The two next lines should be modified as soon as rc_file
- # conforms with common rules. Be aware that it could induce
- # issues if value starts with '
- if var:
- key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
- value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
- os.environ[key] = value
-
-
-def get_credentials_for_rally():
- creds = get_credentials()
- env_cred_dict = get_env_cred_dict()
- rally_conf = {"type": "ExistingCloud", "admin": {}}
- for key in creds:
- if key == 'auth_url':
- rally_conf[key] = creds[key]
- else:
- rally_conf['admin'][key] = creds[key]
-
- endpoint_types = [('internalURL', 'internal'),
- ('publicURL', 'public'), ('adminURL', 'admin')]
-
- endpoint_type = get_endpoint_type_from_env()
- if endpoint_type is not None:
- cred_key = env_cred_dict.get('OS_ENDPOINT_TYPE')
- for k, v in endpoint_types:
- if endpoint_type == v:
- rally_conf[cred_key] = v
-
- region_name = os.getenv('OS_REGION_NAME')
- if region_name is not None:
- cred_key = env_cred_dict.get('OS_REGION_NAME')
- rally_conf[cred_key] = region_name
-
- cred_key = env_cred_dict.get('OS_CACERT')
- rally_conf[cred_key] = os.getenv('OS_CACERT', '')
-
- insecure_key = env_cred_dict.get('OS_INSECURE')
- rally_conf[insecure_key] = os.getenv('OS_INSECURE', '').lower() == 'true'
-
- return rally_conf
-
-
-def get_endpoint_type_from_env():
- endpoint_type = os.environ.get("OS_ENDPOINT_TYPE",
- os.environ.get("OS_INTERFACE"))
- if endpoint_type and "URL" in endpoint_type:
- endpoint_type = endpoint_type.replace("URL", "")
- return endpoint_type
-
-
def get_session_auth(other_creds={}):
loader = loading.get_plugin_loader('password')
creds = get_credentials(other_creds)
count = VM_BOOT_TIMEOUT / SLEEP
for n in range(count, -1, -1):
status = get_instance_status(nova_client, instance)
- if status.lower() == "active":
+ if status is None:
+ time.sleep(SLEEP)
+ continue
+ elif status.lower() == "active":
return instance
elif status.lower() == "error":
logger.error("The instance %s went to ERROR status."
def get_external_net(neutron_client):
+ if (hasattr(CONST, 'EXTERNAL_NETWORK')):
+ return CONST.__getattribute__('EXTERNAL_NETWORK')
for network in neutron_client.list_networks()['networks']:
if network['router:external']:
return network['name']
def get_external_net_id(neutron_client):
+ if (hasattr(CONST, 'EXTERNAL_NETWORK')):
+ networks = neutron_client.list_networks(
+ name=CONST.__getattribute__('EXTERNAL_NETWORK'))
+ net_id = networks['networks'][0]['id']
+ return net_id
for network in neutron_client.list_networks()['networks']:
if network['router:external']:
return network['id']
return id
-def create_glance_image(glance_client, image_name, file_path, disk="qcow2",
- container="bare", public="public"):
+def create_glance_image(glance_client,
+ image_name,
+ file_path,
+ disk="qcow2",
+ extra_properties={},
+ container="bare",
+ public="public"):
if not os.path.isfile(file_path):
logger.error("Error: file %s does not exist." % file_path)
return None
image = glance_client.images.create(name=image_name,
visibility=public,
disk_format=disk,
- container_format=container)
+ container_format=container,
+ **extra_properties)
image_id = image.id
with open(file_path) as image_data:
glance_client.images.upload(image_id, image_data)
return None
-def get_or_create_image(name, path, format):
+def get_or_create_image(name, path, format, extra_properties):
image_exists = False
glance_client = get_glance_client()
image_exists = True
else:
logger.info("Creating image '%s' from '%s'..." % (name, path))
- image_id = create_glance_image(glance_client, name, path, format)
+ image_id = create_glance_image(glance_client,
+ name,
+ path,
+ format,
+ extra_properties)
if not image_id:
logger.error("Failed to create a Glance image...")
else:
return None
-def list_volume_types(cinder_client, public=True, private=True):
- try:
- volume_types = cinder_client.volume_types.list()
- if not public:
- volume_types = [vt for vt in volume_types if not vt.is_public]
- if not private:
- volume_types = [vt for vt in volume_types if vt.is_public]
- return volume_types
- except Exception as e:
- logger.error("Error [list_volume_types(cinder_client)]: %s" % e)
- return None
-
-
-def create_volume_type(cinder_client, name):
- try:
- volume_type = cinder_client.volume_types.create(name)
- return volume_type
- except Exception as e:
- logger.error("Error [create_volume_type(cinder_client, '%s')]: %s"
- % (name, e))
- return None
-
-
def update_cinder_quota(cinder_client, tenant_id, vols_quota,
snapshots_quota, gigabytes_quota):
quotas_values = {"volumes": vols_quota,
return False
-def delete_volume_type(cinder_client, volume_type):
- try:
- cinder_client.volume_types.delete(volume_type)
- return True
- except Exception as e:
- logger.error("Error [delete_volume_type(cinder_client, '%s')]: %s"
- % (volume_type, e))
- return False
-
-
# *********************************************
# KEYSTONE
# *********************************************
def create_tenant(keystone_client, tenant_name, tenant_description):
try:
if is_keystone_v3():
- domain_name = CONST.__getattribute__('OS_PROJECT_DOMAIN_NAME')
+ domain_name = os.environ['OS_PROJECT_DOMAIN_NAME']
domain_id = get_domain_id(keystone_client, domain_name)
tenant = keystone_client.projects.create(
name=tenant_name,
Code Review / functest.git / blobdiff