22
22
import bzrlib.bzrdir as bzrdir
23
23
import bzrlib.errors as errors
24
from bzrlib.osutils import pathjoin, lexists
24
from bzrlib.osutils import pathjoin, lexists, set_or_unset_env
25
25
from bzrlib.tests import TestCaseWithTransport, TestCase, TestSkipped
26
26
import bzrlib.transport
27
from bzrlib.transport import get_transport
27
28
import bzrlib.transport.http
28
29
from bzrlib.workingtree import WorkingTree
115
116
### FIXME decide and then test for all transports. RBC20051208
118
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
119
class SFTPTransportTestRelativeRoot(TestCaseWithSFTPServer):
119
120
"""Test the SFTP transport with homedir based relative paths."""
122
123
self._get_remote_is_absolute = False
123
super(SFTPTransportTestRelative, self).setUp()
124
super(SFTPTransportTestRelativeRoot, self).setUp()
125
126
def test__remote_path_relative_root(self):
126
127
# relative paths are preserved
127
128
t = self.get_transport('')
129
# the remote path should be ''
130
self.assertEqual('', t._path)
128
131
self.assertEqual('a', t._remote_path('a'))
193
196
self.fail('expected exception not raised')
194
197
except TransportError, e:
195
self.assertEquals(str(e),
196
'Transport error: ~janneke: invalid port number ')
198
self.assertEquals(str(e),
200
'invalid port number ~janneke in url:\n'
201
'sftp://lilypond.org:~janneke/public_html/bzr/gub ')
203
def test_get_paramiko_vendor(self):
204
"""Test that if no 'ssh' is available we get builtin paramiko"""
205
from bzrlib.transport import ssh
206
# set '.' as the only location in the path, forcing no 'ssh' to exist
207
orig_vendor = ssh._ssh_vendor
208
orig_path = set_or_unset_env('PATH', '.')
210
# No vendor defined yet, query for one
211
ssh._ssh_vendor = None
212
vendor = ssh._get_ssh_vendor()
213
self.assertIsInstance(vendor, ssh.ParamikoVendor)
215
set_or_unset_env('PATH', orig_path)
216
ssh._ssh_vendor = orig_vendor
218
def test_abspath_root_sibling_server(self):
219
from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
220
server = SFTPSiblingAbsoluteServer()
223
transport = get_transport(server.get_url())
224
self.assertFalse(transport.abspath('/').endswith('/~/'))
225
self.assertTrue(transport.abspath('/').endswith('/'))
199
231
class SFTPBranchTest(TestCaseWithSFTPServer):
271
303
self._test_vendor = vendor
273
305
def test_connection_paramiko(self):
274
self.set_vendor('none')
306
from bzrlib.transport import ssh
307
self.set_vendor(ssh.ParamikoVendor())
275
308
t = self.get_transport()
276
309
self.assertEqual('foobar\n', t.get('a_file').read())
296
329
if not paramiko_loaded:
297
330
raise TestSkipped('you must have paramiko to run this test')
298
331
super(SSHVendorBadConnection, self).setUp()
299
import bzrlib.transport.sftp
301
self._transport_sftp = bzrlib.transport.sftp
332
import bzrlib.transport.ssh
303
334
# open a random port, so we know nobody else is using it
304
335
# but don't actually listen on the port.
306
337
s.bind(('localhost', 0))
307
338
self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
309
orig_vendor = bzrlib.transport.sftp._ssh_vendor
340
orig_vendor = bzrlib.transport.ssh._ssh_vendor
311
bzrlib.transport.sftp._ssh_vendor = orig_vendor
342
bzrlib.transport.ssh._ssh_vendor = orig_vendor
313
344
self.addCleanup(reset)
315
346
def set_vendor(self, vendor):
316
self._transport_sftp._ssh_vendor = vendor
347
import bzrlib.transport.ssh
348
bzrlib.transport.ssh._ssh_vendor = vendor
318
350
def test_bad_connection_paramiko(self):
319
351
"""Test that a real connection attempt raises the right error"""
320
self.set_vendor('none')
352
from bzrlib.transport import ssh
353
self.set_vendor(ssh.ParamikoVendor())
321
354
self.assertRaises(errors.ConnectionError,
322
355
bzrlib.transport.get_transport, self.bogus_url)