Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
This method instantiates a :class:`subprocess.Popen` object based on
the defaults defined by :class:`ExternalCommand` and the overrides
configured by the caller. What happens then depends on :attr:`asynchronous`:
- If :attr:`asynchronous` is set :func:`start()` starts the external command
but doesn't wait for it to end (use :func:`wait()` for that).
- If :attr:`asynchronous` isn't set the ``communicate()`` method on the
:attr:`subprocess` object is called to synchronously execute the
external command.
"""
if self.is_running:
# If the external command is currently running reset() will leak
# file descriptors and besides that it doesn't really make sense to
# re-use ExternalCommand objects in this way.
raise ValueError(compact("""
External command is already running! (you need to explicitly
terminate, kill or wait for the running process before you can
re-use the ExternalCommand object)
"""))
# Prepare the keyword arguments to subprocess.Popen().
kw = dict(args=self.command_line,
bufsize=self.buffer_size,
cwd=self.directory,
env=os.environ.copy())
kw['env'].update(self.environment)
# Prepare the command's standard input/output/error streams.
kw['stdin'] = self.stdin_stream.prepare_input()
kw['stdout'] = self.stdout_stream.prepare_output(self.stdout_file, self.capture)
kw['stderr'] = (subprocess.STDOUT if self.merge_streams else
self.stderr_stream.prepare_output(self.stderr_file, self.capture_stderr))
if self.retry and not self.asynchronous:
def ensure_working_tree(self):
"""
Make sure the local repository has working tree support.
:raises: :exc:`~vcs_repo_mgr.exceptions.MissingWorkingTreeError` when
the local repository doesn't support a working tree.
"""
if not self.supports_working_tree:
raise MissingWorkingTreeError(compact("""
A working tree is required but the local {friendly_name}
repository at {directory} doesn't support a working tree!
""", friendly_name=self.friendly_name, directory=format_path(self.local)))
:param exception: An :exc:`~executor.ExternalCommandFailed` object.
:returns: :data:`True` if the operator has interactively resolved any
merge conflicts (and as such the merge error doesn't need to
be propagated), :data:`False` otherwise.
This method checks whether :data:`sys.stdin` is connected to a terminal
to decide whether interaction with an operator is possible. If it is
then an interactive terminal prompt is used to ask the operator to
resolve the merge conflict(s). If the operator confirms the prompt, the
merge error is swallowed instead of propagated. When :data:`sys.stdin`
is not connected to a terminal or the operator denies the prompt the
merge error is propagated.
"""
if connected_to_terminal(sys.stdin):
logger.info(compact("""
It seems that I'm connected to a terminal so I'll give you a
chance to interactively fix the merge conflict(s) in order to
avoid propagating the merge error. Please mark or stage your
changes but don't commit the result just yet (it will be done
for you).
"""))
while True:
if prompt_for_confirmation("Ignore merge error because you've resolved all conflicts?"):
if self.merge_conflicts:
logger.warning("I'm still seeing merge conflicts, please double check! (%s)",
concatenate(self.merge_conflicts))
else:
# The operator resolved all conflicts.
return True
else:
# The operator wants us to propagate the error.
def get_push_command(self, remote=None, revision=None):
"""Get the command to push changes from the local repository to a remote repository."""
if revision:
raise NotImplementedError(compact("""
Bazaar repository support doesn't include
the ability to push specific revisions!
"""))
command = ['bzr', 'push']
if remote:
command.append(remote)
return command
using :func:`~humanfriendly.coerce_boolean()`) when available, otherwise
:data:`SPHINX_ACTIVE` determines the default value.
Usage notes are only injected when Sphinx is running because of performance.
It's nothing critical of course, but modifying hundreds or thousands of
docstrings that no one is going to look at seems rather pointless :-).
"""
NOTHING = object()
"""A unique object instance used to detect missing attributes."""
CUSTOM_PROPERTY_NOTE = compact("""
The :attr:`{name}` property is a :class:`~{type}`.
""")
DYNAMIC_PROPERTY_NOTE = compact("""
The :attr:`{name}` property is a :class:`~{type}`.
""")
ENVIRONMENT_PROPERTY_NOTE = compact("""
If the environment variable ``${variable}`` is set it overrides the
computed value of this property.
""")
REQUIRED_PROPERTY_NOTE = compact("""
You are required to provide a value for this property by calling the
constructor of the class that defines the property with a keyword argument
named `{name}` (unless a custom constructor is defined, in this case please
refer to the documentation of that constructor).
""")
KEY_PROPERTY_NOTE = compact("""
def ensure_exists(self):
"""Make sure the location exists."""
if not self.context.is_directory(self.directory):
# This can also happen when we don't have permission to one of the
# parent directories so we'll point that out in the error message
# when it seems applicable (so as not to confuse users).
if self.context.have_superuser_privileges:
msg = "The directory %s doesn't exist!"
raise ValueError(msg % self)
else:
raise ValueError(compact("""
The directory {location} isn't accessible, most likely
because it doesn't exist or because of permissions. If
you're sure the directory exists you can use the
--use-sudo option.
""", location=self))
def proxy_method(self, *args, **kw):
if not hasattr(self, 'output'):
raise TypeError(compact("""
The old calling interface is only supported when
merged=True and start_capture() has been called!
"""))
real_method = getattr(self.output, name)
return real_method(*args, **kw)
# Get the docstring of the real method.
It's nothing critical of course, but modifying hundreds or thousands of
docstrings that no one is going to look at seems rather pointless :-).
"""
NOTHING = object()
"""A unique object instance used to detect missing attributes."""
CUSTOM_PROPERTY_NOTE = compact("""
The :attr:`{name}` property is a :class:`~{type}`.
""")
DYNAMIC_PROPERTY_NOTE = compact("""
The :attr:`{name}` property is a :class:`~{type}`.
""")
ENVIRONMENT_PROPERTY_NOTE = compact("""
If the environment variable ``${variable}`` is set it overrides the
computed value of this property.
""")
REQUIRED_PROPERTY_NOTE = compact("""
You are required to provide a value for this property by calling the
constructor of the class that defines the property with a keyword argument
named `{name}` (unless a custom constructor is defined, in this case please
refer to the documentation of that constructor).
""")
KEY_PROPERTY_NOTE = compact("""
Once this property has been assigned a value you are not allowed to assign
a new value to the property.
""")