← 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: Andrew Bennetts
  • Date: 2010-01-12 03:53:21 UTC
  • mfrom: (4948 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4964.
  • Revision ID: andrew.bennetts@canonical.com-20100112035321-hofpz5p10224ryj3
Merge lp:bzr, resolving conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
13
13
#
14
14
# You should have received a copy of the GNU General Public License
15
15
# along with this program; if not, write to the Free Software
16
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
 
18
18
import os
19
19
import socket
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,
33
27
    errors,
34
28
    tests,
35
29
    transport as _mod_transport,
 
30
    ui,
36
31
    )
37
32
from bzrlib.osutils import (
38
33
    pathjoin,
40
35
    set_or_unset_env,
41
36
    )
42
37
from bzrlib.tests import (
 
38
    features,
43
39
    TestCaseWithTransport,
44
40
    TestCase,
45
41
    TestSkipped,
48
44
from bzrlib.transport import get_transport
49
45
import bzrlib.transport.http
50
46
 
51
 
if paramiko_loaded:
 
47
if features.paramiko.available():
52
48
    from bzrlib.transport import sftp as _mod_sftp
53
49
    from bzrlib.transport.sftp import (
54
50
        SFTPAbsoluteServer,
75
71
 
76
72
    def setUp(self):
77
73
        super(TestCaseWithSFTPServer, self).setUp()
78
 
        if not paramiko_loaded:
79
 
            raise TestSkipped('you must have paramiko to run this test')
 
74
        self.requireFeature(features.paramiko)
80
75
        set_test_transport_to_sftp(self)
81
76
 
82
77
 
164
159
class SFTPNonServerTest(TestCase):
165
160
    def setUp(self):
166
161
        TestCase.setUp(self)
167
 
        if not paramiko_loaded:
168
 
            raise TestSkipped('you must have paramiko to run this test')
 
162
        self.requireFeature(features.paramiko)
169
163
 
170
164
    def test_parse_url_with_home_dir(self):
171
165
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative')
198
192
    def test_abspath_root_sibling_server(self):
199
193
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
200
194
        server = SFTPSiblingAbsoluteServer()
201
 
        server.setUp()
 
195
        server.start_server()
202
196
        try:
203
197
            transport = get_transport(server.get_url())
204
198
            self.assertFalse(transport.abspath('/').endswith('/~/'))
205
199
            self.assertTrue(transport.abspath('/').endswith('/'))
206
200
            del transport
207
201
        finally:
208
 
            server.tearDown()
 
202
            server.stop_server()
209
203
 
210
204
 
211
205
class SFTPBranchTest(TestCaseWithSFTPServer):
258
252
      None:       If 'ssh' exists on the machine, then it will be spawned as a
259
253
                  child process.
260
254
    """
261
 
    
 
255
 
262
256
    def setUp(self):
263
257
        super(SSHVendorConnection, self).setUp()
264
258
        from bzrlib.transport.sftp import SFTPFullAbsoluteServer
306
300
    """
307
301
 
308
302
    def setUp(self):
309
 
        if not paramiko_loaded:
310
 
            raise TestSkipped('you must have paramiko to run this test')
 
303
        self.requireFeature(features.paramiko)
311
304
        super(SSHVendorBadConnection, self).setUp()
312
305
        import bzrlib.transport.ssh
313
306
 
368
361
    """Test that the testing SFTPServer's latency knob works."""
369
362
 
370
363
    def test_latency_knob_slows_transport(self):
371
 
        # change the latency knob to 500ms. We take about 40ms for a 
 
364
        # change the latency knob to 500ms. We take about 40ms for a
372
365
        # loopback connection ordinarily.
373
366
        start_time = time.time()
374
367
        self.get_server().add_latency = 0.5
419
412
 
420
413
    def setUp(self):
421
414
        TestCase.setUp(self)
422
 
        if not paramiko_loaded:
423
 
            raise TestSkipped('you must have paramiko to run this test')
 
415
        self.requireFeature(features.paramiko)
424
416
 
425
417
    def test_delay(self):
426
418
        from bzrlib.transport.sftp import SocketDelay
481
473
class Test_SFTPReadvHelper(tests.TestCase):
482
474
 
483
475
    def checkGetRequests(self, expected_requests, offsets):
484
 
        if not paramiko_loaded:
485
 
            raise TestSkipped('you must have paramiko to run this test')
 
476
        self.requireFeature(features.paramiko)
486
477
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
487
478
            _null_report_activity)
488
479
        self.assertEqual(expected_requests, helper._get_requests())
502
493
                               (42000, 24000)])
503
494
 
504
495
    def checkRequestAndYield(self, expected, data, offsets):
505
 
        if not paramiko_loaded:
506
 
            raise TestSkipped('you must have paramiko to run this test')
 
496
        self.requireFeature(features.paramiko)
507
497
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
508
498
            _null_report_activity)
509
499
        data_f = ReadvFile(data)
546
536
    def test_sftp_is_none_if_no_config(self):
547
537
        t = self.get_transport_for_connection(set_config=False)
548
538
        self.assertIs(None, t._get_credentials()[0])
 
539
 
 
540
    def test_sftp_doesnt_prompt_username(self):
 
541
        stdout = tests.StringIOWrapper()
 
542
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n', stdout=stdout)
 
543
        t = self.get_transport_for_connection(set_config=False)
 
544
        self.assertIs(None, t._get_credentials()[0])
 
545
        # No prompts should've been printed, stdin shouldn't have been read
 
546
        self.assertEquals("", stdout.getvalue())
 
547
        self.assertEquals(0, ui.ui_factory.stdin.tell())