/brz/remove-bazaar

#

# 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

#

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

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

 

"""Tests for smart transport"""

 

 

    def __init__(self, vendor):

        self.vendor = vendor

    def close(self):

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

    def get_filelike_channels(self):

        return self.vendor.read_from, self.vendor.write_to

 

 

class _InvalidHostnameFeature(tests.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

        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)

 

    def test_simple_pipes_client_get_concurrent_requests(self):

        # the simple_pipes client does not support pipelined requests:

        # 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

            None, output, 'base')

        client_medium._accept_bytes('abc')

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

    def test_simple_pipes_client_disconnect_does_nothing(self):

        # calling disconnect does nothing.

        input = StringIO()

        self.assertFalse(input.closed)

        self.assertFalse(output.closed)

        self.assertEqual('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.

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

        client_medium.disconnect()

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

    def test_simple_pipes_client_supports__flush(self):

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

        # flush calls made to it.

        from StringIO import StringIO # get regular StringIO

            '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.

            ('close', ),

            ],

            vendor.calls)

    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.

                          1)

 

    def test_ssh_client_supports__flush(self):

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

        # flush calls made to it.

        from StringIO import StringIO # get regular StringIO

        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.

        t.join()

        sock.close()

        self.assertEqual(['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

        # 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.

 

class TestSmartClientStreamMediumRequest(tests.TestCase):

    """Tests the for SmartClientStreamMediumRequest.

    mediums: TCP, SSH, SimplePipes, so we only test it once, and then test that

    those three mediums implement the interface it expects.

    """

 

    def test_accept_bytes_after_finished_writing_errors(self):

        # WritingCompleted to prevent bad assumptions on stream environments

        # breaking the needs of message-based environments.

        output = StringIO()

            None, None, 'base')

        request = medium.SmartClientStreamMediumRequest(client_medium)

        self.assertRaises(errors.WritingNotComplete, request.finished_reading)

    def test_read_bytes(self):

        # read bytes should invoke _read_bytes on the stream medium.

        # we test this by using the SimplePipes medium - the most trivial one

        # faulty implementation could poke at the pipe variables them selves,

        # but we trust that this will be caught as it will break the integration

        # smoke tests.

        self.assertRaises(errors.WritingNotComplete, request.read_bytes, None)

 

    def test_read_bytes_after_finished_reading_errors(self):

        # ReadingCompleted to prevent bad assumptions on stream environments

        # breaking the needs of message-based environments.

        output = StringIO()

 

 

class SampleRequest(object):

    def __init__(self, expected_bytes):

        self.accepted_bytes = ''

        self._finished_reading = False

 

    def portable_socket_pair(self):

        """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)

        server_sock, addr = listen_sock.accept()

        listen_sock.close()

        return server_sock, client_sock

    def test_smart_query_version(self):

        """Feed a canned query version to a server"""

        # wire-to-wire, using the whole stack

        # wire-to-wire, using the whole stack, with a UTF-8 filename.

        transport = memory.MemoryTransport('memory:///')

        utf8_filename = u'testfile\N{INTERROBANG}'.encode('utf-8')

        transport.put_bytes(utf8_filename, 'contents\nof\nfile\n')

        from_server = StringIO()

        server = medium.SmartServerPipeStreamMedium(

            to_server, from_server, transport)

        server = medium.SmartServerPipeStreamMedium(to_server, from_server, None)

        server._serve_one_request(SampleRequest('x'))

        self.assertTrue(server.finished)

    def test_socket_stream_shutdown_detection(self):

        server_sock, client_sock = self.portable_socket_pair()

        client_sock.close()

            server_sock, None)

        server._serve_one_request(SampleRequest('x'))

        self.assertTrue(server.finished)

    def test_socket_stream_incomplete_request(self):

        """The medium should still construct the right protocol version even if

        the initial read only reads part of the request.

        client_sock.sendall(rest_of_request_bytes)

        server._serve_one_request(server_protocol)

        server_sock.close()

                         "Not a version 2 response to 'hello' request.")

        self.assertEqual('', client_sock.recv(1))

 

    def test_pipe_like_stream_with_two_requests(self):

        # If two requests are read in one go, then two calls to

        # _serve_one_request should still process both of them as if they had

        sample_request_bytes = 'command\n'

        to_server = StringIO(sample_request_bytes * 2)

        from_server = StringIO()

        self.assertEqual('', from_server.getvalue())

        self.assertEqual(sample_request_bytes, second_protocol.accepted_bytes)

        self.assertFalse(server.finished)

    def test_socket_stream_with_two_requests(self):

        # If two requests are read in one go, then two calls to

        # _serve_one_request should still process both of them as if they had

        sample_request_bytes = 'command\n'

        server_sock, client_sock = self.portable_socket_pair()

        server = medium.SmartServerSocketStreamMedium(

        self.assertEqual('', from_server.getvalue())

        self.assertTrue(self.closed)

        self.assertTrue(server.finished)

    def test_socket_stream_error_handling(self):

        server_sock, client_sock = self.portable_socket_pair()

        server = medium.SmartServerSocketStreamMedium(

        # closed.

        self.assertEqual('', client_sock.recv(1))

        self.assertTrue(server.finished)

    def test_pipe_like_stream_keyboard_interrupt_handling(self):

        to_server = StringIO('')

        from_server = StringIO()

        # Any empty request (i.e. no bytes) is detected as protocol version one.

        server_protocol = self.build_protocol_pipe_like('')

        self.assertProtocolOne(server_protocol)

    def test_socket_like_build_protocol_empty_bytes(self):

        # Any empty request (i.e. no bytes) is detected as protocol version one.

        server_protocol = self.build_protocol_socket('')

        self.assertEqual(

            protocol.build_server_protocol_three, protocol_factory)

        self.assertEqual('extra bytes', remainder)

    def test_version_two(self):

        result = medium._get_protocol_factory_for_bytes(

            'bzr request 2\nextra bytes')

        self.assertEqual(

            protocol.SmartServerRequestProtocolTwo, protocol_factory)

        self.assertEqual('extra bytes', remainder)

    def test_version_one(self):

        """Version one requests have no version markers."""

        result = medium._get_protocol_factory_for_bytes('anything\n')

        self.assertEqual(

            protocol.SmartServerRequestProtocolOne, protocol_factory)

        self.assertEqual('anything\n', remainder)

 

class TestSmartTCPServer(tests.TestCase):

 

    All of these tests are run with a server running on another thread serving

    a MemoryTransport, and a connection to it already open.

 

    """

 

        """Setup the server.

 

        :param readonly: Create a readonly server.

        """

        if not backing_transport:

        else:

            self.backing_transport = backing_transport

        if readonly:

        self.server.start_background_thread('-' + self.id())

        self.transport = remote.RemoteTCPTransport(self.server.get_url())

        self.addCleanup(self.tearDownServer)

 

    def tearDownServer(self):

        if getattr(self, 'transport', None):

            del self.transport

        if getattr(self, 'server', None):

            self.server.stop_background_thread()

            del self.server

 

 

 

    def test_server_setup_teardown(self):

        """It should be safe to teardown the server with no requests."""

        server = self.server

        transport = remote.RemoteTCPTransport(self.server.get_url())

        self.tearDownServer()

 

    def test_server_closes_listening_sock_on_shutdown_after_request(self):

        """The server should close its listening socket when it's stopped."""

        server = self.server

        self.transport.has('.')

        self.tearDownServer()

 

    def setUp(self):

        super(WritableEndToEndTests, self).setUp()

 

    def test_start_tcp_server(self):

        url = self.server.get_url()

        self._captureVar('BZR_NO_SMART_VFS', None)

        err = self.assertRaises(

            errors.NoSuchFile, self.transport.get, 'not%20a%20file')

 

    def test_simple_clone_conn(self):

        """Test that cloning reuses the same connection."""

    def test_mkdir_error_readonly(self):

        """TransportNotPossible should be preserved from the backing transport."""

        self._captureVar('BZR_NO_SMART_VFS', None)

        self.assertRaises(errors.TransportNotPossible, self.transport.mkdir,

            'foo')

 

        self.hook_calls = []

        server.SmartTCPServer.hooks.install_named_hook('server_started',

            self.capture_server_call, None)

        # at this point, the server will be starting a thread up.

        # there is no indicator at the moment, so bodge it by doing a request.

        self.transport.has('.')

        self.hook_calls = []

        server.SmartTCPServer.hooks.install_named_hook('server_started',

            self.capture_server_call, None)

        # at this point, the server will be starting a thread up.

        # there is no indicator at the moment, so bodge it by doing a request.

        self.transport.has('.')

        self.hook_calls = []

        server.SmartTCPServer.hooks.install_named_hook('server_stopped',

            self.capture_server_call, None)

        result = [([self.backing_transport.base], self.transport.base)]

        # check the stopping message isn't emitted up front.

        self.assertEqual([], self.hook_calls)

        self.hook_calls = []

        server.SmartTCPServer.hooks.install_named_hook('server_stopped',

            self.capture_server_call, None)

        result = [(

            [self.backing_transport.base, self.backing_transport.external_url()]

            , self.transport.base)]

    Note: these tests are rudimentary versions of the command object tests in

    test_smart.py.

    """

    def test_hello(self):

        cmd = _mod_request.HelloRequest(None, '/')

        response = cmd.execute()

        self.assertEqual(('ok', '2'), response.args)

        self.assertEqual(None, response.body)

    def test_get_bundle(self):

        from bzrlib.bundle import serializer

        wt = self.make_branch_and_tree('.')

        self.build_tree_contents([('hello', 'hello world')])

        wt.add('hello')

        rev_id = wt.commit('add hello')

        cmd = _mod_request.GetBundleRequest(self.get_transport(), '/')

        response = cmd.execute('.', rev_id)

        bundle = serializer.read_bundle(StringIO(response.body))

 

    def test_hello(self):

        handler = self.build_handler(None)

        self.assertEqual(('ok', '2'), handler.response.args)

        self.assertEqual(None, handler.response.body)

    def test_disable_vfs_handler_classes_via_environment(self):

        # VFS handler classes will raise an error from "execute" if

        # BZR_NO_SMART_VFS is set.

        """The response for a read-only error is ('ReadOnlyError')."""

        handler = self.build_handler(self.get_readonly_transport())

        # send a mkdir for foo, with no explicit mode - should fail.

        # and the failure should be an explicit ReadOnlyError

        self.assertEqual(("ReadOnlyError", ), handler.response.args)

        # XXX: TODO: test that other TransportNotPossible errors are

    def test_hello_has_finished_body_on_dispatch(self):

        """The 'hello' command should set finished_reading."""

        handler = self.build_handler(None)

        self.assertTrue(handler.finished_reading)

        self.assertNotEqual(None, handler.response)

 

    def test_put_bytes_non_atomic(self):

        """'put_...' should set finished_reading after reading the bytes."""

        handler = self.build_handler(self.get_transport())

        self.assertFalse(handler.finished_reading)

        handler.accept_body('1234')

        self.assertFalse(handler.finished_reading)

        self.assertTrue(handler.finished_reading)

        self.assertEqual(('ok', ), handler.response.args)

        self.assertEqual(None, handler.response.body)

    def test_readv_accept_body(self):

        """'readv' should set finished_reading after reading offsets."""

        self.build_tree(['a-file'])

        handler = self.build_handler(self.get_readonly_transport())

        self.assertFalse(handler.finished_reading)

        handler.accept_body('2,')

        self.assertFalse(handler.finished_reading)

        """'readv' when a short read occurs sets the response appropriately."""

        self.build_tree(['a-file'])

        handler = self.build_handler(self.get_readonly_transport())

        # read beyond the end of the file.

        handler.accept_body('100,1')

        handler.end_of_body()

 

 

class TestRemoteTransport(tests.TestCase):

    def test_use_connection_factory(self):

        # We want to be able to pass a client as a parameter to RemoteTransport.

        input = StringIO('ok\n3\nbardone\n')

    def assertOffsetSerialisation(self, expected_offsets, expected_serialised,

        requester):

        """Check that smart (de)serialises offsets as expected.

        We check both serialisation and deserialisation at the same time

        to ensure that the round tripping cannot skew: both directions should

        be as expected.

        :param expected_offsets: a readv offset list.

        :param expected_seralised: an expected serial form of the offsets.

        """

        ex = self.assertRaises(errors.ConnectionReset,

            response_handler.read_response_tuple)

        self.assertEqual("Connection closed: "

 

    def test_server_offset_serialisation(self):

        """The Smart protocol serialises offsets as a comma and \n string.

 

    def test_query_version(self):

        """query_version on a SmartClientProtocolOne should return a number.

        The protocol provides the query_version because the domain level clients

        may all need to be able to probe for capabilities.

        """

        # What we really want to test here is that SmartClientProtocolOne calls

        # accept_bytes(tuple_based_encoding_of_hello) and reads and parses the

        # the error if the response is a non-understood version.

        input = StringIO('ok\x012\n')

        output = StringIO()

 

    def test_query_version(self):

        """query_version on a SmartClientProtocolTwo should return a number.

        The protocol provides the query_version because the domain level clients

        may all need to be able to probe for capabilities.

        """

        # What we really want to test here is that SmartClientProtocolTwo calls

        # accept_bytes(tuple_based_encoding_of_hello) and reads and parses the

        # the error if the response is a non-understood version.

        input = StringIO(self.response_marker + 'success\nok\x012\n')

        output = StringIO()

        self.assertEqual(4, smart_protocol.next_read_size())

 

 

class LoggingMessageHandler(object):

 

    def __init__(self):

        self.assertEqual(0, smart_protocol.next_read_size())

        self.assertEqual('', smart_protocol.unused_data)

 

    def make_protocol_expecting_message_part(self):

        headers = '\0\0\0\x02de'  # length-prefixed, bencoded empty dict

        message_handler = LoggingMessageHandler()

        return response_handler

 

    def test_interrupted_by_error(self):

        response_handler = self.make_response_handler(interrupted_body_stream)

        stream = response_handler.read_streamed_body()

        exc = self.assertRaises(errors.ErrorFromSmartServer, stream.next)

 

    def test_interrupted_by_connection_lost(self):

        interrupted_body_stream = (

        self.calls.append(('end_received',))

        self.finished_reading = True

 

 

    def accept_body(self, bytes):

        self.calls.append(('accept_body', bytes))

    def test_call_with_body_stream_error(self):

        """call_with_body_stream will abort the streamed body with an

        error if the stream raises an error during iteration.

        The resulting request will still be a complete message.

        """

        requester, output = self.make_client_encoder_and_output()

            yield 'aaa'

            yield 'bbb'

            raise Exception('Boom!')

        self.assertEquals(

            'bzr message 3 (bzr 1.6)\n' # protocol version

            '\x00\x00\x00\x02de' # headers

        self.calls.append('finished_writing')

 

 

class TestResponseEncodingProtocolThree(tests.TestCase):

 

    def make_response_encoder(self):

            # end of message

            'e')

 

 

class TestResponseEncoderBufferingProtocolThree(tests.TestCase):

    """Tests for buffering of responses.

    """

 

    def setUp(self):

        self.writes = []

        self.responder = protocol.ProtocolThreeResponder(self.writes.append)

 

        self.assertEqual(

            expected_count, len(self.writes),

            "Too many writes: %r" % (self.writes,))

    def test_send_error_writes_just_once(self):

        """An error response is written to the medium all at once."""

        self.responder.send_error(Exception('An exception string.'))

        self.responder.send_response(response)

        self.assertWriteCount(1)

 

        # Construct a response with stream with 2 chunks in it.

        response = _mod_request.SuccessfulSmartServerResponse(

            ('arg', 'arg'), body_stream=['chunk1', 'chunk2'])

        self.responder.send_response(response)

 

 

class TestSmartClientUnicode(tests.TestCase):

 

class MockMedium(medium.SmartClientMedium):

    """A mock medium that can be used to test _SmartClient.

    It can be given a series of requests to expect (and responses it should

    return for them).  It can also be told when the client is expected to

    disconnect a medium.  Expectations must be satisfied in the order they are

        super(MockMedium, self).__init__('dummy base')

        self._mock_request = _MockMediumRequest(self)

        self._expected_events = []

    def expect_request(self, request_bytes, response_bytes,

                       allow_partial_read=False):

        """Expect 'request_bytes' to be sent, and reply with 'response_bytes'.

        called to send the request.  Similarly, no assumption is made about how

        many times read_bytes/read_line are called by protocol code to read a

        response.  e.g.::

            request.accept_bytes('ab')

            request.accept_bytes('cd')

            request.finished_writing()

 

        and::

            request.accept_bytes('abcd')

            request.finished_writing()

 

    def test_first_response_is_error(self):

        """If the server replies with an error, then the version detection

        should be complete.

        This test is very similar to test_version_two_server, but catches a bug

        we had in the case where the first reply was an error response.

        """

 

class LengthPrefixedBodyDecoder(tests.TestCase):

 

    # something similar to the ProtocolBase method.

 

    def test_construct(self):

        self.assertEqual(1, decoder.next_read_size())

        self.assertEqual('', decoder.read_pending_data())

        self.assertEqual('blarg', decoder.unused_data)

    def test_accept_bytes_all_at_once_with_excess(self):

        decoder = protocol.LengthPrefixedBodyDecoder()

        decoder.accept_bytes('1\nadone\nunused')

 

class TestChunkedBodyDecoder(tests.TestCase):

    """Tests for ChunkedBodyDecoder.

    This is the body decoder used for protocol version two.

    """

 

        self.assertTrue(decoder.finished_reading)

        self.assertEqual(chunk_content, decoder.read_next_chunk())