/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: Andrew Bennetts
  • Date: 2008-02-12 01:56:09 UTC
  • mto: (3245.4.1 Version 3 implementation.)
  • mto: This revision was merged to the branch mainline in revision 3428.
  • Revision ID: andrew.bennetts@canonical.com-20080212015609-loaxv5te0eb6uot8
Make the general request handler dispatch version 3 events to the specific request handler (i.e. to the SmartServerRequest instance).

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
2053
2053
        client_protocol = protocol.SmartClientRequestProtocolThree(request)
2054
2054
 
2055
2055
 
 
2056
class NoOpRequest(request.SmartServerRequest):
 
2057
 
 
2058
    def do(self):
 
2059
        return request.SuccessfulSmartServerResponse(())
 
2060
 
 
2061
dummy_registry = {'ARG': NoOpRequest}
 
2062
 
 
2063
 
2056
2064
class TestServerProtocolThree(TestSmartProtocol):
2057
2065
    """Tests for v3 of the server-side protocol."""
2058
2066
 
2064
2072
        argument, no body.
2065
2073
        """
2066
2074
        output = StringIO()
2067
 
        protocol_version = "bzr request 3\n"
2068
2075
        headers = '\0\0\0\x02de'  # length-prefixed, bencoded empty dict
2069
2076
        args = '\0\0\0\x07l3:ARGe' # length-prefixed, bencoded list: ['ARG']
2070
2077
        body = 'n'
2071
2078
        request_bytes = headers + args + body
2072
 
        smart_protocol = self.server_protocol_class(None, output.write)
 
2079
        smart_protocol = self.server_protocol_class(None, output.write,
 
2080
            dummy_registry)
2073
2081
        smart_protocol.accept_bytes(request_bytes)
2074
2082
        self.assertEqual(0, smart_protocol.next_read_size())
2075
2083
        self.assertEqual('', smart_protocol.excess_buffer)
2081
2089
        That is, return a protocol object that is waiting to receive a body.
2082
2090
        """
2083
2091
        output = StringIO()
2084
 
        protocol_version = "bzr request 3\n"
2085
2092
        headers = '\0\0\0\x02de'  # length-prefixed, bencoded empty dict
2086
2093
        args = '\0\0\0\x07l3:ARGe' # length-prefixed, bencoded list: ['ARG']
2087
2094
        request_bytes = headers + args
2088
 
        smart_protocol = self.server_protocol_class(None, output.write)
 
2095
        smart_protocol = self.server_protocol_class(None, output.write,
 
2096
            dummy_registry)
2089
2097
        smart_protocol.accept_bytes(request_bytes)
2090
2098
        return smart_protocol
2091
2099
 
2092
2100
    def assertBodyParsingBehaviour(self, calls, protocol_bytes):
2093
2101
        """Assert that the given bytes cause an exact sequence of calls to the
2094
 
        request handler.
 
2102
        request handler, followed by an end_received call.
2095
2103
        """
 
2104
        calls = calls + [('end_received',)]
2096
2105
        smart_protocol = self.make_protocol_expecting_body()
2097
2106
        smart_protocol.request_handler = InstrumentedRequestHandler()
2098
2107
        smart_protocol.accept_bytes(protocol_bytes)
2128
2137
            's' # body kind
2129
2138
            't' # stream terminator
2130
2139
            )
2131
 
        # XXX: No calls to the request handler in this case?  That's slightly
2132
 
        # odd.
2133
2140
        self.assertBodyParsingBehaviour([], body)
2134
2141
 
2135
2142
    def test_request_chunked_body_one_chunks(self):
2183
2190
    def prefixed_body_received(self, body_bytes):
2184
2191
        self.calls.append(('prefixed_body_received', body_bytes))
2185
2192
 
 
2193
    def end_received(self):
 
2194
        self.calls.append(('end_received',))
 
2195
 
2186
2196
 
2187
2197
 
2188
2198
#class TestProtocolTestCoverage(tests.TestCase):