fix interactive ssh and tests

Signed-off-by: a.lipay <a.lipay@yadro.com>
This commit is contained in:
a.lipay 2022-11-09 18:24:32 +03:00 committed by Alipayy
parent 8016ad4b86
commit cbfcbb559c
2 changed files with 9 additions and 6 deletions

View file

@ -142,7 +142,9 @@ class SSHShell(Shell):
@log_command @log_command
def _exec_interactive(self, command: str, options: CommandOptions) -> CommandResult: def _exec_interactive(self, command: str, options: CommandOptions) -> CommandResult:
stdin, stdout, stderr = self._connection.exec_command(command, timeout=options.timeout) stdin, stdout, stderr = self._connection.exec_command(
command, timeout=options.timeout, get_pty=True
)
for interactive_input in options.interactive_inputs: for interactive_input in options.interactive_inputs:
input = interactive_input.input input = interactive_input.input
if not input.endswith("\n"): if not input.endswith("\n"):

View file

@ -41,9 +41,8 @@ class TestSSHShellInteractive(TestCase):
f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs) f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs)
) )
# TODO: we have inconsistency with local shell here, ssh does not echo input into stdout
self.assertEqual(0, result.return_code) self.assertEqual(0, result.return_code)
self.assertEqual(["Password:", "test"], get_output_lines(result)) self.assertEqual(["Password: test", "test"], get_output_lines(result))
self.assertEqual("", result.stderr) self.assertEqual("", result.stderr)
def test_command_with_several_prompts(self): def test_command_with_several_prompts(self):
@ -60,9 +59,10 @@ class TestSSHShellInteractive(TestCase):
f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs) f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs)
) )
# TODO: we have inconsistency with local shell here, ssh does not echo input into stdout
self.assertEqual(0, result.return_code) self.assertEqual(0, result.return_code)
self.assertEqual(["Input1:", "test1", "Input2:", "test2"], get_output_lines(result)) self.assertEqual(
["Input1: test1", "test1", "Input2: test2", "test2"], get_output_lines(result)
)
self.assertEqual("", result.stderr) self.assertEqual("", result.stderr)
def test_invalid_command_with_check(self): def test_invalid_command_with_check(self):
@ -73,7 +73,7 @@ class TestSSHShellInteractive(TestCase):
self.shell.exec(f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs)) self.shell.exec(f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs))
error = format_error_details(raised.exception) error = format_error_details(raised.exception)
self.assertIn("Error", error) self.assertIn("SyntaxError", error)
self.assertIn("return code: 1", error) self.assertIn("return code: 1", error)
def test_invalid_command_without_check(self): def test_invalid_command_without_check(self):
@ -84,6 +84,7 @@ class TestSSHShellInteractive(TestCase):
f'python3 -c "{script}"', f'python3 -c "{script}"',
CommandOptions(interactive_inputs=inputs, check=False), CommandOptions(interactive_inputs=inputs, check=False),
) )
self.assertIn("SyntaxError", result.stdout)
self.assertEqual(1, result.return_code) self.assertEqual(1, result.return_code)
def test_non_existing_binary(self): def test_non_existing_binary(self):