# Path: paramiko/ssh_exception.py
# class ProxyCommandFailure(SSHException):
#     """
#     The "ProxyCommand" found in the .ssh/config file returned an error.
# 
#     :param str command: The command line that is generating this exception.
#     :param str error: The error captured from the proxy command output.
#     """
# 
#     def __init__(self, command, error):
#         SSHException.__init__(self, command, error)
#         self.command = command
#         self.error = error
# 
#     def __str__(self):
#         return 'ProxyCommand("{}") returned nonzero exit status: {}'.format(
#             self.command, self.error
#         )
# 
# Path: paramiko/proxy.py
# class ProxyCommand(ClosingContextManager):
#     """
#     Wraps a subprocess running ProxyCommand-driven programs.
# 
#     This class implements a the socket-like interface needed by the
#     `.Transport` and `.Packetizer` classes. Using this class instead of a
#     regular socket makes it possible to talk with a Popen'd command that will
#     proxy traffic between the client and a server hosted in another machine.
# 
#     Instances of this class may be used as context managers.
#     """
# 
#     def __init__(self, command_line):
#         """
#         Create a new CommandProxy instance. The instance created by this
#         class can be passed as an argument to the `.Transport` class.
# 
#         :param str command_line:
#             the command that should be executed and used as the proxy.
#         """
#         if subprocess is None:
#             raise subprocess_import_error
#         self.cmd = shlex.split(command_line)
#         self.process = subprocess.Popen(
#             self.cmd,
#             stdin=subprocess.PIPE,
#             stdout=subprocess.PIPE,
#             stderr=subprocess.PIPE,
#             bufsize=0,
#         )
#         self.timeout = None
# 
#     def send(self, content):
#         """
#         Write the content received from the SSH client to the standard
#         input of the forked command.
# 
#         :param str content: string to be sent to the forked command
#         """
#         try:
#             self.process.stdin.write(content)
#         except IOError as e:
#             # There was a problem with the child process. It probably
#             # died and we can't proceed. The best option here is to
#             # raise an exception informing the user that the informed
#             # ProxyCommand is not working.
#             raise ProxyCommandFailure(" ".join(self.cmd), e.strerror)
#         return len(content)
# 
#     def recv(self, size):
#         """
#         Read from the standard output of the forked program.
# 
#         :param int size: how many chars should be read
# 
#         :return: the string of bytes read, which may be shorter than requested
#         """
#         try:
#             buffer = b""
#             start = time.time()
#             while len(buffer) < size:
#                 select_timeout = None
#                 if self.timeout is not None:
#                     elapsed = time.time() - start
#                     if elapsed >= self.timeout:
#                         raise socket.timeout()
#                     select_timeout = self.timeout - elapsed
# 
#                 r, w, x = select([self.process.stdout], [], [], select_timeout)
#                 if r and r[0] == self.process.stdout:
#                     buffer += os.read(
#                         self.process.stdout.fileno(), size - len(buffer)
#                     )
#             return buffer
#         except socket.timeout:
#             if buffer:
#                 # Don't raise socket.timeout, return partial result instead
#                 return buffer
#             raise  # socket.timeout is a subclass of IOError
#         except IOError as e:
#             raise ProxyCommandFailure(" ".join(self.cmd), e.strerror)
# 
#     def close(self):
#         os.kill(self.process.pid, signal.SIGTERM)
# 
#     @property
#     def closed(self):
#         return self.process.returncode is not None
# 
#     @property
#     def _closed(self):
#         # Concession to Python 3 socket-like API
#         return self.closed
# 
#     def settimeout(self, timeout):
#         self.timeout = timeout


import signal
import socket
from mock import patch
from pytest import raises
from paramiko import ProxyCommand, ProxyCommandFailure
        stdout = Popen.return_value.stdout
        select.return_value = [stdout], None, None
        fileno = stdout.fileno.return_value
        os_read.side_effect = [b"was", socket.timeout]
        proxy = ProxyCommand("hi")
        data = proxy.recv(5)
        assert data == b"was"  # not b"waste"
        assert os_read.call_args[0] == (fileno, 2)

    @patch("paramiko.proxy.subprocess.Popen")
    @patch("paramiko.proxy.os.read")
    @patch("paramiko.proxy.select")
    def test_recv_raises_timeout_if_nothing_read(self, select, os_read, Popen):
        stdout = Popen.return_value.stdout
        select.return_value = [stdout], None, None
        fileno = stdout.fileno.return_value
        os_read.side_effect = socket.timeout
        proxy = ProxyCommand("hi")
        with raises(socket.timeout):
            proxy.recv(5)
        assert os_read.call_args[0] == (fileno, 5)

    @patch("paramiko.proxy.subprocess.Popen")
    @patch("paramiko.proxy.os.read")
    @patch("paramiko.proxy.select")
    def test_recv_raises_ProxyCommandFailure_on_non_timeout_error(
        self, select, os_read, Popen
    ):
        select.return_value = [Popen.return_value.stdout], None, None
        os_read.side_effect = IOError(0, "whoops")

# Path: paramiko/ssh_exception.py
# class ProxyCommandFailure(SSHException):
#     """
#     The "ProxyCommand" found in the .ssh/config file returned an error.
# 
#     :param str command: The command line that is generating this exception.
#     :param str error: The error captured from the proxy command output.
#     """
# 
#     def __init__(self, command, error):
#         SSHException.__init__(self, command, error)
#         self.command = command
#         self.error = error
# 
#     def __str__(self):
#         return 'ProxyCommand("{}") returned nonzero exit status: {}'.format(
#             self.command, self.error
#         )
# 
# Path: paramiko/proxy.py
# class ProxyCommand(ClosingContextManager):
#     """
#     Wraps a subprocess running ProxyCommand-driven programs.
# 
#     This class implements a the socket-like interface needed by the
#     `.Transport` and `.Packetizer` classes. Using this class instead of a
#     regular socket makes it possible to talk with a Popen'd command that will
#     proxy traffic between the client and a server hosted in another machine.
# 
#     Instances of this class may be used as context managers.
#     """
# 
#     def __init__(self, command_line):
#         """
#         Create a new CommandProxy instance. The instance created by this
#         class can be passed as an argument to the `.Transport` class.
# 
#         :param str command_line:
#             the command that should be executed and used as the proxy.
#         """
#         if subprocess is None:
#             raise subprocess_import_error
#         self.cmd = shlex.split(command_line)
#         self.process = subprocess.Popen(
#             self.cmd,
#             stdin=subprocess.PIPE,
#             stdout=subprocess.PIPE,
#             stderr=subprocess.PIPE,
#             bufsize=0,
#         )
#         self.timeout = None
# 
#     def send(self, content):
#         """
#         Write the content received from the SSH client to the standard
#         input of the forked command.
# 
#         :param str content: string to be sent to the forked command
#         """
#         try:
#             self.process.stdin.write(content)
#         except IOError as e:
#             # There was a problem with the child process. It probably
#             # died and we can't proceed. The best option here is to
#             # raise an exception informing the user that the informed
#             # ProxyCommand is not working.
#             raise ProxyCommandFailure(" ".join(self.cmd), e.strerror)
#         return len(content)
# 
#     def recv(self, size):
#         """
#         Read from the standard output of the forked program.
# 
#         :param int size: how many chars should be read
# 
#         :return: the string of bytes read, which may be shorter than requested
#         """
#         try:
#             buffer = b""
#             start = time.time()
#             while len(buffer) < size:
#                 select_timeout = None
#                 if self.timeout is not None:
#                     elapsed = time.time() - start
#                     if elapsed >= self.timeout:
#                         raise socket.timeout()
#                     select_timeout = self.timeout - elapsed
# 
#                 r, w, x = select([self.process.stdout], [], [], select_timeout)
#                 if r and r[0] == self.process.stdout:
#                     buffer += os.read(
#                         self.process.stdout.fileno(), size - len(buffer)
#                     )
#             return buffer
#         except socket.timeout:
#             if buffer:
#                 # Don't raise socket.timeout, return partial result instead
#                 return buffer
#             raise  # socket.timeout is a subclass of IOError
#         except IOError as e:
#             raise ProxyCommandFailure(" ".join(self.cmd), e.strerror)
# 
#     def close(self):
#         os.kill(self.process.pid, signal.SIGTERM)
# 
#     @property
#     def closed(self):
#         return self.process.returncode is not None
# 
#     @property
#     def _closed(self):
#         # Concession to Python 3 socket-like API
#         return self.closed
# 
#     def settimeout(self, timeout):
#         self.timeout = timeout


# Path: tests/test_proxy.py
import signal
import socket
from mock import patch
from pytest import raises
from paramiko import ProxyCommand, ProxyCommandFailure

        stdout = Popen.return_value.stdout
        select.return_value = [stdout], None, None
        fileno = stdout.fileno.return_value
        os_read.side_effect = [b"was", socket.timeout]
        proxy = ProxyCommand("hi")
        data = proxy.recv(5)
        assert data == b"was"  # not b"waste"
        assert os_read.call_args[0] == (fileno, 2)

    @patch("paramiko.proxy.subprocess.Popen")
    @patch("paramiko.proxy.os.read")
    @patch("paramiko.proxy.select")
    def test_recv_raises_timeout_if_nothing_read(self, select, os_read, Popen):
        stdout = Popen.return_value.stdout
        select.return_value = [stdout], None, None
        fileno = stdout.fileno.return_value
        os_read.side_effect = socket.timeout
        proxy = ProxyCommand("hi")
        with raises(socket.timeout):
            proxy.recv(5)
        assert os_read.call_args[0] == (fileno, 5)

    @patch("paramiko.proxy.subprocess.Popen")
    @patch("paramiko.proxy.os.read")
    @patch("paramiko.proxy.select")
    def test_recv_raises_ProxyCommandFailure_on_non_timeout_error(
        self, select, os_read, Popen
    ):
        select.return_value = [Popen.return_value.stdout], None, None
        os_read.side_effect = IOError(0, "whoops")

        with raises(ProxyCommandFailure) as info: