← 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

  • Committer: John Arbash Meinel
  • Date: 2010-02-17 17:11:16 UTC
  • mfrom: (4797.2.17 2.1)
  • mto: (4797.2.18 2.1)
  • mto: This revision was merged to the branch mainline in revision 5055.
  • Revision ID: john@arbash-meinel.com-20100217171116-h7t9223ystbnx5h8
merge bzr.2.1 in preparation for NEWS entry.

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,
49
44
from bzrlib.transport import get_transport
50
45
import bzrlib.transport.http
51
46
 
52
 
if paramiko_loaded:
 
47
if features.paramiko.available():
53
48
    from bzrlib.transport import sftp as _mod_sftp
54
 
    from bzrlib.transport.sftp import (
55
 
        SFTPAbsoluteServer,
56
 
        SFTPHomeDirServer,
57
 
        SFTPTransport,
58
 
        )
 
49
    from bzrlib.tests import stub_sftp
59
50
 
60
51
from bzrlib.workingtree import WorkingTree
61
52
 
65
56
    if getattr(testcase, '_get_remote_is_absolute', None) is None:
66
57
        testcase._get_remote_is_absolute = True
67
58
    if testcase._get_remote_is_absolute:
68
 
        testcase.transport_server = SFTPAbsoluteServer
 
59
        testcase.transport_server = stub_sftp.SFTPAbsoluteServer
69
60
    else:
70
 
        testcase.transport_server = SFTPHomeDirServer
 
61
        testcase.transport_server = stub_sftp.SFTPHomeDirServer
71
62
    testcase.transport_readonly_server = HttpServer
72
63
 
73
64
 
76
67
 
77
68
    def setUp(self):
78
69
        super(TestCaseWithSFTPServer, self).setUp()
79
 
        if not paramiko_loaded:
80
 
            raise TestSkipped('you must have paramiko to run this test')
 
70
        self.requireFeature(features.paramiko)
81
71
        set_test_transport_to_sftp(self)
82
72
 
83
73
 
165
155
class SFTPNonServerTest(TestCase):
166
156
    def setUp(self):
167
157
        TestCase.setUp(self)
168
 
        if not paramiko_loaded:
169
 
            raise TestSkipped('you must have paramiko to run this test')
 
158
        self.requireFeature(features.paramiko)
170
159
 
171
160
    def test_parse_url_with_home_dir(self):
172
 
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative')
 
161
        s = _mod_sftp.SFTPTransport(
 
162
            'sftp://ro%62ey:h%40t@example.com:2222/~/relative')
173
163
        self.assertEquals(s._host, 'example.com')
174
164
        self.assertEquals(s._port, 2222)
175
165
        self.assertEquals(s._user, 'robey')
177
167
        self.assertEquals(s._path, '/~/relative/')
178
168
 
179
169
    def test_relpath(self):
180
 
        s = SFTPTransport('sftp://user@host.com/abs/path')
 
170
        s = _mod_sftp.SFTPTransport('sftp://user@host.com/abs/path')
181
171
        self.assertRaises(errors.PathNotChild, s.relpath,
182
172
                          'sftp://user@host.com/~/rel/path/sub')
183
173
 
197
187
            ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
198
188
 
199
189
    def test_abspath_root_sibling_server(self):
200
 
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
201
 
        server = SFTPSiblingAbsoluteServer()
202
 
        server.setUp()
 
190
        server = stub_sftp.SFTPSiblingAbsoluteServer()
 
191
        server.start_server()
203
192
        try:
204
193
            transport = get_transport(server.get_url())
205
194
            self.assertFalse(transport.abspath('/').endswith('/~/'))
206
195
            self.assertTrue(transport.abspath('/').endswith('/'))
207
196
            del transport
208
197
        finally:
209
 
            server.tearDown()
 
198
            server.stop_server()
210
199
 
211
200
 
212
201
class SFTPBranchTest(TestCaseWithSFTPServer):
262
251
 
263
252
    def setUp(self):
264
253
        super(SSHVendorConnection, self).setUp()
265
 
        from bzrlib.transport.sftp import SFTPFullAbsoluteServer
266
254
 
267
255
        def create_server():
268
256
            """Just a wrapper so that when created, it will set _vendor"""
269
257
            # SFTPFullAbsoluteServer can handle any vendor,
270
258
            # it just needs to be set between the time it is instantiated
271
259
            # and the time .setUp() is called
272
 
            server = SFTPFullAbsoluteServer()
 
260
            server = stub_sftp.SFTPFullAbsoluteServer()
273
261
            server._vendor = self._test_vendor
274
262
            return server
275
263
        self._test_vendor = 'loopback'
307
295
    """
308
296
 
309
297
    def setUp(self):
310
 
        if not paramiko_loaded:
311
 
            raise TestSkipped('you must have paramiko to run this test')
 
298
        self.requireFeature(features.paramiko)
312
299
        super(SSHVendorBadConnection, self).setUp()
313
300
        import bzrlib.transport.ssh
314
301
 
420
407
 
421
408
    def setUp(self):
422
409
        TestCase.setUp(self)
423
 
        if not paramiko_loaded:
424
 
            raise TestSkipped('you must have paramiko to run this test')
 
410
        self.requireFeature(features.paramiko)
425
411
 
426
412
    def test_delay(self):
427
 
        from bzrlib.transport.sftp import SocketDelay
428
413
        sending = FakeSocket()
429
 
        receiving = SocketDelay(sending, 0.1, bandwidth=1000000,
430
 
                                really_sleep=False)
 
414
        receiving = stub_sftp.SocketDelay(sending, 0.1, bandwidth=1000000,
 
415
                                          really_sleep=False)
431
416
        # check that simulated time is charged only per round-trip:
432
 
        t1 = SocketDelay.simulated_time
 
417
        t1 = stub_sftp.SocketDelay.simulated_time
433
418
        receiving.send("connect1")
434
419
        self.assertEqual(sending.recv(1024), "connect1")
435
 
        t2 = SocketDelay.simulated_time
 
420
        t2 = stub_sftp.SocketDelay.simulated_time
436
421
        self.assertAlmostEqual(t2 - t1, 0.1)
437
422
        receiving.send("connect2")
438
423
        self.assertEqual(sending.recv(1024), "connect2")
439
424
        sending.send("hello")
440
425
        self.assertEqual(receiving.recv(1024), "hello")
441
 
        t3 = SocketDelay.simulated_time
 
426
        t3 = stub_sftp.SocketDelay.simulated_time
442
427
        self.assertAlmostEqual(t3 - t2, 0.1)
443
428
        sending.send("hello")
444
429
        self.assertEqual(receiving.recv(1024), "hello")
446
431
        self.assertEqual(receiving.recv(1024), "hello")
447
432
        sending.send("hello")
448
433
        self.assertEqual(receiving.recv(1024), "hello")
449
 
        t4 = SocketDelay.simulated_time
 
434
        t4 = stub_sftp.SocketDelay.simulated_time
450
435
        self.assertAlmostEqual(t4, t3)
451
436
 
452
437
    def test_bandwidth(self):
453
 
        from bzrlib.transport.sftp import SocketDelay
454
438
        sending = FakeSocket()
455
 
        receiving = SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
456
 
                                really_sleep=False)
 
439
        receiving = stub_sftp.SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
 
440
                                          really_sleep=False)
457
441
        # check that simulated time is charged only per round-trip:
458
 
        t1 = SocketDelay.simulated_time
 
442
        t1 = stub_sftp.SocketDelay.simulated_time
459
443
        receiving.send("connect")
460
444
        self.assertEqual(sending.recv(1024), "connect")
461
445
        sending.send("a" * 100)
462
446
        self.assertEqual(receiving.recv(1024), "a" * 100)
463
 
        t2 = SocketDelay.simulated_time
 
447
        t2 = stub_sftp.SocketDelay.simulated_time
464
448
        self.assertAlmostEqual(t2 - t1, 100 + 7)
465
449
 
466
450
 
482
466
class Test_SFTPReadvHelper(tests.TestCase):
483
467
 
484
468
    def checkGetRequests(self, expected_requests, offsets):
485
 
        if not paramiko_loaded:
486
 
            raise TestSkipped('you must have paramiko to run this test')
 
469
        self.requireFeature(features.paramiko)
487
470
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
488
471
            _null_report_activity)
489
472
        self.assertEqual(expected_requests, helper._get_requests())
503
486
                               (42000, 24000)])
504
487
 
505
488
    def checkRequestAndYield(self, expected, data, offsets):
506
 
        if not paramiko_loaded:
507
 
            raise TestSkipped('you must have paramiko to run this test')
 
489
        self.requireFeature(features.paramiko)
508
490
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
509
491
            _null_report_activity)
510
492
        data_f = ReadvFile(data)