# pylint: disable=too-many-instance-attributes
"""TempestCommon testcases implementation class."""
- TEMPEST_RESULTS_DIR = os.path.join(
- getattr(config.CONF, 'dir_results'), 'tempest')
-
visibility = 'public'
def __init__(self, **kwargs):
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'tempest'
super(TempestCommon, self).__init__(**kwargs)
- self.mode = ""
- self.option = []
self.verifier_id = conf_utils.get_verifier_id()
self.verifier_repo_dir = conf_utils.get_verifier_repo_dir(
self.verifier_id)
self.deployment_dir = conf_utils.get_verifier_deployment_dir(
self.verifier_id, self.deployment_id)
self.verification_id = None
- self.res_dir = TempestCommon.TEMPEST_RESULTS_DIR
+ self.res_dir = os.path.join(
+ getattr(config.CONF, 'dir_results'), self.case_name)
self.raw_list = os.path.join(self.res_dir, 'test_raw_list.txt')
self.list = os.path.join(self.res_dir, 'test_list.txt')
self.conf_file = None
shutil.copyfile(conf_file,
os.path.join(res_dir, 'tempest.conf'))
- def generate_test_list(self):
+ def generate_test_list(self, **kwargs):
"""Generate test list based on the test mode."""
LOGGER.debug("Generating test case list...")
self.backup_tempest_config(self.conf_file, '/etc')
- if self.mode == 'custom':
+ if kwargs.get('mode') == 'custom':
if os.path.isfile(conf_utils.TEMPEST_CUSTOM):
shutil.copyfile(
conf_utils.TEMPEST_CUSTOM, self.list)
raise Exception("Tempest test list file %s NOT found."
% conf_utils.TEMPEST_CUSTOM)
else:
- if self.mode == 'smoke':
- testr_mode = r"'^tempest\.(api|scenario).*\[.*\bsmoke\b.*\]$'"
- elif self.mode == 'full':
- testr_mode = r"'^tempest\.'"
- else:
- testr_mode = self.mode
- cmd = "(cd {0}; stestr list {1} >{2} 2>/dev/null)".format(
+ testr_mode = kwargs.get(
+ 'mode', r'^tempest\.(api|scenario).*\[.*\bsmoke\b.*\]$')
+ cmd = "(cd {0}; stestr list '{1}' >{2} 2>/dev/null)".format(
self.verifier_repo_dir, testr_mode, self.list)
output = subprocess.check_output(cmd, shell=True)
LOGGER.info("%s\n%s", cmd, output)
result_file.write(str(cases_line) + '\n')
result_file.close()
- def run_verifier_tests(self):
+ def run_verifier_tests(self, **kwargs):
"""Execute tempest test cases."""
cmd = ["rally", "verify", "start", "--load-list",
self.list]
- cmd.extend(self.option)
+ cmd.extend(kwargs.get('option', []))
LOGGER.info("Starting Tempest test suite: '%s'.", cmd)
f_stdout = open(
subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
+ def update_rally_regex(self, rally_conf='/etc/rally/rally.conf'):
+ """Set image name as tempest img_name_regex"""
+ rconfig = configparser.RawConfigParser()
+ rconfig.read(rally_conf)
+ rconfig.add_section('tempest')
+ rconfig.set('tempest', 'img_name_regex', '^{}$'.format(
+ self.image.name))
+ with open(rally_conf, 'wb') as config_file:
+ rconfig.write(config_file)
+
def configure(self, **kwargs): # pylint: disable=unused-argument
"""
Create all openstack resources for tempest-based testcases and write
def run(self, **kwargs):
self.start_time = time.time()
try:
- super(TempestCommon, self).run(**kwargs)
+ assert super(TempestCommon, self).run(
+ **kwargs) == testcase.TestCase.EX_OK
+ self.update_rally_regex()
self.configure(**kwargs)
- self.generate_test_list()
+ self.generate_test_list(**kwargs)
self.apply_tempest_blacklist()
- self.run_verifier_tests()
+ self.run_verifier_tests(**kwargs)
self.parse_verifier_result()
self.generate_report()
res = testcase.TestCase.EX_OK
except Exception: # pylint: disable=broad-except
LOGGER.exception('Error with run')
+ self.result = 0
res = testcase.TestCase.EX_RUN_ERROR
self.stop_time = time.time()
return res
Cleanup all OpenStack objects. Should be called on completion.
"""
super(TempestCommon, self).clean()
- self.cloud.delete_image(self.image_alt)
- self.orig_cloud.delete_flavor(self.flavor_alt.id)
-
-
-class TempestSmokeSerial(TempestCommon):
- """Tempest smoke serial testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'tempest_smoke_serial'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "smoke"
- self.option = ["--concurrency", "1"]
-
-
-class TempestNeutronTrunk(TempestCommon):
- """Tempest neutron trunk testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'neutron_trunk'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "'neutron_tempest_plugin.(api|scenario).test_trunk'"
- self.res_dir = os.path.join(
- getattr(config.CONF, 'dir_results'), 'neutron_trunk')
- self.raw_list = os.path.join(self.res_dir, 'test_raw_list.txt')
- self.list = os.path.join(self.res_dir, 'test_list.txt')
-
- def configure(self, **kwargs):
- super(TempestNeutronTrunk, self).configure(**kwargs)
- rconfig = configparser.RawConfigParser()
- rconfig.read(self.conf_file)
- rconfig.set('network-feature-enabled', 'api_extensions', 'all')
- with open(self.conf_file, 'wb') as config_file:
- rconfig.write(config_file)
-
-
-class TempestBarbican(TempestCommon):
- """Tempest Barbican testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'barbican'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "'barbican_tempest_plugin.tests.(api|scenario)'"
- self.res_dir = os.path.join(
- getattr(config.CONF, 'dir_results'), 'barbican')
- self.raw_list = os.path.join(self.res_dir, 'test_raw_list.txt')
- self.list = os.path.join(self.res_dir, 'test_list.txt')
-
-
-class TempestSmokeParallel(TempestCommon):
- """Tempest smoke parallel testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'tempest_smoke_parallel'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "smoke"
-
-
-class TempestFullParallel(TempestCommon):
- """Tempest full parallel testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'tempest_full_parallel'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "full"
-
-
-class TempestCustom(TempestCommon):
- """Tempest custom testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'tempest_custom'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "custom"
- self.option = ["--concurrency", "1"]
-
-
-class TempestDefcore(TempestCommon):
- """Tempest Defcore testcase implementation."""
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'tempest_defcore'
- TempestCommon.__init__(self, **kwargs)
- self.mode = "defcore"
- self.option = ["--concurrency", "1"]
+ if self.image_alt:
+ self.cloud.delete_image(self.image_alt)
+ if self.flavor_alt:
+ self.orig_cloud.delete_flavor(self.flavor_alt.id)
Code Review / functest.git / blobdiff