'''CDS management tests'''
from __future__ import print_function
from os.path import basename
import random
import re
import logging
import nose
from stitches.expect import CTRL_C, Expect
from rhui3_tests_lib.conmgr import ConMgr, SUDO_USER_NAME, SUDO_USER_KEY
from rhui3_tests_lib.helpers import Helpers
from rhui3_tests_lib.rhuimanager import RHUIManager
from rhui3_tests_lib.rhuimanager_instance import RHUIManagerInstance, NoSuchInstance
logging.basicConfig(level=logging.DEBUG)
CDS_HOSTNAMES = ConMgr.get_cds_hostnames()
RHUA = ConMgr.connect()
CDS = [ConMgr.connect(host) for host in CDS_HOSTNAMES]
[docs]def setup():
'''
announce the beginning of the test run
'''
print("*** Running %s: *** " % basename(__file__))
[docs]def test_01_initial_run():
'''
log in to RHUI
'''
RHUIManager.initial_run(RHUA)
[docs]def test_02_list_empty_cds():
'''
check if there are no CDSs
'''
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(cds_list, [])
[docs]def test_03_add_cds():
'''
add all known CDSs
'''
for cds in CDS_HOSTNAMES:
RHUIManagerInstance.add_instance(RHUA, "cds", cds)
[docs]def test_04_list_cds():
'''
list CDSs, expect as many as there are in /etc/hosts
'''
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(len(cds_list), len(CDS_HOSTNAMES))
[docs]def test_05_readd_cds():
'''
add one of the CDSs again (reapply the configuration)
'''
# choose a random CDS hostname from the list
RHUIManagerInstance.add_instance(RHUA, "cds", random.choice(CDS_HOSTNAMES), update=True)
[docs]def test_06_list_cds():
'''
check if the CDSs are still tracked
'''
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(len(cds_list), len(CDS_HOSTNAMES))
[docs]def test_07_delete_nonexisting_cds():
'''
try deleting an untracked CDS, should be rejected (by rhui3_tests_lib)
'''
nose.tools.assert_raises(NoSuchInstance,
RHUIManagerInstance.delete,
RHUA,
"cds",
[CDS_HOSTNAMES[0].replace("cds", "cdsfoo")])
[docs]def test_08_delete_cds():
'''
delete all CDSs
'''
RHUIManagerInstance.delete_all(RHUA, "cds")
[docs]def test_09_list_cds():
'''
list CDSs, expect none
'''
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(cds_list, [])
[docs]def test_10_check_cleanup():
'''
check if Apache was stopped and the remote file system unmounted on all CDSs
'''
service = "httpd"
mdir = "/var/lib/rhui/remote_share"
dirty_hosts = dict()
errors = []
dirty_hosts["httpd"] = [cds.hostname for cds in CDS if Helpers.check_service(cds, service)]
dirty_hosts["mount"] = [cds.hostname for cds in CDS if Helpers.check_mountpoint(cds, mdir)]
if dirty_hosts["httpd"]:
errors.append("Apache is still running on %s" % dirty_hosts["httpd"])
if dirty_hosts["mount"]:
errors.append("The remote file system is still mounted on %s" % dirty_hosts["mount"])
nose.tools.ok_(not errors, msg=errors)
[docs]def test_11_add_cds_uppercase():
'''
add (and delete) a CDS with uppercase characters
'''
# for RHBZ#1572623
# choose a random CDS hostname from the list
cds_up = random.choice(CDS_HOSTNAMES).replace("cds", "CDS")
RHUIManagerInstance.add_instance(RHUA, "cds", cds_up)
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(len(cds_list), 1)
RHUIManagerInstance.delete(RHUA, "cds", [cds_up])
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(len(cds_list), 0)
[docs]def test_12_delete_unreachable():
'''
add a CDS, make it unreachable, and see if it can still be deleted from the RHUA
'''
# for RHBZ#1639996
# choose a random CDS hostname from the list
cds = random.choice(CDS_HOSTNAMES)
RHUIManagerInstance.add_instance(RHUA, "cds", cds)
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_not_equal(cds_list, [])
Helpers.break_hostname(RHUA, cds)
# delete it
RHUIManagerInstance.delete(RHUA, "cds", [cds])
# check it
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_equal(cds_list, [])
Helpers.unbreak_hostname(RHUA)
# the node remains configured (RHUI mount point, httpd)... unconfigure it properly
RHUIManagerInstance.add_instance(RHUA, "cds", cds)
RHUIManagerInstance.delete(RHUA, "cds", [cds])
[docs]def test_13_delete_select_0():
'''
add a CDS and see if no issue occurs if it and "a zeroth" (ghost) CDSs are selected for deletion
'''
# for RHBZ#1305612
# choose a random CDS and add it
cds = random.choice(CDS_HOSTNAMES)
RHUIManagerInstance.add_instance(RHUA, "cds", cds)
cds_list = RHUIManagerInstance.list(RHUA, "cds")
nose.tools.assert_not_equal(cds_list, [])
# try the deletion
RHUIManager.screen(RHUA, "cds")
Expect.enter(RHUA, "d")
Expect.expect(RHUA, "Enter value")
Expect.enter(RHUA, "0")
Expect.expect(RHUA, "Enter value")
Expect.enter(RHUA, "1")
Expect.expect(RHUA, "Enter value")
Expect.enter(RHUA, "c")
state = Expect.expect_list(RHUA,
[(re.compile(".*Are you sure.*", re.DOTALL), 1),
(re.compile(".*An unexpected error.*", re.DOTALL), 2)])
if state == 1:
Expect.enter(RHUA, "y")
RHUIManager.quit(RHUA, timeout=180)
else:
Expect.enter(RHUA, "q")
# the CDS list ought to be empty now; if not, delete the CDS and fail
cds_list = RHUIManagerInstance.list(RHUA, "cds")
if cds_list:
RHUIManagerInstance.delete_all(RHUA, "cds")
raise AssertionError("The CDS list is not empty after the deletion attempt: %s." % cds_list)
[docs]def test_14_verbose_reporting():
'''
check if a failure is reported properly (if puppet is verbose)
'''
# for RHBZ#1751378
# choose a random CDS and open port 443 on it, which will later prevent Apache from starting
cds = random.choice(CDS)
Expect.enter(cds, "ncat -l 443 --keep-open")
# try adding the CDS and check for the specific error message in the output
error_msg = "change from stopped to running failed"
RHUIManager.screen(RHUA, "cds")
Expect.enter(RHUA, "a")
Expect.expect(RHUA, "Hostname")
Expect.enter(RHUA, cds.hostname)
Expect.expect(RHUA, "Username")
Expect.enter(RHUA, SUDO_USER_NAME)
Expect.expect(RHUA, "Absolute")
Expect.enter(RHUA, SUDO_USER_KEY)
Expect.expect(RHUA, "Proceed")
Expect.enter(RHUA, "y")
state = Expect.expect_list(RHUA,
[(re.compile(".*%s.*" % error_msg, re.DOTALL), 1),
(re.compile(".*Aborting.*", re.DOTALL), 2),
(re.compile(".*was successfully configured.*", re.DOTALL), 3)],
timeout=180)
# quit rhui-manager, clean up, fail if the error message didn't appear
Expect.enter(RHUA, "q")
Expect.enter(cds, CTRL_C)
cds_list = RHUIManagerInstance.list(RHUA, "cds")
if cds_list:
RHUIManagerInstance.delete_all(RHUA, "cds")
if state != 1:
raise AssertionError("The expected error message was not seen in rhui-manager's output.")
[docs]def teardown():
'''
announce the end of the test run
'''
print("*** Finished running %s. *** " % basename(__file__))