FIX Background tests

Current solution:
Implemented:
1) first time it is executed it should start the process and verify that
   ping returns some successful results
2) other times the test is executed it should simply report it has been
   successful since the last time it has been executed

Change-Id: I02b13e9dd710710f6883655de62da9f2f0b1d187
This commit is contained in:
r 2022-02-12 01:16:58 +02:00 committed by Federico Ressi
parent 95febbd80f
commit 46075f0df6
4 changed files with 39 additions and 15 deletions

View File

@ -468,7 +468,8 @@ def get_vm_ping_log_files(glob_ping_log_pattern='tobiko_ping_results/ping_'
def rename_ping_staistics_file_to_checked(filepath):
"""append _checked to a ping statistics file once finished it's check"""
os.rename(filepath, f'{filepath}_checked')
check_time = time.strftime("%Y_%m_%d-%H-%M-%S")
os.rename(filepath, f'{filepath}_checked_{check_time}')
def check_ping_statistics(failure_limit=10):
@ -485,8 +486,7 @@ def check_ping_statistics(failure_limit=10):
ping_line = json.loads(ping_line.rstrip())
if ping_line['transmitted'] != ping_line['received']:
failure_counter += 1
LOG.debug(f'found ping failure to :'
f' {ping_line["destination"]}')
LOG.info(f'found ping failure: {ping_line}')
if failure_counter >= failure_limit:
rename_ping_staistics_file_to_checked(filename)
tobiko.fail(f'{failure_counter} pings failure found '

View File

@ -541,26 +541,40 @@ def get_bg_procs_pids(bg_process_name):
def check_or_start_background_process(bg_function=None,
bg_process_name=None,
check_function=None, **kwargs):
""" Check if process exists, if so stop the process,
then execute some check logic i.e. a check function.
check_function=None,
**kwargs):
""" Check if process exists, if so restart the process,
execute some check logic i.e. a check function.
if the process by name isn't running,
start a separate process i.e a background function
start a new separate process i.e a background function
params:
bg_process_name= process name
bg_function: function name
check_function: function name """
procs_running_list = get_bg_procs_pids(bg_process_name)
if procs_running_list:
stop_process(procs_running_list)
# in any case test is still running, check for failures:
# execute process check i.e. go over process results file
# truncate the log file and restart the background process
LOG.info(f'running a check function: {check_function} '
f'on results of processes: {bg_process_name}')
check_function()
else: # if background process is not present , start one:
# if we want to terminate the specific background process by
# name, close it, otherwise the check will continue to run in the
# background
stop_process(procs_running_list)
LOG.info('checked and stopped previous background processes and log '
'starting a new background process ')
else:
# First time the test is run:
# if background process by specific name is not present ,
# start one in the background:
LOG.info(f'No previous background processes found:'
f' {bg_process_name}, starting a new background process '
f'of function: {bg_function}')
start_background_process(bg_function=bg_function,
bg_process_name=bg_process_name, **kwargs)
start_background_process(bg_function=bg_function,
bg_process_name=bg_process_name, **kwargs)
# check test is not failing from the start
check_function()

View File

@ -125,9 +125,6 @@ class DisruptTripleoNodesTest(testtools.TestCase):
def test_0vercloud_health_check(self):
OvercloudHealthCheck.run_before(skip_mac_table_size_test=False)
def test_check_background_vm_ping(self):
nova.check_or_start_background_vm_ping()
def test_hard_reboot_controllers_recovery(self):
OvercloudHealthCheck.run_before()
cloud_disruptions.reset_all_controller_nodes()

View File

@ -23,6 +23,8 @@ from tobiko.openstack import nova
from tobiko.openstack import stacks
from tobiko.shell import ping
from tobiko.shell import sh
from tobiko.tripleo import undercloud
from tobiko.tripleo import nova as tripleo_nova
@pytest.mark.minimal
@ -57,6 +59,17 @@ class NetworkTest(testtools.TestCase):
self.assertEqual(self.stack.network_stack.ha,
gateway['ha'])
@pytest.mark.background
@undercloud.skip_if_missing_undercloud
def test_check_background_vm_ping(self):
""" Tests that are designed to run in the background ,
then collect results.
Logic: checks if process exists, if so stop the process,
then execute some check logic i.e. a check function.
if the process by name isn't running,
start a separate process i.e a background function"""
tripleo_nova.check_or_start_background_vm_ping()
@pytest.mark.migrate_server
class SameHostNetworkTest(NetworkTest):