← 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

Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
126
126
 
127
127
 
128
128
class SmartTCPTests(tests.TestCase):
129
 
    """Tests for connection to TCP server.
130
 
    
 
129
    """Tests for connection/end to end behaviour using the TCP server.
 
130
 
131
131
    All of these tests are run with a server running on another thread serving
132
132
    a MemoryTransport, and a connection to it already open.
 
133
 
 
134
    the server is obtained by calling self.setUpServer(readonly=False).
133
135
    """
134
136
 
135
 
    def setUp(self):
136
 
        super(SmartTCPTests, self).setUp()
 
137
    def setUpServer(self, readonly=False):
 
138
        """Setup the server.
 
139
 
 
140
        :param readonly: Create a readonly server.
 
141
        """
137
142
        self.backing_transport = memory.MemoryTransport()
 
143
        if readonly:
 
144
            self.real_backing_transport = self.backing_transport
 
145
            self.backing_transport = get_transport("readonly+" + self.backing_transport.abspath('.'))
138
146
        self.server = smart.SmartTCPServer(self.backing_transport)
139
147
        self.server.start_background_thread()
140
148
        self.transport = smart.SmartTCPTransport(self.server.get_url())
146
154
            self.server.stop_background_thread()
147
155
        super(SmartTCPTests, self).tearDown()
148
156
        
 
157
 
 
158
class WritableEndToEndTests(SmartTCPTests):
 
159
    """Client to server tests that require a writable transport."""
 
160
 
 
161
    def setUp(self):
 
162
        super(WritableEndToEndTests, self).setUp()
 
163
        self.setUpServer()
 
164
 
149
165
    def test_start_tcp_server(self):
150
166
        url = self.server.get_url()
151
167
        self.assertContainsRe(url, r'^bzr://127\.0\.0\.1:[0-9]{2,}/')
161
177
        self.backing_transport.put_bytes("foo", "contents\nof\nfoo\n")
162
178
        fp = self.transport.get("foo")
163
179
        self.assertEqual('contents\nof\nfoo\n', fp.read())
164
 
        
 
180
 
165
181
    def test_get_error_enoent(self):
166
182
        """Error reported from server getting nonexistent file."""
167
183
        # The path in a raised NoSuchFile exception should be the precise path
215
231
        result_dir = bzrdir.BzrDir.open_containing_from_transport(transport)
216
232
 
217
233
 
 
234
class ReadOnlyEndToEndTests(SmartTCPTests):
 
235
    """Tests from the client to the server using a readonly backing transport."""
 
236
 
 
237
    def test_mkdir_error_readonly(self):
 
238
        """TransportNotPossible should be preserved from the backing transport."""
 
239
        self.setUpServer(readonly=True)
 
240
        self.assertRaises(errors.TransportNotPossible, self.transport.mkdir,
 
241
            'foo')
 
242
        
 
243
 
218
244
class SmartServerTests(tests.TestCaseWithTransport):
219
245
    """Test that call directly into the server logic, bypassing the network."""
220
246
 
235
261
        response = server.dispatch_command('get_bundle', ('.', rev_id))
236
262
        bundle = serializer.read_bundle(StringIO(response.body))
237
263
 
 
264
    def test_readonly_exception_becomes_transport_not_possible(self):
 
265
        """The response for a read-only error is ('ReadOnlyError')."""
 
266
        server = smart.SmartServer(self.get_readonly_transport())
 
267
        # send a mkdir for foo, with no explicit mode - should fail.
 
268
        response = server.dispatch_command('mkdir', ('foo', ''))
 
269
        # and the failure should be an explicit ReadOnlyError
 
270
        self.assertEqual(("ReadOnlyError", ), response.args)
 
271
        # XXX: TODO: test that other TransportNotPossible errors are
 
272
        # presented as TransportNotPossible - not possible to do that
 
273
        # until I figure out how to trigger that relatively cleanly via
 
274
        # the api. RBC 20060918
 
275
 
238
276
 
239
277
class SmartTransportRegistration(tests.TestCase):
240
278
 
275
313
        # The only call to _call should have been to get /foo.
276
314
        self.assertEqual([('_call', ('get', '/foo'))], client._calls)
277
315
 
 
316
    def test__translate_error_readonly(self):
 
317
        """Sending a ReadOnlyError to _translate_error raises TransportNotPossible."""
 
318
        client = FakeClient()
 
319
        transport = smart.SmartTransport('bzr://localhost/', client=client)
 
320
        self.assertRaises(errors.TransportNotPossible,
 
321
            transport._translate_error, ("ReadOnlyError", ))
 
322
 
278
323
 
279
324
class InstrumentedClient(smart.SmartStreamClient):
280
325
    """A smart client whose writes are stored to a supplied list."""