diff --git a/src/neofs_testlib/shell/local_shell.py b/src/neofs_testlib/shell/local_shell.py index b20988c..f16339f 100644 --- a/src/neofs_testlib/shell/local_shell.py +++ b/src/neofs_testlib/shell/local_shell.py @@ -35,53 +35,35 @@ class LocalShell(Shell): def _exec_interactive(self, command: str, options: CommandOptions) -> CommandResult: start_time = datetime.utcnow() log_file = tempfile.TemporaryFile() # File is reliable cross-platform way to capture output - result = None - command_process = None try: command_process = pexpect.spawn(command, timeout=options.timeout) - command_process.delaybeforesend = 1 - command_process.logfile_read = log_file + except (pexpect.ExceptionPexpect, OSError) as exc: + raise RuntimeError(f"Command: {command}") from exc + command_process.delaybeforesend = 1 + command_process.logfile_read = log_file + + try: for interactive_input in options.interactive_inputs: command_process.expect(interactive_input.prompt_pattern) command_process.sendline(interactive_input.input) - - result = self._get_pexpect_process_result(command_process, command) - if options.check and result.return_code != 0: - raise RuntimeError( - f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}" - ) - - return result - except pexpect.ExceptionPexpect as exc: - result = self._get_pexpect_process_result(command_process, command) - message = ( - f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}" - ) + except (pexpect.ExceptionPexpect, OSError) as exc: if options.check: - raise RuntimeError(message) from exc - else: - logger.exception(message) - return result - except OSError as exc: - result = self._get_pexpect_process_result(command_process, command) - message = ( - f"Command: {command}\nreturn code: {result.return_code}\nOutput: {exc.strerror}" - ) - if options.check: - raise RuntimeError(message) from exc - else: - logger.exception(message) - return result - except Exception: - result = self._get_pexpect_process_result(command_process, command) - raise + raise RuntimeError(f"Command: {command}") from exc finally: + result = self._get_pexpect_process_result(command_process) log_file.close() end_time = datetime.utcnow() self._report_command_result(command, start_time, end_time, result) + if options.check and result.return_code != 0: + raise RuntimeError( + f"Command: {command}\nreturn code: {result.return_code}\n" + f"Output: {result.stdout}" + ) + return result + def _exec_non_interactive(self, command: str, options: CommandOptions) -> CommandResult: start_time = datetime.utcnow() result = None @@ -99,13 +81,16 @@ class LocalShell(Shell): result = CommandResult( stdout=command_process.stdout or "", - stderr=command_process.stderr or "", + stderr="", return_code=command_process.returncode, ) - return result except subprocess.CalledProcessError as exc: # TODO: always set check flag to false and capture command result normally - result = self._get_failing_command_result(command) + result = CommandResult( + stdout=exc.stdout or "", + stderr="", + return_code=exc.returncode, + ) raise RuntimeError( f"Command: {command}\nError:\n" f"return code: {exc.returncode}\n" @@ -113,29 +98,15 @@ class LocalShell(Shell): ) from exc except OSError as exc: raise RuntimeError(f"Command: {command}\nOutput: {exc.strerror}") from exc - except Exception as exc: - result = self._get_failing_command_result(command) - raise finally: end_time = datetime.utcnow() self._report_command_result(command, start_time, end_time, result) + return result - def _get_failing_command_result(self, command: str) -> CommandResult: - return_code, cmd_output = subprocess.getstatusoutput(command) - return CommandResult(stdout=cmd_output, stderr="", return_code=return_code) - - def _get_pexpect_process_result( - self, command_process: Optional[pexpect.spawn], command: str - ) -> CommandResult: - """Captures output of the process. - - If command process is not None, captures output of this process. - If command process is None, then command fails when we attempt to start it, in this case - we use regular non-interactive process to get it's output. + def _get_pexpect_process_result(self, command_process: pexpect.spawn) -> CommandResult: + """ + Captures output of the process. """ - if command_process is None: - return self._get_failing_command_result(command) - # Wait for child process to end it's work if command_process.isalive(): command_process.expect(pexpect.EOF) diff --git a/tests/test_local_shell.py b/tests/test_local_shell.py index f9c2d89..de1e22f 100644 --- a/tests/test_local_shell.py +++ b/tests/test_local_shell.py @@ -72,7 +72,7 @@ class TestLocalShellInteractive(TestCase): self.shell.exec("not-a-command", CommandOptions(interactive_inputs=inputs)) error = format_error_details(exc.exception) - self.assertIn("return code: 127", error) + self.assertIn("The command was not found", error) class TestLocalShellNonInteractive(TestCase):