Upload folder using huggingface_hub

096c926 verified almost 2 years ago

12.9 kB

import contextlib
import os
import shlex
import shutil
import sys
import tempfile

from . import formatting
from . import termui
from . import utils
from ._compat import iteritems
from ._compat import PY2
from ._compat import string_types


if PY2:
    from cStringIO import StringIO
else:
    import io
    from ._compat import _find_binary_reader


class EchoingStdin(object):
    def __init__(self, input, output):
        self._input = input
        self._output = output

    def __getattr__(self, x):
        return getattr(self._input, x)

    def _echo(self, rv):
        self._output.write(rv)
        return rv

    def read(self, n=-1):
        return self._echo(self._input.read(n))

    def readline(self, n=-1):
        return self._echo(self._input.readline(n))

    def readlines(self):
        return [self._echo(x) for x in self._input.readlines()]

    def __iter__(self):
        return iter(self._echo(x) for x in self._input)

    def __repr__(self):
        return repr(self._input)


def make_input_stream(input, charset):
    # Is already an input stream.
    if hasattr(input, "read"):
        if PY2:
            return input
        rv = _find_binary_reader(input)
        if rv is not None:
            return rv
        raise TypeError("Could not find binary reader for input stream.")

    if input is None:
        input = b""
    elif not isinstance(input, bytes):
        input = input.encode(charset)
    if PY2:
        return StringIO(input)
    return io.BytesIO(input)


class Result(object):
    """Holds the captured result of an invoked CLI script."""

    def __init__(
        self, runner, stdout_bytes, stderr_bytes, exit_code, exception, exc_info=None
    ):
        #: The runner that created the result
        self.runner = runner
        #: The standard output as bytes.
        self.stdout_bytes = stdout_bytes
        #: The standard error as bytes, or None if not available
        self.stderr_bytes = stderr_bytes
        #: The exit code as integer.
        self.exit_code = exit_code
        #: The exception that happened if one did.
        self.exception = exception
        #: The traceback
        self.exc_info = exc_info

    @property
    def output(self):
        """The (standard) output as unicode string."""
        return self.stdout

    @property
    def stdout(self):
        """The standard output as unicode string."""
        return self.stdout_bytes.decode(self.runner.charset, "replace").replace(
            "\r\n", "\n"
        )

    @property
    def stderr(self):
        """The standard error as unicode string."""
        if self.stderr_bytes is None:
            raise ValueError("stderr not separately captured")
        return self.stderr_bytes.decode(self.runner.charset, "replace").replace(
            "\r\n", "\n"
        )

    def __repr__(self):
        return "<{} {}>".format(
            type(self).__name__, repr(self.exception) if self.exception else "okay"
        )


class CliRunner(object):
    """The CLI runner provides functionality to invoke a Click command line
    script for unittesting purposes in a isolated environment.  This only
    works in single-threaded systems without any concurrency as it changes the
    global interpreter state.
    :param charset: the character set for the input and output data.  This is
                    UTF-8 by default and should not be changed currently as
                    the reporting to Click only works in Python 2 properly.
    :param env: a dictionary with environment variables for overriding.
    :param echo_stdin: if this is set to `True`, then reading from stdin writes
                       to stdout.  This is useful for showing examples in
                       some circumstances.  Note that regular prompts
                       will automatically echo the input.
    :param mix_stderr: if this is set to `False`, then stdout and stderr are
                       preserved as independent streams.  This is useful for
                       Unix-philosophy apps that have predictable stdout and
                       noisy stderr, such that each may be measured
                       independently
    """

    def __init__(self, charset=None, env=None, echo_stdin=False, mix_stderr=True):
        if charset is None:
            charset = "utf-8"
        self.charset = charset
        self.env = env or {}
        self.echo_stdin = echo_stdin
        self.mix_stderr = mix_stderr

    def get_default_prog_name(self, cli):
        """Given a command object it will return the default program name
        for it.  The default is the `name` attribute or ``"root"`` if not
        set.
        """
        return cli.name or "root"

    def make_env(self, overrides=None):
        """Returns the environment overrides for invoking a script."""
        rv = dict(self.env)
        if overrides:
            rv.update(overrides)
        return rv

    @contextlib.contextmanager
    def isolation(self, input=None, env=None, color=False):
        """A context manager that sets up the isolation for invoking of a
        command line tool.  This sets up stdin with the given input data
        and `os.environ` with the overrides from the given dictionary.
        This also rebinds some internals in Click to be mocked (like the
        prompt functionality).
        This is automatically done in the :meth:`invoke` method.
        .. versionadded:: 4.0
           The ``color`` parameter was added.
        :param input: the input stream to put into sys.stdin.
        :param env: the environment overrides as dictionary.
        :param color: whether the output should contain color codes. The
                      application can still override this explicitly.
        """
        input = make_input_stream(input, self.charset)

        old_stdin = sys.stdin
        old_stdout = sys.stdout
        old_stderr = sys.stderr
        old_forced_width = formatting.FORCED_WIDTH
        formatting.FORCED_WIDTH = 80

        env = self.make_env(env)

        if PY2:
            bytes_output = StringIO()
            if self.echo_stdin:
                input = EchoingStdin(input, bytes_output)
            sys.stdout = bytes_output
            if not self.mix_stderr:
                bytes_error = StringIO()
                sys.stderr = bytes_error
        else:
            bytes_output = io.BytesIO()
            if self.echo_stdin:
                input = EchoingStdin(input, bytes_output)
            input = io.TextIOWrapper(input, encoding=self.charset)
            sys.stdout = io.TextIOWrapper(bytes_output, encoding=self.charset)
            if not self.mix_stderr:
                bytes_error = io.BytesIO()
                sys.stderr = io.TextIOWrapper(bytes_error, encoding=self.charset)

        if self.mix_stderr:
            sys.stderr = sys.stdout

        sys.stdin = input

        def visible_input(prompt=None):
            sys.stdout.write(prompt or "")
            val = input.readline().rstrip("\r\n")
            sys.stdout.write("{}\n".format(val))
            sys.stdout.flush()
            return val

        def hidden_input(prompt=None):
            sys.stdout.write("{}\n".format(prompt or ""))
            sys.stdout.flush()
            return input.readline().rstrip("\r\n")

        def _getchar(echo):
            char = sys.stdin.read(1)
            if echo:
                sys.stdout.write(char)
                sys.stdout.flush()
            return char

        default_color = color

        def should_strip_ansi(stream=None, color=None):
            if color is None:
                return not default_color
            return not color

        old_visible_prompt_func = termui.visible_prompt_func
        old_hidden_prompt_func = termui.hidden_prompt_func
        old__getchar_func = termui._getchar
        old_should_strip_ansi = utils.should_strip_ansi
        termui.visible_prompt_func = visible_input
        termui.hidden_prompt_func = hidden_input
        termui._getchar = _getchar
        utils.should_strip_ansi = should_strip_ansi

        old_env = {}
        try:
            for key, value in iteritems(env):
                old_env[key] = os.environ.get(key)
                if value is None:
                    try:
                        del os.environ[key]
                    except Exception:
                        pass
                else:
                    os.environ[key] = value
            yield (bytes_output, not self.mix_stderr and bytes_error)
        finally:
            for key, value in iteritems(old_env):
                if value is None:
                    try:
                        del os.environ[key]
                    except Exception:
                        pass
                else:
                    os.environ[key] = value
            sys.stdout = old_stdout
            sys.stderr = old_stderr
            sys.stdin = old_stdin
            termui.visible_prompt_func = old_visible_prompt_func
            termui.hidden_prompt_func = old_hidden_prompt_func
            termui._getchar = old__getchar_func
            utils.should_strip_ansi = old_should_strip_ansi
            formatting.FORCED_WIDTH = old_forced_width

    def invoke(
        self,
        cli,
        args=None,
        input=None,
        env=None,
        catch_exceptions=True,
        color=False,
        **extra
    ):
        """Invokes a command in an isolated environment.  The arguments are
        forwarded directly to the command line script, the `extra` keyword
        arguments are passed to the :meth:`~clickpkg.Command.main` function of
        the command.
        This returns a :class:`Result` object.
        .. versionadded:: 3.0
           The ``catch_exceptions`` parameter was added.
        .. versionchanged:: 3.0
           The result object now has an `exc_info` attribute with the
           traceback if available.
        .. versionadded:: 4.0
           The ``color`` parameter was added.
        :param cli: the command to invoke
        :param args: the arguments to invoke. It may be given as an iterable
                     or a string. When given as string it will be interpreted
                     as a Unix shell command. More details at
                     :func:`shlex.split`.
        :param input: the input data for `sys.stdin`.
        :param env: the environment overrides.
        :param catch_exceptions: Whether to catch any other exceptions than
                                 ``SystemExit``.
        :param extra: the keyword arguments to pass to :meth:`main`.
        :param color: whether the output should contain color codes. The
                      application can still override this explicitly.
        """
        exc_info = None
        with self.isolation(input=input, env=env, color=color) as outstreams:
            exception = None
            exit_code = 0

            if isinstance(args, string_types):
                args = shlex.split(args)

            try:
                prog_name = extra.pop("prog_name")
            except KeyError:
                prog_name = self.get_default_prog_name(cli)

            try:
                cli.main(args=args or (), prog_name=prog_name, **extra)
            except SystemExit as e:
                exc_info = sys.exc_info()
                exit_code = e.code
                if exit_code is None:
                    exit_code = 0

                if exit_code != 0:
                    exception = e

                if not isinstance(exit_code, int):
                    sys.stdout.write(str(exit_code))
                    sys.stdout.write("\n")
                    exit_code = 1

            except Exception as e:
                if not catch_exceptions:
                    raise
                exception = e
                exit_code = 1
                exc_info = sys.exc_info()
            finally:
                sys.stdout.flush()
                stdout = outstreams[0].getvalue()
                if self.mix_stderr:
                    stderr = None
                else:
                    stderr = outstreams[1].getvalue()

        return Result(
            runner=self,
            stdout_bytes=stdout,
            stderr_bytes=stderr,
            exit_code=exit_code,
            exception=exception,
            exc_info=exc_info,
        )

    @contextlib.contextmanager
    def isolated_filesystem(self):
        """A context manager that creates a temporary folder and changes
        the current working directory to it for isolated filesystem tests.
        """
        cwd = os.getcwd()
        t = tempfile.mkdtemp()
        os.chdir(t)
        try:
            yield t
        finally:
            os.chdir(cwd)
            try:
                shutil.rmtree(t)
            except (OSError, IOError):  # noqa: B014
                pass

	import contextlib
	import os
	import shlex
	import shutil
	import sys
	import tempfile

	from . import formatting
	from . import termui
	from . import utils
	from ._compat import iteritems
	from ._compat import PY2
	from ._compat import string_types


	if PY2:
	from cStringIO import StringIO
	else:
	import io
	from ._compat import _find_binary_reader


	class EchoingStdin(object):
	def __init__(self, input, output):
	self._input = input
	self._output = output

	def __getattr__(self, x):
	return getattr(self._input, x)

	def _echo(self, rv):
	self._output.write(rv)
	return rv

	def read(self, n=-1):
	return self._echo(self._input.read(n))

	def readline(self, n=-1):
	return self._echo(self._input.readline(n))

	def readlines(self):
	return [self._echo(x) for x in self._input.readlines()]

	def __iter__(self):
	return iter(self._echo(x) for x in self._input)

	def __repr__(self):
	return repr(self._input)


	def make_input_stream(input, charset):
	# Is already an input stream.
	if hasattr(input, "read"):
	if PY2:
	return input
	rv = _find_binary_reader(input)
	if rv is not None:
	return rv
	raise TypeError("Could not find binary reader for input stream.")

	if input is None:
	input = b""
	elif not isinstance(input, bytes):
	input = input.encode(charset)
	if PY2:
	return StringIO(input)
	return io.BytesIO(input)


	class Result(object):
	"""Holds the captured result of an invoked CLI script."""

	def __init__(
	self, runner, stdout_bytes, stderr_bytes, exit_code, exception, exc_info=None
	):
	#: The runner that created the result
	self.runner = runner
	#: The standard output as bytes.
	self.stdout_bytes = stdout_bytes
	#: The standard error as bytes, or None if not available
	self.stderr_bytes = stderr_bytes
	#: The exit code as integer.
	self.exit_code = exit_code
	#: The exception that happened if one did.
	self.exception = exception
	#: The traceback
	self.exc_info = exc_info

	@property
	def output(self):
	"""The (standard) output as unicode string."""
	return self.stdout

	@property
	def stdout(self):
	"""The standard output as unicode string."""
	return self.stdout_bytes.decode(self.runner.charset, "replace").replace(
	"\r\n", "\n"
	)

	@property
	def stderr(self):
	"""The standard error as unicode string."""
	if self.stderr_bytes is None:
	raise ValueError("stderr not separately captured")
	return self.stderr_bytes.decode(self.runner.charset, "replace").replace(
	"\r\n", "\n"
	)

	def __repr__(self):
	return "<{} {}>".format(
	type(self).__name__, repr(self.exception) if self.exception else "okay"
	)


	class CliRunner(object):
	"""The CLI runner provides functionality to invoke a Click command line
	script for unittesting purposes in a isolated environment. This only
	works in single-threaded systems without any concurrency as it changes the
	global interpreter state.

	:param charset: the character set for the input and output data. This is
	UTF-8 by default and should not be changed currently as
	the reporting to Click only works in Python 2 properly.
	:param env: a dictionary with environment variables for overriding.
	:param echo_stdin: if this is set to `True`, then reading from stdin writes
	to stdout. This is useful for showing examples in
	some circumstances. Note that regular prompts
	will automatically echo the input.
	:param mix_stderr: if this is set to `False`, then stdout and stderr are
	preserved as independent streams. This is useful for
	Unix-philosophy apps that have predictable stdout and
	noisy stderr, such that each may be measured
	independently
	"""

	def __init__(self, charset=None, env=None, echo_stdin=False, mix_stderr=True):
	if charset is None:
	charset = "utf-8"
	self.charset = charset
	self.env = env or {}
	self.echo_stdin = echo_stdin
	self.mix_stderr = mix_stderr

	def get_default_prog_name(self, cli):
	"""Given a command object it will return the default program name
	for it. The default is the `name` attribute or ``"root"`` if not
	set.
	"""
	return cli.name or "root"

	def make_env(self, overrides=None):
	"""Returns the environment overrides for invoking a script."""
	rv = dict(self.env)
	if overrides:
	rv.update(overrides)
	return rv

	@contextlib.contextmanager
	def isolation(self, input=None, env=None, color=False):
	"""A context manager that sets up the isolation for invoking of a
	command line tool. This sets up stdin with the given input data
	and `os.environ` with the overrides from the given dictionary.
	This also rebinds some internals in Click to be mocked (like the
	prompt functionality).

	This is automatically done in the :meth:`invoke` method.

	.. versionadded:: 4.0
	The ``color`` parameter was added.

	:param input: the input stream to put into sys.stdin.
	:param env: the environment overrides as dictionary.
	:param color: whether the output should contain color codes. The
	application can still override this explicitly.
	"""
	input = make_input_stream(input, self.charset)

	old_stdin = sys.stdin
	old_stdout = sys.stdout
	old_stderr = sys.stderr
	old_forced_width = formatting.FORCED_WIDTH
	formatting.FORCED_WIDTH = 80

	env = self.make_env(env)

	if PY2:
	bytes_output = StringIO()
	if self.echo_stdin:
	input = EchoingStdin(input, bytes_output)
	sys.stdout = bytes_output
	if not self.mix_stderr:
	bytes_error = StringIO()
	sys.stderr = bytes_error
	else:
	bytes_output = io.BytesIO()
	if self.echo_stdin:
	input = EchoingStdin(input, bytes_output)
	input = io.TextIOWrapper(input, encoding=self.charset)
	sys.stdout = io.TextIOWrapper(bytes_output, encoding=self.charset)
	if not self.mix_stderr:
	bytes_error = io.BytesIO()
	sys.stderr = io.TextIOWrapper(bytes_error, encoding=self.charset)

	if self.mix_stderr:
	sys.stderr = sys.stdout

	sys.stdin = input

	def visible_input(prompt=None):
	sys.stdout.write(prompt or "")
	val = input.readline().rstrip("\r\n")
	sys.stdout.write("{}\n".format(val))
	sys.stdout.flush()
	return val

	def hidden_input(prompt=None):
	sys.stdout.write("{}\n".format(prompt or ""))
	sys.stdout.flush()
	return input.readline().rstrip("\r\n")

	def _getchar(echo):
	char = sys.stdin.read(1)
	if echo:
	sys.stdout.write(char)
	sys.stdout.flush()
	return char

	default_color = color

	def should_strip_ansi(stream=None, color=None):
	if color is None:
	return not default_color
	return not color

	old_visible_prompt_func = termui.visible_prompt_func
	old_hidden_prompt_func = termui.hidden_prompt_func
	old__getchar_func = termui._getchar
	old_should_strip_ansi = utils.should_strip_ansi
	termui.visible_prompt_func = visible_input
	termui.hidden_prompt_func = hidden_input
	termui._getchar = _getchar
	utils.should_strip_ansi = should_strip_ansi

	old_env = {}
	try:
	for key, value in iteritems(env):
	old_env[key] = os.environ.get(key)
	if value is None:
	try:
	del os.environ[key]
	except Exception:
	pass
	else:
	os.environ[key] = value
	yield (bytes_output, not self.mix_stderr and bytes_error)
	finally:
	for key, value in iteritems(old_env):
	if value is None:
	try:
	del os.environ[key]
	except Exception:
	pass
	else:
	os.environ[key] = value
	sys.stdout = old_stdout
	sys.stderr = old_stderr
	sys.stdin = old_stdin
	termui.visible_prompt_func = old_visible_prompt_func
	termui.hidden_prompt_func = old_hidden_prompt_func
	termui._getchar = old__getchar_func
	utils.should_strip_ansi = old_should_strip_ansi
	formatting.FORCED_WIDTH = old_forced_width

	def invoke(
	self,
	cli,
	args=None,
	input=None,
	env=None,
	catch_exceptions=True,
	color=False,
	**extra
	):
	"""Invokes a command in an isolated environment. The arguments are
	forwarded directly to the command line script, the `extra` keyword
	arguments are passed to the :meth:`~clickpkg.Command.main` function of
	the command.

	This returns a :class:`Result` object.

	.. versionadded:: 3.0
	The ``catch_exceptions`` parameter was added.

	.. versionchanged:: 3.0
	The result object now has an `exc_info` attribute with the
	traceback if available.

	.. versionadded:: 4.0
	The ``color`` parameter was added.

	:param cli: the command to invoke
	:param args: the arguments to invoke. It may be given as an iterable
	or a string. When given as string it will be interpreted
	as a Unix shell command. More details at
	:func:`shlex.split`.
	:param input: the input data for `sys.stdin`.
	:param env: the environment overrides.
	:param catch_exceptions: Whether to catch any other exceptions than
	``SystemExit``.
	:param extra: the keyword arguments to pass to :meth:`main`.
	:param color: whether the output should contain color codes. The
	application can still override this explicitly.
	"""
	exc_info = None
	with self.isolation(input=input, env=env, color=color) as outstreams:
	exception = None
	exit_code = 0

	if isinstance(args, string_types):
	args = shlex.split(args)

	try:
	prog_name = extra.pop("prog_name")
	except KeyError:
	prog_name = self.get_default_prog_name(cli)

	try:
	cli.main(args=args or (), prog_name=prog_name, **extra)
	except SystemExit as e:
	exc_info = sys.exc_info()
	exit_code = e.code
	if exit_code is None:
	exit_code = 0

	if exit_code != 0:
	exception = e

	if not isinstance(exit_code, int):
	sys.stdout.write(str(exit_code))
	sys.stdout.write("\n")
	exit_code = 1

	except Exception as e:
	if not catch_exceptions:
	raise
	exception = e
	exit_code = 1
	exc_info = sys.exc_info()
	finally:
	sys.stdout.flush()
	stdout = outstreams[0].getvalue()
	if self.mix_stderr:
	stderr = None
	else:
	stderr = outstreams[1].getvalue()

	return Result(
	runner=self,
	stdout_bytes=stdout,
	stderr_bytes=stderr,
	exit_code=exit_code,
	exception=exception,
	exc_info=exc_info,
	)

	@contextlib.contextmanager
	def isolated_filesystem(self):
	"""A context manager that creates a temporary folder and changes
	the current working directory to it for isolated filesystem tests.
	"""
	cwd = os.getcwd()
	t = tempfile.mkdtemp()
	os.chdir(t)
	try:
	yield t
	finally:
	os.chdir(cwd)
	try:
	shutil.rmtree(t)
	except (OSError, IOError): # noqa: B014
	pass