import argparse
import os
-import subprocess
import sys
import functest.ci.tier_builder as tb
import functest.utils.clean_openstack as clean_os
import functest.utils.functest_logger as ft_logger
+import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as os_utils
cmd = ("%s%s" % (EXEC_SCRIPT, flags))
logger.debug("Executing command '%s'" % cmd)
- p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
+ result = ft_utils.execute_command(cmd, logger, exit_on_error=False)
- while p.poll() is None:
- line = p.stdout.readline().rstrip()
- logger.debug(line)
-
- if p.returncode != 0:
+ if result != 0:
logger.error("The test case '%s' failed. Cleaning and exiting."
% test_name)
if CLEAN_FLAG:
from neutronclient.v2_0 import client as neutronclient
import functest.utils.functest_logger as ft_logger
-import functest.utils.functest_utils as functest_utils
-import functest.utils.openstack_utils as openstack_utils
+import functest.utils.functest_utils as ft_utils
+import functest.utils.openstack_utils as os_utils
modes = ['full', 'smoke', 'baremetal', 'compute', 'data_processing',
'identity', 'image', 'network', 'object_storage', 'orchestration',
# TODO move DB creds into config file
url = TEST_DB + "/results"
- installer = functest_utils.get_installer_type(logger)
- scenario = functest_utils.get_scenario(logger)
- version = functest_utils.get_version(logger)
- pod_name = functest_utils.get_pod_name(logger)
+ installer = ft_utils.get_installer_type(logger)
+ scenario = ft_utils.get_scenario(logger)
+ version = ft_utils.get_version(logger)
+ pod_name = ft_utils.get_pod_name(logger)
logger.info("Pushing results to DB: '%s'." % url)
def create_tempest_resources():
- ks_creds = openstack_utils.get_credentials("keystone")
+ ks_creds = os_utils.get_credentials("keystone")
logger.info("Creating tenant and user for Tempest suite")
keystone = ksclient.Client(**ks_creds)
- tenant_id = openstack_utils.create_tenant(keystone,
- TENANT_NAME,
- TENANT_DESCRIPTION)
+ tenant_id = os_utils.create_tenant(keystone,
+ TENANT_NAME,
+ TENANT_DESCRIPTION)
if tenant_id == '':
logger.error("Error : Failed to create %s tenant" % TENANT_NAME)
- user_id = openstack_utils.create_user(keystone, USER_NAME, USER_PASSWORD,
- None, tenant_id)
+ user_id = os_utils.create_user(keystone, USER_NAME, USER_PASSWORD,
+ None, tenant_id)
if user_id == '':
logger.error("Error : Failed to create %s user" % USER_NAME)
def free_tempest_resources():
- ks_creds = openstack_utils.get_credentials("keystone")
+ ks_creds = os_utils.get_credentials("keystone")
logger.info("Deleting tenant and user for Tempest suite)")
keystone = ksclient.Client(**ks_creds)
- user_id = openstack_utils.get_user_id(keystone, USER_NAME)
+ user_id = os_utils.get_user_id(keystone, USER_NAME)
if user_id == '':
logger.error("Error : Failed to get id of %s user" % USER_NAME)
else:
- if not openstack_utils.delete_user(keystone, user_id):
+ if not os_utils.delete_user(keystone, user_id):
logger.error("Error : Failed to delete %s user" % USER_NAME)
- tenant_id = openstack_utils.get_tenant_id(keystone, TENANT_NAME)
+ tenant_id = os_utils.get_tenant_id(keystone, TENANT_NAME)
if tenant_id == '':
logger.error("Error : Failed to get id of %s tenant" % TENANT_NAME)
else:
- if not openstack_utils.delete_tenant(keystone, tenant_id):
+ if not os_utils.delete_tenant(keystone, tenant_id):
logger.error("Error : Failed to delete %s tenant" % TENANT_NAME)
logger.debug("Generating tempest.conf file...")
cmd = "rally verify genconfig"
- functest_utils.execute_command(cmd, logger)
+ ft_utils.execute_command(cmd, logger)
logger.debug("Resolving deployment UUID and directory...")
cmd = "rally deployment list | awk '/" + DEPLOYMENT_MAME + "/ {print $2}'"
cmd = "cd " + deployment_dir + ";"
if mode == 'smoke':
cmd += "testr list-tests smoke >" + TEMPEST_LIST_FILE + ";cd"
- functest_utils.execute_command(cmd, logger)
+ ft_utils.execute_command(cmd, logger)
elif mode == 'full':
cmd += "testr list-tests >" + TEMPEST_LIST_FILE + ";cd"
- functest_utils.execute_command(cmd, logger)
+ ft_utils.execute_command(cmd, logger)
logger.debug("Updating selected tempest.conf parameters...")
config = ConfigParser.RawConfigParser()
config.read(tempest_conf_file)
private_net_name = ""
- creds_neutron = openstack_utils.get_credentials("neutron")
+ creds_neutron = os_utils.get_credentials("neutron")
neutron_client = neutronclient.Client(**creds_neutron)
- private_net = openstack_utils.get_private_net(neutron_client)
+ private_net = os_utils.get_private_net(neutron_client)
if private_net is None:
logger.error("No shared private networks found.")
else:
CI_DEBUG = os.environ.get("CI_DEBUG")
if CI_DEBUG == "true" or CI_DEBUG == "True":
- subprocess.call(cmd_line, shell=True, stderr=subprocess.STDOUT)
+ ft_utils.execute_command(cmd_line, logger, exit_on_error=True)
else:
header = ("Tempest environment:\n"
" Installer: %s\n Scenario: %s\n Node: %s\n Date: %s\n" %
f_env.close()
cmd_line = "rally verify show"
- subprocess.call(cmd_line, shell=True)
+ ft_utils.execute_command(cmd_line, logger,
+ exit_on_error=True, info=True)
cmd_line = "rally verify list"
logger.debug('Executing command : {}'.format(cmd_line))
+++ /dev/null
-#!/usr/bin/python
-
-import os
-import unittest
-import yaml
-from functest_utils import getTestEnv, isTestRunnable, generateTestcaseList
-
-
-class TestFunctestUtils(unittest.TestCase):
-
- def setUp(self):
- os.environ["INSTALLER_TYPE"] = "fuel"
- os.environ["DEPLOY_SCENARIO"] = "os-odl_l3-ovs-ha"
-
- global functest_yaml
-
- with open("../config_functest.yaml") as f:
- functest_yaml = yaml.safe_load(f)
- f.close()
-
- def test_getTestEnv(self):
-
- env_test = getTestEnv('ovno', functest_yaml)
- self.assertEqual(env_test, {'scenario': 'ocl'})
-
- env_test = getTestEnv('doctor', functest_yaml)
- self.assertEqual(env_test, {'installer': 'fuel'})
-
- env_test = getTestEnv('promise', functest_yaml)
- self.assertEqual(env_test, {'installer': '(fuel)|(joid)'})
-
- env_test = getTestEnv('functest/tempest', functest_yaml)
- self.assertEqual(env_test, None)
-
- env_test = getTestEnv('functest/vims', functest_yaml)
- self.assertEqual(env_test, None)
-
- env_test = getTestEnv('functest/odl', functest_yaml)
- self.assertEqual(env_test, {'scenario': 'odl'})
-
- env_test = getTestEnv('functest/onos', functest_yaml)
- self.assertEqual(env_test, {'scenario': 'onos'})
-
- env_test = getTestEnv('policy-test', functest_yaml)
- self.assertEqual(env_test, {'scenario': 'odl'})
-
- env_test = getTestEnv('foo', functest_yaml)
- self.assertEqual(env_test, '')
-
- def test_isTestRunnable(self):
-
- test = isTestRunnable('ovno', functest_yaml)
- self.assertFalse(test)
-
- test = isTestRunnable('doctor', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('promise', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('functest/onos', functest_yaml)
- self.assertFalse(test)
-
- test = isTestRunnable('functest/odl', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('functest/vping_ssh', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('functest/vping_userdata', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('functest/tempest', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('functest/rally', functest_yaml)
- self.assertTrue(test)
-
- test = isTestRunnable('functest/vims', functest_yaml)
- self.assertTrue(test)
-
- def test_generateTestcaseList(self):
-
- test = generateTestcaseList(functest_yaml)
-
- expected_list = ("vping_ssh vping_userdata tempest odl doctor " +
- "promise policy-test odl-vpn_service-tests vims " +
- "rally ")
- self.assertEqual(test, expected_list)
-
- def tearDown(self):
- os.environ["INSTALLER_TYPE"] = ""
- os.environ["DEPLOY_SCENARIO"] = ""
-
-
-if __name__ == '__main__':
- unittest.main()
class Logger:
def __init__(self, logger_name):
+ CI_DEBUG = os.getenv('CI_DEBUG')
+
self.logger = logging.getLogger(logger_name)
self.logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - '
'%(levelname)s - %(message)s')
ch.setFormatter(formatter)
-
- CI_DEBUG = os.getenv('CI_DEBUG')
-
- ch.setLevel(logging.INFO)
-
if CI_DEBUG is not None and CI_DEBUG.lower() == "true":
ch.setLevel(logging.DEBUG)
-
+ else:
+ ch.setLevel(logging.INFO)
self.logger.addHandler(ch)
+ hdlr = logging.FileHandler('/home/opnfv/functest/results/functest.log')
+ hdlr.setFormatter(formatter)
+ hdlr.setLevel(logging.DEBUG)
+ self.logger.addHandler(hdlr)
+
def getLogger(self):
return self.logger
import shutil
import socket
import subprocess
+import sys
import urllib2
from git import Repo
return nameservers
-def getTestEnv(test, functest_yaml):
- """
- Get the config of the testcase based on functest_config.yaml
- 2 options
- - test = test project e.g; ovno
- - test = testcase e.g. functest/odl
- look for the / to see if it is a test project or a testcase
- """
- try:
- TEST_ENV = functest_yaml.get("test-dependencies")
-
- if test.find("/") < 0:
- config_test = TEST_ENV[test]
- else:
- test_split = test.split("/")
- testproject = test_split[0]
- testcase = test_split[1]
- config_test = TEST_ENV[testproject][testcase]
- except KeyError:
- # if not defined in dependencies => no dependencies
- config_test = ""
- except Exception, e:
- print "Error [getTestEnv]:", e
-
- return config_test
-
-
def get_ci_envvars():
"""
Get the CI env variables
return ci_env_var
-def isTestRunnable(test, functest_yaml):
- """
- Return True if the test is runnable in the current scenario
- """
- # By default we assume that all the tests are always runnable...
- is_runnable = True
- # Retrieve CI environment
- ci_env = get_ci_envvars()
- # Retrieve test environement from config file
- test_env = getTestEnv(test, functest_yaml)
-
- # if test_env not empty => dependencies to be checked
- if test_env is not None and len(test_env) > 0:
- # possible criteria = ["installer", "scenario"]
- # consider test criteria from config file
- # compare towards CI env through CI en variable
- for criteria in test_env:
- if re.search(test_env[criteria], ci_env[criteria]) is None:
- # print "Test "+ test + " cannot be run on the environment"
- is_runnable = False
- return is_runnable
-
-
-def generateTestcaseList(functest_yaml):
- """
- Generate a test file with the runnable test according to
- the current scenario
- """
- test_list = ""
- # get testcases
- testcase_list = functest_yaml.get("test-dependencies")
- projects = testcase_list.keys()
-
- for project in projects:
- testcases = testcase_list[project]
- # 1 or 2 levels for testcases project[/case]l
- # if only project name without controller or scenario
- # => shall be runnable on any controller/scenario
- if testcases is None:
- test_list += project + " "
- else:
- for testcase in testcases:
- if testcase == "installer" or testcase == "scenario":
- # project (1 level)
- if isTestRunnable(project, functest_yaml):
- test_list += project + " "
- else:
- # project/testcase (2 levels)
- thetest = project + "/" + testcase
- if isTestRunnable(thetest, functest_yaml):
- test_list += testcase + " "
-
- # sort the list to execute the test in the right order
- test_order_list = functest_yaml.get("test_exec_priority")
- test_sorted_list = ""
- for test in test_order_list:
- if test_order_list[test] in test_list:
- test_sorted_list += test_order_list[test] + " "
-
- # create a file that could be consumed by run-test.sh
- # this method is used only for CI
- # so it can be run only in container
- # reuse default conf directory to store the list of runnable tests
- file = open("/home/opnfv/functest/conf/testcase-list.txt", 'w')
- file.write(test_sorted_list)
- file.close()
-
- return test_sorted_list
-
-
-def execute_command(cmd, logger=None, exit_on_error=True):
- """
- Execute Linux command
- prints stdout to a file and depending on if there
- is a logger defined, it will print it or not.
- """
+def execute_command(cmd, logger=None, exit_on_error=True, info=False):
if logger:
logger.debug('Executing command : {}'.format(cmd))
- output_file = "output.txt"
- f = open(output_file, 'w+')
- p = subprocess.call(cmd, shell=True, stdout=f, stderr=subprocess.STDOUT)
- f.close()
- f = open(output_file, 'r')
- result = f.read()
- if result != "" and logger:
- logger.debug(result)
- if p == 0:
- return True
- else:
+ p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
+ while p.poll() is None:
+ line = p.stdout.readline().rstrip()
+ if logger:
+ if info:
+ logger.info(line)
+ else:
+ logger.debug(line)
+ else:
+ print line
+ if p.returncode != 0:
if logger:
logger.error("Error when executing command %s" % cmd)
+ else:
+ print("Error when executing command %s" % cmd)
if exit_on_error:
- exit(-1)
- return False
+ sys.exit(1)
+
+ return p.returncode
Code Review / functest-xtesting.git / commitdiff