general:
dir:
# Relative to the path where the repo is cloned:
- vping: functest/opnfv_tests/openstack/vping
+ vping: functest/opnfv_tests/openstack/vping
dir_odl: functest/opnfv_tests/sdn/odl
dir_rally: functest/opnfv_tests/openstack/rally
- dir_tempest_cases: functest/opnfv_tests/openstack/tempest/custom_tests
+ tempest_cases: functest/opnfv_tests/openstack/tempest/custom_tests
dir_vIMS: functest/opnfv_tests/vnf/ims
dir_onos: functest/opnfv_tests/sdn/onos/teston
dir_onos_sfc: functest/opnfv_tests/sdn/onos/sfc
# Absolute path
dir_home: /home/opnfv
- dir_repos: /home/opnfv/repos
- repo_functest: /home/opnfv/repos/functest
+ repos: /home/opnfv/repos
+ repo_functest: /home/opnfv/repos/functest
dir_repo_rally: /home/opnfv/repos/rally
- dir_repo_tempest: /home/opnfv/repos/tempest
+ repo_tempest: /home/opnfv/repos/tempest
dir_repo_releng: /home/opnfv/repos/releng
dir_repo_vims_test: /home/opnfv/repos/vims-test
- dir_repo_sdnvpn: /home/opnfv/repos/sdnvpn
- dir_repo_sfc: /home/opnfv/repos/sfc
+ repo_sdnvpn: /home/opnfv/repos/sdnvpn
+ repo_sfc: /home/opnfv/repos/sfc
dir_repo_onos: /home/opnfv/repos/onos
dir_repo_promise: /home/opnfv/repos/promise
dir_repo_doctor: /home/opnfv/repos/doctor
- dir_repo_copper: /home/opnfv/repos/copper
+ repo_copper: /home/opnfv/repos/copper
dir_repo_ovno: /home/opnfv/repos/ovno
- dir_repo_parser: /home/opnfv/repos/parser
- dir_repo_domino: /home/opnfv/repos/domino
+ repo_parser: /home/opnfv/repos/parser
+ repo_domino: /home/opnfv/repos/domino
dir_repo_snaps: /home/opnfv/repos/snaps
- functest: /home/opnfv/functest
- functest_test: /home/opnfv/repos/functest/functest/opnfv_tests
- results: /home/opnfv/functest/results
- functest_conf: /home/opnfv/functest/conf
- functest_data: /home/opnfv/functest/data
+ functest: /home/opnfv/functest
+ functest_test: /home/opnfv/repos/functest/functest/opnfv_tests
+ results: /home/opnfv/functest/results
+ functest_conf: /home/opnfv/functest/conf
+ functest_data: /home/opnfv/functest/data
dir_vIMS_data: /home/opnfv/functest/data/vIMS/
- dir_rally_inst: /home/opnfv/.rally
+ rally_inst: /home/opnfv/.rally
openstack:
creds: /home/opnfv/functest/conf/openstack.creds
installer_master_username: 'root'
installer_master_password: 'r00tme'
multisite:
- fuel_environment:
+ fuel:
installer_username: 'root'
installer_password: 'r00tme'
- compass_environment:
+ compass:
installer_username: 'root'
installer_password: 'root'
multisite_controller_ip: '10.1.0.50'
+#!/usr/bin/env python
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+#
import json
import re
import urllib2
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as ft_utils
-import functest.utils.functest_constants as ft_constants
-
+from functest.utils.constants import CONST
COL_1_LEN = 25
COL_2_LEN = 15
# and then we can print the url to the specific test result
-class GlobalVariables:
- IS_CI_RUN = ft_constants.IS_CI_RUN
- BUILD_TAG = ft_constants.CI_BUILD_TAG
- INSTALLER = ft_constants.CI_INSTALLER_TYPE
- CI_LOOP = ft_constants.CI_LOOP
- SCENARIO = ft_constants.CI_SCENARIO
-
-
logger = ft_logger.Logger("generate_report").getLogger()
def get_results_from_db():
url = "%s/results?build_tag=%s" % (ft_utils.get_db_url(),
- GlobalVariables.BUILD_TAG)
+ CONST.BUILD_TAG)
logger.debug("Query to rest api: %s" % url)
try:
data = json.load(urllib2.urlopen(url))
'| ' + w2.ljust(COL_2_LEN - 1) +
'| ' + w3.ljust(COL_3_LEN - 1) +
'| ' + w4.ljust(COL_4_LEN - 1))
- if GlobalVariables.IS_CI_RUN:
+ if CONST.IS_CI_RUN:
str += ('| ' + w5.ljust(COL_5_LEN - 1))
str += '|\n'
return str
def print_line_no_columns(str):
TOTAL_LEN = COL_1_LEN + COL_2_LEN + COL_3_LEN + COL_4_LEN + 2
- if GlobalVariables.IS_CI_RUN:
+ if CONST.IS_CI_RUN:
TOTAL_LEN += COL_5_LEN + 1
return ('| ' + str.ljust(TOTAL_LEN) + "|\n")
delimiter + char * COL_2_LEN +
delimiter + char * COL_3_LEN +
delimiter + char * COL_4_LEN)
- if GlobalVariables.IS_CI_RUN:
+ if CONST.IS_CI_RUN:
str += (delimiter + char * COL_5_LEN)
str += '+\n'
return str
def main(args):
executed_test_cases = args
- if GlobalVariables.IS_CI_RUN:
+ if CONST.IS_CI_RUN:
results = get_results_from_db()
if results is not None:
for test in executed_test_cases:
"result": data['result']})
TOTAL_LEN = COL_1_LEN + COL_2_LEN + COL_3_LEN + COL_4_LEN
- if GlobalVariables.IS_CI_RUN:
+ if CONST.IS_CI_RUN:
TOTAL_LEN += COL_5_LEN
MID = TOTAL_LEN / 2
- if GlobalVariables.BUILD_TAG is not None:
- if re.search("daily", GlobalVariables.BUILD_TAG) is not None:
- GlobalVariables.CI_LOOP = "daily"
+ if CONST.BUILD_TAG is not None:
+ if re.search("daily", CONST.BUILD_TAG) is not None:
+ CONST.CI_LOOP = "daily"
else:
- GlobalVariables.CI_LOOP = "weekly"
+ CONST.CI_LOOP = "weekly"
str = ''
str += print_separator('=', delimiter="=")
str += print_line_no_columns(' ')
str += print_line_no_columns(" Deployment description:")
str += print_line_no_columns(" INSTALLER: %s"
- % GlobalVariables.INSTALLER)
- if GlobalVariables.SCENARIO is not None:
+ % CONST.INSTALLER_TYPE)
+ if CONST.DEPLOY_SCENARIO is not None:
str += print_line_no_columns(" SCENARIO: %s"
- % GlobalVariables.SCENARIO)
- if GlobalVariables.BUILD_TAG is not None:
+ % CONST.DEPLOY_SCENARIO)
+ if CONST.BUILD_TAG is not None:
str += print_line_no_columns(" BUILD TAG: %s"
- % GlobalVariables.BUILD_TAG)
- if GlobalVariables.CI_LOOP is not None:
+ % CONST.BUILD_TAG)
+ if CONST.CI_LOOP is not None:
str += print_line_no_columns(" CI LOOP: %s"
- % GlobalVariables.CI_LOOP)
+ % CONST.CI_LOOP)
str += print_line_no_columns(' ')
str += print_separator('=')
- if GlobalVariables.IS_CI_RUN:
+ if CONST.IS_CI_RUN:
str += print_line('TEST CASE', 'TIER', 'DURATION', 'RESULT', 'URL')
else:
str += print_line('TEST CASE', 'TIER', 'DURATION', 'RESULT')
#
+import argparse
import json
import os
import re
import subprocess
import sys
-import argparse
import yaml
+from opnfv.utils import constants as opnfv_constants
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as os_utils
-import functest.utils.functest_constants as ft_constants
-
-from opnfv.utils import constants as opnfv_constants
+from functest.utils.constants import CONST
actions = ['start', 'check']
parser = argparse.ArgumentParser()
logger = ft_logger.Logger("prepare_env").getLogger()
-CONFIG_FUNCTEST_PATH = ft_constants.CONFIG_FUNCTEST_YAML
+CONFIG_FUNCTEST_PATH = CONST.CONFIG_FUNCTEST_YAML
CONFIG_PATCH_PATH = os.path.join(os.path.dirname(
CONFIG_FUNCTEST_PATH), "config_patch.yaml")
print_separator()
logger.info("Checking environment variables...")
- if ft_constants.CI_INSTALLER_TYPE is None:
+ if CONST.INSTALLER_TYPE is None:
logger.warning("The env variable 'INSTALLER_TYPE' is not defined.")
- ft_constants.CI_INSTALLER_TYPE = "undefined"
+ CONST.INSTALLER_TYPE = "undefined"
else:
- if ft_constants.CI_INSTALLER_TYPE not in ft_constants.INSTALLERS:
+ if CONST.INSTALLER_TYPE not in opnfv_constants.INSTALLERS:
logger.warning("INSTALLER_TYPE=%s is not a valid OPNFV installer. "
"Available OPNFV Installers are : %s. "
"Setting INSTALLER_TYPE=undefined."
- % (ft_constants.CI_INSTALLER_TYPE,
- ft_constants.INSTALLERS))
- ft_constants.CI_INSTALLER_TYPE = "undefined"
+ % (CONST.INSTALLER_TYPE,
+ opnfv_constants.INSTALLERS))
+ CONST.INSTALLER_TYPE = "undefined"
else:
logger.info(" INSTALLER_TYPE=%s"
- % ft_constants.CI_INSTALLER_TYPE)
+ % CONST.INSTALLER_TYPE)
- if ft_constants.CI_INSTALLER_IP is None:
+ if CONST.INSTALLER_IP is None:
logger.warning("The env variable 'INSTALLER_IP' is not defined. "
"It is needed to fetch the OpenStack credentials. "
"If the credentials are not provided to the "
"container as a volume, please add this env variable "
"to the 'docker run' command.")
else:
- logger.info(" INSTALLER_IP=%s" % ft_constants.CI_INSTALLER_IP)
+ logger.info(" INSTALLER_IP=%s" % CONST.INSTALLER_IP)
- if ft_constants.CI_SCENARIO is None:
+ if CONST.DEPLOY_SCENARIO is None:
logger.warning("The env variable 'DEPLOY_SCENARIO' is not defined. "
"Setting CI_SCENARIO=undefined.")
- ft_constants.CI_SCENARIO = "undefined"
+ CONST.DEPLOY_SCENARIO = "undefined"
else:
- logger.info(" DEPLOY_SCENARIO=%s" % ft_constants.CI_SCENARIO)
- if ft_constants.CI_DEBUG:
- logger.info(" CI_DEBUG=%s" % ft_constants.CI_DEBUG)
+ logger.info(" DEPLOY_SCENARIO=%s" % CONST.DEPLOY_SCENARIO)
+ if CONST.CI_DEBUG:
+ logger.info(" CI_DEBUG=%s" % CONST.CI_DEBUG)
- if ft_constants.CI_NODE:
- logger.info(" NODE_NAME=%s" % ft_constants.CI_NODE)
+ if CONST.NODE_NAME:
+ logger.info(" NODE_NAME=%s" % CONST.NODE_NAME)
- if ft_constants.CI_BUILD_TAG:
- logger.info(" BUILD_TAG=%s" % ft_constants.CI_BUILD_TAG)
+ if CONST.BUILD_TAG:
+ logger.info(" BUILD_TAG=%s" % CONST.BUILD_TAG)
- if ft_constants.IS_CI_RUN:
- logger.info(" IS_CI_RUN=%s" % ft_constants.IS_CI_RUN)
+ if CONST.IS_CI_RUN:
+ logger.info(" IS_CI_RUN=%s" % CONST.IS_CI_RUN)
def create_directories():
print_separator()
logger.info("Creating needed directories...")
- if not os.path.exists(ft_constants.FUNCTEST_CONF_DIR):
- os.makedirs(ft_constants.FUNCTEST_CONF_DIR)
- logger.info(" %s created." % ft_constants.FUNCTEST_CONF_DIR)
+ if not os.path.exists(CONST.dir_functest_conf):
+ os.makedirs(CONST.dir_functest_conf)
+ logger.info(" %s created." % CONST.dir_functest_conf)
else:
logger.debug(" %s already exists."
- % ft_constants.FUNCTEST_CONF_DIR)
+ % CONST.dir_functest_conf)
- if not os.path.exists(ft_constants.FUNCTEST_DATA_DIR):
- os.makedirs(ft_constants.FUNCTEST_DATA_DIR)
- logger.info(" %s created." % ft_constants.FUNCTEST_DATA_DIR)
+ if not os.path.exists(CONST.dir_functest_data):
+ os.makedirs(CONST.dir_functest_data)
+ logger.info(" %s created." % CONST.dir_functest_data)
else:
logger.debug(" %s already exists."
- % ft_constants.FUNCTEST_DATA_DIR)
+ % CONST.dir_functest_data)
def source_rc_file():
print_separator()
logger.info("Fetching RC file...")
- if ft_constants.OPENSTACK_CREDS is None:
+ if CONST.openstack_creds is None:
logger.warning("The environment variable 'creds' must be set and"
"pointing to the local RC file. Using default: "
"/home/opnfv/functest/conf/openstack.creds ...")
- os.path.join(ft_constants.FUNCTEST_CONF_DIR, 'openstack.creds')
+ os.path.join(CONST.dir_functest_conf, 'openstack.creds')
- if not os.path.isfile(ft_constants.OPENSTACK_CREDS):
+ if not os.path.isfile(CONST.openstack_creds):
logger.info("RC file not provided. "
"Fetching it from the installer...")
- if ft_constants.CI_INSTALLER_IP is None:
+ if CONST.INSTALLER_IP is None:
logger.error("The env variable CI_INSTALLER_IP must be provided in"
" order to fetch the credentials from the installer.")
sys.exit("Missing CI_INSTALLER_IP.")
- if ft_constants.CI_INSTALLER_TYPE not in ft_constants.INSTALLERS:
+ if CONST.INSTALLER_TYPE not in opnfv_constants.INSTALLERS:
logger.error("Cannot fetch credentials. INSTALLER_TYPE=%s is "
"not a valid OPNFV installer. Available "
"installers are : %s." %
- (ft_constants.CI_INSTALLER_TYPE,
+ (CONST.INSTALLER_TYPE,
opnfv_constants.INSTALLERS))
sys.exit("Wrong INSTALLER_TYPE.")
cmd = ("/home/opnfv/repos/releng/utils/fetch_os_creds.sh "
"-d %s -i %s -a %s"
- % (ft_constants.OPENSTACK_CREDS,
- ft_constants.CI_INSTALLER_TYPE,
- ft_constants.CI_INSTALLER_IP))
+ % (CONST.openstack_creds,
+ CONST.INSTALLER_TYPE,
+ CONST.INSTALLER_IP))
logger.debug("Executing command: %s" % cmd)
p = subprocess.Popen(cmd.split(), stdout=subprocess.PIPE)
output = p.communicate()[0]
sys.exit(1)
else:
logger.info("RC file provided in %s."
- % ft_constants.OPENSTACK_CREDS)
- if os.path.getsize(ft_constants.OPENSTACK_CREDS) == 0:
+ % CONST.openstack_creds)
+ if os.path.getsize(CONST.openstack_creds) == 0:
logger.error("The file %s is empty."
- % ft_constants.OPENSTACK_CREDS)
+ % CONST.openstack_creds)
sys.exit(1)
logger.info("Sourcing the OpenStack RC file...")
creds = os_utils.source_credentials(
- ft_constants.OPENSTACK_CREDS)
+ CONST.openstack_creds)
str = ""
for key, value in creds.iteritems():
if re.search("OS_", key):
str += "\n\t\t\t\t\t\t " + key + "=" + value
if key == 'OS_AUTH_URL':
- ft_constants.OS_AUTH_URL = value
+ CONST.OS_AUTH_URL = value
elif key == 'OS_USERNAME':
- ft_constants.OS_USERNAME = value
+ CONST.OS_USERNAME = value
elif key == 'OS_TENANT_NAME':
- ft_constants.OS_TENANT_NAME = value
+ CONST.OS_TENANT_NAME = value
elif key == 'OS_PASSWORD':
- ft_constants.OS_PASSWORD = value
+ CONST.OS_PASSWORD = value
logger.debug("Used credentials: %s" % str)
- logger.debug("OS_AUTH_URL:%s" % ft_constants.OS_AUTH_URL)
- logger.debug("OS_USERNAME:%s" % ft_constants.OS_USERNAME)
- logger.debug("OS_TENANT_NAME:%s" % ft_constants.OS_TENANT_NAME)
- logger.debug("OS_PASSWORD:%s" % ft_constants.OS_PASSWORD)
+ logger.debug("OS_AUTH_URL:%s" % CONST.OS_AUTH_URL)
+ logger.debug("OS_USERNAME:%s" % CONST.OS_USERNAME)
+ logger.debug("OS_TENANT_NAME:%s" % CONST.OS_TENANT_NAME)
+ logger.debug("OS_PASSWORD:%s" % CONST.OS_PASSWORD)
def patch_config_file():
updated = False
for key in functest_patch_yaml:
- if key in ft_constants.CI_SCENARIO:
+ if key in CONST.DEPLOY_SCENARIO:
new_functest_yaml = dict(ft_utils.merge_dicts(
ft_utils.get_functest_yaml(), functest_patch_yaml[key]))
updated = True
def verify_deployment():
print_separator()
logger.info("Verifying OpenStack services...")
- cmd = ("%s/functest/ci/check_os.sh" % ft_constants.FUNCTEST_REPO_DIR)
+ cmd = ("%s/functest/ci/check_os.sh" % CONST.dir_repo_functest)
logger.debug("Executing command: %s" % cmd)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
cmd = "rally deployment destroy opnfv-rally"
ft_utils.execute_command(cmd, error_msg=(
"Deployment %s does not exist."
- % ft_constants.RALLY_DEPLOYMENT_NAME),
+ % CONST.rally_deployment_name),
verbose=False)
rally_conf = os_utils.get_credentials_for_rally()
with open('rally_conf.json', 'w') as fp:
json.dump(rally_conf, fp)
cmd = "rally deployment create --file=rally_conf.json --name="
- cmd += ft_constants.RALLY_DEPLOYMENT_NAME
+ cmd += CONST.rally_deployment_name
ft_utils.execute_command(cmd,
error_msg="Problem creating Rally deployment")
logger.info("Installing tempest from existing repo...")
cmd = ("rally verify install --source " +
- ft_constants.TEMPEST_REPO_DIR +
+ CONST.dir_repo_tempest +
" --system-wide")
ft_utils.execute_command(cmd,
error_msg="Problem installing Tempest.")
def check_environment():
msg_not_active = "The Functest environment is not installed."
- if not os.path.isfile(ft_constants.ENV_FILE):
+ if not os.path.isfile(CONST.env_active):
logger.error(msg_not_active)
sys.exit(1)
- with open(ft_constants.ENV_FILE, "r") as env_file:
+ with open(CONST.env_active, "r") as env_file:
s = env_file.read()
if not re.search("1", s):
logger.error(msg_not_active)
verify_deployment()
install_rally()
- with open(ft_constants.ENV_FILE, "w") as env_file:
+ with open(CONST.env_active, "w") as env_file:
env_file.write("1")
check_environment()
# http://www.apache.org/licenses/LICENSE-2.0
#
+import argparse
import datetime
import importlib
import os
import re
import sys
-import argparse
-
import functest.ci.generate_report as generate_report
import functest.ci.tier_builder as tb
import functest.core.testcase_base as testcase_base
+import functest.utils.functest_constants as ft_constants
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as ft_utils
-import functest.utils.functest_constants as ft_constants
import functest.utils.openstack_clean as os_clean
import functest.utils.openstack_snapshot as os_snapshot
import functest.utils.openstack_utils as os_utils
-
+from functest.utils.constants import CONST
parser = argparse.ArgumentParser()
parser.add_argument("-t", "--test", dest="test", action='store',
""" global variables """
-EXEC_SCRIPT = ("%s/functest/ci/exec_test.sh" % ft_constants.FUNCTEST_REPO_DIR)
+EXEC_SCRIPT = ("%s/functest/ci/exec_test.sh" % CONST.dir_repo_functest)
# This will be the return code of this script. If any of the tests fails,
# this variable will change to -1
def source_rc_file():
- rc_file = ft_constants.OPENSTACK_CREDS
+ rc_file = CONST.openstack_creds
if not os.path.isfile(rc_file):
logger.error("RC file %s does not exist..." % rc_file)
sys.exit(1)
if re.search("OS_", key):
if key == 'OS_AUTH_URL':
ft_constants.OS_AUTH_URL = value
+ CONST.OS_AUTH_URL = value
elif key == 'OS_USERNAME':
ft_constants.OS_USERNAME = value
+ CONST.OS_USERNAME = value
elif key == 'OS_TENANT_NAME':
ft_constants.OS_TENANT_NAME = value
+ CONST.OS_TENANT_NAME = value
elif key == 'OS_PASSWORD':
ft_constants.OS_PASSWORD = value
- logger.debug("OS_AUTH_URL:%s" % ft_constants.OS_AUTH_URL)
- logger.debug("OS_USERNAME:%s" % ft_constants.OS_USERNAME)
- logger.debug("OS_TENANT_NAME:%s" % ft_constants.OS_TENANT_NAME)
- logger.debug("OS_PASSWORD:%s" % ft_constants.OS_PASSWORD)
+ CONST.OS_PASSWORD = value
+ logger.debug("OS_AUTH_URL:%s" % CONST.OS_AUTH_URL)
+ logger.debug("OS_USERNAME:%s" % CONST.OS_USERNAME)
+ logger.debug("OS_TENANT_NAME:%s" % CONST.OS_TENANT_NAME)
+ logger.debug("OS_PASSWORD:%s" % CONST.OS_PASSWORD)
def generate_os_snapshot():
def run_all(tiers):
summary = ""
- BUILD_TAG = ft_constants.CI_BUILD_TAG
- if BUILD_TAG is not None and re.search("daily", BUILD_TAG) is not None:
- CI_LOOP = "daily"
- else:
- CI_LOOP = "weekly"
-
tiers_to_run = []
for tier in tiers.get_tiers():
if (len(tier.get_tests()) != 0 and
- re.search(CI_LOOP, tier.get_ci_loop()) is not None):
+ re.search(CONST.CI_LOOP, tier.get_ci_loop()) is not None):
tiers_to_run.append(tier)
summary += ("\n - %s:\n\t %s"
% (tier.get_name(),
def main():
- CI_INSTALLER_TYPE = ft_constants.CI_INSTALLER_TYPE
- CI_SCENARIO = ft_constants.CI_SCENARIO
+ CI_INSTALLER_TYPE = CONST.INSTALLER_TYPE
+ CI_SCENARIO = CONST.DEPLOY_SCENARIO
- file = ft_constants.FUNCTEST_TESTCASES_YAML
+ file = CONST.functest_testcases_yaml
_tiers = tb.TierBuilder(CI_INSTALLER_TYPE, CI_SCENARIO, file)
if args.noclean:
import click
import git
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
-import functest.utils.functest_constants as ft_constants
class CliEnv:
"it again? [y|n]\n")
while True:
if answer.lower() in ["y", "yes"]:
- os.remove(ft_constants.ENV_FILE)
+ os.remove(CONST.env_active)
break
elif answer.lower() in ["n", "no"]:
return
answer = raw_input("Invalid answer. Please type [y|n]\n")
cmd = ("python %s/functest/ci/prepare_env.py start" %
- ft_constants.FUNCTEST_REPO_DIR)
+ CONST.dir_repo_functest)
ft_utils.execute_command(cmd)
def show(self):
- CI_INSTALLER_TYPE = ft_constants.CI_INSTALLER_TYPE
- if CI_INSTALLER_TYPE is None:
- CI_INSTALLER_TYPE = "Unknown"
- CI_INSTALLER_IP = ft_constants.CI_INSTALLER_IP
- if CI_INSTALLER_IP is None:
- CI_INSTALLER_IP = "Unknown"
- CI_INSTALLER = ("%s, %s" % (CI_INSTALLER_TYPE, CI_INSTALLER_IP))
-
- CI_SCENARIO = ft_constants.CI_SCENARIO
- if CI_SCENARIO is None:
- CI_SCENARIO = "Unknown"
-
- CI_NODE = ft_constants.CI_NODE
- if CI_NODE is None:
- CI_NODE = "Unknown"
-
- repo = git.Repo(ft_constants.FUNCTEST_REPO_DIR)
- branch = repo.head.reference
- GIT_BRANCH = branch.name
- GIT_HASH = branch.commit.hexsha
-
- CI_BUILD_TAG = ft_constants.CI_BUILD_TAG
- if CI_BUILD_TAG is not None:
- CI_BUILD_TAG = CI_BUILD_TAG.lstrip(
+ def _get_value(attr, default='Unknown'):
+ return attr if attr else default
+
+ install_type = _get_value(CONST.INSTALLER_TYPE)
+ installer_ip = _get_value(CONST.INSTALLER_IP)
+ installer_info = ("%s, %s" % (install_type, installer_ip))
+ scenario = _get_value(CONST.DEPLOY_SCENARIO)
+ node = _get_value(CONST.NODE_NAME)
+ repo_h = git.Repo(CONST.dir_repo_functest).head
+ if repo_h.is_detached:
+ git_branch = 'detached from FETCH_HEAD'
+ git_hash = repo_h.commit.hexsha
+ else:
+ branch = repo_h.reference
+ git_branch = branch.name
+ git_hash = branch.commit.hexsha
+ is_debug = _get_value(CONST.CI_DEBUG, 'false')
+ build_tag = CONST.BUILD_TAG
+ if build_tag is not None:
+ build_tag = build_tag.lstrip(
"jenkins-").lstrip("functest").lstrip("-")
- CI_DEBUG = ft_constants.CI_DEBUG
- if CI_DEBUG is None:
- CI_DEBUG = "false"
-
STATUS = "not ready"
if self.status(verbose=False) == 0:
STATUS = "ready"
click.echo("+======================================================+")
click.echo("| Functest Environment info |")
click.echo("+======================================================+")
- click.echo("| INSTALLER: %s|" % CI_INSTALLER.ljust(41))
- click.echo("| SCENARIO: %s|" % CI_SCENARIO.ljust(41))
- click.echo("| POD: %s|" % CI_NODE.ljust(41))
- click.echo("| GIT BRACNH: %s|" % GIT_BRANCH.ljust(41))
- click.echo("| GIT HASH: %s|" % GIT_HASH.ljust(41))
- if CI_BUILD_TAG:
- click.echo("| BUILD TAG: %s|" % CI_BUILD_TAG.ljust(41))
- click.echo("| DEBUG FLAG: %s|" % CI_DEBUG.ljust(41))
+ click.echo("| INSTALLER: %s|" % installer_info.ljust(41))
+ click.echo("| SCENARIO: %s|" % scenario.ljust(41))
+ click.echo("| POD: %s|" % node.ljust(41))
+ click.echo("| GIT BRACNH: %s|" % git_branch.ljust(41))
+ click.echo("| GIT HASH: %s|" % git_hash.ljust(41))
+ if build_tag:
+ click.echo("| BUILD TAG: %s|" % build_tag.ljust(41))
+ click.echo("| DEBUG FLAG: %s|" % is_debug.ljust(41))
click.echo("+------------------------------------------------------+")
click.echo("| STATUS: %s|" % STATUS.ljust(41))
click.echo("+------------------------------------------------------+")
def status(self, verbose=True):
ret_val = 0
- if not os.path.isfile(ft_constants.ENV_FILE):
+ if not os.path.isfile(CONST.env_active):
if verbose:
click.echo("Functest environment is not installed.\n")
ret_val = 1
import click
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_clean as os_clean
import functest.utils.openstack_snapshot as os_snapshot
-import functest.utils.functest_constants as ft_constants
-
-
-OPENSTACK_RC_FILE = ft_constants.OPENSTACK_CREDS
-OPENSTACK_SNAPSHOT_FILE = ft_constants.OPENSTACK_SNAPSHOT_FILE
class CliOpenStack:
def __init__(self):
- self.os_auth_url = ft_constants.OS_AUTH_URL
+ self.os_auth_url = CONST.OS_AUTH_URL
self.endpoint_ip = None
self.endpoint_port = None
- if self.os_auth_url is not None:
+ self.openstack_creds = CONST.openstack_creds
+ self.snapshot_file = CONST.openstack_snapshot_file
+ if self.os_auth_url:
self.endpoint_ip = self.os_auth_url.rsplit("/")[2].rsplit(":")[0]
self.endpoint_port = self.os_auth_url.rsplit("/")[2].rsplit(":")[1]
click.echo("Cannot talk to the endpoint %s\n" % self.endpoint_ip)
exit(0)
- def show_credentials(self):
+ @staticmethod
+ def show_credentials():
for key, value in os.environ.items():
if key.startswith('OS_'):
click.echo("{}={}".format(key, value))
def fetch_credentials(self):
- if os.path.isfile(OPENSTACK_RC_FILE):
+ if os.path.isfile(self.openstack_creds):
answer = raw_input("It seems the RC file is already present. "
"Do you want to overwrite it? [y|n]\n")
while True:
else:
answer = raw_input("Invalid answer. Please type [y|n]\n")
- CI_INSTALLER_TYPE = ft_constants.CI_INSTALLER_TYPE
- if CI_INSTALLER_TYPE is None:
+ installer_type = CONST.INSTALLER_TYPE
+ if installer_type is None:
click.echo("The environment variable 'INSTALLER_TYPE' is not"
"defined. Please export it")
- CI_INSTALLER_IP = ft_constants.CI_INSTALLER_IP
- if CI_INSTALLER_IP is None:
+ installer_ip = CONST.INSTALLER_IP
+ if installer_ip is None:
click.echo("The environment variable 'INSTALLER_IP' is not"
"defined. Please export it")
cmd = ("%s/releng/utils/fetch_os_creds.sh -d %s -i %s -a %s"
- % (ft_constants.REPOS_DIR,
- OPENSTACK_RC_FILE,
- CI_INSTALLER_TYPE,
- CI_INSTALLER_IP))
+ % (CONST.dir_repos,
+ self.openstack_creds,
+ installer_type,
+ installer_ip))
click.echo("Fetching credentials from installer node '%s' with IP=%s.."
- % (CI_INSTALLER_TYPE, CI_INSTALLER_IP))
+ % (installer_type, installer_ip))
ft_utils.execute_command(cmd, verbose=False)
def check(self):
self.ping_endpoint()
- cmd = ft_constants.FUNCTEST_REPO_DIR + "/functest/ci/check_os.sh"
+ cmd = CONST.dir_repo_functest + "/functest/ci/check_os.sh"
ft_utils.execute_command(cmd, verbose=False)
def snapshot_create(self):
self.ping_endpoint()
- if os.path.isfile(OPENSTACK_SNAPSHOT_FILE):
+ if os.path.isfile(self.snapshot_file):
answer = raw_input("It seems there is already an OpenStack "
"snapshot. Do you want to overwrite it with "
"the current OpenStack status? [y|n]\n")
os_snapshot.main()
def snapshot_show(self):
- if not os.path.isfile(OPENSTACK_SNAPSHOT_FILE):
+ if not os.path.isfile(self.snapshot_file):
click.echo("There is no OpenStack snapshot created. To create "
"one run the command "
"'functest openstack snapshot-create'")
return
- with open(OPENSTACK_SNAPSHOT_FILE, 'r') as yaml_file:
+ with open(self.snapshot_file, 'r') as yaml_file:
click.echo("\n%s"
% yaml_file.read())
def clean(self):
self.ping_endpoint()
- if not os.path.isfile(OPENSTACK_SNAPSHOT_FILE):
+ if not os.path.isfile(self.snapshot_file):
click.echo("Not possible to clean OpenStack without a snapshot. "
"This could cause problems. "
"Run first the command "
import click
import functest.ci.tier_builder as tb
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
import functest.utils.functest_vacation as vacation
-import functest.utils.functest_constants as ft_constants
class CliTestcase:
def __init__(self):
- CI_INSTALLER_TYPE = ft_constants.CI_INSTALLER_TYPE
- CI_SCENARIO = ft_constants.CI_SCENARIO
- testcases = ft_constants.FUNCTEST_TESTCASES_YAML
-
- self.tiers = tb.TierBuilder(CI_INSTALLER_TYPE, CI_SCENARIO, testcases)
+ self.tiers = tb.TierBuilder(CONST.INSTALLER_TYPE,
+ CONST.DEPLOY_SCENARIO,
+ CONST.functest_testcases_yaml)
def list(self):
summary = ""
click.echo(description)
- def run(self, testname, noclean=False):
+ @staticmethod
+ def run(testname, noclean=False):
if testname == 'vacation':
vacation.main()
- elif not os.path.isfile(ft_constants.ENV_FILE):
+ elif not os.path.isfile(CONST.env_active):
click.echo("Functest environment is not ready. "
"Run first 'functest env prepare'")
else:
for test in tests:
if noclean:
cmd = ("python %s/functest/ci/run_tests.py "
- "-n -t %s" % (ft_constants.FUNCTEST_REPO_DIR, test))
+ "-n -t %s" % (CONST.dir_repo_functest, test))
else:
cmd = ("python %s/functest/ci/run_tests.py "
- "-t %s" % (ft_constants.FUNCTEST_REPO_DIR, test))
+ "-t %s" % (CONST.dir_repo_functest, test))
ft_utils.execute_command(cmd)
import click
import functest.ci.tier_builder as tb
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
-import functest.utils.functest_constants as ft_constants
class CliTier:
def __init__(self):
- CI_INSTALLER_TYPE = ft_constants.CI_INSTALLER_TYPE
- CI_SCENARIO = ft_constants.CI_SCENARIO
- testcases = ft_constants.FUNCTEST_TESTCASES_YAML
- self.tiers = tb.TierBuilder(CI_INSTALLER_TYPE, CI_SCENARIO, testcases)
+ self.tiers = tb.TierBuilder(CONST.INSTALLER_TYPE,
+ CONST.DEPLOY_SCENARIO,
+ CONST.functest_testcases_yaml)
def list(self):
summary = ""
tests = tier.get_test_names()
click.echo("Test cases in tier '%s':\n %s\n" % (tiername, tests))
- def run(self, tiername, noclean=False):
- if not os.path.isfile(ft_constants.ENV_FILE):
+ @staticmethod
+ def run(tiername, noclean=False):
+ if not os.path.isfile(CONST.env_active):
click.echo("Functest environment is not ready. "
"Run first 'functest env prepare'")
else:
if noclean:
cmd = ("python %s/functest/ci/run_tests.py "
- "-n -t %s" % (ft_constants.FUNCTEST_REPO_DIR, tiername))
+ "-n -t %s" % (CONST.dir_repo_functest, tiername))
else:
cmd = ("python %s/functest/ci/run_tests.py "
- "-t %s" % (ft_constants.FUNCTEST_REPO_DIR, tiername))
+ "-t %s" % (CONST.dir_repo_functest, tiername))
ft_utils.execute_command(cmd)
import testcase_base as base
import functest.utils.functest_utils as ft_utils
import functest.utils.functest_logger as ft_logger
+from functest.utils.constants import CONST
class FeatureBase(base.TestcaseBase):
self.project_name = project
self.case_name = case
self.cmd = cmd
- self.repo = self.get_conf('general.dir.{}'.format(repo))
+ self.repo = CONST.__getattribute__(repo)
self.result_file = self.get_result_file()
self.logger = ft_logger.Logger(project).getLogger()
return exit_code
def get_result_file(self):
- dir = self.get_conf('general.dir.results')
- return "{}/{}.log".format(dir, self.project_name)
+ return "{}/{}.log".format(CONST.dir_results, self.project_name)
def log_results(self):
ft_utils.logger_test_results(self.project_name,
self.case_name,
self.criteria,
self.details)
-
- @staticmethod
- def get_conf(parameter):
- return ft_utils.get_functest_config(parameter)
# 0.3: add report flag to push results when needed
# 0.4: refactoring to match Test abstraction class
-import argparse
-import os
-import sys
-import time
+import functest.core.feature_base as base
-import functest.core.testcase_base as testcase_base
-import functest.utils.functest_constants as ft_constants
-import functest.utils.functest_logger as ft_logger
-import functest.utils.functest_utils as ft_utils
-
-
-class DominoCases(testcase_base.TestcaseBase):
- DOMINO_REPO = ft_constants.DOMINO_REPO_DIR
- RESULTS_DIR = ft_constants.FUNCTEST_RESULTS_DIR
- logger = ft_logger.Logger("domino").getLogger()
+class Domino(base.FeatureBase):
def __init__(self):
- super(DominoCases, self).__init__()
- self.project_name = "domino"
- self.case_name = "domino-multinode"
-
- def main(self, **kwargs):
- cmd = 'cd %s && ./tests/run_multinode.sh' % self.DOMINO_REPO
- log_file = os.path.join(self.RESULTS_DIR, "domino.log")
- start_time = time.time()
-
- ret = ft_utils.execute_command(cmd,
- output_file=log_file)
-
- stop_time = time.time()
- if ret == 0:
- self.logger.info("domino OK")
- status = 'PASS'
- else:
- self.logger.info("domino FAILED")
- status = "FAIL"
-
- # report status only if tests run (FAIL OR PASS)
- self.criteria = status
- self.start_time = start_time
- self.stop_time = stop_time
- self.details = {}
-
- def run(self):
- kwargs = {}
- return self.main(**kwargs)
-
-if __name__ == '__main__':
- parser = argparse.ArgumentParser()
- parser.add_argument("-r", "--report",
- help="Create json result file",
- action="store_true")
- args = vars(parser.parse_args())
- domino = DominoCases()
- try:
- result = domino.main(**args)
- if result != testcase_base.TestcaseBase.EX_OK:
- sys.exit(result)
- if args['report']:
- sys.exit(domino.push_to_db())
- except Exception:
- sys.exit(testcase_base.TestcaseBase.EX_RUN_ERROR)
+ super(Domino, self).__init__(project='domino',
+ case='domino-multinode',
+ repo='dir_repo_domino')
+ self.cmd = 'cd %s && ./tests/run_multinode.sh' % self.repo
#
# http://www.apache.org/licenses/LICENSE-2.0
#
-
import functest.core.feature_base as base
super(OpenDaylightSFC, self).__init__(project='sfc',
case='functest-odl-sfc"',
repo='dir_repo_sfc')
- self.cmd = 'cd %s/tests/functest && python ./run_tests.py' % self.repo
+ dir_sfc_functest = '{}/sfc/tests/functest'.format(self.repo)
+ self.cmd = 'cd %s && python ./run_tests.py' % dir_sfc_functest
#
# http://www.apache.org/licenses/LICENSE-2.0
#
+import functest.core.feature_base as base
-import argparse
-import os
-import sys
-import time
-
-import functest.core.testcase_base as testcase_base
-import functest.utils.functest_constants as ft_constants
-import functest.utils.functest_logger as ft_logger
-import functest.utils.functest_utils as ft_utils
-
-
-class SdnVpnTests(testcase_base.TestcaseBase):
- SDNVPN_REPO_TESTS = os.path.join(
- ft_constants.SDNVPN_REPO_DIR, "tests/functest")
- logger = ft_logger.Logger("sdnvpn").getLogger()
+class SdnVpnTests(base.FeatureBase):
def __init__(self):
- super(SdnVpnTests, self).__init__()
- self.project_name = "sdnvpn"
- self.case_name = "bgpvpn"
-
- def main(self, **kwargs):
- os.chdir(self.SDNVPN_REPO_TESTS)
- cmd = 'run_tests.py'
- log_file = os.path.join(
- ft_constants.FUNCTEST_RESULTS_DIR, "sdnvpn.log")
- start_time = time.time()
-
- ret = ft_utils.execute_command(cmd,
- output_file=log_file)
-
- stop_time = time.time()
- if ret == 0:
- self.logger.info("%s OK" % self.case_name)
- status = 'PASS'
- else:
- self.logger.info("%s FAILED" % self.case_name)
- status = "FAIL"
-
- # report status only if tests run (FAIL OR PASS)
- self.criteria = status
- self.start_time = start_time
- self.stop_time = stop_time
- self.details = {}
-
- def run(self):
- kwargs = {}
- return self.main(**kwargs)
-
-
-if __name__ == '__main__':
- parser = argparse.ArgumentParser()
- parser.add_argument("-r", "--report",
- help="Create json result file",
- action="store_true")
- args = vars(parser.parse_args())
- sdnvpn = SdnVpnTests()
- try:
- result = sdnvpn.main(**args)
- if result != testcase_base.TestcaseBase.EX_OK:
- sys.exit(result)
- if args['report']:
- sys.exit(sdnvpn.push_to_db())
- except Exception:
- sys.exit(testcase_base.TestcaseBase.EX_RUN_ERROR)
+ super(SdnVpnTests, self).__init__(project='sdnvpn',
+ case='bgpvpn',
+ repo='dir_repo_sdnvpn')
+ dir_sfc_functest = '{}/sdnvpn/test/functest'.format(self.repo)
+ self.cmd = 'cd %s && python ./run_tests.py' % dir_sfc_functest
import re
import shutil
-import functest.utils.functest_constants as ft_constants
-import functest.utils.functest_utils as ft_utils
import opnfv.utils.constants as releng_constants
+import functest.utils.functest_utils as ft_utils
+from functest.utils.constants import CONST
IMAGE_ID_ALT = None
FLAVOR_ID_ALT = None
-REPO_PATH = ft_constants.FUNCTEST_REPO_DIR
-GLANCE_IMAGE_PATH = os.path.join(ft_constants.FUNCTEST_DATA_DIR,
- ft_constants.GLANCE_IMAGE_FILENAME)
-TEMPEST_TEST_LIST_DIR = ft_constants.TEMPEST_TEST_LIST_DIR
-TEMPEST_RESULTS_DIR = os.path.join(ft_constants.FUNCTEST_RESULTS_DIR,
+REPO_PATH = CONST.dir_repo_functest
+GLANCE_IMAGE_PATH = os.path.join(CONST.dir_functest_data,
+ CONST.openstack_image_file_name)
+TEMPEST_TEST_LIST_DIR = CONST.dir_tempest_cases
+TEMPEST_RESULTS_DIR = os.path.join(CONST.dir_results,
'tempest')
TEMPEST_CUSTOM = os.path.join(REPO_PATH, TEMPEST_TEST_LIST_DIR,
'test_list.txt')
TEMPEST_RAW_LIST = os.path.join(TEMPEST_RESULTS_DIR, 'test_raw_list.txt')
TEMPEST_LIST = os.path.join(TEMPEST_RESULTS_DIR, 'test_list.txt')
-CI_INSTALLER_TYPE = ft_constants.CI_INSTALLER_TYPE
-CI_INSTALLER_IP = ft_constants.CI_INSTALLER_IP
+CI_INSTALLER_TYPE = CONST.INSTALLER_TYPE
+CI_INSTALLER_IP = CONST.INSTALLER_IP
def configure_tempest(logger, deployment_dir, IMAGE_ID=None, FLAVOR_ID=None):
config.set(
'compute',
'fixed_network_name',
- ft_constants.TEMPEST_PRIVATE_NET_NAME)
- if ft_constants.TEMPEST_USE_CUSTOM_IMAGES:
+ CONST.tempest_private_net_name)
+ if CONST.tempest_use_custom_images:
if IMAGE_ID is not None:
config.set('compute', 'image_ref', IMAGE_ID)
if IMAGE_ID_ALT is not None:
config.set('compute', 'image_ref_alt', IMAGE_ID_ALT)
- if ft_constants.TEMPEST_USE_CUSTOM_FLAVORS:
+ if CONST.tempest_use_custom_flavors:
if FLAVOR_ID is not None:
config.set('compute', 'flavor_ref', FLAVOR_ID)
if FLAVOR_ID_ALT is not None:
config.set('compute', 'flavor_ref_alt', FLAVOR_ID_ALT)
- config.set('identity', 'tenant_name', ft_constants.TEMPEST_TENANT_NAME)
- config.set('identity', 'username', ft_constants.TEMPEST_USER_NAME)
- config.set('identity', 'password', ft_constants.TEMPEST_USER_PASSWORD)
+ config.set('identity', 'tenant_name', CONST.tempest_identity_tenant_name)
+ config.set('identity', 'username', CONST.tempest_identity_user_name)
+ config.set('identity', 'password', CONST.tempest_identity_user_password)
config.set(
- 'validation', 'ssh_timeout', ft_constants.TEMPEST_SSH_TIMEOUT)
+ 'validation', 'ssh_timeout', CONST.tempest_validation_ssh_timeout)
- if ft_constants.OS_ENDPOINT_TYPE is not None:
+ if CONST.OS_ENDPOINT_TYPE is not None:
services_list = ['compute',
'volume',
'image',
if service not in sections:
config.add_section(service)
config.set(service, 'endpoint_type',
- ft_constants.OS_ENDPOINT_TYPE)
+ CONST.OS_ENDPOINT_TYPE)
with open(tempest_conf_file, 'wb') as config_file:
config.write(config_file)
kingbird_conf_path = "/etc/kingbird/kingbird.conf"
installer_type = CI_INSTALLER_TYPE
installer_ip = CI_INSTALLER_IP
- installer_username = ft_utils.get_functest_config(
- "multisite." + installer_type +
- "_environment.installer_username")
- installer_password = ft_utils.get_functest_config(
- "multisite." + installer_type +
- "_environment.installer_password")
+ installer_username = CONST.__getattribute__(
+ 'multisite_{}_installer_username'.format(installer_type))
+ installer_password = CONST.__getattribute__(
+ 'multisite_{}_installer_password'.format(installer_type))
ssh_options = ("-o UserKnownHostsFile=/dev/null -o "
"StrictHostKeyChecking=no")
import subprocess
import time
+import opnfv.utils.constants as releng_constants
import yaml
import conf_utils
import functest.utils.functest_logger as ft_logger
import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as os_utils
-import functest.utils.functest_constants as ft_constants
-import opnfv.utils.constants as releng_constants
+from functest.utils.constants import CONST
""" logging configuration """
logger = ft_logger.Logger("Tempest").getLogger()
self.OPTION = ""
self.FLAVOR_ID = None
self.IMAGE_ID = None
- self.DEPLOYMENT_DIR = ft_utils.get_deployment_dir()
-
- def read_file(self, filename):
+ self.DEPLOYMENT_DIR = self.get_deployment_dir()
+
+ @staticmethod
+ def get_deployment_dir():
+ """
+ Returns current Rally deployment directory
+ """
+ cmd = ("rally deployment list | awk '/" +
+ CONST.rally_deployment_name +
+ "/ {print $2}'")
+ p = subprocess.Popen(cmd, shell=True,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
+ deployment_uuid = p.stdout.readline().rstrip()
+ if deployment_uuid == "":
+ logger.error("Rally deployment not found.")
+ exit(-1)
+ return os.path.join(CONST.dir_rally_inst,
+ "tempest/for-deployment-" + deployment_uuid)
+
+ @staticmethod
+ def read_file(filename):
with open(filename) as src:
return [line.strip() for line in src.readlines()]
logger.debug("Creating tenant and user for Tempest suite")
tenant_id = os_utils.create_tenant(
keystone_client,
- ft_constants.TEMPEST_TENANT_NAME,
- ft_constants.TEMPEST_TENANT_DESCRIPTION)
+ CONST.tempest_identity_tenant_name,
+ CONST.tempest_identity_tenant_description)
if not tenant_id:
logger.error("Error : Failed to create %s tenant"
- % ft_constants.TEMPEST_TENANT_NAME)
+ % CONST.tempest_identity_tenant_name)
user_id = os_utils.create_user(keystone_client,
- ft_constants.TEMPEST_USER_NAME,
- ft_constants.TEMPEST_USER_PASSWORD,
+ CONST.tempest_identity_user_name,
+ CONST.tempest_identity_user_password,
None, tenant_id)
if not user_id:
logger.error("Error : Failed to create %s user" %
- ft_constants.TEMPEST_USER_NAME)
+ CONST.tempest_identity_user_name)
logger.debug("Creating private network for Tempest suite")
network_dic = \
os_utils.create_shared_network_full(
- ft_constants.TEMPEST_PRIVATE_NET_NAME,
- ft_constants.TEMPEST_PRIVATE_SUBNET_NAME,
- ft_constants.TEMPEST_ROUTER_NAME,
- ft_constants.TEMPEST_PRIVATE_SUBNET_CIDR)
+ CONST.tempest_private_net_name,
+ CONST.tempest_private_subnet_name,
+ CONST.tempest_router_name,
+ CONST.tempest_private_subnet_cidr)
if not network_dic:
return releng_constants.EXIT_RUN_ERROR
- if ft_constants.TEMPEST_USE_CUSTOM_IMAGES:
+ if CONST.tempest_use_custom_images:
# adding alternative image should be trivial should we need it
logger.debug("Creating image for Tempest suite")
_, self.IMAGE_ID = os_utils.get_or_create_image(
- ft_constants.GLANCE_IMAGE_NAME, conf_utils.GLANCE_IMAGE_PATH,
- ft_constants.GLANCE_IMAGE_FORMAT)
+ CONST.openstack_image_name, conf_utils.GLANCE_IMAGE_PATH,
+ CONST.openstack_image_disk_format)
if not self.IMAGE_ID:
return releng_constants.EXIT_RUN_ERROR
- if ft_constants.TEMPEST_USE_CUSTOM_FLAVORS:
+ if CONST.tempest_use_custom_flavors:
# adding alternative flavor should be trivial should we need it
logger.debug("Creating flavor for Tempest suite")
_, self.FLAVOR_ID = os_utils.get_or_create_flavor(
- ft_constants.FLAVOR_NAME,
- ft_constants.FLAVOR_RAM,
- ft_constants.FLAVOR_DISK,
- ft_constants.FLAVOR_VCPUS)
+ CONST.openstack_flavor_name,
+ CONST.openstack_flavor_ram,
+ CONST.openstack_flavor_disk,
+ CONST.openstack_flavor_vcpus)
if not self.FLAVOR_ID:
return releng_constants.EXIT_RUN_ERROR
result_file = open(conf_utils.TEMPEST_LIST, 'w')
black_tests = []
try:
- installer_type = ft_constants.CI_INSTALLER_TYPE
- deploy_scenario = ft_constants.CI_SCENARIO
+ installer_type = CONST.INSTALLER_TYPE
+ deploy_scenario = CONST.DEPLOY_SCENARIO
if (bool(installer_type) * bool(deploy_scenario)):
# if INSTALLER_TYPE and DEPLOY_SCENARIO are set we read the
# file
header = ("Tempest environment:\n"
" Installer: %s\n Scenario: %s\n Node: %s\n Date: %s\n" %
- (ft_constants.CI_INSTALLER_TYPE,
- ft_constants.CI_SCENARIO,
- ft_constants.CI_NODE,
+ (CONST.INSTALLER_TYPE,
+ CONST.DEPLOY_SCENARIO,
+ CONST.NODE_NAME,
time.strftime("%a %b %d %H:%M:%S %Z %Y")))
f_stdout = open(conf_utils.TEMPEST_RESULTS_DIR + "/tempest.log", 'w+')
try:
with open(self.config_functest) as f:
self.functest_yaml = yaml.safe_load(f)
- self.parse(None, self.functest_yaml)
+ self._parse(None, self.functest_yaml)
except:
raise Exception('Parse {} failed'.format(self.config_functest))
+ self._set_others()
- def parse(self, attr_now, left_parametes):
+ def _parse(self, attr_now, left_parametes):
for param_n, param_v in left_parametes.iteritems():
- attr_further = self.get_attr_further(attr_now, param_n)
+ attr_further = self._get_attr_further(attr_now, param_n)
if not isinstance(param_v, dict):
self.__setattr__(attr_further, param_v)
else:
- self.parse(attr_further, param_v)
+ self._parse(attr_further, param_v)
- def get_attr_further(self, attr_now, next):
+ def _get_attr_further(self, attr_now, next):
return attr_now if next == 'general' else (
'{}_{}'.format(attr_now, next) if attr_now else next)
+ def _set_others(self):
+ self.env_active = os.path.join(self.dir_functest_conf, "env_active")
+
+
CONF = Config()
import os
+import re
default_envs = {
'NODE_NAME': 'unknown_pod',
'DEPLOY_TYPE': 'virt',
'INSTALLER_TYPE': None,
'INSTALLER_IP': None,
- 'BUILD_TAG': None
+ 'BUILD_TAG': None,
+ 'OS_ENDPOINT_TYPE': None,
+ 'OS_AUTH_URL': None
}
for k, v in default_envs.iteritems():
if k not in os.environ:
self.__setattr__(k, v)
+ self._set_ci_run()
+ self._set_ci_loop()
+
+ def _set_ci_run(self):
+ if self.BUILD_TAG:
+ self.IS_CI_RUN = True
+ else:
+ self.IS_CI_RUN = False
+
+ def _set_ci_loop(self):
+ if self.BUILD_TAG and re.search("daily", self.BUILD_TAG):
+ self.CI_LOOP = "daily"
+ else:
+ self.CI_LOOP = "weekly"
ENV = Environment()
# http://www.apache.org/licenses/LICENSE-2.0
#
import os
-import functest.utils.functest_utils as ft_utils
+
import functest.utils.functest_logger as ft_logger
+import functest.utils.functest_utils as ft_utils
logger = ft_logger.Logger("functest_constants").getLogger()
HOME = get_value('general.dir.dir_home', 'HOME')
-REPOS_DIR = get_value('general.dir.dir_repos', 'REPOS_DIR')
+REPOS_DIR = get_value('general.dir.repos', 'REPOS_DIR')
FUNCTEST_BASE_DIR = get_value('general.dir.functest',
'FUNCTEST_BASE_DIR')
FUNCTEST_REPO_DIR = get_value('general.dir.repo_functest',
'FUNCTEST_TESTCASES_YAML')
RALLY_DEPLOYMENT_NAME = get_value('rally.deployment_name',
'RALLY_DEPLOYMENT_NAME')
-TEMPEST_REPO_DIR = get_value('general.dir.dir_repo_tempest',
+TEMPEST_REPO_DIR = get_value('general.dir.repo_tempest',
'TEMPEST_REPO_DIR')
ENV_FILE = os.path.join(FUNCTEST_CONF_DIR, "env_active")
OPENSTACK_SNAPSHOT_FILE = get_value('general.openstack.snapshot_file',
'OPENSTACK_SNAPSHOT_FILE')
-DOMINO_REPO_DIR = get_value('general.dir.dir_repo_domino',
+DOMINO_REPO_DIR = get_value('general.dir.repo_domino',
'DOMINO_REPO_DIR')
-SDNVPN_REPO_DIR = get_value('general.dir.dir_repo_sdnvpn',
+SDNVPN_REPO_DIR = get_value('general.dir.repo_sdnvpn',
'SDNVPN_REPO_DIR')
-SFC_REPO_DIR = get_value('general.dir.dir_repo_sfc',
+SFC_REPO_DIR = get_value('general.dir.repo_sfc',
'SFC_REPO_DIR')
ONOS_SFC_IMAGE_NAME = get_value('onos_sfc.image_name',
RALLY_PRIVATE_SUBNET_CIDR = get_value('rally.subnet_cidr',
'RALLY_PRIVATE_SUBNET_CIDR')
RALLY_ROUTER_NAME = get_value('rally.router_name', 'RALLY_ROUTER_NAME')
-RALLY_INSTALLATION_DIR = get_value('general.dir.dir_rally_inst',
+RALLY_INSTALLATION_DIR = get_value('general.dir.rally_inst',
'RALLY_INSTALLATION_DIR')
GLANCE_IMAGE_NAME = get_value('general.openstack.image_name',
'GLANCE_IMAGE_NAME')
'TEMPEST_USE_CUSTOM_IMAGES')
TEMPEST_USE_CUSTOM_FLAVORS = get_value('tempest.use_custom_flavors',
'TEMPEST_USE_CUSTOM_FLAVORS')
-TEMPEST_TEST_LIST_DIR = get_value('general.dir.dir_tempest_cases',
+TEMPEST_TEST_LIST_DIR = get_value('general.dir.tempest_cases',
'TEMPEST_TEST_LIST_DIR')
NAME_VM_1 = get_value('vping.vm_name_1', 'NAME_VM_1')
NAME_VM_2 = get_value('vping.vm_name_2', 'NAME_VM_2')
'PROMISE_ROUTER_NAME')
DOCTOR_REPO_DIR = get_value('general.dir.dir_repo_doctor',
'DOCTOR_REPO_DIR')
-COPPER_REPO_DIR = get_value('general.dir.dir_repo_copper',
+COPPER_REPO_DIR = get_value('general.dir.repo_copper',
'COPPER_REPO_DIR')
EXAMPLE_INSTANCE_NAME = get_value('example.example_vm_name',
'EXAMPLE_INSTANCE_NAME')
CW_INPUTS = get_value('vIMS.clearwater.inputs', 'CW_INPUTS')
CW_REQUIERMENTS = get_value('vIMS.clearwater.requierments',
'CW_REQUIERMENTS')
-PARSER_REPO_DIR = get_value('general.dir.dir_repo_parser',
+PARSER_REPO_DIR = get_value('general.dir.repo_parser',
'PARSER_REPO_DIR')
return returncode
-def get_deployment_dir():
- """
- Returns current Rally deployment directory
- """
- deployment_name = get_functest_config('rally.deployment_name')
- rally_dir = get_functest_config('general.dir.dir_rally_inst')
- cmd = ("rally deployment list | awk '/" + deployment_name +
- "/ {print $2}'")
- p = subprocess.Popen(cmd, shell=True,
- stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT)
- deployment_uuid = p.stdout.readline().rstrip()
- if deployment_uuid == "":
- logger.error("Rally deployment not found.")
- exit(-1)
- deployment_dir = (rally_dir + "/tempest/for-deployment-" +
- deployment_uuid)
- return deployment_dir
-
-
def get_dict_by_test(testname):
with open(get_testcases_file_dir()) as f:
testcases_yaml = yaml.safe_load(f)
#
import time
+
+import yaml
+
import functest.utils.functest_logger as ft_logger
import functest.utils.openstack_utils as os_utils
-import yaml
-import functest.utils.functest_constants as ft_constants
+from functest.utils.constants import CONST
logger = ft_logger.Logger("openstack_clean").getLogger()
-OS_SNAPSHOT_FILE = ft_constants.OPENSTACK_SNAPSHOT_FILE
+OS_SNAPSHOT_FILE = CONST.openstack_snapshot_file
def separator():
# http://www.apache.org/licenses/LICENSE-2.0
#
+import yaml
+
import functest.utils.functest_logger as ft_logger
import functest.utils.openstack_utils as os_utils
-import yaml
-import functest.utils.functest_constants as ft_constants
+from functest.utils.constants import CONST
logger = ft_logger.Logger("openstack_snapshot").getLogger()
-OS_SNAPSHOT_FILE = ft_constants.OPENSTACK_SNAPSHOT_FILE
+OS_SNAPSHOT_FILE = CONST.openstack_snapshot_file
def separator():
Code Review / functest.git / commitdiff