← 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: Tarmac
  • Author(s): Vincent Ladeuil
  • Date: 2017-01-30 14:42:05 UTC
  • mfrom: (6620.1.1 trunk)
  • Revision ID: tarmac-20170130144205-r8fh2xpmiuxyozpv
Merge  2.7 into trunk including fix for bug #1657238 [r=vila]

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
1
 
# Copyright (C) 2005-2011 Robey Pointer <robey@lag.net>
 
1
# Copyright (C) 2005-2012, 2016 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 time
22
22
 
23
23
from bzrlib import (
24
 
    bzrdir,
25
24
    config,
 
25
    controldir,
26
26
    errors,
27
27
    tests,
28
28
    transport as _mod_transport,
81
81
        l.unlock()
82
82
        self.assertFalse(lexists('bogus.write-lock'))
83
83
 
84
 
        open('something.write-lock', 'wb').write('fake lock\n')
 
84
        with open('something.write-lock', 'wb') as f: f.write('fake lock\n')
85
85
        self.assertRaises(LockError, t.lock_write, 'something')
86
86
        os.remove('something.write-lock')
87
87
 
140
140
    def test__remote_path_relative_root(self):
141
141
        # relative paths are preserved
142
142
        t = self.get_transport('')
143
 
        self.assertEqual('/~/', t._path)
 
143
        self.assertEqual('/~/', t._parsed_url.path)
144
144
        # the remote path should be relative to home dir
145
145
        # (i.e. not begining with a '/')
146
146
        self.assertEqual('a', t._remote_path('a'))
147
147
 
148
148
 
149
149
class SFTPNonServerTest(TestCase):
 
150
 
150
151
    def setUp(self):
151
 
        TestCase.setUp(self)
 
152
        super(SFTPNonServerTest, self).setUp()
152
153
        self.requireFeature(features.paramiko)
153
154
 
154
155
    def test_parse_url_with_home_dir(self):
155
156
        s = _mod_sftp.SFTPTransport(
156
157
            'sftp://ro%62ey:h%40t@example.com:2222/~/relative')
157
 
        self.assertEquals(s._host, 'example.com')
158
 
        self.assertEquals(s._port, 2222)
159
 
        self.assertEquals(s._user, 'robey')
160
 
        self.assertEquals(s._password, 'h@t')
161
 
        self.assertEquals(s._path, '/~/relative/')
 
158
        self.assertEqual(s._parsed_url.host, 'example.com')
 
159
        self.assertEqual(s._parsed_url.port, 2222)
 
160
        self.assertEqual(s._parsed_url.user, 'robey')
 
161
        self.assertEqual(s._parsed_url.password, 'h@t')
 
162
        self.assertEqual(s._parsed_url.path, '/~/relative/')
162
163
 
163
164
    def test_relpath(self):
164
165
        s = _mod_sftp.SFTPTransport('sftp://user@host.com/abs/path')
180
181
        server.start_server()
181
182
        self.addCleanup(server.stop_server)
182
183
 
183
 
        transport = _mod_transport.get_transport(server.get_url())
 
184
        transport = _mod_transport.get_transport_from_url(server.get_url())
184
185
        self.assertFalse(transport.abspath('/').endswith('/~/'))
185
186
        self.assertTrue(transport.abspath('/').endswith('/'))
186
187
        del transport
191
192
 
192
193
    def test_push_support(self):
193
194
        self.build_tree(['a/', 'a/foo'])
194
 
        t = bzrdir.BzrDir.create_standalone_workingtree('a')
 
195
        t = controldir.ControlDir.create_standalone_workingtree('a')
195
196
        b = t.branch
196
197
        t.add('foo')
197
198
        t.commit('foo', rev_id='a1')
198
199
 
199
 
        b2 = bzrdir.BzrDir.create_branch_and_repo(self.get_url('/b'))
 
200
        b2 = controldir.ControlDir.create_branch_and_repo(self.get_url('/b'))
200
201
        b2.pull(b)
201
202
 
202
 
        self.assertEquals(b2.revision_history(), ['a1'])
 
203
        self.assertEqual(b2.last_revision(), 'a1')
203
204
 
204
 
        open('a/foo', 'wt').write('something new in foo\n')
 
205
        with open('a/foo', 'wt') as f: f.write('something new in foo\n')
205
206
        t.commit('new', rev_id='a2')
206
207
        b2.pull(b)
207
208
 
208
 
        self.assertEquals(b2.revision_history(), ['a1', 'a2'])
 
209
        self.assertEqual(b2.last_revision(), 'a2')
209
210
 
210
211
 
211
212
class SSHVendorConnection(TestCaseWithSFTPServer):
279
280
        self.addCleanup(s.close)
280
281
        self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
281
282
 
282
 
    def set_vendor(self, vendor):
 
283
    def set_vendor(self, vendor, subprocess_stderr=None):
283
284
        from bzrlib.transport import ssh
284
285
        self.overrideAttr(ssh._ssh_vendor_manager, '_cached_ssh_vendor', vendor)
 
286
        if subprocess_stderr is not None:
 
287
            self.overrideAttr(ssh.SubprocessVendor, "_stderr_target",
 
288
                subprocess_stderr)
285
289
 
286
290
    def test_bad_connection_paramiko(self):
287
291
        """Test that a real connection attempt raises the right error"""
288
292
        from bzrlib.transport import ssh
289
293
        self.set_vendor(ssh.ParamikoVendor())
290
 
        t = _mod_transport.get_transport(self.bogus_url)
 
294
        t = _mod_transport.get_transport_from_url(self.bogus_url)
291
295
        self.assertRaises(errors.ConnectionError, t.get, 'foobar')
292
296
 
293
297
    def test_bad_connection_ssh(self):
294
298
        """None => auto-detect vendor"""
295
 
        self.set_vendor(None)
296
 
        # This is how I would normally test the connection code
297
 
        # it makes it very clear what we are testing.
298
 
        # However, 'ssh' will create stipple on the output, so instead
299
 
        # I'm using run_bzr_subprocess, and parsing the output
300
 
        # try:
301
 
        #     t = _mod_transport.get_transport(self.bogus_url)
302
 
        # except errors.ConnectionError:
303
 
        #     # Correct error
304
 
        #     pass
305
 
        # except errors.NameError, e:
306
 
        #     if 'SSHException' in str(e):
307
 
        #         raise TestSkipped('Known NameError bug in paramiko 1.6.1')
308
 
        #     raise
309
 
        # else:
310
 
        #     self.fail('Excepted ConnectionError to be raised')
311
 
 
312
 
        out, err = self.run_bzr_subprocess(['log', self.bogus_url], retcode=3)
313
 
        self.assertEqual('', out)
314
 
        if "NameError: global name 'SSHException'" in err:
315
 
            # We aren't fixing this bug, because it is a bug in
316
 
            # paramiko, but we know about it, so we don't have to
317
 
            # fail the test
318
 
            raise TestSkipped('Known NameError bug with paramiko-1.6.1')
319
 
        self.assertContainsRe(err, r'bzr: ERROR: Unable to connect to SSH host'
320
 
                                   r' 127\.0\.0\.1:\d+; ')
 
299
        f = file(os.devnull, "wb")
 
300
        self.addCleanup(f.close)
 
301
        self.set_vendor(None, f)
 
302
        t = _mod_transport.get_transport_from_url(self.bogus_url)
 
303
        try:
 
304
            self.assertRaises(errors.ConnectionError, t.get, 'foobar')
 
305
        except NameError, e:
 
306
            if "global name 'SSHException'" in str(e):
 
307
                self.knownFailure('Known NameError bug in paramiko 1.6.1')
 
308
            raise
321
309
 
322
310
 
323
311
class SFTPLatencyKnob(TestCaseWithSFTPServer):
374
362
class TestSocketDelay(TestCase):
375
363
 
376
364
    def setUp(self):
377
 
        TestCase.setUp(self)
 
365
        super(TestSocketDelay, self).setUp()
378
366
        self.requireFeature(features.paramiko)
379
367
 
380
368
    def test_delay(self):
488
476
            conf._get_config().update(
489
477
                {'sftptest': {'scheme': 'ssh', 'port': port, 'user': 'bar'}})
490
478
            conf._save()
491
 
        t = _mod_transport.get_transport('sftp://localhost:%d' % port)
 
479
        t = _mod_transport.get_transport_from_url(
 
480
            'sftp://localhost:%d' % port)
492
481
        # force a connection to be performed.
493
482
        t.has('foo')
494
483
        return t
507
496
        t = self.get_transport_for_connection(set_config=False)
508
497
        self.assertIs(None, t._get_credentials()[0])
509
498
        # No prompts should've been printed, stdin shouldn't have been read
510
 
        self.assertEquals("", stdout.getvalue())
511
 
        self.assertEquals(0, ui.ui_factory.stdin.tell())
 
499
        self.assertEqual("", stdout.getvalue())
 
500
        self.assertEqual(0, ui.ui_factory.stdin.tell())