Execute shell command locally. Args: command: Shell command to execute. input_spec: Input specification. Returns: Tuple of (stdout, stderr, exit_code).
(
self, command: str,
input_spec: ExecutionInput)
| 689 | raise |
| 690 | |
| 691 | async def _local_execute_shell( |
| 692 | self, command: str, |
| 693 | input_spec: ExecutionInput) -> tuple[str, str, int]: |
| 694 | """ |
| 695 | Execute shell command locally. |
| 696 | |
| 697 | Args: |
| 698 | command: Shell command to execute. |
| 699 | input_spec: Input specification. |
| 700 | |
| 701 | Returns: |
| 702 | Tuple of (stdout, stderr, exit_code). |
| 703 | """ |
| 704 | shell_exec = self._get_shell_executable() |
| 705 | |
| 706 | # Build full command with environment exports |
| 707 | if platform.system() == 'Windows': |
| 708 | # Windows: use set for environment |
| 709 | env_cmds = [f'set {k}={v}' for k, v in input_spec.env_vars.items()] |
| 710 | full_cmd = ' && '.join(env_cmds |
| 711 | + [command]) if env_cmds else command |
| 712 | cmd = shell_exec + [full_cmd] |
| 713 | else: |
| 714 | # Unix: use export |
| 715 | env_cmds = [ |
| 716 | f"export {k}='{v}'" for k, v in input_spec.env_vars.items() |
| 717 | ] |
| 718 | full_cmd = ' && '.join(env_cmds |
| 719 | + [command]) if env_cmds else command |
| 720 | cmd = shell_exec + [full_cmd] |
| 721 | |
| 722 | # Use working_dir from input_spec for proper resource access |
| 723 | cwd = input_spec.working_dir if input_spec.working_dir else None |
| 724 | |
| 725 | return self._local_run_subprocess( |
| 726 | cmd, |
| 727 | env=input_spec.env_vars, |
| 728 | cwd=cwd, |
| 729 | stdin_input=input_spec.stdin) |
| 730 | |
| 731 | async def _local_execute_javascript( |
| 732 | self, js_code: str, |
no test coverage detected