Add bash prompt check when execute commands (#3117)

lisa/node.py

@@ -112,6 +112,7 @@ def __init__(
         self.capture_boot_time: bool = False
         self.capture_azure_information: bool = False
         self.capture_kernel_config: bool = False
+        self.has_checked_bash_prompt: bool = False
 
     @property
     def shell(self) -> Shell:
@@ -288,6 +289,8 @@ def execute_async(
         encoding: str = "",
     ) -> Process:
         self.initialize()
+        if isinstance(self, RemoteNode):
+            self._check_bash_prompt()
 
         return self._execute(
             cmd,
@@ -739,6 +742,26 @@ def _check_password_and_store_prompt(self) -> None:
         ssh_shell = cast(SshShell, self.shell)
         ssh_shell.password_prompts = password_prompts
 
+    def _check_bash_prompt(self) -> None:
+        # Check if there is bash prompt in stdout of command. If yes, the prompt
+        # should be filtered from the stdout. E.g. image yaseensmarket1645449809728
+        # wordpress-red-hat images.
+        if not self.has_checked_bash_prompt:
+            process = self._execute(f"echo {constants.LISA_TEST_FOR_BASH_PROMPT}")
+            result = process.wait_result(10)
+            if result.stdout.endswith(f"{constants.LISA_TEST_FOR_BASH_PROMPT}"):
+                bash_prompt = result.stdout.replace(
+                    constants.LISA_TEST_FOR_BASH_PROMPT, ""
+                )
+                if bash_prompt:
+                    self.log.debug(
+                        "detected bash prompt, it will be removed from every output: "
+                        f"{bash_prompt}"
+                    )
+                    ssh_shell = cast(SshShell, self.shell)
+                    ssh_shell.bash_prompt = bash_prompt
+                self.has_checked_bash_prompt = True
+
     def _reset_password(self) -> bool:
         from lisa.features import PasswordExtension
 

lisa/util/constants.py

@@ -163,6 +163,7 @@
 
 # Test for command with sudo
 LISA_TEST_FOR_SUDO = "lisa test for sudo"
+LISA_TEST_FOR_BASH_PROMPT = "lisa test for bash prompt"
 
 # linux signals
 SIGINT = 2

lisa/util/process.py

@@ -397,15 +397,9 @@ def wait_result(
         if self._is_posix and self._sudo:
             self._result.stdout = self._filter_sudo_result(self._result.stdout)
 
-        if (
-            isinstance(self._shell, SshShell)
-            and self._shell._inner_shell
-            and self._shell._inner_shell._spur._shell_type
-            == spur.ssh.ShellTypes.minimal
-        ):
-            self._result.stdout = self._filter_profile_error(self._result.stdout)
+        self._result.stdout = self._filter_profile_error(self._result.stdout)
+        self._result.stdout = self._filter_bash_prompt(self._result.stdout)
         self._check_if_need_input_password(self._result.stdout)
-
         self._result.stdout = self._filter_sudo_required_password_info(
             self._result.stdout
         )
@@ -519,6 +513,9 @@ def _filter_profile_error(self, raw_input: str) -> str:
         if (
             isinstance(self._shell, SshShell)
             and self._shell.spawn_initialization_error_string
+            and self._shell._inner_shell
+            and self._shell._inner_shell._spur._shell_type
+            == spur.ssh.ShellTypes.minimal
         ):
             raw_input = raw_input.replace(
                 rf"{self._shell.spawn_initialization_error_string}\n", ""
@@ -532,6 +529,16 @@ def _filter_profile_error(self, raw_input: str) -> str:
             )
         return raw_input
 
+    def _filter_bash_prompt(self, raw_input: str) -> str:
+        # some images have bash prompt in stdout, remove it.
+        # E.g. yaseensmarket1645449809728 wordpress-red-hat
+        # ----------------------------------------------------------------------
+        # Use the this command 'sudo bash ~/getcert.sh' to install a certificate
+        # ----------------------------------------------------------------------
+        if isinstance(self._shell, SshShell) and self._shell.bash_prompt:
+            raw_input = raw_input.replace(self._shell.bash_prompt, "")
+        return raw_input
+
     def _check_if_need_input_password(self, raw_input: str) -> None:
         # Check if the stdout contains "[sudo] password for .*: " and
         # "sudo: timed out reading password" strings. If so, raise exception

lisa/util/shell.py

@@ -223,6 +223,7 @@ def __init__(self, connection_info: schema.ConnectionInfo) -> None:
         self._jump_box_sock: Any = None
         self.is_sudo_required_password: bool = False
         self.password_prompts: List[str] = []
+        self.bash_prompt: str = ""
         self.spawn_initialization_error_string = ""
 
         paramiko_logger = logging.getLogger("paramiko")