from cinderclient import client as cinderclient
import functest.utils.functest_logger as ft_logger
+import functest.utils.functest_utils as ft_utils
from glanceclient import client as glanceclient
from keystoneclient.v2_0 import client as keystoneclient
from neutronclient.v2_0 import client as neutronclient
def create_flavor(nova_client, flavor_name, ram, disk, vcpus):
try:
flavor = nova_client.flavors.create(flavor_name, ram, vcpus, disk)
+ try:
+ extra_specs = ft_utils.get_functest_config(
+ 'general.flavor_extra_specs')
+ flavor.update(extra_specs)
+ except ValueError:
+ # flavor extra specs are not configured, therefore skip the update
+ pass
+
except Exception, e:
logger.error("Error [create_flavor(nova_client, '%s', '%s', '%s', "
"'%s')]: %s" % (flavor_name, ram, disk, vcpus, e))
return network_dic
+def create_shared_network_full(net_name, subnt_name, router_name, subnet_cidr):
+ neutron_client = get_neutron_client()
+
+ network_dic = create_network_full(neutron_client,
+ net_name,
+ subnt_name,
+ router_name,
+ subnet_cidr)
+ if network_dic:
+ if not update_neutron_net(neutron_client,
+ network_dic['net_id'],
+ shared=True):
+ logger.error("Failed to update network %s..." % net_name)
+ return None
+ else:
+ logger.debug("Network '%s' is available..." % net_name)
+ else:
+ logger.error("Network %s creation failed" % net_name)
+ return None
+ return network_dic
+
+
def create_bgpvpn(neutron_client, **kwargs):
# route_distinguishers
# route_targets
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
# *********************************************
if logger:
logger.info("Creating image '%s' from '%s'..." % (image_name,
file_path))
+ try:
+ properties = ft_utils.get_functest_config(
+ 'general.image_properties')
+ except ValueError:
+ # image properties are not configured
+ # therefore don't add any properties
+ properties = {}
with open(file_path) as fimage:
image = glance_client.images.create(name=image_name,
is_public=public,
disk_format=disk,
container_format=container,
+ properties=properties,
data=fimage)
image_id = image.id
return image_id
return None
+def get_or_create_image(name, path, format):
+ image_exists = False
+ glance_client = get_glance_client()
+
+ image_id = get_image_id(glance_client, name)
+ if image_id != '':
+ logger.info("Using existing image '%s'..." % name)
+ image_exists = True
+ else:
+ logger.info("Creating image '%s' from '%s'..." % (name, path))
+ image_id = create_glance_image(glance_client, name, path, format)
+ if not image_id:
+ logger.error("Failed to create a Glance image...")
+ else:
+ logger.debug("Image '%s' with ID=%s created successfully."
+ % (name, image_id))
+
+ return image_exists, image_id
+
+
def delete_glance_image(nova_client, image_id):
try:
nova_client.images.delete(image_id)
enabled=True)
return tenant.id
except Exception, e:
- logger.error("Error [create_tenant(cinder_client, '%s', '%s')]: %s"
+ logger.error("Error [create_tenant(keystone_client, '%s', '%s')]: %s"
% (tenant_name, tenant_description, e))
return None