TEST_DB = functest_yaml.get("results").get("test_db_url")
NAME_VM_1 = functest_yaml.get("vping").get("vm_name_1")
NAME_VM_2 = functest_yaml.get("vping").get("vm_name_2")
-IP_1 = functest_yaml.get("vping").get("ip_1")
-IP_2 = functest_yaml.get("vping").get("ip_2")
# GLANCE_IMAGE_NAME = functest_yaml.get("general"). \
# get("openstack").get("image_name")
GLANCE_IMAGE_NAME = "functest-vping"
NEUTRON_ROUTER_NAME = functest_yaml.get("vping"). \
get("vping_router_name")
+SECGROUP_NAME = "vPing-sg"
+SECGROUP_DESCR = "Security group for vPing test case"
def pMsg(value):
def create_private_neutron_net(neutron):
- neutron.format = 'json'
- logger.info('Creating neutron network %s...' % NEUTRON_PRIVATE_NET_NAME)
- network_id = functest_utils. \
- create_neutron_net(neutron, NEUTRON_PRIVATE_NET_NAME)
+ # Check if the network already exists
+ network_id = functest_utils.get_network_id(neutron,NEUTRON_PRIVATE_NET_NAME)
+ subnet_id = functest_utils.get_subnet_id(neutron,NEUTRON_PRIVATE_SUBNET_NAME)
+ router_id = functest_utils.get_router_id(neutron,NEUTRON_ROUTER_NAME)
- if not network_id:
- return False
- logger.debug("Network '%s' created successfully" % network_id)
- logger.debug('Creating Subnet....')
- subnet_id = functest_utils. \
- create_neutron_subnet(neutron,
- NEUTRON_PRIVATE_SUBNET_NAME,
- NEUTRON_PRIVATE_SUBNET_CIDR,
- network_id)
- if not subnet_id:
- return False
- logger.debug("Subnet '%s' created successfully" % subnet_id)
- logger.debug('Creating Router...')
- router_id = functest_utils. \
- create_neutron_router(neutron, NEUTRON_ROUTER_NAME)
+ if network_id != '' and subnet_id != '' and router_id != '' :
+ logger.info("Using existing network '%s'..." % NEUTRON_PRIVATE_NET_NAME)
+ else:
+ neutron.format = 'json'
+ logger.info('Creating neutron network %s...' % NEUTRON_PRIVATE_NET_NAME)
+ network_id = functest_utils. \
+ create_neutron_net(neutron, NEUTRON_PRIVATE_NET_NAME)
- if not router_id:
- return False
+ if not network_id:
+ return False
+ logger.debug("Network '%s' created successfully" % network_id)
+ logger.debug('Creating Subnet....')
+ subnet_id = functest_utils. \
+ create_neutron_subnet(neutron,
+ NEUTRON_PRIVATE_SUBNET_NAME,
+ NEUTRON_PRIVATE_SUBNET_CIDR,
+ network_id)
+ if not subnet_id:
+ return False
+ logger.debug("Subnet '%s' created successfully" % subnet_id)
+ logger.debug('Creating Router...')
+ router_id = functest_utils. \
+ create_neutron_router(neutron, NEUTRON_ROUTER_NAME)
- logger.debug("Router '%s' created successfully" % router_id)
- logger.debug('Adding router to subnet...')
+ if not router_id:
+ return False
- if not functest_utils.add_interface_router(neutron, router_id, subnet_id):
- return False
- logger.debug("Interface added successfully.")
+ logger.debug("Router '%s' created successfully" % router_id)
+ logger.debug('Adding router to subnet...')
- logger.debug('Adding gateway to router...')
- if not functest_utils.add_gateway_router(neutron, router_id):
- return False
- logger.debug("Gateway added successfully.")
+ if not functest_utils.add_interface_router(neutron, router_id, subnet_id):
+ return False
+ logger.debug("Interface added successfully.")
+
+ logger.debug('Adding gateway to router...')
+ if not functest_utils.add_gateway_router(neutron, router_id):
+ return False
+ logger.debug("Gateway added successfully.")
network_dic = {'net_id': network_id,
'subnet_id': subnet_id,
'router_id': router_id}
return network_dic
+def create_security_group(neutron_client):
+ sg_id = functest_utils.get_security_group_id(neutron_client, SECGROUP_NAME)
+ if sg_id != '':
+ logger.info("Using existing security group '%s'..." % SECGROUP_NAME)
+ else:
+ logger.info("Creating security group '%s'..." % SECGROUP_NAME)
+ SECGROUP = functest_utils.create_security_group(neutron_client,
+ SECGROUP_NAME,
+ SECGROUP_DESCR)
+ if not SECGROUP:
+ logger.error("Failed to create the security group...")
+ return False
+
+ sg_id = SECGROUP['id']
+
+ logger.debug("Security group '%s' with ID=%s created successfully." %\
+ (SECGROUP['name'], sg_id))
+
+ logger.debug("Adding ICMP rules in security group '%s'..." % SECGROUP_NAME)
+ if not functest_utils.create_secgroup_rule(neutron_client, sg_id, \
+ 'ingress', 'icmp'):
+ logger.error("Failed to create the security group rule...")
+ return False
+
+ logger.debug("Adding SSH rules in security group '%s'..." % SECGROUP_NAME)
+ if not functest_utils.create_secgroup_rule(neutron_client, sg_id, \
+ 'ingress', 'tcp', '22', '22'):
+ logger.error("Failed to create the security group rule...")
+ return False
+
+ if not functest_utils.create_secgroup_rule(neutron_client, sg_id, \
+ 'egress', 'tcp', '22', '22'):
+ logger.error("Failed to create the security group rule...")
+ return False
+ return sg_id
-def cleanup(nova, neutron, image_id, network_dic, port_id1, port_id2, secgroup_id):
+def cleanup(nova, neutron, image_id, network_dic, sg_id):
if args.noclean:
logger.debug("The OpenStack resources are not deleted.")
return True
subnet_id = network_dic["subnet_id"]
router_id = network_dic["router_id"]
- if not functest_utils.delete_neutron_port(neutron, port_id1):
- logger.error("Unable to remove port '%s'" % port_id1)
- return False
- logger.debug("Port '%s' removed successfully" % port_id1)
-
- if not functest_utils.delete_neutron_port(neutron, port_id2):
- logger.error("Unable to remove port '%s'" % port_id2)
- return False
- logger.debug("Port '%s' removed successfully" % port_id2)
-
if not functest_utils.remove_interface_router(neutron, router_id,
subnet_id):
logger.error("Unable to remove subnet '%s' from router '%s'" % (
logger.debug(
"Network '%s' deleted successfully" % NEUTRON_PRIVATE_NET_NAME)
- if not functest_utils.delete_security_group(neutron, secgroup_id):
- logger.error("Unable to delete security group '%s'" % secgroup_id)
+ if not functest_utils.delete_security_group(neutron, sg_id):
+ logger.error("Unable to delete security group '%s'" % sg_id)
return False
logger.debug(
- "Security group '%s' deleted successfully" % secgroup_id)
+ "Security group '%s' deleted successfully" % sg_id)
return True
token=keystone_client.auth_token)
EXIT_CODE = -1
- image = None
+ image_id = None
flavor = None
- logger.debug("Creating image '%s' from '%s'..." % (GLANCE_IMAGE_NAME,
+ # Check if the given image exists
+ image_id = functest_utils.get_image_id(glance_client, GLANCE_IMAGE_NAME)
+ if image_id != '':
+ logger.info("Using existing image '%s'..." % GLANCE_IMAGE_NAME)
+ else:
+ logger.info("Creating image '%s' from '%s'..." % (GLANCE_IMAGE_NAME,
GLANCE_IMAGE_PATH))
- image_id = functest_utils.create_glance_image(glance_client,
- GLANCE_IMAGE_NAME,
- GLANCE_IMAGE_PATH)
- if not image_id:
- logger.error("Failed to create a Glance image...")
- return(EXIT_CODE)
- logger.debug("Image '%s' with ID=%s created successfully." %\
+ image_id = functest_utils.create_glance_image(glance_client,
+ GLANCE_IMAGE_NAME,
+ GLANCE_IMAGE_PATH)
+ if not image_id:
+ logger.error("Failed to create a Glance image...")
+ return(EXIT_CODE)
+ logger.debug("Image '%s' with ID=%s created successfully." %\
(GLANCE_IMAGE_NAME, image_id))
- # Check if the given image exists
- image = functest_utils.get_image_id(glance_client, GLANCE_IMAGE_NAME)
- if image == '':
- logger.error("ERROR: Glance image '%s' not found." % GLANCE_IMAGE_NAME)
- logger.info("Available images are: ")
- pMsg(nova_client.images.list())
- return(EXIT_CODE)
+
network_dic = create_private_neutron_net(neutron_client)
+
if not network_dic:
logger.error(
"There has been a problem when creating the neutron network")
network_id = network_dic["net_id"]
- # Check if the given flavor exists
+ sg_id = create_security_group(neutron_client)
+ # Check if the given flavor exists
try:
flavor = nova_client.flavors.find(name=FLAVOR)
- logger.info("Flavor found '%s'" % FLAVOR)
+ logger.info("Using existing Flavor '%s'" % FLAVOR)
except:
logger.error("Flavor '%s' not found." % FLAVOR)
logger.info("Available flavors are: ")
logger.info("Instance %s found. Deleting..." % server.name)
server.delete()
- SECGROUP_NAME = "vPing-sg"
- SECGROUP_DESCR = "Security group for vPing test case"
- logger.info("Creating security group '%s'..." % SECGROUP_NAME)
- SECGROUP = functest_utils.create_security_group(neutron_client,
- SECGROUP_NAME,
- SECGROUP_DESCR)
- SECGROUP_ID = SECGROUP['id']
- if not SECGROUP:
- logger.error("Failed to create the security group...")
- return(EXIT_CODE)
- logger.debug("Security group '%s' with ID=%s created successfully." %\
- (SECGROUP['name'], SECGROUP_ID))
-
- logger.debug("Adding ICMP rules in security group '%s'..." % SECGROUP_NAME)
- if not functest_utils.create_secgroup_rule(neutron_client, SECGROUP_ID, \
- 'ingress', 'icmp'):
- logger.error("Failed to create the security group rule...")
- return(EXIT_CODE)
-
- logger.debug("Adding SSH rules in security group '%s'..." % SECGROUP_NAME)
- if not functest_utils.create_secgroup_rule(neutron_client, SECGROUP_ID, \
- 'ingress', 'tcp', '22', '22'):
- logger.error("Failed to create the security group rule...")
- return(EXIT_CODE)
-
- if not functest_utils.create_secgroup_rule(neutron_client, SECGROUP_ID, \
- 'egress', 'tcp', '22', '22'):
- logger.error("Failed to create the security group rule...")
- return(EXIT_CODE)
# boot VM 1
# basic boot
'%Y-%m-%d %H:%M:%S')))
# create VM
- logger.debug("Creating port 'vping-port-1' with IP %s..." % IP_1)
- port_id1 = functest_utils.create_neutron_port(neutron_client,
- "vping-port-1", network_id,
- IP_1)
- if not port_id1:
- logger.error("Unable to create port.")
- return(EXIT_CODE)
-
- logger.info("Creating instance '%s' with IP %s..." % (NAME_VM_1, IP_1))
+ logger.info("Creating instance '%s'..." % NAME_VM_1)
logger.debug(
"Configuration:\n name=%s \n flavor=%s \n image=%s \n "
- "network=%s \n" % (NAME_VM_1, flavor, image, network_id))
+ "network=%s \n" % (NAME_VM_1, flavor, image_id, network_id))
vm1 = nova_client.servers.create(
name=NAME_VM_1,
flavor=flavor,
- image=image,
- # nics = [{"net-id": network_id, "v4-fixed-ip": IP_1}]
- nics=[{"port-id": port_id1}]
+ image=image_id,
+ nics=[{"net-id": network_id}]
)
# wait until VM status is active
if not waitVmActive(nova_client, vm1):
logger.error("Instance '%s' cannot be booted. Status is '%s'" % (
NAME_VM_1, functest_utils.get_instance_status(nova_client, vm1)))
- cleanup(nova_client, neutron_client, image_id, network_dic,
- port_id1, SECGROUP_ID)
+ cleanup(nova_client, neutron_client, image_id, network_dic, sg_id)
return (EXIT_CODE)
else:
logger.info("Instance '%s' is ACTIVE." % NAME_VM_1)
# theoretically there is only one IP address so we take the
# first element of the table
# Dangerous! To be improved!
- # test_ip = server.networks.get(NEUTRON_PRIVATE_NET_NAME)[0]
- test_ip = IP_1
+ test_ip = vm1.networks.get(NEUTRON_PRIVATE_NET_NAME)[0]
logger.debug("Instance '%s' got %s" % (NAME_VM_1, test_ip))
logger.info("Adding '%s' to security group '%s'..." % (NAME_VM_1, SECGROUP_NAME))
- functest_utils.add_secgroup_to_instance(nova_client, vm1.id, SECGROUP['id'])
+ functest_utils.add_secgroup_to_instance(nova_client, vm1.id, sg_id)
# boot VM 2
# we will boot then execute a ping script with cloud-init
# create VM
- logger.debug("Creating port 'vping-port-2' with IP %s..." % IP_2)
- port_id2 = functest_utils.create_neutron_port(neutron_client,
- "vping-port-2", network_id,
- IP_2)
-
- if not port_id2:
- logger.error("Unable to create port.")
- return(EXIT_CODE)
-
- logger.info("Creating instance '%s' with IP %s..." % (NAME_VM_2, IP_2))
+ logger.info("Creating instance '%s'..." % NAME_VM_2)
logger.debug(
"Configuration:\n name=%s \n flavor=%s \n image=%s \n "
- "network=%s \n" % (NAME_VM_2, flavor, image, network_id))
+ "network=%s \n" % (NAME_VM_2, flavor, image_id, network_id))
vm2 = nova_client.servers.create(
name=NAME_VM_2,
flavor=flavor,
- image=image,
- nics=[{"port-id": port_id2}]
+ image=image_id,
+ nics=[{"net-id": network_id}]
)
if not waitVmActive(nova_client, vm2):
logger.error("Instance '%s' cannot be booted. Status is '%s'" % (
NAME_VM_2, functest_utils.get_instance_status(nova_client, vm2)))
- cleanup(nova_client, neutron_client, image_id, network_dic,
- port_id1, port_id2, SECGROUP_ID)
+ cleanup(nova_client, neutron_client, image_id, network_dic, sg_id)
return (EXIT_CODE)
else:
logger.info("Instance '%s' is ACTIVE." % NAME_VM_2)
logger.info("Adding '%s' to security group '%s'..." % (NAME_VM_2, SECGROUP_NAME))
- functest_utils.add_secgroup_to_instance(nova_client, vm2.id, SECGROUP['id'])
+ functest_utils.add_secgroup_to_instance(nova_client, vm2.id, sg_id)
logger.info("Creating floating IP for VM '%s'..." % NAME_VM_2)
floatip = functest_utils.create_floating_ip(neutron_client)
if floatip == None:
logger.error("Cannot create floating IP.")
- cleanup(nova_client, neutron_client, image_id, network_dic,
- port_id1, port_id2, SECGROUP_ID)
+ cleanup(nova_client, neutron_client, image_id, network_dic, sg_id)
return (EXIT_CODE)
logger.info("Floating IP created: '%s'" % floatip)
logger.info("Associating floating ip: '%s' to VM '%s' " % (floatip, NAME_VM_2))
if not functest_utils.add_floating_ip(nova_client, vm2.id, floatip):
logger.error("Cannot associate floating IP to VM.")
- cleanup(nova_client, neutron_client, image_id, network_dic,
- port_id1, port_id2, SECGROUP_ID)
+ cleanup(nova_client, neutron_client, image_id, network_dic, sg_id)
return (EXIT_CODE)
logger.info("Trying to establish SSH connection to %s..." % floatip)
if timeout == 0: # 300 sec timeout (5 min)
logger.error("Cannot establish connection to IP '%s'. Aborting" % floatip)
- cleanup(nova_client, neutron_client, image_id, network_dic,
- port_id1, port_id2, SECGROUP_ID)
+ cleanup(nova_client, neutron_client, image_id, network_dic, sg_id)
return (EXIT_CODE)
scp = SCPClient(ssh.get_transport())
logger.error("Cannot SCP the file '%s' to VM '%s'" % (ping_script,floatip))
- #SSH_OPTS="-o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no "+\
- # "-o LogLevel=quiet"
-
- #cmd1 = "sshpass -p 'cubswin:)' scp " + SSH_OPTS + " " + \
- # ping_script + " cirros@"+floatip+":~/ping.sh"
- #cmd2 = "sshpass -p 'cubswin:)' ssh " + SSH_OPTS + \
- # " cirros@"+floatip+" 'chmod 755 ~/ping.sh '"
- #cmd3 = "sshpass -p 'cubswin:)' ssh " + SSH_OPTS + \
- # " cirros@"+floatip+" '~/ping.sh "+IP_1+"'"
-
cmd = 'chmod 755 ~/ping.sh'
(stdin, stdout, stderr) = ssh.exec_command(cmd)
for line in stdout.readlines():
sec = 0
duration = 0
- cmd = '~/ping.sh ' + IP_1
+ cmd = '~/ping.sh ' + test_ip
flag = False
while True:
time.sleep(1)
break
if flag :
break
- logger.debug("Pinging %s. Waiting for response..." % IP_1)
+ logger.debug("Pinging %s. Waiting for response..." % test_ip)
sec += 1
- cleanup(nova_client, neutron_client, image_id, network_dic,
- port_id1, port_id2, SECGROUP_ID)
+ cleanup(nova_client, neutron_client, image_id, network_dic, sg_id)
test_status = "NOK"
if EXIT_CODE == 0:
Code Review / functest.git / commitdiff