/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

  • Committer: Aaron Bentley
  • Date: 2006-09-28 13:48:10 UTC
  • mfrom: (2049 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2268.
  • Revision ID: abentley@panoramicfeedback.com-20060928134810-2c8ae086a4a70f43
Merge bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
21
21
 
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
29
30
 
115
116
        ### FIXME decide and then test for all transports. RBC20051208
116
117
 
117
118
 
118
 
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
 
119
class SFTPTransportTestRelativeRoot(TestCaseWithSFTPServer):
119
120
    """Test the SFTP transport with homedir based relative paths."""
120
121
 
121
122
    def setUp(self):
122
123
        self._get_remote_is_absolute = False
123
 
        super(SFTPTransportTestRelative, self).setUp()
 
124
        super(SFTPTransportTestRelativeRoot, self).setUp()
124
125
 
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'))
129
132
 
130
133
 
192
195
                              clone_from=fake)
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),
 
199
                    'Transport error: '
 
200
                    'invalid port number ~janneke in url:\n'
 
201
                    'sftp://lilypond.org:~janneke/public_html/bzr/gub ')
 
202
 
 
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', '.')
 
209
        try:
 
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)
 
214
        finally:
 
215
            set_or_unset_env('PATH', orig_path)
 
216
            ssh._ssh_vendor = orig_vendor
 
217
 
 
218
    def test_abspath_root_sibling_server(self):
 
219
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
 
220
        server = SFTPSiblingAbsoluteServer()
 
221
        server.setUp()
 
222
        try:
 
223
            transport = get_transport(server.get_url())
 
224
            self.assertFalse(transport.abspath('/').endswith('/~/'))
 
225
            self.assertTrue(transport.abspath('/').endswith('/'))
 
226
            del transport
 
227
        finally:
 
228
            server.tearDown()
197
229
 
198
230
 
199
231
class SFTPBranchTest(TestCaseWithSFTPServer):
271
303
        self._test_vendor = vendor
272
304
 
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())
277
310
 
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
300
 
 
301
 
        self._transport_sftp = bzrlib.transport.sftp
 
332
        import bzrlib.transport.ssh
302
333
 
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()
308
339
 
309
 
        orig_vendor = bzrlib.transport.sftp._ssh_vendor
 
340
        orig_vendor = bzrlib.transport.ssh._ssh_vendor
310
341
        def reset():
311
 
            bzrlib.transport.sftp._ssh_vendor = orig_vendor
 
342
            bzrlib.transport.ssh._ssh_vendor = orig_vendor
312
343
            s.close()
313
344
        self.addCleanup(reset)
314
345
 
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
317
349
 
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)
323
356
 
402
435
 
403
436
    def setUp(self):
404
437
        TestCase.setUp(self)
 
438
        if not paramiko_loaded:
 
439
            raise TestSkipped('you must have paramiko to run this test')
405
440
 
406
441
    def test_delay(self):
407
442
        from bzrlib.transport.sftp import SocketDelay