← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-11-20 22:44:10 UTC
  • mfrom: (2018.2.37 smart server unicode)
  • Revision ID: pqm@pqm.ubuntu.com-20061120224410-8a8116b29a11c1ff
(Andrew Bennetts) Fix some bzr smart server tests, support sending/receiving byte strings for the smart server

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
596
596
        self.assertEqual('ok\0011\n',
597
597
                         from_server.getvalue())
598
598
 
599
 
    def test_canned_get_response(self):
 
599
    def test_response_to_canned_get(self):
600
600
        transport = memory.MemoryTransport('memory:///')
601
601
        transport.put_bytes('testfile', 'contents\nof\nfile\n')
602
602
        to_server = StringIO('get\001./testfile\n')
612
612
                         'done\n',
613
613
                         from_server.getvalue())
614
614
 
 
615
    def test_response_to_canned_get_of_utf8(self):
 
616
        # wire-to-wire, using the whole stack, with a UTF-8 filename.
 
617
        transport = memory.MemoryTransport('memory:///')
 
618
        utf8_filename = u'testfile\N{INTERROBANG}'.encode('utf-8')
 
619
        transport.put_bytes(utf8_filename, 'contents\nof\nfile\n')
 
620
        to_server = StringIO('get\001' + utf8_filename + '\n')
 
621
        from_server = StringIO()
 
622
        server = smart.SmartServerPipeStreamMedium(
 
623
            to_server, from_server, transport)
 
624
        protocol = smart.SmartServerRequestProtocolOne(transport,
 
625
                from_server.write)
 
626
        server._serve_one_request(protocol)
 
627
        self.assertEqual('ok\n'
 
628
                         '17\n'
 
629
                         'contents\nof\nfile\n'
 
630
                         'done\n',
 
631
                         from_server.getvalue())
 
632
 
615
633
    def test_pipe_like_stream_with_bulk_data(self):
616
634
        sample_request_bytes = 'command\n9\nbulk datadone\n'
617
635
        to_server = StringIO(sample_request_bytes)
1110
1128
        self.assertOffsetSerialisation([(1,2), (3,4), (100, 200)],
1111
1129
            '1,2\n3,4\n100,200', self.client_protocol, self.smart_server_request)
1112
1130
 
1113
 
    def test_accept_bytes_to_protocol(self):
 
1131
    def test_accept_bytes_of_bad_request_to_protocol(self):
1114
1132
        out_stream = StringIO()
1115
1133
        protocol = smart.SmartServerRequestProtocolOne(None, out_stream.write)
1116
1134
        protocol.accept_bytes('abc')
1117
1135
        self.assertEqual('abc', protocol.in_buffer)
1118
1136
        protocol.accept_bytes('\n')
1119
1137
        self.assertEqual("error\x01Generic bzr smart protocol error: bad request"
1120
 
            " u'abc'\n", out_stream.getvalue())
1121
 
        self.assertTrue(protocol.has_dispatched)
1122
 
        self.assertEqual(1, protocol.next_read_size())
1123
 
 
1124
 
    def test_accept_bytes_with_invalid_utf8_to_protocol(self):
1125
 
        out_stream = StringIO()
1126
 
        protocol = smart.SmartServerRequestProtocolOne(None, out_stream.write)
1127
 
        # the byte 0xdd is not a valid UTF-8 string.
1128
 
        protocol.accept_bytes('\xdd\n')
1129
 
        self.assertEqual(
1130
 
            "error\x01Generic bzr smart protocol error: "
1131
 
            "one or more arguments of request '\\xdd\\n' are not valid UTF-8\n",
1132
 
            out_stream.getvalue())
1133
 
        self.assertTrue(protocol.has_dispatched)
1134
 
        self.assertEqual(1, protocol.next_read_size())
 
1138
            " 'abc'\n", out_stream.getvalue())
 
1139
        self.assertTrue(protocol.has_dispatched)
 
1140
        self.assertEqual(0, protocol.next_read_size())
1135
1141
 
1136
1142
    def test_accept_body_bytes_to_protocol(self):
1137
1143
        protocol = self.build_protocol_waiting_for_body()
1184
1190
        self.assertEqual("hello\n", protocol.excess_buffer)
1185
1191
        self.assertEqual("", protocol.in_buffer)
1186
1192
 
1187
 
    def test_sync_with_request_sets_finished_reading(self):
1188
 
        protocol = smart.SmartServerRequestProtocolOne(None, None)
1189
 
        request = smart.SmartServerRequestHandler(None)
1190
 
        protocol.sync_with_request(request)
 
1193
    def test__send_response_sets_finished_reading(self):
 
1194
        protocol = smart.SmartServerRequestProtocolOne(None, lambda x: None)
1191
1195
        self.assertEqual(1, protocol.next_read_size())
1192
 
        request.finished_reading = True
1193
 
        protocol.sync_with_request(request)
 
1196
        protocol._send_response(('x',))
1194
1197
        self.assertEqual(0, protocol.next_read_size())
1195
1198
 
1196
1199
    def test_query_version(self):