/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

First attempt to merge .dev and resolve the conflicts (but tests are 
failing)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005 Robey Pointer <robey@lag.net>, Canonical Ltd
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>
 
2
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
2
3
#
3
4
# This program is free software; you can redistribute it and/or modify
4
5
# it under the terms of the GNU General Public License as published by
16
17
 
17
18
import os
18
19
import socket
 
20
import sys
19
21
import threading
20
22
import time
21
23
 
22
 
import bzrlib.bzrdir as bzrdir
23
 
import bzrlib.errors as errors
24
 
from bzrlib.osutils import pathjoin, lexists
25
 
from bzrlib.tests import TestCaseWithTransport, TestCase, TestSkipped
26
 
import bzrlib.transport
27
 
import bzrlib.transport.http
28
 
from bzrlib.workingtree import WorkingTree
29
 
 
30
24
try:
31
25
    import paramiko
32
26
    paramiko_loaded = True
33
27
except ImportError:
34
28
    paramiko_loaded = False
35
29
 
 
30
from bzrlib import (
 
31
    bzrdir,
 
32
    errors,
 
33
    )
 
34
from bzrlib.osutils import (
 
35
    pathjoin,
 
36
    lexists,
 
37
    set_or_unset_env,
 
38
    )
 
39
from bzrlib.tests import (
 
40
    TestCaseWithTransport,
 
41
    TestCase,
 
42
    TestSkipped,
 
43
    )
 
44
from bzrlib.tests.http_server import HttpServer
 
45
from bzrlib.transport import get_transport
 
46
import bzrlib.transport.http
 
47
 
 
48
if paramiko_loaded:
 
49
    from bzrlib.transport.sftp import (
 
50
        SFTPAbsoluteServer,
 
51
        SFTPHomeDirServer,
 
52
        SFTPTransport,
 
53
        )
 
54
 
 
55
from bzrlib.workingtree import WorkingTree
 
56
 
36
57
 
37
58
def set_test_transport_to_sftp(testcase):
38
59
    """A helper to set transports on test case instances."""
39
 
    from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
40
60
    if getattr(testcase, '_get_remote_is_absolute', None) is None:
41
61
        testcase._get_remote_is_absolute = True
42
62
    if testcase._get_remote_is_absolute:
43
63
        testcase.transport_server = SFTPAbsoluteServer
44
64
    else:
45
65
        testcase.transport_server = SFTPHomeDirServer
46
 
    testcase.transport_readonly_server = bzrlib.transport.http.HttpServer
 
66
    testcase.transport_readonly_server = HttpServer
47
67
 
48
68
 
49
69
class TestCaseWithSFTPServer(TestCaseWithTransport):
53
73
        super(TestCaseWithSFTPServer, self).setUp()
54
74
        if not paramiko_loaded:
55
75
            raise TestSkipped('you must have paramiko to run this test')
56
 
        set_test_transport_to_sftp(self) 
57
 
 
58
 
    def get_transport(self, path=None):
59
 
        """Return a transport relative to self._test_root."""
60
 
        return bzrlib.transport.get_transport(self.get_url(path))
 
76
        set_test_transport_to_sftp(self)
61
77
 
62
78
 
63
79
class SFTPLockTests (TestCaseWithSFTPServer):
87
103
        l.unlock()
88
104
        l2.unlock()
89
105
 
90
 
    def test_multiple_connections(self):
91
 
        t = self.get_transport()
92
 
        self.assertTrue('sftpserver - new connection' in self.get_server().logs)
93
 
        self.get_server().logs = []
94
 
        # The second request should reuse the first connection
95
 
        # SingleListener only allows for a single connection,
96
 
        # So the next line fails unless the connection is reused
97
 
        t2 = self.get_transport()
98
 
        self.assertEquals(self.get_server().logs, [])
99
 
 
100
106
 
101
107
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
102
108
    """Test the SFTP transport with homedir based relative paths."""
103
109
 
104
110
    def test__remote_path(self):
 
111
        if sys.platform == 'darwin':
 
112
            # This test is about sftp absolute path handling. There is already
 
113
            # (in this test) a TODO about windows needing an absolute path
 
114
            # without drive letter. To me, using self.test_dir is a trick to
 
115
            # get an absolute path for comparison purposes.  That fails for OSX
 
116
            # because the sftp server doesn't resolve the links (and it doesn't
 
117
            # have to). --vila 20070924
 
118
            self.knownFailure('Mac OSX symlinks /tmp to /private/tmp,'
 
119
                              ' testing against self.test_dir'
 
120
                              ' is not appropriate')
105
121
        t = self.get_transport()
 
122
        # This test require unix-like absolute path
 
123
        test_dir = self.test_dir
 
124
        if sys.platform == 'win32':
 
125
            # using hack suggested by John Meinel.
 
126
            # TODO: write another mock server for this test
 
127
            #       and use absolute path without drive letter
 
128
            test_dir = '/' + test_dir
106
129
        # try what is currently used:
107
130
        # remote path = self._abspath(relpath)
108
 
        self.assertEqual(self.test_dir + '/relative', t._remote_path('relative'))
 
131
        self.assertIsSameRealPath(test_dir + '/relative',
 
132
                                  t._remote_path('relative'))
109
133
        # we dont os.path.join because windows gives us the wrong path
110
 
        root_segments = self.test_dir.split('/')
 
134
        root_segments = test_dir.split('/')
111
135
        root_parent = '/'.join(root_segments[:-1])
112
136
        # .. should be honoured
113
 
        self.assertEqual(root_parent + '/sibling', t._remote_path('../sibling'))
 
137
        self.assertIsSameRealPath(root_parent + '/sibling',
 
138
                                  t._remote_path('../sibling'))
114
139
        # /  should be illegal ?
115
140
        ### FIXME decide and then test for all transports. RBC20051208
116
141
 
117
142
 
118
 
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
 
143
class SFTPTransportTestRelativeRoot(TestCaseWithSFTPServer):
119
144
    """Test the SFTP transport with homedir based relative paths."""
120
145
 
121
146
    def setUp(self):
 
147
        # Only SFTPHomeDirServer is tested here
122
148
        self._get_remote_is_absolute = False
123
 
        super(SFTPTransportTestRelative, self).setUp()
 
149
        super(SFTPTransportTestRelativeRoot, self).setUp()
124
150
 
125
151
    def test__remote_path_relative_root(self):
126
152
        # relative paths are preserved
127
153
        t = self.get_transport('')
 
154
        self.assertEqual('/~/', t._path)
 
155
        # the remote path should be relative to home dir
 
156
        # (i.e. not begining with a '/')
128
157
        self.assertEqual('a', t._remote_path('a'))
129
158
 
130
159
 
131
 
class FakeSFTPTransport (object):
132
 
    _sftp = object()
133
 
fake = FakeSFTPTransport()
134
 
 
135
 
 
136
160
class SFTPNonServerTest(TestCase):
137
161
    def setUp(self):
138
162
        TestCase.setUp(self)
139
163
        if not paramiko_loaded:
140
164
            raise TestSkipped('you must have paramiko to run this test')
141
165
 
142
 
    def test_parse_url(self):
143
 
        from bzrlib.transport.sftp import SFTPTransport
144
 
        s = SFTPTransport('sftp://simple.example.com/home/source', clone_from=fake)
145
 
        self.assertEquals(s._host, 'simple.example.com')
146
 
        self.assertEquals(s._port, None)
147
 
        self.assertEquals(s._path, '/home/source')
148
 
        self.failUnless(s._password is None)
149
 
 
150
 
        self.assertEquals(s.base, 'sftp://simple.example.com/home/source/')
151
 
 
152
 
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative', clone_from=fake)
 
166
    def test_parse_url_with_home_dir(self):
 
167
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative')
153
168
        self.assertEquals(s._host, 'example.com')
154
169
        self.assertEquals(s._port, 2222)
155
 
        self.assertEquals(s._username, 'robey')
 
170
        self.assertEquals(s._user, 'robey')
156
171
        self.assertEquals(s._password, 'h@t')
157
 
        self.assertEquals(s._path, 'relative')
158
 
 
159
 
        # Base should not keep track of the password
160
 
        self.assertEquals(s.base, 'sftp://robey@example.com:2222/~/relative/')
 
172
        self.assertEquals(s._path, '/~/relative/')
161
173
 
162
174
    def test_relpath(self):
163
 
        from bzrlib.transport.sftp import SFTPTransport
164
 
        from bzrlib.errors import PathNotChild
165
 
 
166
 
        s = SFTPTransport('sftp://user@host.com/abs/path', clone_from=fake)
167
 
        self.assertEquals(s.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
168
 
        # Can't test this one, because we actually get an AssertionError
169
 
        # TODO: Consider raising an exception rather than an assert
170
 
        #self.assertRaises(PathNotChild, s.relpath, 'http://user@host.com/abs/path/sub')
171
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user2@host.com/abs/path/sub')
172
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@otherhost.com/abs/path/sub')
173
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com:33/abs/path/sub')
174
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com/~/rel/path/sub')
175
 
 
176
 
        # Make sure it works when we don't supply a username
177
 
        s = SFTPTransport('sftp://host.com/abs/path', clone_from=fake)
178
 
        self.assertEquals(s.relpath('sftp://host.com/abs/path/sub'), 'sub')
179
 
 
180
 
        # Make sure it works when parts of the path will be url encoded
181
 
        # TODO: These may be incorrect, we might need to urllib.urlencode() before
182
 
        # we pass the paths into the SFTPTransport constructor
183
 
        s = SFTPTransport('sftp://host.com/dev/,path', clone_from=fake)
184
 
        self.assertEquals(s.relpath('sftp://host.com/dev/,path/sub'), 'sub')
185
 
        s = SFTPTransport('sftp://host.com/dev/%path', clone_from=fake)
186
 
        self.assertEquals(s.relpath('sftp://host.com/dev/%path/sub'), 'sub')
187
 
 
188
 
    def test_parse_invalid_url(self):
189
 
        from bzrlib.transport.sftp import SFTPTransport, TransportError
190
 
        try:
191
 
            s = SFTPTransport('sftp://lilypond.org:~janneke/public_html/bzr/gub',
192
 
                              clone_from=fake)
193
 
            self.fail('expected exception not raised')
194
 
        except TransportError, e:
195
 
            self.assertEquals(str(e), 
196
 
                    'Transport error: ~janneke: invalid port number ')
 
175
        s = SFTPTransport('sftp://user@host.com/abs/path')
 
176
        self.assertRaises(errors.PathNotChild, s.relpath,
 
177
                          'sftp://user@host.com/~/rel/path/sub')
 
178
 
 
179
    def test_get_paramiko_vendor(self):
 
180
        """Test that if no 'ssh' is available we get builtin paramiko"""
 
181
        from bzrlib.transport import ssh
 
182
        # set '.' as the only location in the path, forcing no 'ssh' to exist
 
183
        orig_vendor = ssh._ssh_vendor_manager._cached_ssh_vendor
 
184
        orig_path = set_or_unset_env('PATH', '.')
 
185
        try:
 
186
            # No vendor defined yet, query for one
 
187
            ssh._ssh_vendor_manager.clear_cache()
 
188
            vendor = ssh._get_ssh_vendor()
 
189
            self.assertIsInstance(vendor, ssh.ParamikoVendor)
 
190
        finally:
 
191
            set_or_unset_env('PATH', orig_path)
 
192
            ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
 
193
 
 
194
    def test_abspath_root_sibling_server(self):
 
195
        from bzrlib.transport.sftp import SFTPSiblingAbsoluteServer
 
196
        server = SFTPSiblingAbsoluteServer()
 
197
        server.setUp()
 
198
        try:
 
199
            transport = get_transport(server.get_url())
 
200
            self.assertFalse(transport.abspath('/').endswith('/~/'))
 
201
            self.assertTrue(transport.abspath('/').endswith('/'))
 
202
            del transport
 
203
        finally:
 
204
            server.tearDown()
197
205
 
198
206
 
199
207
class SFTPBranchTest(TestCaseWithSFTPServer):
260
268
            server._vendor = self._test_vendor
261
269
            return server
262
270
        self._test_vendor = 'loopback'
263
 
        self.transport_server = create_server
 
271
        self.vfs_transport_server = create_server
264
272
        f = open('a_file', 'wb')
265
273
        try:
266
274
            f.write('foobar\n')
305
313
        s.bind(('localhost', 0))
306
314
        self.bogus_url = 'sftp://%s:%s/' % s.getsockname()
307
315
 
308
 
        orig_vendor = bzrlib.transport.ssh._ssh_vendor
 
316
        orig_vendor = bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor
309
317
        def reset():
310
 
            bzrlib.transport.ssh._ssh_vendor = orig_vendor
 
318
            bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor = orig_vendor
311
319
            s.close()
312
320
        self.addCleanup(reset)
313
321
 
314
322
    def set_vendor(self, vendor):
315
323
        import bzrlib.transport.ssh
316
 
        bzrlib.transport.ssh._ssh_vendor = vendor
 
324
        bzrlib.transport.ssh._ssh_vendor_manager._cached_ssh_vendor = vendor
317
325
 
318
326
    def test_bad_connection_paramiko(self):
319
327
        """Test that a real connection attempt raises the right error"""
320
328
        from bzrlib.transport import ssh
321
329
        self.set_vendor(ssh.ParamikoVendor())
322
 
        self.assertRaises(errors.ConnectionError,
323
 
                          bzrlib.transport.get_transport, self.bogus_url)
 
330
        t = bzrlib.transport.get_transport(self.bogus_url)
 
331
        self.assertRaises(errors.ConnectionError, t.get, 'foobar')
324
332
 
325
333
    def test_bad_connection_ssh(self):
326
334
        """None => auto-detect vendor"""
341
349
        # else:
342
350
        #     self.fail('Excepted ConnectionError to be raised')
343
351
 
344
 
        out, err = self.run_bzr_subprocess('log', self.bogus_url, retcode=3)
 
352
        out, err = self.run_bzr_subprocess(['log', self.bogus_url], retcode=3)
345
353
        self.assertEqual('', out)
346
354
        if "NameError: global name 'SSHException'" in err:
347
355
            # We aren't fixing this bug, because it is a bug in
348
356
            # paramiko, but we know about it, so we don't have to
349
357
            # fail the test
350
358
            raise TestSkipped('Known NameError bug with paramiko-1.6.1')
351
 
        self.assertContainsRe(err, 'Connection error')
 
359
        self.assertContainsRe(err, r'bzr: ERROR: Unable to connect to SSH host'
 
360
                                   r' 127\.0\.0\.1:\d+; ')
352
361
 
353
362
 
354
363
class SFTPLatencyKnob(TestCaseWithSFTPServer):
360
369
        start_time = time.time()
361
370
        self.get_server().add_latency = 0.5
362
371
        transport = self.get_transport()
 
372
        transport.has('not me') # Force connection by issuing a request
363
373
        with_latency_knob_time = time.time() - start_time
364
374
        self.assertTrue(with_latency_knob_time > 0.4)
365
375
 
366
376
    def test_default(self):
367
377
        # This test is potentially brittle: under extremely high machine load
368
378
        # it could fail, but that is quite unlikely
 
379
        raise TestSkipped('Timing-sensitive test')
369
380
        start_time = time.time()
370
381
        transport = self.get_transport()
 
382
        transport.has('not me') # Force connection by issuing a request
371
383
        regular_time = time.time() - start_time
372
384
        self.assertTrue(regular_time < 0.5)
373
385
 
403
415
 
404
416
    def setUp(self):
405
417
        TestCase.setUp(self)
 
418
        if not paramiko_loaded:
 
419
            raise TestSkipped('you must have paramiko to run this test')
406
420
 
407
421
    def test_delay(self):
408
422
        from bzrlib.transport.sftp import SocketDelay