+ 

+ Encoding and bytes/text

+ -----------------------

+ 

+ When writing files or issuing commands, bytestrings are passed through

+ unchanged, and text strings (``unicode`` in Python 2) are encoded using

+ a configurable encoding (``utf-8`` by default).

+ 

+ When reading files, bytestrings are returned by default,

+ but an encoding can be given to get a test string.

+ 

+ For command output, separate ``stdout_bytes`` and ``stdout_text`` attributes

+ are provided.

+ The latter uses a configurable encoding (``utf-8` by default).

+ 

+ 

+     command_prelude = b''

+     def put_file_contents(self, filename, contents, encoding='utf-8'):

+         self.transport.put_file_contents(filename, contents, encoding=encoding)

+                     cwd=None, bg=False, encoding='utf-8'):

+         :param bg: If True, runs command in background.

+                    In this case, either the result should be used in a ``with``

+                    statement, or ``wait()`` should be called explicitly

+                    when the command is finished.

+         :param encoding: Encoding for the resulting Command instance's

+                          ``stdout_text`` and ``stderr_text``, and for

+                          ``stdin_text``, ``argv``, etc. if they are not

+                          bytestrings already.

+         def encode(string):

+             if not isinstance(string, bytes):

+                 return string.encode(encoding)

+             else:

+                 return string

+ 

+         command = self.transport.start_shell(argv, log_stdout=log_stdout,

+                                              encoding=encoding)

+         command.stdin.write(b'cd %s\n' % shell_quote(encode(cwd)))

+             quoted = shell_quote(encode(self.env_sh_path))

+             command.stdin.write(b'. %s\n' % quoted)

+             command.stdin.write(encode(self.command_prelude))

+ 

+         if stdin_text:

+             command.stdin.write(b"echo -e ")

+             command.stdin.write(_echo_quote(encode(stdin_text)))

+             command.stdin.write(b" | ")

+             command.stdin.write(b'(')

+             command.stdin.write(encode(argv))

+             command.stdin.write(b')')

+                 command.stdin.write(shell_quote(encode(arg)))

+                 command.stdin.write(b' ')

+         command.stdin.write(b'\nexit\n')

+         command.raiseonerr = raiseonerr

+             command.wait()

+ def _echo_quote(bytestring):

+     """Encode a bytestring for use with bash & "echo -e"

+     """

+     bytestring = bytestring.replace(b"\\", br"\\")

+     bytestring = bytestring.replace(b"\0", br"\x00")

+     bytestring = bytestring.replace(b"'", br"'\''")

+     return b"'" + bytestring + b"'"

+ 

+     command_prelude = b'set -e\n'

+ DEFAULT = object()

+ 

+         """Read the named remote file and return the contents

+ 

+         The string will be decoded using the given encoding;

+         if encoding is None (default), it will be returned as a bytestring.

+         """

+     def put_file_contents(self, filename, contents, encoding='utf-8'):

+         """Write the given string (or bytestring) to the named remote file

+ 

+         The contents string will be encoded using the given encoding

+         (default: ``'utf-8'``), unless aleady a bytestring.

+         """

+     def start_shell(self, argv, log_stdout=True, encoding=None):

+         :param encoding: Encoding for the resulting Command's ``stdout_text``

+                          and ``stderr_text``.

+ 

+         Note that ``shell.stdin`` uses bytes I/O.

+         contents = self.get_file_contents(remotepath, encoding=None)

+         self.put_file_contents(remotepath, contents, encoding=None)

+ class _decoded_output_property(object):

+     """Descriptor for on-demand decoding of a Command's output stream

+     """

+     def __init__(self, name):

+         self.name = name

+ 

+     def __set_name__(self, cls, name):

+         # Sanity check (called only on Python 3.6+).

+         # This property expects to handle attributes named '<foo>_text'.

+         assert name == self.name + '_text'

+ 

+     def __get__(self, instance, cls=None):

+         if instance is None:

+             return self

+         else:

+             bytestring = getattr(instance, self.name + '_bytes')

+             print(bytestring, instance.encoding)

+             decoded = bytestring.decode(instance.encoding)

+             setattr(instance, self.name + '_text', decoded)

+             return decoded

+ 

+ 

+     After calling wait(), ``stdout_bytes`` and ``stderr_bytes`` attributes will

+     be bytestrings containing the output, and ``returncode`` will contain the

+ 

+     The ``stdout_text`` and ``stdout_text`` will be the corresponding output

+     decoded using the given ``encoding`` (default: ``'utf-8'``).

+     These are decoded on-demand; do not access them if a command

+     produces binary output.

+ 

+     A Command may be used as a context manager (in the ``with`` statement).

+     Exiting the context will automatically call ``wait()``.

+     This raises an exception if the exit code is not 0, unless the

+     ``raiseonerr`` attribute is set to false before exiting the context.

+                  get_logger=None, encoding='utf-8'):

+         self.encoding = encoding

+         self.raiseonerr = True

+ 

+     stdout_text = _decoded_output_property('stdout')

+     stderr_text = _decoded_output_property('stderr')

+     def wait(self, raiseonerr=DEFAULT):

+         Raises an exception if the exit code is not 0, unless ``raiseonerr`` is

+ 

+         When ``raiseonerr`` is not specified as argument, the ``raiseonerr``

+         attribute is used.

+         if raiseonerr is DEFAULT:

+             raiseonerr = self.raiseonerr

+ 

+     def __enter__(self):

+         return self

+ 

+     def __exit__(self, *exc_info):

+         self.wait(raiseonerr=self.raiseonerr)

+ 

+         with self.sftp_open(filename, 'rb') as f:

+     def put_file_contents(self, filename, contents, encoding=None):

+         if encoding and not isinstance(contents, bytes):

+             contents = contents.encode(encoding)

+         with self.sftp_open(filename, 'wb') as f:

+     def start_shell(self, argv, log_stdout=True, encoding='utf-8'):

+                           get_logger=self.host.config.get_logger,

+                           encoding=encoding)

+     def start_shell(self, argv, log_stdout=True, encoding='utf-8'):

+         command = self._run(['bash'], argv=argv, log_stdout=log_stdout,

+                             encoding=encoding)

+     def _run(self, command, log_stdout=True, argv=None, collect_output=True,

+              encoding='utf-8'):

+                           get_logger=self.host.config.get_logger,

+                           encoding=encoding)

+     def put_file_contents(self, filename, contents, encoding='utf-8'):

+         if encoding and not isinstance(contents, bytes):

+             contents = contents.encode(encoding)

+         assert cmd.stdout_bytes == contents

+             result = cmd.stdout_bytes

+                                          get_logger=get_logger,

+                                          encoding=encoding)

+         self._use_bytes = (encoding is None)

+ 

+             if self._use_bytes:

+         self.stdin = self._ssh.makefile('wb')

+         stdout = self._ssh.makefile('rb')

+         stderr = self._ssh.makefile_stderr('rb')

+         self.stdout_bytes = b''.join(self._stdout_lines)

+         self.stderr_bytes = b''.join(self._stderr_lines)

+ 

+                     log.debug(line.rstrip(b'\n').decode('utf-8',

+                                                         errors='replace'))

+ def shell_quote(bytestring):

+     """Quotes a bytestring for the Bash shell"""

+     return b"'" + bytestring.replace(b"'", b"'\\''") + b"'"

+ import sys

+         # Run dummy command first; this should catch spurious SSH messages.

+         host.run_command(['echo', 'hello', 'world'])

+         # Now, run the actual command

+     def test_get_put_file_contents_bytes(self, multihost, tmpdir):

+         host = multihost.host

+         filename = str(tmpdir.join('test-bytes.txt'))

+         testbytes = u'test \0 \N{WHITE SMILING FACE}'.encode('utf-8')

+         with _first_command(host):

+             host.put_file_contents(filename, testbytes, encoding=None)

+         result = host.get_file_contents(filename, encoding=None)

+         assert result == testbytes

+ 

+     @pytest.mark.parametrize('encoding', ('utf-8', 'utf-16'))

+     def test_put_file_contents_utf(self, multihost, tmpdir, encoding):

+         host = multihost.host

+         filename = str(tmpdir.join('test-{}.txt'.format(encoding)))

+         teststring = u'test \N{WHITE SMILING FACE}'

+         with _first_command(host):

+             host.put_file_contents(filename, teststring, encoding=encoding)

+         result = host.get_file_contents(filename, encoding=None)

+         assert result == teststring.encode(encoding)

+         with open(filename, 'rb') as f:

+             assert f.read() == teststring.encode(encoding)

+ 

+     @pytest.mark.parametrize('encoding', ('utf-8', 'utf-16'))

+     def test_get_file_contents_encoding(self, multihost, tmpdir, encoding):

+         host = multihost.host

+         filename = str(tmpdir.join('test-{}.txt'.format(encoding)))

+         teststring = u'test \N{WHITE SMILING FACE}'

+         with open(filename, 'wb') as f:

+             f.write(teststring.encode(encoding))

+         result = host.get_file_contents(filename, encoding=encoding)

+         assert result == teststring

+         assert type(result) == type(u'')

+ 

+     def test_escaping(self, multihost, tmpdir):

+         host = multihost.host

+         test_file_path = str(tmpdir.join('testfile.txt'))

+ 

+         stdin_text = '"test", test, "test", $test, '

+         stdin_text += ''.join(chr(x) for x in range(32, 127))

+         stdin_text += r', \x66\0111\x00, '

+         stdin_text += ''.join('\\' + chr(x) for x in range(32, 127))

+         tee = host.run_command(

+             ["tee", test_file_path],

+             stdin_text=stdin_text,

+             raiseonerr=False,

+         )

+         print(tee.stderr_text)

+         assert tee.stdout_text == stdin_text + '\n'

+         with open(test_file_path, "r") as f:

+             assert f.read() == tee.stdout_text

+ 

+     def test_escaping_binary(self, multihost, tmpdir):

+         host = multihost.host

+         test_file_path = str(tmpdir.join('testfile.txt'))

+ 

+         stdin_bytes = b'"test", test, "test", $test, '

+         stdin_bytes += bytes(range(0, 256))

+         stdin_bytes += br', \x66\0111\x00'

+         tee = host.run_command(

+             ["tee", test_file_path],

+             stdin_text=stdin_bytes,

+             raiseonerr=False,

+         )

+         assert tee.stdout_bytes == stdin_bytes + b'\n'

+         with open(test_file_path, "rb") as f:

+             assert f.read() == tee.stdout_bytes

+ 

+     def test_background_explicit_wait(self, multihost, tmpdir):

+         host = multihost.host

+ 

+         pipe_filename = str(tmpdir.join('test.pipe'))

+ 

+         with _first_command(host):

+             host.run_command(['mkfifo', pipe_filename])

+ 

+         cat = host.run_command(['cat', pipe_filename], bg=True)

+         host.run_command('cat > ' + pipe_filename, stdin_text='expected value')

+ 

+         cat.wait()

+         assert cat.stdout_text == 'expected value\n'

+         assert cat.returncode == 0

+ 

+     def test_background_context(self, multihost, tmpdir):

+         host = multihost.host

+ 

+         pipe_filename = str(tmpdir.join('test.pipe'))

+ 

+         with _first_command(host):

+             host.run_command(['mkfifo', pipe_filename])

+ 

+         with host.run_command(['cat', pipe_filename], bg=True) as cat:

+             host.run_command('cat > ' + pipe_filename,

+                              stdin_text='expected value')

+ 

+         assert cat.stdout_text == 'expected value\n'

+         assert cat.returncode == 0

+ 

+ 

+     def test_background_raiseonerr_false(self, multihost, tmpdir):

+         host = multihost.host

+         with _first_command(host):

+             false = host.run_command(['false'], raiseonerr=False, bg=True)

+ 

+         assert false.returncode != 0

+ 

+ 

+     def test_background_raiseonerr_with(self, multihost, tmpdir):

+         host = multihost.host

+         with _first_command(host):

+             with pytest.raises(CalledProcessError):

+                 with host.run_command(['false'], raiseonerr=True, bg=True):

+                     pass

+ 

+     def test_background_raiseonerr_wait(self, multihost, tmpdir):

+         host = multihost.host

+         with _first_command(host):

+             false = host.run_command(['false'], raiseonerr=True, bg=True)

+ 

+             with pytest.raises(CalledProcessError):

+                 false.wait()

+ 

+ 

+ 

+ @pytest.mark.needs_ssh

+ class TestLocalhostBadConnection(object):

+ # Tox (http://tox.testrun.org/) is a tool for running tests

+ # in multiple virtualenvs. This configuration file will run the

+ # test suite on all supported python versions. To use it, "pip install tox"

+ # and then run "tox" from this directory.

+ 

+ [tox]

+ envlist = py2,py36

+ minver = 1.8

+ 

+ [testenv]

+ deps =

+     pytest

+     paramiko

+ commands = python -m pytest -vv test_pytestmultihost/