from keystoneclient import client as keystoneclient
from neutronclient.neutron import client as neutronclient
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
logger = logging.getLogger(__name__)
'OS_PROJECT_NAME': 'project_name',
'OS_ENDPOINT_TYPE': 'endpoint_type',
'OS_REGION_NAME': 'region_name',
- 'OS_CACERT': 'https_cacert'
+ 'OS_CACERT': 'https_cacert',
+ 'OS_INSECURE': 'https_insecure'
}
return env_cred_dict
def source_credentials(rc_file):
with open(rc_file, "r") as f:
for line in f:
- var = line.rstrip('"\n').replace('export ', '').split("=")
+ 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 '
- key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
- value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
- os.environ[key] = value
+ if var:
+ key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
+ value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
+ os.environ[key] = value
def get_credentials_for_rally():
endpoint_types = [('internalURL', 'internal'),
('publicURL', 'public'), ('adminURL', 'admin')]
- endpoint_type = os.getenv('OS_ENDPOINT_TYPE')
+ 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 == k:
+ if endpoint_type == v:
rally_conf[cred_key] = v
region_name = os.getenv('OS_REGION_NAME')
cred_key = env_cred_dict.get('OS_REGION_NAME')
rally_conf[cred_key] = region_name
- cacert = os.getenv('OS_CACERT')
- if cacert is not None:
- cred_key = env_cred_dict.get('OS_CACERT')
- rally_conf[cred_key] = cacert
+ 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)
return auth
-def get_endpoint(service_type, endpoint_type='publicURL'):
+def get_endpoint(service_type, interface='public'):
auth = get_session_auth()
return get_session().get_endpoint(auth=auth,
service_type=service_type,
- endpoint_type=endpoint_type)
+ interface=interface)
def get_session(other_creds={}):
auth = get_session_auth(other_creds)
- cacert = os.getenv('OS_CACERT')
- if cacert is not None:
- if not os.path.isfile(cacert):
- raise Exception("The 'OS_CACERT' environment"
- "variable is set to %s but the file"
- "does not exist.", cacert)
-
- return session.Session(auth=auth, verify=cacert)
+ https_cacert = os.getenv('OS_CACERT', '')
+ https_insecure = os.getenv('OS_INSECURE', '').lower() == 'true'
+ return session.Session(auth=auth,
+ verify=(https_cacert or not https_insecure))
# *********************************************
def get_keystone_client(other_creds={}):
sess = get_session(other_creds)
- return keystoneclient.Client(get_keystone_client_version(), session=sess)
+ return keystoneclient.Client(get_keystone_client_version(),
+ session=sess,
+ interface=os.getenv('OS_INTERFACE', 'admin'))
def get_nova_client_version():
return heatclient.Client(get_heat_client_version(), session=sess)
+def download_and_add_image_on_glance(glance, image_name, image_url, data_dir):
+ try:
+ dest_path = data_dir
+ if not os.path.exists(dest_path):
+ os.makedirs(dest_path)
+ file_name = image_url.rsplit('/')[-1]
+ if not ft_utils.download_url(image_url, dest_path):
+ return False
+ except Exception:
+ raise Exception("Impossible to download image from {}".format(
+ image_url))
+
+ try:
+ image = create_glance_image(
+ glance, image_name, dest_path + file_name)
+ if not image:
+ return False
+ else:
+ return image
+ except Exception:
+ raise Exception("Impossible to put image {} in glance".format(
+ image_name))
+
+
# *********************************************
# NOVA
# *********************************************
try:
instances = nova_client.servers.list(search_opts={'all_tenants': 1})
return instances
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_instances(nova_client)]: %s" % e)
return None
try:
instance = nova_client.servers.get(instance.id)
return instance.status
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_instance_status(nova_client)]: %s" % e)
return None
try:
instance = nova_client.servers.find(name=instance_name)
return instance
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_instance_by_name(nova_client, '%s')]: %s"
% (instance_name, e))
return None
try:
aggregates = nova_client.aggregates.list()
return aggregates
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_aggregates(nova_client)]: %s" % e)
return None
aggregates = get_aggregates(nova_client)
_id = [ag.id for ag in aggregates if ag.name == aggregate_name][0]
return _id
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_aggregate_id(nova_client, %s)]:"
" %s" % (aggregate_name, e))
return None
try:
availability_zones = nova_client.availability_zones.list()
return availability_zones
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_availability_zones(nova_client)]: %s" % e)
return None
try:
az_names = [az.zoneName for az in get_availability_zones(nova_client)]
return az_names
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_availability_zone_names(nova_client)]:"
" %s" % e)
return None
# flavor extra specs are not configured, therefore skip the update
pass
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_flavor(nova_client, '%s', '%s', '%s', "
"'%s')]: %s" % (flavor_name, ram, disk, vcpus, e))
return None
flavor_id = create_flavor(
nova_client, flavor_name, ram, disk, vcpus, public=public)
if not flavor_id:
- logger.error("Failed to create flavor '%s'..." % (flavor_name))
+ raise Exception("Failed to create flavor '%s'..." % (flavor_name))
else:
logger.debug("Flavor '%s' with ID=%s created successfully."
% (flavor_name, flavor_id))
return flavor_exists, flavor_id
-def get_floating_ips(nova_client):
+def get_floating_ips(neutron_client):
try:
- floating_ips = nova_client.floating_ips.list()
- return floating_ips
- except Exception, e:
- logger.error("Error [get_floating_ips(nova_client)]: %s" % e)
+ floating_ips = neutron_client.list_floatingips()
+ return floating_ips['floatingips']
+ except Exception as e:
+ logger.error("Error [get_floating_ips(neutron_client)]: %s" % e)
return None
if hypervisor.state == "up":
nodes.append(hypervisor.hypervisor_hostname)
return nodes
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_hypervisors(nova_client)]: %s" % e)
return None
try:
nova_client.aggregates.create(aggregate_name, av_zone)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_aggregate(nova_client, %s, %s)]: %s"
% (aggregate_name, av_zone, e))
return None
aggregate_id = get_aggregate_id(nova_client, aggregate_name)
nova_client.aggregates.add_host(aggregate_id, compute_host)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [add_host_to_aggregate(nova_client, %s, %s)]: %s"
% (aggregate_name, compute_host, e))
return None
create_aggregate(nova_client, aggregate_name, av_zone)
add_host_to_aggregate(nova_client, aggregate_name, compute_host)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_aggregate_with_host("
"nova_client, %s, %s, %s)]: %s"
% (aggregate_name, av_zone, compute_host, e))
ip_json = neutron_client.create_floatingip({'floatingip': props})
fip_addr = ip_json['floatingip']['floating_ip_address']
fip_id = ip_json['floatingip']['id']
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_floating_ip(neutron_client)]: %s" % e)
return None
return {'fip_addr': fip_addr, 'fip_id': fip_id}
try:
nova_client.servers.add_floating_ip(server_id, floatingip_addr)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [add_floating_ip(nova_client, '%s', '%s')]: %s"
% (server_id, floatingip_addr, e))
return False
try:
nova_client.servers.force_delete(instance_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_instance(nova_client, '%s')]: %s"
% (instance_id, e))
return False
-def delete_floating_ip(nova_client, floatingip_id):
+def delete_floating_ip(neutron_client, floatingip_id):
try:
- nova_client.floating_ips.delete(floatingip_id)
+ neutron_client.delete_floatingip(floatingip_id)
return True
- except Exception, e:
- logger.error("Error [delete_floating_ip(nova_client, '%s')]: %s"
+ except Exception as e:
+ logger.error("Error [delete_floating_ip(neutron_client, '%s')]: %s"
% (floatingip_id, e))
return False
aggregate_id = get_aggregate_id(nova_client, aggregate_name)
nova_client.aggregates.remove_host(aggregate_id, compute_host)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [remove_host_from_aggregate(nova_client, %s, %s)]:"
" %s" % (aggregate_name, compute_host, e))
return False
remove_hosts_from_aggregate(nova_client, aggregate_name)
nova_client.aggregates.delete(aggregate_name)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_aggregate(nova_client, %s)]: %s"
% (aggregate_name, e))
return False
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']
network = neutron_client.create_network(body=json_body)
network_dict = network['network']
return network_dict['id']
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_neutron_net(neutron_client, '%s')]: %s"
% (name, e))
return None
-def create_neutron_subnet(neutron_client, name, cidr, net_id):
+def create_neutron_subnet(neutron_client, name, cidr, net_id,
+ dns=['8.8.8.8', '8.8.4.4']):
json_body = {'subnets': [{'name': name, 'cidr': cidr,
- 'ip_version': 4, 'network_id': net_id}]}
+ 'ip_version': 4, 'network_id': net_id,
+ 'dns_nameservers': dns}]}
+
try:
subnet = neutron_client.create_subnet(body=json_body)
return subnet['subnets'][0]['id']
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_neutron_subnet(neutron_client, '%s', "
"'%s', '%s')]: %s" % (name, cidr, net_id, e))
return None
try:
router = neutron_client.create_router(json_body)
return router['router']['id']
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_neutron_router(neutron_client, '%s')]: %s"
% (name, e))
return None
try:
port = neutron_client.create_port(body=json_body)
return port['port']['id']
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_neutron_port(neutron_client, '%s', '%s', "
"'%s')]: %s" % (name, network_id, ip, e))
return None
try:
neutron_client.update_network(network_id, body=json_body)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [update_neutron_net(neutron_client, '%s', '%s')]: "
"%s" % (network_id, str(shared), e))
return False
port = neutron_client.update_port(port=port_id,
body=json_body)
return port['port']['id']
- except Exception, e:
+ except Exception as e:
logger.error("Error [update_neutron_port(neutron_client, '%s', '%s')]:"
" %s" % (port_id, device_owner, e))
return None
try:
neutron_client.add_interface_router(router=router_id, body=json_body)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [add_interface_router(neutron_client, '%s', "
"'%s')]: %s" % (router_id, subnet_id, e))
return False
try:
neutron_client.add_gateway_router(router_id, router_dict)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [add_gateway_router(neutron_client, '%s')]: %s"
% (router_id, e))
return False
try:
neutron_client.delete_network(network_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_neutron_net(neutron_client, '%s')]: %s"
% (network_id, e))
return False
try:
neutron_client.delete_subnet(subnet_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_neutron_subnet(neutron_client, '%s')]: %s"
% (subnet_id, e))
return False
try:
neutron_client.delete_router(router=router_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_neutron_router(neutron_client, '%s')]: %s"
% (router_id, e))
return False
try:
neutron_client.delete_port(port_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_neutron_port(neutron_client, '%s')]: %s"
% (port_id, e))
return False
neutron_client.remove_interface_router(router=router_id,
body=json_body)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [remove_interface_router(neutron_client, '%s', "
"'%s')]: %s" % (router_id, subnet_id, e))
return False
try:
neutron_client.remove_gateway_router(router_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [remove_gateway_router(neutron_client, '%s')]: %s"
% (router_id, e))
return False
net_name,
subnet_name,
router_name,
- cidr):
+ cidr,
+ dns=['8.8.8.8', '8.8.4.4']):
# Check if the network already exists
network_id = get_network_id(neutron_client, net_name)
logger.debug("Network '%s' created successfully" % network_id)
logger.debug('Creating Subnet....')
subnet_id = create_neutron_subnet(neutron_client, subnet_name,
- cidr, network_id)
+ cidr, network_id, dns)
if not subnet_id:
return None
return network_dic
-def create_bgpvpn(neutron_client, **kwargs):
- # route_distinguishers
- # route_targets
- json_body = {"bgpvpn": kwargs}
- return neutron_client.create_bgpvpn(json_body)
-
-
-def create_network_association(neutron_client, bgpvpn_id, neutron_network_id):
- json_body = {"network_association": {"network_id": neutron_network_id}}
- return neutron_client.create_network_association(bgpvpn_id, json_body)
-
-
-def create_router_association(neutron_client, bgpvpn_id, router_id):
- json_body = {"router_association": {"router_id": router_id}}
- return neutron_client.create_router_association(bgpvpn_id, json_body)
-
-
-def update_bgpvpn(neutron_client, bgpvpn_id, **kwargs):
- json_body = {"bgpvpn": kwargs}
- return neutron_client.update_bgpvpn(bgpvpn_id, json_body)
-
-
-def delete_bgpvpn(neutron_client, bgpvpn_id):
- return neutron_client.delete_bgpvpn(bgpvpn_id)
-
-
-def get_bgpvpn(neutron_client, bgpvpn_id):
- return neutron_client.show_bgpvpn(bgpvpn_id)
-
-
-def get_bgpvpn_routers(neutron_client, bgpvpn_id):
- return get_bgpvpn(neutron_client, bgpvpn_id)['bgpvpn']['routers']
-
-
-def get_bgpvpn_networks(neutron_client, bgpvpn_id):
- return get_bgpvpn(neutron_client, bgpvpn_id)['bgpvpn']['networks']
-
# *********************************************
# SEC GROUPS
# *********************************************
security_groups = neutron_client.list_security_groups()[
'security_groups']
return security_groups
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_security_groups(neutron_client)]: %s" % e)
return None
try:
secgroup = neutron_client.create_security_group(json_body)
return secgroup['security_group']
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_security_group(neutron_client, '%s', "
"'%s')]: %s" % (sg_name, sg_description, e))
return None
security_rules = [rule for rule in security_rules
if rule["security_group_id"] == sg_id]
return security_rules
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_security_group_rules(neutron_client, sg_id)]:"
" %s" % e)
return None
try:
security_rules = get_security_group_rules(neutron_client, sg_id)
security_rules = [rule for rule in security_rules
- if (rule["direction"].lower() == direction
- and rule["protocol"].lower() == protocol
- and rule["port_range_min"] == port_min
- and rule["port_range_max"] == port_max)]
+ if (rule["direction"].lower() == direction and
+ rule["protocol"].lower() == protocol and
+ rule["port_range_min"] == port_min and
+ rule["port_range_max"] == port_max)]
if len(security_rules) == 0:
return True
else:
return False
- except Exception, e:
+ except Exception as e:
logger.error("Error [check_security_group_rules("
" neutron_client, sg_id, direction,"
" protocol, port_min=None, port_max=None)]: "
try:
nova_client.servers.add_security_group(instance_id, secgroup_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [add_secgroup_to_instance(nova_client, '%s', "
"'%s')]: %s" % (instance_id, secgroup_id, e))
return False
neutron_client.update_quota(tenant_id=tenant_id,
body=json_body)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [update_sg_quota(neutron_client, '%s', '%s', "
"'%s')]: %s" % (tenant_id, sg_quota, sg_rule_quota, e))
return False
try:
neutron_client.delete_security_group(secgroup_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_security_group(neutron_client, '%s')]: %s"
% (secgroup_id, e))
return False
# *********************************************
# GLANCE
# *********************************************
-def get_images(nova_client):
+def get_images(glance_client):
try:
- images = nova_client.images.list()
+ images = glance_client.images.list()
return images
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_images]: %s" % e)
return None
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 image_id
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_glance_image(glance_client, '%s', '%s', "
"'%s')]: %s" % (image_name, file_path, public, e))
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 image_exists, image_id
-def delete_glance_image(nova_client, image_id):
+def delete_glance_image(glance_client, image_id):
try:
- nova_client.images.delete(image_id)
+ glance_client.images.delete(image_id)
return True
- except Exception, e:
- logger.error("Error [delete_glance_image(nova_client, '%s')]: %s"
+ except Exception as e:
+ logger.error("Error [delete_glance_image(glance_client, '%s')]: %s"
% (image_id, e))
return False
try:
volumes = cinder_client.volumes.list(search_opts={'all_tenants': 1})
return volumes
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_volumes(cinder_client)]: %s" % e)
return None
if not private:
volume_types = [vt for vt in volume_types if vt.is_public]
return volume_types
- except Exception, e:
+ except Exception as e:
logger.error("Error [list_volume_types(cinder_client)]: %s" % e)
return None
try:
volume_type = cinder_client.volume_types.create(name)
return volume_type
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_volume_type(cinder_client, '%s')]: %s"
% (name, e))
return None
try:
cinder_client.quotas.update(tenant_id, **quotas_values)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [update_cinder_quota(cinder_client, '%s', '%s', "
"'%s' '%s')]: %s" % (tenant_id, vols_quota,
snapshots_quota, gigabytes_quota, e))
else:
cinder_client.volumes.delete(volume_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_volume(cinder_client, '%s', '%s')]: %s"
% (volume_id, str(forced), e))
return False
try:
cinder_client.volume_types.delete(volume_type)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_volume_type(cinder_client, '%s')]: %s"
% (volume_type, e))
return False
else:
tenants = keystone_client.tenants.list()
return tenants
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_tenants(keystone_client)]: %s" % e)
return None
try:
users = keystone_client.users.list()
return users
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_users(keystone_client)]: %s" % e)
return None
return id
+def get_domain_id(keystone_client, domain_name):
+ domains = keystone_client.domains.list()
+ id = ''
+ for d in domains:
+ if d.name == domain_name:
+ id = d.id
+ break
+ return id
+
+
def create_tenant(keystone_client, tenant_name, tenant_description):
try:
if is_keystone_v3():
+ domain_name = CONST.__getattribute__('OS_PROJECT_DOMAIN_NAME')
+ domain_id = get_domain_id(keystone_client, domain_name)
tenant = keystone_client.projects.create(
name=tenant_name,
description=tenant_description,
- domain="default",
+ domain=domain_id,
enabled=True)
else:
tenant = keystone_client.tenants.create(tenant_name,
tenant_description,
enabled=True)
return tenant.id
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_tenant(keystone_client, '%s', '%s')]: %s"
% (tenant_name, tenant_description, e))
return None
return tenant_id
+def get_or_create_tenant_for_vnf(keystone_client, tenant_name,
+ tenant_description):
+ """Get or Create a Tenant
+
+ Args:
+ keystone_client: keystone client reference
+ tenant_name: the name of the tenant
+ tenant_description: the description of the tenant
+
+ return False if tenant retrieved though get
+ return True if tenant created
+ raise Exception if error during processing
+ """
+ try:
+ tenant_id = get_tenant_id(keystone_client, tenant_name)
+ if not tenant_id:
+ tenant_id = create_tenant(keystone_client, tenant_name,
+ tenant_description)
+ return True
+ else:
+ return False
+ except:
+ raise Exception("Impossible to create a Tenant for the VNF {}".format(
+ tenant_name))
+
+
def create_user(keystone_client, user_name, user_password,
user_email, tenant_id):
try:
tenant_id,
enabled=True)
return user.id
- except Exception, e:
+ except Exception as e:
logger.error("Error [create_user(keystone_client, '%s', '%s', '%s'"
"'%s')]: %s" % (user_name, user_password,
user_email, tenant_id, e))
return user_id
+def get_or_create_user_for_vnf(keystone_client, vnf_ref):
+ """Get or Create user for VNF
+
+ Args:
+ keystone_client: keystone client reference
+ vnf_ref: VNF reference used as user name & password, tenant name
+
+ return False if user retrieved through get
+ return True if user created
+ raise Exception if error during processing
+ """
+ try:
+ user_id = get_user_id(keystone_client, vnf_ref)
+ tenant_id = get_tenant_id(keystone_client, vnf_ref)
+ created = False
+ if not user_id:
+ user_id = create_user(keystone_client, vnf_ref, vnf_ref,
+ "", tenant_id)
+ created = True
+ try:
+ role_id = get_role_id(keystone_client, 'admin')
+ tenant_id = get_tenant_id(keystone_client, vnf_ref)
+ add_role_user(keystone_client, user_id, role_id, tenant_id)
+ except:
+ logger.warn("Cannot associate user to role admin on tenant")
+ return created
+ except:
+ raise Exception("Impossible to create a user for the VNF {}".format(
+ vnf_ref))
+
+
def add_role_user(keystone_client, user_id, role_id, tenant_id):
try:
if is_keystone_v3():
else:
keystone_client.roles.add_user_role(user_id, role_id, tenant_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [add_role_user(keystone_client, '%s', '%s'"
"'%s')]: %s " % (user_id, role_id, tenant_id, e))
return False
else:
keystone_client.tenants.delete(tenant_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_tenant(keystone_client, '%s')]: %s"
% (tenant_id, e))
return False
try:
keystone_client.users.delete(user_id)
return True
- except Exception, e:
+ except Exception as e:
logger.error("Error [delete_user(keystone_client, '%s')]: %s"
% (user_id, e))
return False
try:
resources = heat_client.resources.get(stack_id, resource)
return resources
- except Exception, e:
+ except Exception as e:
logger.error("Error [get_resource]: %s" % e)
return None
+
+
+# *********************************************
+# TEMPEST
+# *********************************************
+def init_tempest_cleanup(tempest_config_dir=None,
+ tempest_config_filename='tempest.conf',
+ output_file=None):
+ """
+ Initialize the Tempest Cleanup utility.
+ See https://docs.openstack.org/tempest/latest/cleanup.html for docs.
+
+ :param tempest_config_dir: The directory where the Tempest config file is
+ located. If not specified, we let Tempest pick both the directory
+ and the filename (i.e. second parameter is ignored)
+ :param tempest_config_filename: The filename of the Tempest config file
+ :param output_file: Optional file where to save output
+ """
+ # The Tempest cleanup utility currently offers no cmd argument to specify
+ # the config file, therefore it has to be configured with env variables
+ env = None
+ if tempest_config_dir:
+ env = os.environ.copy()
+ env['TEMPEST_CONFIG_DIR'] = tempest_config_dir
+ env['TEMPEST_CONFIG'] = tempest_config_filename
+
+ # If this command fails, an exception must be raised to stop the script
+ # otherwise the later cleanup would destroy also other resources
+ cmd_line = "tempest cleanup --init-saved-state"
+ ft_utils.execute_command_raise(cmd_line, env=env, output_file=output_file,
+ error_msg="Tempest cleanup init failed")
+
+
+def perform_tempest_cleanup(tempest_config_dir=None,
+ tempest_config_filename='tempest.conf',
+ output_file=None):
+ """
+ Perform cleanup using the Tempest Cleanup utility.
+ See https://docs.openstack.org/tempest/latest/cleanup.html for docs.
+
+ :param tempest_config_dir: The directory where the Tempest config file is
+ located. If not specified, we let Tempest pick both the directory
+ and the filename (i.e. second parameter is ignored)
+ :param tempest_config_filename: The filename of the Tempest config file
+ :param output_file: Optional file where to save output
+ """
+ # The Tempest cleanup utility currently offers no cmd argument to specify
+ # the config file, therefore it has to be configured with env variables
+ env = None
+ if tempest_config_dir:
+ env = os.environ.copy()
+ env['TEMPEST_CONFIG_DIR'] = tempest_config_dir
+ env['TEMPEST_CONFIG'] = tempest_config_filename
+
+ # If this command fails, an exception must be raised to stop the script
+ # otherwise the later cleanup would destroy also other resources
+ cmd_line = "tempest cleanup"
+ ft_utils.execute_command(cmd_line, env=env, output_file=output_file,
+ error_msg="Tempest cleanup failed")
Code Review / functest.git / blobdiff