starlingx/test
Code Issues Proposed changes

Supporting get_pty in ssh_connection.send()

Browse Source 
Change-Id: I515a8701cfc4af56a81f089caf256618a4e03612
Signed-off-by: croy <Christian.Roy@windriver.com>
This commit is contained in:
croy
2025-04-15 15:01:34 -04:00
parent 5e87812d9d
commit ba6a7a18c5
1 changed files with 9 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
framework/ssh/ssh_connection.py
View File

@@ -136,7 +136,7 @@ class SSHConnection:
return is_connection_success return is_connection_success
def send(self, cmd: str, reconnect_timeout: int = 600) -> str: def send(self, cmd: str, reconnect_timeout: int = 600, get_pty: bool = False) -> str:
""" """
Send a command to the SSH session and return the output. Send a command to the SSH session and return the output.
@@ -146,11 +146,12 @@ class SSHConnection:
Args: Args:
cmd (str): The command to execute. cmd (str): The command to execute.
reconnect_timeout (int): Time in seconds to retry the connection. reconnect_timeout (int): Time in seconds to retry the connection.
get_pty (bool): Defaults to False. Whether to request a terminal when running a 'send' command.
Returns: Returns:
str: The output of the command. str: The output of the command.
""" """
return self._execute_command("SEND", cmd, reconnect_timeout=reconnect_timeout) return self._execute_command("SEND", cmd, reconnect_timeout=reconnect_timeout, get_pty=get_pty)
def send_as_sudo(self, cmd: str, reconnect_timeout: int = 600) -> str: def send_as_sudo(self, cmd: str, reconnect_timeout: int = 600) -> str:
""" """
@@ -187,6 +188,7 @@ class SSHConnection:
cmd: str, cmd: str,
reconnect_timeout: int = 600, reconnect_timeout: int = 600,
prompts: List[PromptResponse] = None, prompts: List[PromptResponse] = None,
get_pty: bool = False,
) -> str: ) -> str:
""" """
Executes the given action with the given command. Executes the given action with the given command.
@@ -198,6 +200,7 @@ class SSHConnection:
cmd (str): The command to run. cmd (str): The command to run.
reconnect_timeout (int): The time in seconds to wait for SSH connection. reconnect_timeout (int): The time in seconds to wait for SSH connection.
prompts (List[PromptResponse], optional): Expected prompts, if any. prompts (List[PromptResponse], optional): Expected prompts, if any.
get_pty (bool): Defaults to False. Whether to request a terminal when running a 'send' command.
Returns: Returns:
str: The output of the command. str: The output of the command.
@@ -223,7 +226,7 @@ class SSHConnection:
thread_manager = ThreadManager(timeout=reconnect_timeout / 10) thread_manager = ThreadManager(timeout=reconnect_timeout / 10)
if action == "SEND": if action == "SEND":
thread_manager.start_thread("SSH_Command", self._send, cmd) thread_manager.start_thread("SSH_Command", self._send, cmd, get_pty)
elif action == "SEND_SUDO": elif action == "SEND_SUDO":
thread_manager.start_thread("SSH_Command", self._send_as_sudo, cmd) thread_manager.start_thread("SSH_Command", self._send_as_sudo, cmd)
elif action == "SEND_EXPECT_PROMPTS": elif action == "SEND_EXPECT_PROMPTS":
@@ -244,20 +247,21 @@ class SSHConnection:
time.sleep(refresh_timeout) time.sleep(refresh_timeout)
self.is_connected = False self.is_connected = False
def _send(self, cmd: str, timeout: int = 30) -> str: def _send(self, cmd: str, timeout: int = 30, get_pty: bool = False) -> str:
""" """
Sends the given command with the specified timeout. Sends the given command with the specified timeout.
Args: Args:
cmd (str): The command to send. cmd (str): The command to send.
timeout (int): The timeout in seconds for command execution. timeout (int): The timeout in seconds for command execution.
get_pty (bool): Defaults to False. Whether to request a terminal when running a 'send' command.
Returns: Returns:
str: The output of the command. str: The output of the command.
""" """
get_logger().log_ssh(cmd) get_logger().log_ssh(cmd)
stdin, stdout, stderr = self.client.exec_command(cmd, timeout=timeout) stdin, stdout, stderr = self.client.exec_command(cmd, timeout=timeout, get_pty=get_pty)
stdout.channel.set_combine_stderr(True) stdout.channel.set_combine_stderr(True)
self.last_return_code = stdout.channel.recv_exit_status() self.last_return_code = stdout.channel.recv_exit_status()