/breezy/unstable

To get this branch, use:
bzr branch https://code.breezy-vcs.org/breezy/unstable

« back to all changes in this revision

Viewing changes to breezy/tests/test_sftp_transport.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-24 01:39:33 UTC
  • mfrom: (3815.3776.6)
  • Revision ID: jelmer@jelmer.uk-20170524013933-ir4y4tqtrsiz2ka2
New upstream snapshot.

Show diffs side-by-side

added added

removed removed

Lines of Context:
20
20
import sys
21
21
import time
22
22
 
23
 
from bzrlib import (
 
23
from breezy import (
24
24
    config,
25
25
    controldir,
26
26
    errors,
28
28
    transport as _mod_transport,
29
29
    ui,
30
30
    )
31
 
from bzrlib.osutils import (
 
31
from breezy.osutils import (
32
32
    lexists,
33
33
    )
34
 
from bzrlib.tests import (
 
34
from breezy.tests import (
35
35
    features,
36
36
    TestCaseWithTransport,
37
37
    TestCase,
38
38
    TestSkipped,
39
39
    )
40
 
from bzrlib.tests.http_server import HttpServer
41
 
import bzrlib.transport.http
 
40
from breezy.tests.http_server import HttpServer
 
41
import breezy.transport.http
42
42
 
43
43
if features.paramiko.available():
44
 
    from bzrlib.transport import sftp as _mod_sftp
45
 
    from bzrlib.tests import stub_sftp
 
44
    from breezy.transport import sftp as _mod_sftp
 
45
    from breezy.tests import stub_sftp
46
46
 
47
47
 
48
48
def set_test_transport_to_sftp(testcase):
68
68
class SFTPLockTests(TestCaseWithSFTPServer):
69
69
 
70
70
    def test_sftp_locks(self):
71
 
        from bzrlib.errors import LockError
 
71
        from breezy.errors import LockError
72
72
        t = self.get_transport()
73
73
 
74
74
        l = t.lock_write('bogus')
168
168
 
169
169
    def test_get_paramiko_vendor(self):
170
170
        """Test that if no 'ssh' is available we get builtin paramiko"""
171
 
        from bzrlib.transport import ssh
 
171
        from breezy.transport import ssh
172
172
        # set '.' as the only location in the path, forcing no 'ssh' to exist
173
173
        self.overrideAttr(ssh, '_ssh_vendor_manager')
174
174
        self.overrideEnv('PATH', '.')
247
247
        self._test_vendor = vendor
248
248
 
249
249
    def test_connection_paramiko(self):
250
 
        from bzrlib.transport import ssh
 
250
        from breezy.transport import ssh
251
251
        self.set_vendor(ssh.ParamikoVendor())
252
252
        t = self.get_transport()
253
253
        self.assertEqual('foobar\n', t.get('a_file').read())
281
281
        self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
282
282
 
283
283
    def set_vendor(self, vendor, subprocess_stderr=None):
284
 
        from bzrlib.transport import ssh
 
284
        from breezy.transport import ssh
285
285
        self.overrideAttr(ssh._ssh_vendor_manager, '_cached_ssh_vendor', vendor)
286
286
        if subprocess_stderr is not None:
287
287
            self.overrideAttr(ssh.SubprocessVendor, "_stderr_target",
289
289
 
290
290
    def test_bad_connection_paramiko(self):
291
291
        """Test that a real connection attempt raises the right error"""
292
 
        from bzrlib.transport import ssh
 
292
        from breezy.transport import ssh
293
293
        self.set_vendor(ssh.ParamikoVendor())
294
294
        t = _mod_transport.get_transport_from_url(self.bogus_url)
295
295
        self.assertRaises(errors.ConnectionError, t.get, 'foobar')
302
302
        t = _mod_transport.get_transport_from_url(self.bogus_url)
303
303
        try:
304
304
            self.assertRaises(errors.ConnectionError, t.get, 'foobar')
305
 
        except NameError, e:
 
305
        except NameError as e:
306
306
            if "global name 'SSHException'" in str(e):
307
307
                self.knownFailure('Known NameError bug in paramiko 1.6.1')
308
308
            raise
491
491
        self.assertIs(None, t._get_credentials()[0])
492
492
 
493
493
    def test_sftp_doesnt_prompt_username(self):
494
 
        stdout = tests.StringIOWrapper()
495
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n', stdout=stdout)
 
494
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
496
495
        t = self.get_transport_for_connection(set_config=False)
497
496
        self.assertIs(None, t._get_credentials()[0])
498
497
        # No prompts should've been printed, stdin shouldn't have been read
499
 
        self.assertEqual("", stdout.getvalue())
 
498
        self.assertEqual("", ui.ui_factory.stdout.getvalue())
500
499
        self.assertEqual(0, ui.ui_factory.stdin.tell())