"""TempestCommon testcases implementation class."""
visibility = 'public'
+ filename_alt = '/home/opnfv/functest/images/cirros-0.4.0-x86_64-disk.img'
def __init__(self, **kwargs):
if "case_name" not in kwargs:
self.conf_file = None
self.image_alt = None
self.flavor_alt = None
+ self.services = []
+ try:
+ self.services = kwargs['run']['args']['services']
+ except Exception: # pylint: disable=broad-except
+ pass
+ self.neutron_extensions = []
+ try:
+ self.neutron_extensions = kwargs['run']['args'][
+ 'neutron_extensions']
+ except Exception: # pylint: disable=broad-except
+ pass
+
+ def check_services(self):
+ """Check the mandatory services."""
+ for service in self.services:
+ try:
+ self.cloud.search_services(service)[0]
+ except Exception: # pylint: disable=broad-except
+ self.is_skipped = True
+ break
+
+ def check_extensions(self):
+ """Check the mandatory network extensions."""
+ extensions = self.cloud.get_network_extensions()
+ for network_extension in self.neutron_extensions:
+ if network_extension not in extensions:
+ LOGGER.warning(
+ "Cannot find Neutron extension: %s", network_extension)
+ self.is_skipped = True
+ break
+
+ def check_requirements(self):
+ self.check_services()
+ self.check_extensions()
@staticmethod
def read_file(filename):
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
for line in proc.stdout:
+ LOGGER.info(line.rstrip())
new_line = line.replace(' ', '').split('|')
if 'Tests' in new_line:
break
- LOGGER.info(line)
if 'Testscount' in new_line:
result['num_tests'] = int(new_line[2])
elif 'Success' in new_line:
f_stdout = open(
os.path.join(self.res_dir, "tempest.log"), 'w+')
- f_stderr = open(
- os.path.join(self.res_dir,
- "tempest-error.log"), 'w+')
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
- stderr=f_stderr,
+ stderr=subprocess.STDOUT,
bufsize=1)
with proc.stdout:
for line in iter(proc.stdout.readline, b''):
if re.search(r"\} tempest\.", line):
- LOGGER.info(line.replace('\n', ''))
+ LOGGER.info(line.rstrip())
elif re.search(r'(?=\(UUID=(.*)\))', line):
self.verification_id = re.search(
r'(?=\(UUID=(.*)\))', line).group(1)
- LOGGER.info('Verification UUID: %s', self.verification_id)
f_stdout.write(line)
proc.wait()
-
f_stdout.close()
- f_stderr.close()
if self.verification_id is None:
raise Exception('Verification UUID not found')
+ LOGGER.info('Verification UUID: %s', self.verification_id)
def parse_verifier_result(self):
"""Parse and save test results."""
return
with open(os.path.join(self.res_dir,
- "tempest-error.log"), 'r') as logfile:
+ "tempest.log"), 'r') as logfile:
output = logfile.read()
success_testcases = []
"""Set image name as tempest img_name_regex"""
rconfig = configparser.RawConfigParser()
rconfig.read(rally_conf)
+ if not rconfig.has_section('tempest'):
+ 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 update_default_role(self, rally_conf='/etc/rally/rally.conf'):
+ """Detect and update the default role if required"""
+ role = self.get_default_role(self.cloud)
+ if not role:
+ return
+ rconfig = configparser.RawConfigParser()
+ rconfig.read(rally_conf)
+ if not rconfig.has_section('tempest'):
+ rconfig.add_section('tempest')
+ rconfig.set('tempest', 'swift_operator_role', role.name)
+ with open(rally_conf, 'wb') as config_file:
+ rconfig.write(config_file)
+
+ def update_rally_logs(self, rally_conf='/etc/rally/rally.conf'):
+ """Print rally logs in res dir"""
+ if not os.path.exists(self.res_dir):
+ os.makedirs(self.res_dir)
+ rconfig = configparser.RawConfigParser()
+ rconfig.read(rally_conf)
+ rconfig.set('DEFAULT', 'log-file', 'rally.log')
+ rconfig.set('DEFAULT', 'log_dir', self.res_dir)
+ with open(rally_conf, 'wb') as config_file:
+ rconfig.write(config_file)
+
+ @staticmethod
+ def clean_rally_conf(rally_conf='/etc/rally/rally.conf'):
+ """Clean Rally config"""
+ rconfig = configparser.RawConfigParser()
+ rconfig.read(rally_conf)
+ if rconfig.has_option('tempest', 'img_name_regex'):
+ rconfig.remove_option('tempest', 'img_name_regex')
+ if rconfig.has_option('tempest', 'swift_operator_role'):
+ rconfig.remove_option('tempest', 'swift_operator_role')
+ if rconfig.has_option('DEFAULT', 'log-file'):
+ rconfig.remove_option('DEFAULT', 'log-file')
+ if rconfig.has_option('DEFAULT', 'log_dir'):
+ rconfig.remove_option('DEFAULT', 'log_dir')
+ 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
os.makedirs(self.res_dir)
compute_cnt = len(self.cloud.list_hypervisors())
- self.image_alt = self.publish_image(
- '{}-img_alt_{}'.format(self.case_name, self.guid))
+ self.image_alt = self.publish_image_alt()
self.flavor_alt = self.create_flavor_alt()
LOGGER.debug("flavor: %s", self.flavor_alt)
assert super(TempestCommon, self).run(
**kwargs) == testcase.TestCase.EX_OK
self.update_rally_regex()
+ self.update_default_role()
+ self.update_rally_logs()
self.configure(**kwargs)
self.generate_test_list(**kwargs)
self.apply_tempest_blacklist()
"""
Cleanup all OpenStack objects. Should be called on completion.
"""
- super(TempestCommon, self).clean()
+ self.clean_rally_conf()
if self.image_alt:
self.cloud.delete_image(self.image_alt)
if self.flavor_alt:
self.orig_cloud.delete_flavor(self.flavor_alt.id)
+ super(TempestCommon, self).clean()
Code Review / functest.git / blobdiff