+
+
+# ----------------------------------------------------------
+#
+# CI UTILS
+#
+# -----------------------------------------------------------
+def get_git_branch(repo_path):
+ """
+ Get git branch name
+ """
+ repo = Repo(repo_path)
+ branch = repo.active_branch
+ return branch.name
+
+
+def get_installer_type(logger=None):
+ """
+ Get installer type (fuel, apex, joid, compass)
+ """
+ try:
+ installer = os.environ['INSTALLER_TYPE']
+ except KeyError:
+ if logger:
+ logger.error("Impossible to retrieve the installer type")
+ installer = "Unknown_installer"
+
+ return installer
+
+
+def get_scenario(logger=None):
+ """
+ Get scenario
+ """
+ try:
+ scenario = os.environ['DEPLOY_SCENARIO']
+ except KeyError:
+ if logger:
+ logger.error("Impossible to retrieve the scenario")
+ scenario = "Unknown_scenario"
+
+ return scenario
+
+
+def get_pod_name(logger=None):
+ """
+ Get PoD Name from env variable NODE_NAME
+ """
+ try:
+ return os.environ['NODE_NAME']
+ except KeyError:
+ if logger:
+ logger.error(
+ "Unable to retrieve the POD name from environment.Using pod name 'unknown-pod'")
+ return "unknown-pod"
+
+
+def push_results_to_db(db_url, case_name, logger, pod_name,
+ version, payload):
+ """
+ POST results to the Result target DB
+ """
+ url = db_url + "/results"
+ installer = get_installer_type(logger)
+ params = {"project_name": "functest", "case_name": case_name,
+ "pod_name": pod_name, "installer": installer,
+ "version": version, "details": payload}
+
+ headers = {'Content-Type': 'application/json'}
+ try:
+ r = requests.post(url, data=json.dumps(params), headers=headers)
+ if logger:
+ logger.debug(r)
+ return True
+ except Exception, e:
+ print "Error []:", e
+ return False
+
+
+def get_resolvconf_ns():
+ """
+ Get nameservers from current resolv.conf
+ """
+ nameservers = []
+ rconf = open("/etc/resolv.conf", "r")
+ line = rconf.readline()
+ while line:
+ ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line)
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ if ip:
+ result = sock.connect_ex((ip.group(),53))
+ if result == 0:
+ nameservers.append(ip.group())
+ line = rconf.readline()
+ 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 []:", e
+
+ return config_test
+
+
+def get_ci_envvars():
+ """
+ Get the CI env variables
+ """
+ ci_env_var = {
+ "installer": os.environ.get('INSTALLER_TYPE'),
+ "scenario": os.environ.get('DEPLOY_SCENARIO')}
+ 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):