import re
import sys
import time
+import yaml
import novaclient.v2.client as novaclient
from neutronclient.v2_0 import client as neutronclient
from cinderclient import client as cinderclient
parser = argparse.ArgumentParser()
-parser.add_argument("repo_path", help="Path to the repository")
parser.add_argument("-d", "--debug", help="Debug mode", action="store_true")
args = parser.parse_args()
else:
ch.setLevel(logging.INFO)
-sys.path.append(args.repo_path + "testcases/")
-import functest_utils
-
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
logger.addHandler(ch)
-default_images = ['TestVM']
-default_networks = ['net04', 'net04_ext']
-default_routers = ['router04']
-default_users = ["heat", "heat-cfn", "cinder", "nova", "swift", "glance",
- "neutron", "admin", "fuel_stats_user"]
-default_tenants = ["admin", "services","service"]
-default_security_groups = ['default']
+REPO_PATH=os.environ['repos_dir']+'/functest/'
+if not os.path.exists(REPO_PATH):
+ logger.error("Functest repository directory not found '%s'" % REPO_PATH)
+ exit(-1)
+sys.path.append(REPO_PATH + "testcases/")
+import functest_utils
+
+with open(REPO_PATH+"testcases/VIM/OpenStack/CI/libraries/os_defaults.yaml") as f:
+ defaults_yaml = yaml.safe_load(f)
+f.close()
+
+installer = os.environ["INSTALLER_TYPE"]
+
+default_images = defaults_yaml.get(installer).get("images")
+default_networks = defaults_yaml.get(installer).get("networks") +\
+ defaults_yaml.get("common").get("networks")
+default_routers = defaults_yaml.get(installer).get("routers") +\
+ defaults_yaml.get("common").get("routers")
+default_security_groups = defaults_yaml.get(installer).get("security_groups")
+default_users = defaults_yaml.get(installer).get("users")
+default_tenants = defaults_yaml.get(installer).get("tenants")
def separator():
logger.debug("-------------------------------------------")
def remove_instances(nova_client):
logger.info("Removing Nova instances...")
instances = functest_utils.get_instances(nova_client)
- if len(instances) == 0:
+ if instances is None or len(instances) == 0:
logger.debug("No instances found.")
return
logger.info(" > ERROR: There has been a problem removing the "
"instance %s..." % instance_id)
+ timeout = 50
+ while timeout > 0:
+ instances = functest_utils.get_instances(nova_client)
+ if instances is None or len(instances) == 0:
+ break
+ else:
+ logger.debug("Waiting for instances to be terminated...")
+ timeout -= 1
+ time.sleep(1)
+
def remove_images(nova_client):
logger.info("Removing Glance images...")
images = functest_utils.get_images(nova_client)
- if len(images) == 0:
+ if images is None or len(images) == 0:
logger.debug("No images found.")
return
logger.debug(" > this is a default image and will NOT be deleted.")
-def remove_volumes(cinder_client, nova_client):
+def remove_volumes(cinder_client):
logger.info("Removing Cinder volumes...")
-
- timeout = 50
- while timeout > 0:
- instances = functest_utils.get_instances(nova_client)
- if len(instances) == 0:
- break
- else:
- logger.debug("Waiting for instances to be terminated...")
- timeout -= 1
- time.sleep(1)
-
volumes = functest_utils.get_volumes(cinder_client)
- if len(volumes) == 0:
+ if volumes is None or len(volumes) == 0:
logger.debug("No volumes found.")
return
def remove_floatingips(nova_client):
logger.info("Removing floating IPs...")
floatingips = functest_utils.get_floating_ips(nova_client)
- if len(floatingips) == 0:
+ if floatingips is None or len(floatingips) == 0:
logger.debug("No floating IPs found.")
return
logger.info(" > ERROR: There has been a problem removing the "
"floating IP %s..." % fip_id)
+ timeout = 50
+ while timeout > 0:
+ floatingips = functest_utils.get_floating_ips(nova_client)
+ if floatingips is None or len(floatingips) == 0:
+ break
+ else:
+ logger.debug("Waiting for floating ips to be released...")
+ timeout -= 1
+ time.sleep(1)
+
def remove_networks(neutron_client):
logger.info("Removing Neutron objects")
networks = functest_utils.get_network_list(neutron_client)
if networks == None:
logger.debug("There are no networks in the deployment. ")
- return
+ else:
+ logger.debug("Existing networks:")
+ for network in networks:
+ net_id = network['id']
+ net_name = network['name']
+ logger.debug(" '%s', ID=%s " %(net_name,net_id))
+ if net_name not in default_networks:
+ logger.debug(" > this is not a default network and will be deleted.")
+ network_ids.append(net_id)
+ else:
+ logger.debug(" > this is a default network and will NOT be deleted.")
- logger.debug("Existing networks:")
- for network in networks:
- net_id = network['id']
- net_name = network['name']
- logger.debug(" '%s', ID=%s " %(net_name,net_id))
- if net_name not in default_networks:
- logger.debug(" > this is not a default network and will be deleted.")
- network_ids.append(net_id)
- else:
- logger.debug(" > this is a default network and will NOT be deleted.")
+ #delete ports
+ ports = functest_utils.get_port_list(neutron_client)
+ if ports is None:
+ logger.debug("There are no ports in the deployment. ")
+ else:
+ remove_ports(neutron_client, ports, network_ids)
+ #remove routers
+ routers = functest_utils.get_router_list(neutron_client)
+ if routers is None:
+ logger.debug("There are no routers in the deployment. ")
+ else:
+ remove_routers(neutron_client, routers)
- #remove interfaces router and delete ports
- ports = functest_utils.get_port_list(neutron_client)
+ #remove networks
+ if network_ids != None:
+ for net_id in network_ids:
+ logger.debug("Removing network %s ..." % net_id)
+ if functest_utils.delete_neutron_net(neutron_client, net_id):
+ logger.debug(" > Done!")
+ else:
+ logger.info(" > ERROR: There has been a problem removing the "
+ "network %s..." % net_id)
+
+
+def remove_ports(neutron_client, ports, network_ids):
for port in ports:
if port['network_id'] in network_ids:
port_id = port['id']
try:
subnet_id = port['fixed_ips'][0]['subnet_id']
except:
- logger.info(" > ERROR: Error removing port %s." % port_id)
+ logger.info(" > WARNING: Port %s does not contain 'fixed_ips'" % port_id)
print port
- print ports
- continue
-
router_id = port['device_id']
- if port['device_owner'] == 'network:router_interface':
+ if len(port['fixed_ips']) == 0 and router_id == '':
+ logger.debug("Removing port %s ..." % port_id)
+ if functest_utils.delete_neutron_port(neutron_client, port_id):
+ logger.debug(" > Done!")
+ else:
+ logger.info(" > ERROR: There has been a problem removing the "
+ "port %s ..." %port_id)
+
+ elif port['device_owner'] == 'network:router_interface':
logger.debug("Detaching port %s (subnet %s) from router %s ..."
% (port_id,subnet_id,router_id))
if functest_utils.remove_interface_router(neutron_client,
else:
logger.info(" > ERROR: There has been a problem removing the "
"interface %s from router %s..." %(subnet_id,router_id))
- #print port
else:
+ logger.debug("Clearing device_owner for port %s ..." % port_id)
+ functest_utils.update_neutron_port(neutron_client,
+ port_id,
+ device_owner='clear')
logger.debug("Removing port %s ..." % port_id)
if functest_utils.delete_neutron_port(neutron_client, port_id):
logger.debug(" > Done!")
else:
- logger.info(" > ERROR: There has been a problem removing the "
- "port %s ..." %port_id)
- #print port
+ logger.debug(" > Port %s could not be removed directly" % port_id)
- #remove routers
- routers = functest_utils.get_router_list(neutron_client)
+
+def remove_routers(neutron_client, routers):
for router in routers:
router_id = router['id']
router_name = router['name']
else:
logger.info(" > ERROR: There has been a problem removing "
"the gateway...")
- #print router
-
else:
logger.debug("Router is not connected to anything. Ready to remove...")
- logger.debug("Removing router %s(%s) ..." % (router_name,router_id))
+ logger.debug("Removing router %s(%s) ..." % (router_name, router_id))
if functest_utils.delete_neutron_router(neutron_client, router_id):
logger.debug(" > Done!")
else:
logger.info(" > ERROR: There has been a problem removing the "
- "router '%s'(%s)..." % (router_name,router_id))
-
-
- #remove networks
- for net_id in network_ids:
- logger.debug("Removing network %s ..." % net_id)
- if functest_utils.delete_neutron_net(neutron_client, net_id):
- logger.debug(" > Done!")
- else:
- logger.info(" > ERROR: There has been a problem removing the "
- "network %s..." % net_id)
+ "router '%s'(%s)..." % (router_name, router_id))
def remove_security_groups(neutron_client):
logger.info("Removing Security groups...")
secgroups = functest_utils.get_security_groups(neutron_client)
- if len(secgroups) == 0:
+ if secgroups is None or len(secgroups) == 0:
logger.debug("No security groups found.")
return
separator()
remove_images(nova_client)
separator()
- remove_volumes(cinder_client, nova_client)
+ remove_volumes(cinder_client)
separator()
remove_floatingips(nova_client)
separator()