Newer
Older
Return the patched environment required to run a test as a string that
can be prepended to a shell command.
"""
return ' '.join(f'{k}="{v}"' for k,v in self._patched_env.items())
def script(self, code, intercept_shell=True, timeout=60, timeout_signal=signal.SIGKILL):
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
"""
Execute a shell script passed as an argument in bash.
For instance, the following snippet:
mountdir = pathlib.Path('/tmp')
file01 = 'file01'
ShellClient().script(
f'''
expected_pathname={mountdir / file01}
if [[ -e ${{expected_pathname}} ]];
then
exit 0
fi
exit 1
''')
transforms into:
bash -c '
expected_pathname=/tmp/file01
if [[ -e ${expected_pathname} ]];
then
exit 0
fi
exit 1
'
Note that since we are using Python's f-strings, for variable
expansions to work correctly, they need to be defined with double
braces, e.g. ${{expected_pathname}}.
Parameters
----------
code: `str`
The script code to be passed to 'bash -c'.
intercept_shell: `bool`
Controls whether the shell executing the script should be
executed with LD_PRELOAD=libgkfwd_intercept.so (default: True).
timeout: `int`
How much time, in seconds, we should give the process to complete.
If the process does not finish within the timeout, it will be sent
the signal defined by `timeout_signal`.
timeout_signal: `int`
The signal to be sent to the process if `timeout` is not None.
Default value: signal.SIGKILL
Returns
-------
A sh.RunningCommand instance that allows interacting with
the finished process.
"""
logger.debug(f"running bash")
logger.debug(f"cmd: bash -c '{code}'")
logger.debug(f"timeout: {timeout} seconds")
logger.debug(f"timeout_signal: {signal.Signals(timeout_signal).name}")
logger.debug(f"patched env: {self._patched_env}")
# 'sh' raises an exception if the return code is not zero;
# since we'd rather check for return codes explictly, we
# whitelist all exit codes from 1 to 255 as 'ok' using the
# _ok_code argument
return self._cmd('-c',
code,
_env = (self._env if intercept_shell else os.environ),
# _out=sys.stdout,
# _err=sys.stderr,
_timeout=timeout,
_timeout_signal=timeout_signal,
# _ok_code=list(range(0, 256))
def run(self, cmd, *args, timeout=60, timeout_signal=signal.SIGKILL):
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
"""
Execute a shell command with arguments.
For example, the following snippet:
mountdir = pathlib.Path('/tmp')
file01 = 'file01'
ShellClient().stat('--terse', mountdir / file01)
transforms into:
bash -c 'stat --terse /tmp/file01'
Parameters:
-----------
cmd: `str`
The command to execute.
args: `list`
The list of arguments for the command.
timeout: `number`
How much time, in seconds, we should give the process to complete.
If the process does not finish within the timeout, it will be sent
the signal defined by `timeout_signal`.
timeout_signal: `int`
The signal to be sent to the process if `timeout` is not None.
Default value: signal.SIGKILL
Returns
-------
A ShellCommand instance that allows interacting with the finished
process. Note that ShellCommand wraps sh.RunningCommand and adds s
extra properties to it.
"""
bash_c_args = f"{cmd} {' '.join(str(a) for a in args)}"
logger.debug(f"running bash")
logger.debug(f"cmd: bash -c '{bash_c_args}'")
logger.debug(f"timeout: {timeout} seconds")
logger.debug(f"timeout_signal: {signal.Signals(timeout_signal).name}")
logger.debug(f"patched env:\n{pformat(self._patched_env)}")
# 'sh' raises an exception if the return code is not zero;
# since we'd rather check for return codes explictly, we
# whitelist all exit codes from 1 to 255 as 'ok' using the
# _ok_code argument
proc = self._cmd('-c',
bash_c_args,
_env = self._env,
# _out=sys.stdout,
# _err=sys.stderr,
_timeout=timeout,
_timeout_signal=timeout_signal,
# _ok_code=list(range(0, 256))