(self, normalized_command: str)
| 415 | ) |
| 416 | |
| 417 | def _execute_local(self, normalized_command: str) -> ShellCommandResult: |
| 418 | try: |
| 419 | completed = subprocess.run( |
| 420 | normalized_command, |
| 421 | shell=True, |
| 422 | executable="/bin/bash", |
| 423 | stdout=subprocess.PIPE, |
| 424 | stderr=subprocess.STDOUT, |
| 425 | text=True, |
| 426 | timeout=self.runtime.ssh.timeout, |
| 427 | ) |
| 428 | output = completed.stdout or "" |
| 429 | trimmed_output = self._trim_output(normalized_command, output.rstrip()) |
| 430 | return ShellCommandResult( |
| 431 | command=normalized_command, |
| 432 | output=trimmed_output, |
| 433 | exit_code=completed.returncode, |
| 434 | timed_out=False, |
| 435 | ) |
| 436 | except subprocess.TimeoutExpired as exc: |
| 437 | output = exc.stdout or "" |
| 438 | trimmed_output = self._trim_output(normalized_command, str(output).rstrip()) |
| 439 | if trimmed_output: |
| 440 | trimmed_output = f"{trimmed_output}\nCommand execution timeout!" |
| 441 | else: |
| 442 | trimmed_output = "Command execution timeout!" |
| 443 | return ShellCommandResult( |
| 444 | command=normalized_command, |
| 445 | output=trimmed_output, |
| 446 | exit_code=124, |
| 447 | timed_out=True, |
| 448 | ) |
| 449 | |
| 450 | def execute(self, command: str) -> ShellCommandResult: |
| 451 | normalized_command = self._normalize_command(command) |
no test coverage detected