Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def stdout(self, value):
if self._stdout is None:
self._stdout = value
elif value is None:
pass
else:
safe_close(value)
msg = "Multiple redirections for stdout for {0!r}"
msg = msg.format(" ".join(self.args))
raise XonshError(msg)
for name, param in inspect.signature(f).parameters.items():
if (
param.kind == param.POSITIONAL_ONLY
or param.kind == param.POSITIONAL_OR_KEYWORD
):
numargs += 1
elif name in ALIAS_KWARG_NAMES and param.kind == param.KEYWORD_ONLY:
numargs += 1
if numargs < 6:
return functools.partial(PROXIES[numargs], f)
elif numargs == 6:
# don't need to partial.
return f
else:
e = "Expected proxy with 6 or fewer arguments for {}, not {}"
raise XonshError(e.format(", ".join(ALIAS_KWARG_NAMES), numargs))
# redirect to fd
if dest.startswith("&"):
try:
dest = int(dest[1:])
if loc is None:
loc, dest = dest, "" # NOQA
else:
e = "Unrecognized redirection command: {}".format(r)
raise XonshError(e)
except (ValueError, XonshError):
raise
except Exception:
pass
mode = _MODES.get(mode, None)
if mode == "r" and (len(orig) > 0 or len(dest) > 0):
raise XonshError("Unrecognized redirection command: {}".format(r))
elif mode in _WRITE_MODES and len(dest) > 0:
raise XonshError("Unrecognized redirection command: {}".format(r))
return orig, mode, dest
spec = SubprocSpec.build(cmd, captured=captured)
spec.pipeline_index = i
specs.append(spec)
i += 1
# now modify the subprocs based on the redirects.
for i, redirect in enumerate(redirects):
if redirect == "|":
# these should remain integer file descriptors, and not Python
# file objects since they connect processes.
r, w = os.pipe()
specs[i].stdout = w
specs[i + 1].stdin = r
elif redirect == "&" and i == len(redirects) - 1:
specs[-1].background = True
else:
raise XonshError("unrecognized redirect {0!r}".format(redirect))
# Apply boundary conditions
_update_last_spec(specs[-1])
return specs
spec = SubprocSpec.build(cmd, captured=captured)
spec.pipeline_index = i
specs.append(spec)
i += 1
# now modify the subprocs based on the redirects.
for i, redirect in enumerate(redirects):
if redirect == "|":
# these should remain integer file descriptors, and not Python
# file objects since they connect processes.
r, w = os.pipe()
specs[i].stdout = w
specs[i + 1].stdin = r
elif redirect == "&" and i == len(redirects) - 1:
specs[-1].background = True
else:
raise XonshError("unrecognized redirect {0!r}".format(redirect))
# Apply boundary conditions
_update_last_spec(specs[-1])
return specs
# redirect to fd
if dest.startswith("&"):
try:
dest = int(dest[1:])
if loc is None:
loc, dest = dest, "" # NOQA
else:
e = "Unrecognized redirection command: {}".format(r)
raise XonshError(e)
except (ValueError, XonshError):
raise
except Exception:
pass
mode = _MODES.get(mode, None)
if mode == "r" and (len(orig) > 0 or len(dest) > 0):
raise XonshError("Unrecognized redirection command: {}".format(r))
elif mode in _WRITE_MODES and len(dest) > 0:
raise XonshError("Unrecognized redirection command: {}".format(r))
return orig, mode, dest
for name, param in inspect.signature(f).parameters.items():
if (
param.kind == param.POSITIONAL_ONLY
or param.kind == param.POSITIONAL_OR_KEYWORD
):
numargs += 1
elif name in ALIAS_KWARG_NAMES and param.kind == param.KEYWORD_ONLY:
numargs += 1
if numargs < 6:
return functools.partial(PROXIES[numargs], f)
elif numargs == 6:
# don't need to partial.
return f
else:
e = "Expected proxy with 6 or fewer arguments for {}, not {}"
raise XonshError(e.format(", ".join(ALIAS_KWARG_NAMES), numargs))
def _parse_redirects(r, loc=None):
"""returns origin, mode, destination tuple"""
orig, mode, dest = _REDIR_REGEX.match(r).groups()
# redirect to fd
if dest.startswith("&"):
try:
dest = int(dest[1:])
if loc is None:
loc, dest = dest, "" # NOQA
else:
e = "Unrecognized redirection command: {}".format(r)
raise XonshError(e)
except (ValueError, XonshError):
raise
except Exception:
pass
mode = _MODES.get(mode, None)
if mode == "r" and (len(orig) > 0 or len(dest) > 0):
raise XonshError("Unrecognized redirection command: {}".format(r))
elif mode in _WRITE_MODES and len(dest) > 0:
raise XonshError("Unrecognized redirection command: {}".format(r))
return orig, mode, dest