← 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: Vincent Ladeuil
  • Date: 2012-03-13 17:25:29 UTC
  • mfrom: (6499 +trunk)
  • mto: This revision was merged to the branch mainline in revision 6501.
  • Revision ID: v.ladeuil+lp@free.fr-20120313172529-i0suyjnepsor25i7
Merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
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'))
154
154
    def test_parse_url_with_home_dir(self):
155
155
        s = _mod_sftp.SFTPTransport(
156
156
            '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/')
 
157
        self.assertEquals(s._parsed_url.host, 'example.com')
 
158
        self.assertEquals(s._parsed_url.port, 2222)
 
159
        self.assertEquals(s._parsed_url.user, 'robey')
 
160
        self.assertEquals(s._parsed_url.password, 'h@t')
 
161
        self.assertEquals(s._parsed_url.path, '/~/relative/')
162
162
 
163
163
    def test_relpath(self):
164
164
        s = _mod_sftp.SFTPTransport('sftp://user@host.com/abs/path')
180
180
        server.start_server()
181
181
        self.addCleanup(server.stop_server)
182
182
 
183
 
        transport = _mod_transport.get_transport(server.get_url())
 
183
        transport = _mod_transport.get_transport_from_url(server.get_url())
184
184
        self.assertFalse(transport.abspath('/').endswith('/~/'))
185
185
        self.assertTrue(transport.abspath('/').endswith('/'))
186
186
        del transport
191
191
 
192
192
    def test_push_support(self):
193
193
        self.build_tree(['a/', 'a/foo'])
194
 
        t = bzrdir.BzrDir.create_standalone_workingtree('a')
 
194
        t = controldir.ControlDir.create_standalone_workingtree('a')
195
195
        b = t.branch
196
196
        t.add('foo')
197
197
        t.commit('foo', rev_id='a1')
198
198
 
199
 
        b2 = bzrdir.BzrDir.create_branch_and_repo(self.get_url('/b'))
 
199
        b2 = controldir.ControlDir.create_branch_and_repo(self.get_url('/b'))
200
200
        b2.pull(b)
201
201
 
202
 
        self.assertEquals(b2.revision_history(), ['a1'])
 
202
        self.assertEquals(b2.last_revision(), 'a1')
203
203
 
204
 
        open('a/foo', 'wt').write('something new in foo\n')
 
204
        with open('a/foo', 'wt') as f: f.write('something new in foo\n')
205
205
        t.commit('new', rev_id='a2')
206
206
        b2.pull(b)
207
207
 
208
 
        self.assertEquals(b2.revision_history(), ['a1', 'a2'])
 
208
        self.assertEquals(b2.last_revision(), 'a2')
209
209
 
210
210
 
211
211
class SSHVendorConnection(TestCaseWithSFTPServer):
279
279
        self.addCleanup(s.close)
280
280
        self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
281
281
 
282
 
    def set_vendor(self, vendor):
 
282
    def set_vendor(self, vendor, subprocess_stderr=None):
283
283
        from bzrlib.transport import ssh
284
284
        self.overrideAttr(ssh._ssh_vendor_manager, '_cached_ssh_vendor', vendor)
 
285
        if subprocess_stderr is not None:
 
286
            self.overrideAttr(ssh.SubprocessVendor, "_stderr_target",
 
287
                subprocess_stderr)
285
288
 
286
289
    def test_bad_connection_paramiko(self):
287
290
        """Test that a real connection attempt raises the right error"""
288
291
        from bzrlib.transport import ssh
289
292
        self.set_vendor(ssh.ParamikoVendor())
290
 
        t = _mod_transport.get_transport(self.bogus_url)
 
293
        t = _mod_transport.get_transport_from_url(self.bogus_url)
291
294
        self.assertRaises(errors.ConnectionError, t.get, 'foobar')
292
295
 
293
296
    def test_bad_connection_ssh(self):
294
297
        """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+; ')
 
298
        f = file(os.devnull, "wb")
 
299
        self.addCleanup(f.close)
 
300
        self.set_vendor(None, f)
 
301
        t = _mod_transport.get_transport_from_url(self.bogus_url)
 
302
        try:
 
303
            self.assertRaises(errors.ConnectionError, t.get, 'foobar')
 
304
        except NameError, e:
 
305
            if "global name 'SSHException'" in str(e):
 
306
                self.knownFailure('Known NameError bug in paramiko 1.6.1')
 
307
            raise
321
308
 
322
309
 
323
310
class SFTPLatencyKnob(TestCaseWithSFTPServer):
488
475
            conf._get_config().update(
489
476
                {'sftptest': {'scheme': 'ssh', 'port': port, 'user': 'bar'}})
490
477
            conf._save()
491
 
        t = _mod_transport.get_transport('sftp://localhost:%d' % port)
 
478
        t = _mod_transport.get_transport_from_url(
 
479
            'sftp://localhost:%d' % port)
492
480
        # force a connection to be performed.
493
481
        t.has('foo')
494
482
        return t