/brz/remove-bazaar

# Copyright (C) 2006-2016 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tests for smart transport"""

# all of this deals with byte strings so this is safe

import doctest

import errno

import os

import socket

import subprocess

import sys

import threading

import time

from testtools.matchers import DocTestMatches

import breezy

from .. import (

    controldir,

    debug,

    errors,

    osutils,

    tests,

    transport as _mod_transport,

    urlutils,

    )

from ..bzr import (

    bzrdir,

    )

from ..bzr.smart import (

    client,

    medium,

    message,

    protocol,

    request as _mod_request,

    server as _mod_server,

    vfs,

    )

from . import (

    features,

    test_smart,

    test_server,

    )

from ..sixish import (

    BytesIO,

    int2byte,

    )

from ..transport import (

    http,

    local,

    memory,

    remote,

    ssh,

    )

def create_file_pipes():

    r, w = os.pipe()

    # These must be opened without buffering, or we get undefined results

    rf = os.fdopen(r, 'rb', 0)

    wf = os.fdopen(w, 'wb', 0)

    return rf, wf

def portable_socket_pair():

    """Return a pair of TCP sockets connected to each other.

    Unlike socket.socketpair, this should work on Windows.

    """

    listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    listen_sock.bind(('127.0.0.1', 0))

    listen_sock.listen(1)

    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    client_sock.connect(listen_sock.getsockname())

    server_sock, addr = listen_sock.accept()

    listen_sock.close()

    return server_sock, client_sock

class BytesIOSSHVendor(object):

    """A SSH vendor that uses BytesIO to buffer writes and answer reads."""

    def __init__(self, read_from, write_to):

        self.read_from = read_from

        self.write_to = write_to

        self.calls = []

    def connect_ssh(self, username, password, host, port, command):

        self.calls.append(('connect_ssh', username, password, host, port,

                           command))

        return BytesIOSSHConnection(self)

class FirstRejectedBytesIOSSHVendor(BytesIOSSHVendor):

    """The first connection will be considered closed.

    The second connection will succeed normally.

    """

    def __init__(self, read_from, write_to, fail_at_write=True):

        super(FirstRejectedBytesIOSSHVendor, self).__init__(read_from,

                                                            write_to)

        self.fail_at_write = fail_at_write

        self._first = True

    def connect_ssh(self, username, password, host, port, command):

        self.calls.append(('connect_ssh', username, password, host, port,

                           command))

        if self._first:

            self._first = False

            return ClosedSSHConnection(self)

        return BytesIOSSHConnection(self)

class BytesIOSSHConnection(ssh.SSHConnection):

    """A SSH connection that uses BytesIO to buffer writes and answer reads."""

    def __init__(self, vendor):

        self.vendor = vendor

    def close(self):

        self.vendor.calls.append(('close', ))

        self.vendor.read_from.close()

        self.vendor.write_to.close()

    def get_sock_or_pipes(self):

        return 'pipes', (self.vendor.read_from, self.vendor.write_to)

class ClosedSSHConnection(ssh.SSHConnection):

    """An SSH connection that just has closed channels."""

    def __init__(self, vendor):

        self.vendor = vendor

    def close(self):

        self.vendor.calls.append(('close', ))

    def get_sock_or_pipes(self):

        # We create matching pipes, and then close the ssh side

        bzr_read, ssh_write = create_file_pipes()

        # We always fail when bzr goes to read

        ssh_write.close()

        if self.vendor.fail_at_write:

            # If set, we'll also fail when bzr goes to write

            ssh_read, bzr_write = create_file_pipes()

            ssh_read.close()

        else:

            bzr_write = self.vendor.write_to

        return 'pipes', (bzr_read, bzr_write)

class _InvalidHostnameFeature(features.Feature):

    """Does 'non_existent.invalid' fail to resolve?

    RFC 2606 states that .invalid is reserved for invalid domain names, and

    also underscores are not a valid character in domain names.  Despite this,

    it's possible a badly misconfigured name server might decide to always

    return an address for any name, so this feature allows us to distinguish a

    broken system from a broken test.

    """

    def _probe(self):

        try:

            socket.gethostbyname('non_existent.invalid')

        except socket.gaierror:

            # The host name failed to resolve.  Good.

            return True

        else:

            return False

    def feature_name(self):

        return 'invalid hostname'

InvalidHostnameFeature = _InvalidHostnameFeature()

class SmartClientMediumTests(tests.TestCase):

    """Tests for SmartClientMedium.

    We should create a test scenario for this: we need a server module that

    construct the test-servers (like make_loopsocket_and_medium), and the list

    of SmartClientMedium classes to test.

    """

    def make_loopsocket_and_medium(self):

        """Create a loopback socket for testing, and a medium aimed at it."""

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        sock.bind(('127.0.0.1', 0))

        sock.listen(1)

        port = sock.getsockname()[1]

        client_medium = medium.SmartTCPClientMedium('127.0.0.1', port, 'base')

        return sock, client_medium

    def receive_bytes_on_server(self, sock, bytes):

        """Accept a connection on sock and read 3 bytes.

        The bytes are appended to the list bytes.

        :return: a Thread which is running to do the accept and recv.

        """

        def _receive_bytes_on_server():

            connection, address = sock.accept()

            bytes.append(osutils.recv_all(connection, 3))

            connection.close()

        t = threading.Thread(target=_receive_bytes_on_server)

        t.start()

        return t

    def test_construct_smart_simple_pipes_client_medium(self):

        # the SimplePipes client medium takes two pipes:

        # readable pipe, writeable pipe.

        # Constructing one should just save these and do nothing.

        # We test this by passing in None.

        client_medium = medium.SmartSimplePipesClientMedium(None, None, None)

    def test_simple_pipes_client_request_type(self):

        # SimplePipesClient should use SmartClientStreamMediumRequest's.

        client_medium = medium.SmartSimplePipesClientMedium(None, None, None)

        request = client_medium.get_request()

        self.assertIsInstance(request, medium.SmartClientStreamMediumRequest)

    def test_simple_pipes_client_get_concurrent_requests(self):

        # the simple_pipes client does not support pipelined requests:

        # but it does support serial requests: we construct one after

        # another is finished. This is a smoke test testing the integration

        # of the SmartClientStreamMediumRequest and the SmartClientStreamMedium

        # classes - as the sibling classes share this logic, they do not have

        # explicit tests for this.

        output = BytesIO()

        client_medium = medium.SmartSimplePipesClientMedium(

            None, output, 'base')

        request = client_medium.get_request()

        request.finished_writing()

        request.finished_reading()

        request2 = client_medium.get_request()

        request2.finished_writing()

        request2.finished_reading()

    def test_simple_pipes_client__accept_bytes_writes_to_writable(self):

        # accept_bytes writes to the writeable pipe.

        output = BytesIO()

        client_medium = medium.SmartSimplePipesClientMedium(

            None, output, 'base')

        client_medium._accept_bytes(b'abc')

        self.assertEqual(b'abc', output.getvalue())

    def test_simple_pipes__accept_bytes_subprocess_closed(self):

        # It is unfortunate that we have to use Popen for this. However,

        # os.pipe() does not behave the same as subprocess.Popen().

        # On Windows, if you use os.pipe() and close the write side,

        # read.read() hangs. On Linux, read.read() returns the empty string.

        p = subprocess.Popen([sys.executable, '-c',

                              'import sys\n'

                              'sys.stdout.write(sys.stdin.read(4))\n'

                              'sys.stdout.close()\n'],

                             stdout=subprocess.PIPE, stdin=subprocess.PIPE, bufsize=0)

        client_medium = medium.SmartSimplePipesClientMedium(

            p.stdout, p.stdin, 'base')

        client_medium._accept_bytes(b'abc\n')

        self.assertEqual(b'abc', client_medium._read_bytes(3))

        p.wait()

        # While writing to the underlying pipe,

        #   Windows py2.6.6 we get IOError(EINVAL)

        #   Lucid py2.6.5, we get IOError(EPIPE)

        # In both cases, it should be wrapped to ConnectionReset

        self.assertRaises(errors.ConnectionReset,

                          client_medium._accept_bytes, b'more')

    def test_simple_pipes__accept_bytes_pipe_closed(self):

        child_read, client_write = create_file_pipes()

        client_medium = medium.SmartSimplePipesClientMedium(

            None, client_write, 'base')

        client_medium._accept_bytes(b'abc\n')

        self.assertEqual(b'abc\n', child_read.read(4))

        # While writing to the underlying pipe,

        #   Windows py2.6.6 we get IOError(EINVAL)

        #   Lucid py2.6.5, we get IOError(EPIPE)

        # In both cases, it should be wrapped to ConnectionReset

        child_read.close()

        self.assertRaises(errors.ConnectionReset,

                          client_medium._accept_bytes, b'more')

    def test_simple_pipes__flush_pipe_closed(self):

        child_read, client_write = create_file_pipes()

        client_medium = medium.SmartSimplePipesClientMedium(

            None, client_write, 'base')

        client_medium._accept_bytes(b'abc\n')

        child_read.close()

        # Even though the pipe is closed, flush on the write side seems to be a

        # no-op, rather than a failure.

        client_medium._flush()

    def test_simple_pipes__flush_subprocess_closed(self):

        p = subprocess.Popen([sys.executable, '-c',

                              'import sys\n'

                              'sys.stdout.write(sys.stdin.read(4))\n'

                              'sys.stdout.close()\n'],

                             stdout=subprocess.PIPE, stdin=subprocess.PIPE, bufsize=0)

        client_medium = medium.SmartSimplePipesClientMedium(

            p.stdout, p.stdin, 'base')

        client_medium._accept_bytes(b'abc\n')

        p.wait()

        # Even though the child process is dead, flush seems to be a no-op.

        client_medium._flush()

    def test_simple_pipes__read_bytes_pipe_closed(self):

        child_read, client_write = create_file_pipes()

        client_medium = medium.SmartSimplePipesClientMedium(

            child_read, client_write, 'base')

        client_medium._accept_bytes(b'abc\n')

        client_write.close()

        self.assertEqual(b'abc\n', client_medium._read_bytes(4))

        self.assertEqual(b'', client_medium._read_bytes(4))

    def test_simple_pipes__read_bytes_subprocess_closed(self):

        p = subprocess.Popen([sys.executable, '-c',

                              'import sys\n'

                              'if sys.platform == "win32":\n'

                              '    import msvcrt, os\n'

                              '    msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)\n'

                              '    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)\n'

                              'sys.stdout.write(sys.stdin.read(4))\n'

                              'sys.stdout.close()\n'],

                             stdout=subprocess.PIPE, stdin=subprocess.PIPE, bufsize=0)

        client_medium = medium.SmartSimplePipesClientMedium(

            p.stdout, p.stdin, 'base')

        client_medium._accept_bytes(b'abc\n')

        p.wait()

        self.assertEqual(b'abc\n', client_medium._read_bytes(4))

        self.assertEqual(b'', client_medium._read_bytes(4))

    def test_simple_pipes_client_disconnect_does_nothing(self):

        # calling disconnect does nothing.

        input = BytesIO()

        output = BytesIO()

        client_medium = medium.SmartSimplePipesClientMedium(

            input, output, 'base')

        # send some bytes to ensure disconnecting after activity still does not

        # close.

        client_medium._accept_bytes(b'abc')

        client_medium.disconnect()

        self.assertFalse(input.closed)

        self.assertFalse(output.closed)

    def test_simple_pipes_client_accept_bytes_after_disconnect(self):

        # calling disconnect on the client does not alter the pipe that

        # accept_bytes writes to.

        input = BytesIO()

        output = BytesIO()

        client_medium = medium.SmartSimplePipesClientMedium(

            input, output, 'base')

        client_medium._accept_bytes(b'abc')

        client_medium.disconnect()

        client_medium._accept_bytes(b'abc')

        self.assertFalse(input.closed)

        self.assertFalse(output.closed)

        self.assertEqual(b'abcabc', output.getvalue())

    def test_simple_pipes_client_ignores_disconnect_when_not_connected(self):

        # Doing a disconnect on a new (and thus unconnected) SimplePipes medium

        # does nothing.

        client_medium = medium.SmartSimplePipesClientMedium(None, None, 'base')

        client_medium.disconnect()

    def test_simple_pipes_client_can_always_read(self):

        # SmartSimplePipesClientMedium is never disconnected, so read_bytes

        # always tries to read from the underlying pipe.

        input = BytesIO(b'abcdef')

        client_medium = medium.SmartSimplePipesClientMedium(

            input, None, 'base')

        self.assertEqual(b'abc', client_medium.read_bytes(3))

        client_medium.disconnect()

        self.assertEqual(b'def', client_medium.read_bytes(3))

    def test_simple_pipes_client_supports__flush(self):

        # invoking _flush on a SimplePipesClient should flush the output

        # pipe. We test this by creating an output pipe that records

        # flush calls made to it.

        from io import BytesIO  # get regular BytesIO

        input = BytesIO()

        output = BytesIO()

        flush_calls = []

        def logging_flush(): flush_calls.append('flush')

        output.flush = logging_flush

        client_medium = medium.SmartSimplePipesClientMedium(

            input, output, 'base')

        # this call is here to ensure we only flush once, not on every

        # _accept_bytes call.

        client_medium._accept_bytes(b'abc')

        client_medium._flush()

        client_medium.disconnect()

        self.assertEqual(['flush'], flush_calls)

    def test_construct_smart_ssh_client_medium(self):

        # the SSH client medium takes:

        # host, port, username, password, vendor

        # Constructing one should just save these and do nothing.

        # we test this by creating a empty bound socket and constructing

        # a medium.

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        sock.bind(('127.0.0.1', 0))

        unopened_port = sock.getsockname()[1]

        # having vendor be invalid means that if it tries to connect via the

        # vendor it will blow up.

        ssh_params = medium.SSHParams('127.0.0.1', unopened_port, None, None)

        client_medium = medium.SmartSSHClientMedium(

            'base', ssh_params, "not a vendor")

        sock.close()

    def test_ssh_client_connects_on_first_use(self):

        # The only thing that initiates a connection from the medium is giving

        # it bytes.

        output = BytesIO()

        vendor = BytesIOSSHVendor(BytesIO(), output)

        ssh_params = medium.SSHParams(

            'a hostname', 'a port', 'a username', 'a password', 'bzr')

        client_medium = medium.SmartSSHClientMedium('base', ssh_params, vendor)

        client_medium._accept_bytes(b'abc')

        self.assertEqual(b'abc', output.getvalue())

        self.assertEqual([('connect_ssh', 'a username', 'a password',

                           'a hostname', 'a port',

                           ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes'])],

                         vendor.calls)

    def test_ssh_client_changes_command_when_bzr_remote_path_passed(self):

        # The only thing that initiates a connection from the medium is giving

        # it bytes.

        output = BytesIO()

        vendor = BytesIOSSHVendor(BytesIO(), output)

        ssh_params = medium.SSHParams(

            'a hostname', 'a port', 'a username', 'a password',

            bzr_remote_path='fugly')

        client_medium = medium.SmartSSHClientMedium('base', ssh_params, vendor)

        client_medium._accept_bytes(b'abc')

        self.assertEqual(b'abc', output.getvalue())

        self.assertEqual([('connect_ssh', 'a username', 'a password',

                           'a hostname', 'a port',

                           ['fugly', 'serve', '--inet', '--directory=/', '--allow-writes'])],

                         vendor.calls)

    def test_ssh_client_disconnect_does_so(self):

        # calling disconnect should disconnect both the read_from and write_to

        # file-like object it from the ssh connection.

        input = BytesIO()

        output = BytesIO()

        vendor = BytesIOSSHVendor(input, output)

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams('a hostname'), vendor)

        client_medium._accept_bytes(b'abc')

        client_medium.disconnect()

        self.assertTrue(input.closed)

        self.assertTrue(output.closed)

        self.assertEqual([

            ('connect_ssh', None, None, 'a hostname', None,

             ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),

            ('close', ),

            ],

            vendor.calls)

    def test_ssh_client_disconnect_allows_reconnection(self):

        # calling disconnect on the client terminates the connection, but should

        # not prevent additional connections occuring.

        # we test this by initiating a second connection after doing a

        # disconnect.

        input = BytesIO()

        output = BytesIO()

        vendor = BytesIOSSHVendor(input, output)

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams('a hostname'), vendor)

        client_medium._accept_bytes(b'abc')

        client_medium.disconnect()

        # the disconnect has closed output, so we need a new output for the

        # new connection to write to.

        input2 = BytesIO()

        output2 = BytesIO()

        vendor.read_from = input2

        vendor.write_to = output2

        client_medium._accept_bytes(b'abc')

        client_medium.disconnect()

        self.assertTrue(input.closed)

        self.assertTrue(output.closed)

        self.assertTrue(input2.closed)

        self.assertTrue(output2.closed)

        self.assertEqual([

            ('connect_ssh', None, None, 'a hostname', None,

             ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),

            ('close', ),

            ('connect_ssh', None, None, 'a hostname', None,

             ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),

            ('close', ),

            ],

            vendor.calls)

    def test_ssh_client_repr(self):

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams("example.com", "4242", "username"))

        self.assertEqual(

            "SmartSSHClientMedium(bzr+ssh://username@example.com:4242/)",

            repr(client_medium))

    def test_ssh_client_repr_no_port(self):

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams("example.com", None, "username"))

        self.assertEqual(

            "SmartSSHClientMedium(bzr+ssh://username@example.com/)",

            repr(client_medium))

    def test_ssh_client_repr_no_username(self):

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams("example.com", None, None))

        self.assertEqual(

            "SmartSSHClientMedium(bzr+ssh://example.com/)",

            repr(client_medium))

    def test_ssh_client_ignores_disconnect_when_not_connected(self):

        # Doing a disconnect on a new (and thus unconnected) SSH medium

        # does not fail.  It's ok to disconnect an unconnected medium.

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams(None))

        client_medium.disconnect()

    def test_ssh_client_raises_on_read_when_not_connected(self):

        # Doing a read on a new (and thus unconnected) SSH medium raises

        # MediumNotConnected.

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams(None))

        self.assertRaises(errors.MediumNotConnected, client_medium.read_bytes,

                          0)

        self.assertRaises(errors.MediumNotConnected, client_medium.read_bytes,

                          1)

    def test_ssh_client_supports__flush(self):

        # invoking _flush on a SSHClientMedium should flush the output

        # pipe. We test this by creating an output pipe that records

        # flush calls made to it.

        from io import BytesIO  # get regular BytesIO

        input = BytesIO()

        output = BytesIO()

        flush_calls = []

        def logging_flush(): flush_calls.append('flush')

        output.flush = logging_flush

        vendor = BytesIOSSHVendor(input, output)

        client_medium = medium.SmartSSHClientMedium(

            'base', medium.SSHParams('a hostname'), vendor=vendor)

        # this call is here to ensure we only flush once, not on every

        # _accept_bytes call.

        client_medium._accept_bytes(b'abc')

        client_medium._flush()

        client_medium.disconnect()

        self.assertEqual(['flush'], flush_calls)

    def test_construct_smart_tcp_client_medium(self):

        # the TCP client medium takes a host and a port.  Constructing it won't

        # connect to anything.

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        sock.bind(('127.0.0.1', 0))

        unopened_port = sock.getsockname()[1]

        client_medium = medium.SmartTCPClientMedium(

            '127.0.0.1', unopened_port, 'base')

        sock.close()

    def test_tcp_client_connects_on_first_use(self):

        # The only thing that initiates a connection from the medium is giving

        # it bytes.

        sock, medium = self.make_loopsocket_and_medium()

        bytes = []

        t = self.receive_bytes_on_server(sock, bytes)

        medium.accept_bytes(b'abc')

        t.join()

        sock.close()

        self.assertEqual([b'abc'], bytes)

    def test_tcp_client_disconnect_does_so(self):

        # calling disconnect on the client terminates the connection.

        # we test this by forcing a short read during a socket.MSG_WAITALL

        # call: write 2 bytes, try to read 3, and then the client disconnects.

        sock, medium = self.make_loopsocket_and_medium()

        bytes = []

        t = self.receive_bytes_on_server(sock, bytes)

        medium.accept_bytes(b'ab')

        medium.disconnect()

        t.join()

        sock.close()

        self.assertEqual([b'ab'], bytes)

        # now disconnect again: this should not do anything, if disconnection

        # really did disconnect.

        medium.disconnect()

    def test_tcp_client_ignores_disconnect_when_not_connected(self):

        # Doing a disconnect on a new (and thus unconnected) TCP medium

        # does not fail.  It's ok to disconnect an unconnected medium.

        client_medium = medium.SmartTCPClientMedium(None, None, None)

        client_medium.disconnect()

    def test_tcp_client_raises_on_read_when_not_connected(self):

        # Doing a read on a new (and thus unconnected) TCP medium raises

        # MediumNotConnected.

        client_medium = medium.SmartTCPClientMedium(None, None, None)

        self.assertRaises(errors.MediumNotConnected,

                          client_medium.read_bytes, 0)

        self.assertRaises(errors.MediumNotConnected,

                          client_medium.read_bytes, 1)

    def test_tcp_client_supports__flush(self):

        # invoking _flush on a TCPClientMedium should do something useful.

        # RBC 20060922 not sure how to test/tell in this case.

        sock, medium = self.make_loopsocket_and_medium()

        bytes = []

        t = self.receive_bytes_on_server(sock, bytes)

        # try with nothing buffered

        medium._flush()

        medium._accept_bytes(b'ab')

        # and with something sent.

        medium._flush()

        medium.disconnect()

        t.join()

        sock.close()

        self.assertEqual([b'ab'], bytes)

        # now disconnect again : this should not do anything, if disconnection

        # really did disconnect.

        medium.disconnect()