diff --git a/src/core/src/bootstrap/Bootstrapper.py b/src/core/src/bootstrap/Bootstrapper.py index 91d7875c..080b9439 100644 --- a/src/core/src/bootstrap/Bootstrapper.py +++ b/src/core/src/bootstrap/Bootstrapper.py @@ -130,42 +130,43 @@ def basic_environment_health_check(self): self.composite_logger.log("Process id: " + str(os.getpid())) # Ensure sudo works in the environment - sudo_check_result = self.check_sudo_status_with_retry() - self.composite_logger.log_debug("Sudo status check: " + str(sudo_check_result) + "\n") + sudo_check_result = self.check_sudo_status_with_attempts() + self.composite_logger.log_debug("[BST] Sudo status check: " + str(sudo_check_result) + "\n") - def check_sudo_status_with_retry(self, raise_if_not_sudo=True): + def check_sudo_status_with_attempts(self, raise_if_not_sudo=True): # type:(bool) -> any - """ retry to invoke sudo check """ - for attempts in range(1, Constants.MAX_CHECK_SUDO_RETRY_COUNT + 1): + """ Attempt(s) up to max six times to invoke sudo check """ + + self.composite_logger.log("[BST] Performing sudo status check... This should complete within 10 seconds.") + for attempts in range(1, Constants.MAX_CHECK_SUDO_ATTEMPTS + 1): try: sudo_status = self.check_sudo_status(raise_if_not_sudo=raise_if_not_sudo) - if sudo_status and attempts > 1: - self.composite_logger.log_debug("Sudo Check Successfully [RetryCount={0}][MaxRetryCount={1}]".format(str(attempts), Constants.MAX_CHECK_SUDO_RETRY_COUNT)) + if sudo_status and attempts >= 1: + self.composite_logger.log_debug("[BST] Sudo status check completed successfully. [Attempt(s)={0}][MaxAttempt(s)={1}]".format(str(attempts), Constants.MAX_CHECK_SUDO_ATTEMPTS)) return sudo_status elif sudo_status is None or sudo_status is False: - if attempts < Constants.MAX_CHECK_SUDO_RETRY_COUNT: - self.composite_logger.log_debug("Retrying sudo status check after a delay of [ElapsedTimeInSeconds={0}][RetryCount={1}]".format(Constants.MAX_CHECK_SUDO_INTERVAL_IN_SEC, str(attempts))) + if attempts < Constants.MAX_CHECK_SUDO_ATTEMPTS: + self.composite_logger.log_debug("[BST] Re-attempt sudo status check after a delay. [ElapsedTimeInSeconds={0}][Attempt(s)={1}]".format(Constants.MAX_CHECK_SUDO_INTERVAL_IN_SEC, str(attempts))) time.sleep(Constants.MAX_CHECK_SUDO_INTERVAL_IN_SEC) continue - elif attempts >= Constants.MAX_CHECK_SUDO_RETRY_COUNT: + elif attempts >= Constants.MAX_CHECK_SUDO_ATTEMPTS: raise except Exception as exception: - if attempts >= Constants.MAX_CHECK_SUDO_RETRY_COUNT: - self.composite_logger.log_error("Customer environment error (sudo failure). [Exception={0}][MaxRetryCount={1}]".format(str(exception), str(attempts))) + if attempts >= Constants.MAX_CHECK_SUDO_ATTEMPTS: + self.composite_logger.log_error("[BST] Customer environment error (sudo failure). [Attempt(s)={0}][MaxAttempt(s)={1}][Exception={2}]".format(str(attempts), Constants.MAX_CHECK_SUDO_ATTEMPTS, str(exception))) if raise_if_not_sudo: raise - self.composite_logger.log_debug("Retrying sudo status check after a delay of [ElapsedTimeInSeconds={0}][RetryCount={1}]".format(Constants.MAX_CHECK_SUDO_INTERVAL_IN_SEC, str(attempts))) + self.composite_logger.log_debug("[BST] Re-attempt sudo status check after a delay. [ElapsedTimeInSeconds={0}][Attempt(s)={1}]".format(Constants.MAX_CHECK_SUDO_INTERVAL_IN_SEC, str(attempts))) time.sleep(Constants.MAX_CHECK_SUDO_INTERVAL_IN_SEC) def check_sudo_status(self, raise_if_not_sudo=True): # type:(bool) -> any """ Checks if we can invoke sudo successfully. """ try: - self.composite_logger.log("Performing sudo status check... This should complete within 10 seconds.") return_code, output = self.env_layer.run_command_output("timeout 10 sudo id && echo True || echo False", False, False) # output should look like either this (bad): # [sudo] password for username: @@ -187,7 +188,7 @@ def check_sudo_status(self, raise_if_not_sudo=True): else: raise Exception("Unexpected sudo check result. Output: " + " ".join(output.split("\n"))) except Exception as exception: - self.composite_logger.log_debug("Sudo status check failed. Please ensure the computer is configured correctly for sudo invocation. " + + self.composite_logger.log_debug("[BST] Sudo status check failed. Please ensure the computer is configured correctly for sudo invocation. " + "Exception details: " + str(exception)) if raise_if_not_sudo: raise diff --git a/src/core/src/bootstrap/Constants.py b/src/core/src/bootstrap/Constants.py index c8e1421d..f1369f66 100644 --- a/src/core/src/bootstrap/Constants.py +++ b/src/core/src/bootstrap/Constants.py @@ -224,7 +224,7 @@ class StatusTruncationConfig(EnumBackport): MAX_ZYPPER_REPO_REFRESH_RETRY_COUNT = 5 MAX_COMPLETE_STATUS_FILES_TO_RETAIN = 10 SET_CHECK_SUDO_STATUS_TRUE = True - MAX_CHECK_SUDO_RETRY_COUNT = 6 + MAX_CHECK_SUDO_ATTEMPTS = 6 MAX_CHECK_SUDO_INTERVAL_IN_SEC = 300 class PackageBatchConfig(EnumBackport): diff --git a/src/core/tests/Test_Bootstrapper.py b/src/core/tests/Test_Bootstrapper.py index 9df4859b..be8a64a8 100644 --- a/src/core/tests/Test_Bootstrapper.py +++ b/src/core/tests/Test_Bootstrapper.py @@ -13,7 +13,14 @@ # limitations under the License. # # Requires Python 2.7+ +import os +import sys import unittest +# Conditional import for StringIO +try: + from StringIO import StringIO # Python 2 +except ImportError: + from io import StringIO # Python 3 from core.src.bootstrap.Constants import Constants from core.tests.library.ArgumentComposer import ArgumentComposer @@ -21,9 +28,12 @@ class TestBootstrapper(unittest.TestCase): + def __init__(self, methodName: str = "runTest"): + super().__init__(methodName) + def setUp(self): self.sudo_check_status_attempts = 0 - Constants.SET_CHECK_SUDO_STATUS_TRUE = False + Constants.SET_CHECK_SUDO_STATUS_TRUE = False # override check_sudo_status in RuntimeCompositor argument_composer = ArgumentComposer() argument_composer.operation = Constants.ASSESSMENT self.argv = argument_composer.get_composed_arguments() @@ -36,22 +46,22 @@ def tearDown(self): # regions mock def mock_false_run_command_output(self, command, no_output=False, chk_err=True): - """Mock a failed sudo check status command output to test retry logic.""" - # Mock failure to trigger retry logic in check_sudo_status + """Mock a failed sudo check status command output to test multiple attempts logic.""" + # Mock failure to trigger multiple attempts logic in check_sudo_status return (1, "[sudo] password for user:\nFalse") def mock_insufficient_run_command_output(self, command, no_output=False, chk_err=True): """Mock an insufficient output line in sudo check status command output.""" - # Mock failure to trigger retry logic in check_sudo_status + # Mock failure to trigger multiple attempts logic in check_sudo_status return (1, "[sudo] password for user:") def mock_unexpected_output_run_command_output(self, command, no_output=False, chk_err=True): """Mock an unexpected output line in sudo check status command output.""" - # Mock failure to trigger retry logic in check_sudo_status + # Mock failure to trigger multiple attempts logic in check_sudo_status return (1, "[sudo] password for user:\nUnexpectedOutput") - def mock_retry_run_command_output(self, command, no_output=False, chk_err=True): - """Mock 3 failed sudo check status attempts followed by a success on the 4th attempt.""" + def mock_run_command_output_with_attempts(self, command, no_output=False, chk_err=True): + """Mock 3 failed sudo check status attempts followed by a success on the 4th attempts.""" self.sudo_check_status_attempts += 1 # Mock failure on the first two attempts @@ -61,22 +71,34 @@ def mock_retry_run_command_output(self, command, no_output=False, chk_err=True): # Mock success (True) on the 3rd attempt elif self.sudo_check_status_attempts == 3: return (0, "uid=0(root) gid=0(root) groups=0(root)\nTrue") + + def mock_get_arguments_configuration(self, argv): + raise Exception("EXCEPTION during patch management core bootstrap:") + + def mock_os_path_exists(self, path): + return True + + def mock_os_path_getsize(self, path): + return Constants.MAX_AUTO_ASSESSMENT_LOGFILE_SIZE_IN_BYTES + 1 + + def mock_os_remove(self, path): + raise Exception("Mocked exception in os.remove") # end regions mock def test_check_sudo_status_all_attempts_failed(self): # Set raise_if_not_sudo=False to test the `return False` all attempts failed self.runtime.env_layer.run_command_output = self.mock_false_run_command_output - result = self.runtime.bootstrapper.check_sudo_status_with_retry(raise_if_not_sudo=False) + result = self.runtime.bootstrapper.check_sudo_status_with_attempts(raise_if_not_sudo=False) - # Verify check_sudo_status_with_retry is False - self.assertEqual(result, None, "Expected check_sudo_status retry to return None after all attempts failed") + # Verify check_sudo_status_with_attempts is False + self.assertEqual(result, None, "Expected check_sudo_status_with_attempts to return None after all attempts failed") def test_check_sudo_status_throw_exception(self): # Set raise_if_not_sudo=True to throw exception) after all retries self.runtime.env_layer.run_command_output = self.mock_false_run_command_output with self.assertRaises(Exception) as context: - self.runtime.bootstrapper.check_sudo_status_with_retry(raise_if_not_sudo=True) + self.runtime.bootstrapper.check_sudo_status_with_attempts(raise_if_not_sudo=True) # Verify exception msg contains the expected failure text self.assertTrue("Unable to invoke sudo successfully" in str(context.exception)) @@ -86,7 +108,7 @@ def test_check_sudo_status_insufficient_output_lines(self): self.runtime.env_layer.run_command_output = self.mock_insufficient_run_command_output with self.assertRaises(Exception) as context: - self.runtime.bootstrapper.check_sudo_status_with_retry() + self.runtime.bootstrapper.check_sudo_status_with_attempts() # Verify exception msg contains the expected failure text self.assertTrue("Unexpected sudo check result" in str(context.exception)) @@ -96,24 +118,73 @@ def test_check_sudo_status_unexpected_output_lines(self): self.runtime.env_layer.run_command_output = self.mock_unexpected_output_run_command_output with self.assertRaises(Exception) as context: - self.runtime.bootstrapper.check_sudo_status_with_retry() + self.runtime.bootstrapper.check_sudo_status_with_attempts() # Verify exception msg contains the expected failure text self.assertTrue("Unexpected sudo check result" in str(context.exception)) def test_check_sudo_status_succeeds_on_third_attempt(self): - # Test retry logic in check sudo status after 2 failed attempts followed by success (true) - self.runtime.env_layer.run_command_output = self.mock_retry_run_command_output + # Test check sudo status after 2 failed attempts followed by success (true) + self.runtime.env_layer.run_command_output = self.mock_run_command_output_with_attempts # Attempt to check sudo status, succeed (true) on the 3rd attempt - result = self.runtime.bootstrapper.check_sudo_status_with_retry(raise_if_not_sudo=True) + result = self.runtime.bootstrapper.check_sudo_status_with_attempts(raise_if_not_sudo=True) # Verify the result is success (True) self.assertTrue(result, "Expected check_sudo_status to succeed on the 3rd attempts") # Verify 3 attempts were made self.assertEqual(self.sudo_check_status_attempts, 3, "Expected exactly 3 attempts in check_sudo_status") - - + + def test_build_out_container_throw_exception(self): + # Test build_out_container throws exception when no container name is provided + + # Save original methods + original_get_arguments_configuration = self.runtime.bootstrapper.configuration_factory.get_arguments_configuration + + # Mock + self.runtime.bootstrapper.configuration_factory.get_arguments_configuration = self.mock_get_arguments_configuration + + # Verify exception + with self.assertRaises(Exception) as context: + self.runtime.bootstrapper.build_out_container() + + # Restore original methods + self.runtime.bootstrapper.configuration_factory.get_arguments_configuration = original_get_arguments_configuration + self.runtime.stop() + + def test_reset_auto_assessment_log_file_if_needed_raise_exception(self): + # Arrange, Capture stdout + captured_output = StringIO() + original_output = sys.stdout + sys.stdout = captured_output # Redirect stdout to the StringIO object + + # Save original methods + self.runtime.bootstrapper.auto_assessment_only = True + original_path_exists = os.path.exists + original_path_getsize = os.path.getsize + original_os_remove = os.remove + + # Mock + os.path.exists = self.mock_os_path_exists + os.path.getsize = self.mock_os_path_getsize + os.remove = self.mock_os_remove + + self.runtime.bootstrapper.reset_auto_assessment_log_file_if_needed() + + # Restore stdout + sys.stdout = original_output + + # Assert + output = captured_output.getvalue() + self.assertIn("INFO: Error while checking/removing auto-assessment log file.", output) # Verify the log output contains the expected text + + # Restore original methods + os.path.exists = original_path_exists + os.path.getsize = original_path_getsize + os.remove = original_os_remove + self.runtime.stop() + + if __name__ == '__main__': unittest.main()