"""Tests for changing the RHUI remote share"""
from os.path import basename
import logging
import nose
from stitches.expect import Expect
import yaml
from rhui4_tests_lib.conmgr import ConMgr
from rhui4_tests_lib.rhuimanager import RHUIManager
from rhui4_tests_lib.rhuimanager_cmdline_instance import RHUIManagerCLIInstance
logging.basicConfig(level=logging.DEBUG)
RHUA_HOSTNAME = ConMgr.get_rhua_hostname()
CDS_HOSTNAME = ConMgr.get_cds_hostnames()[0]
RHUA = ConMgr.connect()
CDS = ConMgr.connect(CDS_HOSTNAME)
ANSWERS = "/root/.rhui/answers.yaml"
ANSWERS_BAK = ANSWERS + ".backup_test"
NEW_FS_OPTIONS = "timeo=100"
REMOTE_SHARE = "/var/lib/rhui/remote_share"
FORCE_FLAG = " --remote-fs-force-change"
def _check_rhui_mountpoint(connection, fs_server, options=""):
"""check the RHUI mountpoint"""
for mount_info_file in ["/proc/mounts", "/etc/fstab"]:
_, stdout, _ = connection.exec_command(f"cat {mount_info_file}")
mounts = stdout.read().decode().splitlines()
matches = [line for line in mounts if REMOTE_SHARE in line]
# there must be only one such share
nose.tools.eq_(len(matches), 1,
msg=f"unexpected matches in {mount_info_file}: {matches}")
# and it must be using the expected FS server
properties = matches[0].split()
actual_share = properties[0]
test = actual_share.startswith(fs_server)
nose.tools.ok_(test,
msg=f"{fs_server} not found in {mount_info_file}, found: {actual_share}")
# if also checking options, find and compare them; options are in the fourth column
if options:
# in /proc/mounts, there are many options added by the NFS kernel module,
# whereas in /etc/fstab the options are standalone
actual_options = properties[3]
test = options in actual_options if mount_info_file == "/proc/mounts" else \
actual_options == options
nose.tools.ok_(test,
msg=f"{options} not found in {mount_info_file}, found: {actual_options}")
[docs]
def setup():
"""announce the beginning of the test run"""
print(f"*** Running {basename(__file__)}: ***")
[docs]
def test_01_add_cds():
"""add a CDS"""
RHUIManager.initial_run(RHUA)
RHUIManagerCLIInstance.add(RHUA, "cds", CDS_HOSTNAME, unsafe=True)
# check that
cds_list = RHUIManagerCLIInstance.list(RHUA, "cds")
nose.tools.ok_(cds_list)
[docs]
def test_02_prep():
"""back up the answers file"""
Expect.expect_retval(RHUA, f"cp {ANSWERS} {ANSWERS_BAK}")
[docs]
def test_03_rerun_installer():
"""rerun the installer with a different remote FS server"""
# the FS server is supposed to be on the RHUA, so let's use the RHUA hostname
# get the actual FS server hostname from the answers file
_, stdout, _ = RHUA.exec_command(f"cat {ANSWERS}")
answers = yaml.safe_load(stdout)
current_fs_server = answers["rhua"]["remote_fs_server"]
fs_hostname = current_fs_server.split(":")[0]
new_fs_server = current_fs_server.replace(fs_hostname, RHUA_HOSTNAME)
# first, check if the installer refuses to rerun without the force flag
installer_cmd = f"rhui-installer --rerun --remote-fs-server {new_fs_server}"
Expect.expect_retval(RHUA, installer_cmd, 1)
# then, use the force flag and rerun
installer_cmd += FORCE_FLAG
Expect.expect_retval(RHUA, installer_cmd, timeout=300)
[docs]
def test_04_check_rhua_mountpoint():
"""check if the new remote share has replaced the old one on the RHUA"""
_check_rhui_mountpoint(RHUA, RHUA_HOSTNAME)
[docs]
def test_05_reinstall_cds():
"""reinstall the CDS"""
RHUIManagerCLIInstance.reinstall(RHUA, "cds", CDS_HOSTNAME)
[docs]
def test_06_check_cds_mountpoint():
"""check if the new remote share is now used on the CDS"""
_check_rhui_mountpoint(CDS, RHUA_HOSTNAME)
[docs]
def test_07_rerun_installer():
"""rerun the installer with different mount options"""
# first check if the installer fails if options change but the force flag isn't used
installer_cmd = f"rhui-installer --rerun --rhua-mount-options {NEW_FS_OPTIONS}"
Expect.expect_retval(RHUA, installer_cmd, 1)
# now with the force flag
installer_cmd += FORCE_FLAG
Expect.expect_retval(RHUA, installer_cmd, timeout=300)
[docs]
def test_08_check_rhua_mountpoint():
"""check if the new options have replaced the old ones on the RHUA"""
_check_rhui_mountpoint(RHUA, RHUA_HOSTNAME, NEW_FS_OPTIONS)
[docs]
def test_09_reinstall_cds():
"""reinstall the CDS"""
RHUIManagerCLIInstance.reinstall(RHUA, "cds", CDS_HOSTNAME)
[docs]
def test_10_check_cds_mountpoint():
"""check if the new options are now used on the CDS"""
_check_rhui_mountpoint(CDS, RHUA_HOSTNAME, NEW_FS_OPTIONS)
[docs]
def test_99_cleanup():
"""clean up: delete the CDS and rerun the installer with the original remote FS"""
RHUIManagerCLIInstance.delete(RHUA, "cds", [CDS_HOSTNAME], force=True)
# get the original FS server hostname from the backed up answers file
_, stdout, _ = RHUA.exec_command(f"cat {ANSWERS_BAK}")
answers = yaml.safe_load(stdout)
original_fs_server = answers["rhua"]["remote_fs_server"]
fs_hostname = original_fs_server.split(":")[0]
original_fs_options = answers["rhua"]["rhua_mount_options"]
installer_cmd = f"rhui-installer --rerun " \
f"--remote-fs-server {original_fs_server} " \
f"--rhua-mount-options {original_fs_options}" \
f"{FORCE_FLAG}"
Expect.expect_retval(RHUA, installer_cmd, timeout=300)
# did it work?
_check_rhui_mountpoint(RHUA, fs_hostname, original_fs_options)
# finish the cleanup
Expect.expect_retval(RHUA, f"rm -f {ANSWERS_BAK}")
ConMgr.remove_ssh_keys(RHUA)
[docs]
def teardown():
"""announce the end of the test run"""
print(f"*** Finished running {basename(__file__)}. ***")