Pytest: Add high availability test cases

Convert the following robot test cases to pytest test cases

    1. 24-Test-HA-Service-List-Display.robot
    2. 29-Test-HA-Service-Group-List-Display.robot
    3. 53-Kill-SM-Services.robot

Signed-off-by: Dongqi Chen <chen.dq@neusoft.com>
Change-Id: Id5528a2c28e7b8dcc5c6c6548380d76d1c1be058
This commit is contained in:
Dongqi Chen 2021-06-24 10:44:24 +08:00
parent a68778f64d
commit 50c6bd26fa
7 changed files with 123 additions and 1 deletions

View File

@ -2,4 +2,4 @@
host=review.opendev.org
port=29418
project=starlingx/test.git
defaultbranch=r/stx.4.0
defaultbranch=devel

View File

@ -0,0 +1,13 @@
from pytest import fixture, skip
from utils.clients import ssh
from testfixtures.resource_mgmt import *
from testfixtures.config_host import *
@fixture(scope='session')
def no_virtual_env():
LOG.fixture_step("(Session) Skip if virtual environment")
con_ssh = ssh.ControllerClient.get_active_controller()
output = con_ssh.exec_sudo_cmd('dmesg | grep -i paravirtualized')[1]
if 'KVM' in output and 'bare hardware' not in output:
skip('Virtual environment does not support')

View File

@ -0,0 +1,12 @@
from pytest import mark
from keywords import system_helper
@mark.hasanity
def test_ha_service_group_lst_display():
"""
Test to check services groups.
"""
state_list = system_helper.get_servicegroups(fields='state')
for state in state_list:
assert 'active' == state or 'standby' == state, "Some Services Are Disabled"

View File

@ -0,0 +1,12 @@
from pytest import mark
from keywords import system_helper
@mark.hasanity
def test_ha_service_list_display():
"""
Test checks for system services.
"""
state_list = system_helper.get_services(field='state')
for state in state_list:
assert 'disable' not in state, "Some Services Are Disabled"

View File

@ -0,0 +1,47 @@
from pytest import mark
from keywords import system_helper, host_helper
from utils.clients import ssh
@mark.hasanity1
def test_kill_critical_process_on_active_controller(no_simplex):
"""
Test case to kill critical process on active controller
"""
# get controller node ip
ip0 = system_helper.get_host_values(host='controller-0', fields='mgmt_ip')[0]
ip1 = system_helper.get_host_values(host='controller-1', fields='mgmt_ip')[0]
con_ssh = ssh.ControllerClient.get_active_controller()
con_ssh.exec_cmd('cd /etc/pmon.d')
out = con_ssh.exec_sudo_cmd('grep -r --color="never" "severity = critical"')[1]
file = out.split('/n')
# get sm.conf file
if system_helper.is_aio_system():
process_file = file[0].split(':')[-2]
else:
process_file = file[1].split(':')[-1]
cmd = "cat {} | grep --color='never' script".format(process_file)
script_name = con_ssh.exec_sudo_cmd(cmd)[1].split('=').strip()
# kill sm process
con_ssh.exec_sudo_cmd('mv {0} {0}~'.format(script_name))
con_ssh.exec_sudo_cmd('pkill {}'.format(process_file.split('.')[0]))
# check if activate node is switched
with host_helper.ssh_to_host(ip1) as ct1_ssh:
code = ct1_ssh.exec_cmd('source /etc/platform/openrc')[0]
assert code == 0, "activate controller did not switch to controller-1"
with host_helper.ssh_to_host(ip0) as ct0_ssh:
ct0_ssh.exec_sudo_cmd('mv {0}~ {0}'.format(script_name))
out = ct0_ssh.exec_cmd('sm-dump --pid | grep vim-services')[1]
assert 'standby' in out, "vim-services status is error"
host_helper.lock_host('controller-0')
host_helper.reboot_hosts('controller-0')
host_helper.unlock_host('controller-0')
host_helper.swact_host('controller-1')

View File

@ -0,0 +1,38 @@
import time
from pytest import mark
from keywords import system_helper, pm_helper
from utils.clients import ssh
from utils.tis_log import LOG
from utils import exceptions
@mark.hasanity3
def test_kill_sm_services():
"""
Test will pick a process and kill/disable it, then wait
up to two minutes to check its status. The process should be
restarted within 60 seconds or so. SM will set the current state of
the service back to enabled-active if it changed the state to
disabled when the process was killed
"""
con_ssh = ssh.ControllerClient.get_active_controller()
process_cmd = "sm-dump --pid | grep enabled-active | awk '{{if($4) print $1}}'"
process_name_lst = con_ssh.exec_sudo_cmd(process_cmd)[1].split()
LOG.info(process_name_lst)
for process_name in process_name_lst:
process_id = pm_helper.get_process_from_sm(process_name)[0]
con_ssh.exec_sudo_cmd("kill -9 {}".format(process_id))
time.sleep(30)
end_time = time.time() + 120
while time.time() < end_time:
new_process_id = pm_helper.get_process_from_sm(process_name)[0]
if new_process_id:
break
time.sleep(10)
else:
msg = "did not get process id within 120s"
raise exceptions.TimeoutException(msg)
assert new_process_id != process_id, 'did not create new process id'