← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

Merge up through 2.2.0.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
1
 
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
1
# Copyright (C) 2005-2010 Robey Pointer <robey@lag.net>
2
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
3
3
#
4
4
# This program is free software; you can redistribute it and/or modify
21
21
import threading
22
22
import time
23
23
 
24
 
try:
25
 
    import paramiko
26
 
    paramiko_loaded = True
27
 
except ImportError:
28
 
    paramiko_loaded = False
29
 
 
30
24
from bzrlib import (
31
25
    bzrdir,
32
26
    config,
41
35
    set_or_unset_env,
42
36
    )
43
37
from bzrlib.tests import (
 
38
    features,
44
39
    TestCaseWithTransport,
45
40
    TestCase,
46
41
    TestSkipped,
47
42
    )
48
43
from bzrlib.tests.http_server import HttpServer
49
 
from bzrlib.transport import get_transport
50
44
import bzrlib.transport.http
51
45
 
52
 
if paramiko_loaded:
 
46
if features.paramiko.available():
53
47
    from bzrlib.transport import sftp as _mod_sftp
54
 
    from bzrlib.transport.sftp import (
55
 
        SFTPAbsoluteServer,
56
 
        SFTPHomeDirServer,
57
 
        SFTPTransport,
58
 
        )
 
48
    from bzrlib.tests import stub_sftp
59
49
 
60
50
from bzrlib.workingtree import WorkingTree
61
51
 
65
55
    if getattr(testcase, '_get_remote_is_absolute', None) is None:
66
56
        testcase._get_remote_is_absolute = True
67
57
    if testcase._get_remote_is_absolute:
68
 
        testcase.transport_server = SFTPAbsoluteServer
 
58
        testcase.transport_server = stub_sftp.SFTPAbsoluteServer
69
59
    else:
70
 
        testcase.transport_server = SFTPHomeDirServer
 
60
        testcase.transport_server = stub_sftp.SFTPHomeDirServer
71
61
    testcase.transport_readonly_server = HttpServer
72
62
 
73
63
 
76
66
 
77
67
    def setUp(self):
78
68
        super(TestCaseWithSFTPServer, self).setUp()
79
 
        if not paramiko_loaded:
80
 
            raise TestSkipped('you must have paramiko to run this test')
 
69
        self.requireFeature(features.paramiko)
81
70
        set_test_transport_to_sftp(self)
82
71
 
83
72
 
165
154
class SFTPNonServerTest(TestCase):
166
155
    def setUp(self):
167
156
        TestCase.setUp(self)
168
 
        if not paramiko_loaded:
169
 
            raise TestSkipped('you must have paramiko to run this test')
 
157
        self.requireFeature(features.paramiko)
170
158
 
171
159
    def test_parse_url_with_home_dir(self):
172
 
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative')
 
160
        s = _mod_sftp.SFTPTransport(
 
161
            'sftp://ro%62ey:h%40t@example.com:2222/~/relative')
173
162
        self.assertEquals(s._host, 'example.com')
174
163
        self.assertEquals(s._port, 2222)
175
164
        self.assertEquals(s._user, 'robey')
177
166
        self.assertEquals(s._path, '/~/relative/')
178
167
 
179
168
    def test_relpath(self):
180
 
        s = SFTPTransport('sftp://user@host.com/abs/path')
 
169
        s = _mod_sftp.SFTPTransport('sftp://user@host.com/abs/path')
181
170
        self.assertRaises(errors.PathNotChild, s.relpath,
182
171
                          'sftp://user@host.com/~/rel/path/sub')
183
172
 
197
186
            ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
198
187
 
199
188
    def test_abspath_root_sibling_server(self):
200
 
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
201
 
        server = SFTPSiblingAbsoluteServer()
202
 
        server.setUp()
 
189
        server = stub_sftp.SFTPSiblingAbsoluteServer()
 
190
        server.start_server()
203
191
        try:
204
 
            transport = get_transport(server.get_url())
 
192
            transport = _mod_transport.get_transport(server.get_url())
205
193
            self.assertFalse(transport.abspath('/').endswith('/~/'))
206
194
            self.assertTrue(transport.abspath('/').endswith('/'))
207
195
            del transport
208
196
        finally:
209
 
            server.tearDown()
 
197
            server.stop_server()
210
198
 
211
199
 
212
200
class SFTPBranchTest(TestCaseWithSFTPServer):
262
250
 
263
251
    def setUp(self):
264
252
        super(SSHVendorConnection, self).setUp()
265
 
        from bzrlib.transport.sftp import SFTPFullAbsoluteServer
266
253
 
267
254
        def create_server():
268
255
            """Just a wrapper so that when created, it will set _vendor"""
269
256
            # SFTPFullAbsoluteServer can handle any vendor,
270
257
            # it just needs to be set between the time it is instantiated
271
258
            # and the time .setUp() is called
272
 
            server = SFTPFullAbsoluteServer()
 
259
            server = stub_sftp.SFTPFullAbsoluteServer()
273
260
            server._vendor = self._test_vendor
274
261
            return server
275
262
        self._test_vendor = 'loopback'
307
294
    """
308
295
 
309
296
    def setUp(self):
310
 
        if not paramiko_loaded:
311
 
            raise TestSkipped('you must have paramiko to run this test')
 
297
        self.requireFeature(features.paramiko)
312
298
        super(SSHVendorBadConnection, self).setUp()
313
 
        import bzrlib.transport.ssh
314
299
 
315
300
        # open a random port, so we know nobody else is using it
316
301
        # but don't actually listen on the port.
317
302
        s = socket.socket()
318
303
        s.bind(('localhost', 0))
 
304
        self.addCleanup(s.close)
319
305
        self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
320
306
 
321
 
        orig_vendor = bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor
322
 
        def reset():
323
 
            bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
324
 
            s.close()
325
 
        self.addCleanup(reset)
326
 
 
327
307
    def set_vendor(self, vendor):
328
 
        import bzrlib.transport.ssh
329
 
        bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor = vendor
 
308
        from bzrlib.transport import ssh
 
309
        self.overrideAttr(ssh._ssh_vendor_manager, '_cached_ssh_vendor', vendor)
330
310
 
331
311
    def test_bad_connection_paramiko(self):
332
312
        """Test that a real connection attempt raises the right error"""
333
313
        from bzrlib.transport import ssh
334
314
        self.set_vendor(ssh.ParamikoVendor())
335
 
        t = bzrlib.transport.get_transport(self.bogus_url)
 
315
        t = _mod_transport.get_transport(self.bogus_url)
336
316
        self.assertRaises(errors.ConnectionError, t.get, 'foobar')
337
317
 
338
318
    def test_bad_connection_ssh(self):
343
323
        # However, 'ssh' will create stipple on the output, so instead
344
324
        # I'm using run_bzr_subprocess, and parsing the output
345
325
        # try:
346
 
        #     t = bzrlib.transport.get_transport(self.bogus_url)
 
326
        #     t = _mod_transport.get_transport(self.bogus_url)
347
327
        # except errors.ConnectionError:
348
328
        #     # Correct error
349
329
        #     pass
420
400
 
421
401
    def setUp(self):
422
402
        TestCase.setUp(self)
423
 
        if not paramiko_loaded:
424
 
            raise TestSkipped('you must have paramiko to run this test')
 
403
        self.requireFeature(features.paramiko)
425
404
 
426
405
    def test_delay(self):
427
 
        from bzrlib.transport.sftp import SocketDelay
428
406
        sending = FakeSocket()
429
 
        receiving = SocketDelay(sending, 0.1, bandwidth=1000000,
430
 
                                really_sleep=False)
 
407
        receiving = stub_sftp.SocketDelay(sending, 0.1, bandwidth=1000000,
 
408
                                          really_sleep=False)
431
409
        # check that simulated time is charged only per round-trip:
432
 
        t1 = SocketDelay.simulated_time
 
410
        t1 = stub_sftp.SocketDelay.simulated_time
433
411
        receiving.send("connect1")
434
412
        self.assertEqual(sending.recv(1024), "connect1")
435
 
        t2 = SocketDelay.simulated_time
 
413
        t2 = stub_sftp.SocketDelay.simulated_time
436
414
        self.assertAlmostEqual(t2 - t1, 0.1)
437
415
        receiving.send("connect2")
438
416
        self.assertEqual(sending.recv(1024), "connect2")
439
417
        sending.send("hello")
440
418
        self.assertEqual(receiving.recv(1024), "hello")
441
 
        t3 = SocketDelay.simulated_time
 
419
        t3 = stub_sftp.SocketDelay.simulated_time
442
420
        self.assertAlmostEqual(t3 - t2, 0.1)
443
421
        sending.send("hello")
444
422
        self.assertEqual(receiving.recv(1024), "hello")
446
424
        self.assertEqual(receiving.recv(1024), "hello")
447
425
        sending.send("hello")
448
426
        self.assertEqual(receiving.recv(1024), "hello")
449
 
        t4 = SocketDelay.simulated_time
 
427
        t4 = stub_sftp.SocketDelay.simulated_time
450
428
        self.assertAlmostEqual(t4, t3)
451
429
 
452
430
    def test_bandwidth(self):
453
 
        from bzrlib.transport.sftp import SocketDelay
454
431
        sending = FakeSocket()
455
 
        receiving = SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
456
 
                                really_sleep=False)
 
432
        receiving = stub_sftp.SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
 
433
                                          really_sleep=False)
457
434
        # check that simulated time is charged only per round-trip:
458
 
        t1 = SocketDelay.simulated_time
 
435
        t1 = stub_sftp.SocketDelay.simulated_time
459
436
        receiving.send("connect")
460
437
        self.assertEqual(sending.recv(1024), "connect")
461
438
        sending.send("a" * 100)
462
439
        self.assertEqual(receiving.recv(1024), "a" * 100)
463
 
        t2 = SocketDelay.simulated_time
 
440
        t2 = stub_sftp.SocketDelay.simulated_time
464
441
        self.assertAlmostEqual(t2 - t1, 100 + 7)
465
442
 
466
443
 
482
459
class Test_SFTPReadvHelper(tests.TestCase):
483
460
 
484
461
    def checkGetRequests(self, expected_requests, offsets):
485
 
        if not paramiko_loaded:
486
 
            raise TestSkipped('you must have paramiko to run this test')
 
462
        self.requireFeature(features.paramiko)
487
463
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
488
464
            _null_report_activity)
489
465
        self.assertEqual(expected_requests, helper._get_requests())
503
479
                               (42000, 24000)])
504
480
 
505
481
    def checkRequestAndYield(self, expected, data, offsets):
506
 
        if not paramiko_loaded:
507
 
            raise TestSkipped('you must have paramiko to run this test')
 
482
        self.requireFeature(features.paramiko)
508
483
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
509
484
            _null_report_activity)
510
485
        data_f = ReadvFile(data)
535
510
            conf._get_config().update(
536
511
                {'sftptest': {'scheme': 'ssh', 'port': port, 'user': 'bar'}})
537
512
            conf._save()
538
 
        t = get_transport('sftp://localhost:%d' % port)
 
513
        t = _mod_transport.get_transport('sftp://localhost:%d' % port)
539
514
        # force a connection to be performed.
540
515
        t.has('foo')
541
516
        return t