225
225
self.assertEquals(b2.revision_history(), ['a1', 'a2'])
228
class SFTPFullHandshakingTest(TestCaseWithSFTPServer):
229
"""Verify that a full-handshake (SSH over loopback TCP) sftp connection works."""
228
class SSHVendorConnection(TestCaseWithSFTPServer):
229
"""Test that the ssh vendors can all connect.
231
Verify that a full-handshake (SSH over loopback TCP) sftp connection works.
233
We have 3 sftp implementations in the test suite:
234
'loopback': Doesn't use ssh, just uses a local socket. Most tests are
235
done this way to save the handshaking time, so it is not
237
'none': This uses paramiko's built-in ssh client and server, and layers
239
None: If 'ssh' exists on the machine, then it will be spawned as a
231
def test_connection(self):
244
super(SSHVendorConnection, self).setUp()
232
245
from bzrlib.transport.sftp import SFTPFullAbsoluteServer
233
self.transport_server = SFTPFullAbsoluteServer
248
"""Just a wrapper so that when created, it will set _vendor"""
249
# SFTPFullAbsoluteServer can handle any vendor,
250
# it just needs to be set between the time it is instantiated
251
# and the time .setUp() is called
252
server = SFTPFullAbsoluteServer()
253
server._vendor = self._test_vendor
255
self._test_vendor = 'loopback'
256
self.transport_server = create_server
257
f = open('a_file', 'wb')
263
def set_vendor(self, vendor):
264
self._test_vendor = vendor
266
def test_connection_paramiko(self):
267
self.set_vendor('none')
268
t = self.get_transport()
269
self.assertEqual('foobar\n', t.get('a_file').read())
271
def test_connection_vendor(self):
272
raise TestSkipped("We don't test spawning real ssh,"
273
" because it prompts for a password."
274
" Enable this test if we figure out"
275
" how to prevent this.")
276
self.set_vendor(None)
277
t = self.get_transport()
278
self.assertEqual('foobar\n', t.get('a_file').read())
281
class SSHVendorBadConnection(TestCaseWithTransport):
282
"""Test that the ssh vendors handle bad connection properly
284
We don't subclass TestCaseWithSFTPServer, because we don't actually
285
need an SFTP connection.
289
if not paramiko_loaded:
290
raise TestSkipped('you must have paramiko to run this test')
291
super(SSHVendorBadConnection, self).setUp()
292
import bzrlib.transport.sftp
294
self._transport_sftp = bzrlib.transport.sftp
296
# open a random port, so we know nobody else is using it
297
# but don't actually listen on the port.
299
s.bind(('localhost', 0))
300
self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
302
orig_vendor = bzrlib.transport.sftp._ssh_vendor
304
bzrlib.transport.sftp._ssh_vendor = orig_vendor
306
self.addCleanup(reset)
308
def set_vendor(self, vendor):
309
self._transport_sftp._ssh_vendor = vendor
311
def test_bad_connection_paramiko(self):
312
"""Test that a real connection attempt raises the right error"""
313
self.set_vendor('none')
314
self.assertRaises(errors.ConnectionError,
315
bzrlib.transport.get_transport, self.bogus_url)
317
def test_bad_connection_ssh(self):
318
"""None => auto-detect vendor"""
319
self.set_vendor(None)
320
# This is how I would normally test the connection code
321
# it makes it very clear what we are testing.
322
# However, 'ssh' will create stipple on the output, so instead
323
# I'm using run_bzr_subprocess, and parsing the output
325
# t = bzrlib.transport.get_transport(self.bogus_url)
326
# except errors.ConnectionError:
329
# except errors.NameError, e:
330
# if 'SSHException' in str(e):
331
# raise TestSkipped('Known NameError bug in paramiko 1.6.1')
334
# self.fail('Excepted ConnectionError to be raised')
336
out, err = self.run_bzr_subprocess('log', self.bogus_url, retcode=3)
337
self.assertEqual('', out)
338
if "NameError: global name 'SSHException'" in err:
339
# We aren't fixing this bug, because it is a bug in
340
# paramiko, but we know about it, so we don't have to
342
raise TestSkipped('Known NameError bug with paramiko-1.6.1')
343
self.assertContainsRe(err, 'Connection error')
added
removed
bzrlib/tests/test_sftp_transport.py
