Bugfix: Fix sudo check logs and wordings

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

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):

src/core/tests/Test_Bootstrapper.py

@@ -13,17 +13,27 @@
 # 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
 from core.tests.library.RuntimeCompositor import RuntimeCompositor
 
 
 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()