Source code for rhui4_tests.test_pulp_workers
"""Tests for changing the number of Pulp Workers"""
from os.path import basename
import logging
import nose
from stitches.expect import Expect
import yaml
from rhui4_tests_lib.conmgr import ConMgr
logging.basicConfig(level=logging.DEBUG)
RHUA_HOSTNAME = ConMgr.get_rhua_hostname()
RHUA = ConMgr.connect()
ANSWERS = "/root/.rhui/answers.yaml"
ANSWERS_BAK = ANSWERS + ".backup_test"
CUSTOM_WORKER_COUNT = 2
UNIT_NAME = "pulpcore-worker"
def _change_worker_count(count, expect_failure=False):
"""helper method to change the number of Pulp workers"""
cmd = f"rhui-installer --rerun --pulp-workers {count}"
Expect.expect_retval(RHUA, cmd, 2 if expect_failure else 0, 600)
def _get_current_pulp_worker_count():
"""get the number of Pulp workers"""
cmd = f"systemctl list-units --no-legend --all '{UNIT_NAME}@*'"
_, stdout, _ = RHUA.exec_command(cmd)
raw_lines = stdout.read().decode().splitlines()
return len(raw_lines)
def _get_orig_pulp_worker_count():
"""get the number of Pulp workers"""
_, stdout, _ = RHUA.exec_command(f"cat {ANSWERS_BAK}")
answers = yaml.safe_load(stdout)
saved_worker_count = answers["rhua"]["pulp_workers"]
return saved_worker_count
[docs]
def setup():
"""announce the beginning of the test run"""
print(f"*** Running {basename(__file__)}: ***")
[docs]
def test_01_prep():
"""back up the answers file"""
Expect.expect_retval(RHUA, f"cp {ANSWERS} {ANSWERS_BAK}")
[docs]
def test_02_change_pulp_worker_count():
"""change the number of Pulp workers"""
_change_worker_count(CUSTOM_WORKER_COUNT)
[docs]
def test_03_check_pulp_worker_count():
"""check if the number of Pulp workers has changed"""
current_count = _get_current_pulp_worker_count()
nose.tools.eq_(current_count, CUSTOM_WORKER_COUNT)
[docs]
def test_04_check_rhui_manager_status():
"""run rhui-manager status and check if workers are ok and their count matches"""
cmd = "rhui-manager --noninteractive status"
_, stdout, _ = RHUA.exec_command(cmd)
output = stdout.read().decode().splitlines()
worker_lines = [line for line in output if line.startswith(UNIT_NAME)]
nose.tools.eq_(len(worker_lines), CUSTOM_WORKER_COUNT)
exit_code = output[-1]
# also, if all the services are OK, the last line in the output will be 0
nose.tools.eq_(exit_code, str(0))
[docs]
def test_05_revert_pulp_worker_count():
"""revert the number of Pulp workers"""
orig_count = _get_orig_pulp_worker_count()
_change_worker_count(orig_count)
[docs]
def test_06_check_pulp_worker_count():
"""check if the number of Pulp workers has been reverted"""
current_count = _get_current_pulp_worker_count()
orig_count = _get_orig_pulp_worker_count()
nose.tools.eq_(current_count, orig_count)
[docs]
def test_07_wrong_count_parameter():
"""check if the installer refuses incorrect Pulp worker count values"""
for bad_count in [0, -1, "foo", ""]:
_change_worker_count(bad_count, True)
[docs]
def test_99_cleanup():
"""clean up: remove the answers file backup"""
Expect.expect_retval(RHUA, f"rm -f {ANSWERS_BAK}")
[docs]
def teardown():
"""announce the end of the test run"""
print(f"*** Finished running {basename(__file__)}. ***")