← 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: Vincent Ladeuil
  • Date: 2010-06-17 09:23:19 UTC
  • mfrom: (5301 +trunk)
  • mto: (5247.1.8 first-try)
  • mto: This revision was merged to the branch mainline in revision 5326.
  • Revision ID: v.ladeuil+lp@free.fr-20100617092319-da2bzdtf3j0voynf
Merge bzr.dev into cleanup

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
46
46
        local,
47
47
        memory,
48
48
        remote,
 
49
        ssh,
49
50
        )
50
51
 
51
52
 
63
64
        return StringIOSSHConnection(self)
64
65
 
65
66
 
66
 
class StringIOSSHConnection(object):
 
67
class StringIOSSHConnection(ssh.SSHConnection):
67
68
    """A SSH connection that uses StringIO to buffer writes and answer reads."""
68
69
 
69
70
    def __init__(self, vendor):
71
72
 
72
73
    def close(self):
73
74
        self.vendor.calls.append(('close', ))
 
75
        self.vendor.read_from.close()
 
76
        self.vendor.write_to.close()
74
77
 
75
 
    def get_filelike_channels(self):
76
 
        return self.vendor.read_from, self.vendor.write_to
 
78
    def get_sock_or_pipes(self):
 
79
        return 'pipes', (self.vendor.read_from, self.vendor.write_to)
77
80
 
78
81
 
79
82
class _InvalidHostnameFeature(tests.Feature):
243
246
        unopened_port = sock.getsockname()[1]
244
247
        # having vendor be invalid means that if it tries to connect via the
245
248
        # vendor it will blow up.
246
 
        client_medium = medium.SmartSSHClientMedium('127.0.0.1', unopened_port,
247
 
            username=None, password=None, base='base', vendor="not a vendor",
248
 
            bzr_remote_path='bzr')
 
249
        ssh_params = medium.SSHParams('127.0.0.1', unopened_port, None, None)
 
250
        client_medium = medium.SmartSSHClientMedium(
 
251
            'base', ssh_params, "not a vendor")
249
252
        sock.close()
250
253
 
251
254
    def test_ssh_client_connects_on_first_use(self):
253
256
        # it bytes.
254
257
        output = StringIO()
255
258
        vendor = StringIOSSHVendor(StringIO(), output)
256
 
        client_medium = medium.SmartSSHClientMedium(
257
 
            'a hostname', 'a port', 'a username', 'a password', 'base', vendor,
258
 
            'bzr')
 
259
        ssh_params = medium.SSHParams(
 
260
            'a hostname', 'a port', 'a username', 'a password', 'bzr')
 
261
        client_medium = medium.SmartSSHClientMedium('base', ssh_params, vendor)
259
262
        client_medium._accept_bytes('abc')
260
263
        self.assertEqual('abc', output.getvalue())
261
264
        self.assertEqual([('connect_ssh', 'a username', 'a password',
268
271
        # it bytes.
269
272
        output = StringIO()
270
273
        vendor = StringIOSSHVendor(StringIO(), output)
271
 
        client_medium = medium.SmartSSHClientMedium('a hostname', 'a port',
272
 
            'a username', 'a password', 'base', vendor, bzr_remote_path='fugly')
 
274
        ssh_params = medium.SSHParams(
 
275
            'a hostname', 'a port', 'a username', 'a password',
 
276
            bzr_remote_path='fugly')
 
277
        client_medium = medium.SmartSSHClientMedium('base', ssh_params, vendor)
273
278
        client_medium._accept_bytes('abc')
274
279
        self.assertEqual('abc', output.getvalue())
275
280
        self.assertEqual([('connect_ssh', 'a username', 'a password',
284
289
        output = StringIO()
285
290
        vendor = StringIOSSHVendor(input, output)
286
291
        client_medium = medium.SmartSSHClientMedium(
287
 
            'a hostname', base='base', vendor=vendor, bzr_remote_path='bzr')
 
292
            'base', medium.SSHParams('a hostname'), vendor)
288
293
        client_medium._accept_bytes('abc')
289
294
        client_medium.disconnect()
290
295
        self.assertTrue(input.closed)
305
310
        output = StringIO()
306
311
        vendor = StringIOSSHVendor(input, output)
307
312
        client_medium = medium.SmartSSHClientMedium(
308
 
            'a hostname', base='base', vendor=vendor, bzr_remote_path='bzr')
 
313
            'base', medium.SSHParams('a hostname'), vendor)
309
314
        client_medium._accept_bytes('abc')
310
315
        client_medium.disconnect()
311
316
        # the disconnect has closed output, so we need a new output for the
334
339
        # Doing a disconnect on a new (and thus unconnected) SSH medium
335
340
        # does not fail.  It's ok to disconnect an unconnected medium.
336
341
        client_medium = medium.SmartSSHClientMedium(
337
 
            None, base='base', bzr_remote_path='bzr')
 
342
            'base', medium.SSHParams(None))
338
343
        client_medium.disconnect()
339
344
 
340
345
    def test_ssh_client_raises_on_read_when_not_connected(self):
341
346
        # Doing a read on a new (and thus unconnected) SSH medium raises
342
347
        # MediumNotConnected.
343
348
        client_medium = medium.SmartSSHClientMedium(
344
 
            None, base='base', bzr_remote_path='bzr')
 
349
            'base', medium.SSHParams(None))
345
350
        self.assertRaises(errors.MediumNotConnected, client_medium.read_bytes,
346
351
                          0)
347
352
        self.assertRaises(errors.MediumNotConnected, client_medium.read_bytes,
359
364
        output.flush = logging_flush
360
365
        vendor = StringIOSSHVendor(input, output)
361
366
        client_medium = medium.SmartSSHClientMedium(
362
 
            'a hostname', base='base', vendor=vendor, bzr_remote_path='bzr')
 
367
            'base', medium.SSHParams('a hostname'), vendor=vendor)
363
368
        # this call is here to ensure we only flush once, not on every
364
369
        # _accept_bytes call.
365
370
        client_medium._accept_bytes('abc')