← 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: Robert Collins
  • Date: 2007-07-25 00:52:21 UTC
  • mfrom: (2650 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2651.
  • Revision ID: robertc@robertcollins.net-20070725005221-0ysm6il5mqnme3wz
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
21
21
import threading
22
22
import time
23
23
 
24
 
import bzrlib.bzrdir as bzrdir
25
 
import bzrlib.errors as errors
26
 
from bzrlib.osutils import pathjoin, lexists, set_or_unset_env
27
 
from bzrlib.tests import TestCaseWithTransport, TestCase, TestSkipped
 
24
from bzrlib import (
 
25
    bzrdir,
 
26
    errors,
 
27
    )
 
28
from bzrlib.osutils import (
 
29
    pathjoin,
 
30
    lexists,
 
31
    set_or_unset_env,
 
32
    )
 
33
from bzrlib.tests import (
 
34
    TestCaseWithTransport,
 
35
    TestCase,
 
36
    TestSkipped,
 
37
    )
28
38
from bzrlib.tests.HttpServer import HttpServer
29
 
import bzrlib.transport
30
39
from bzrlib.transport import get_transport
31
40
import bzrlib.transport.http
 
41
from bzrlib.transport.sftp import (
 
42
    SFTPAbsoluteServer,
 
43
    SFTPHomeDirServer,
 
44
    SFTPTransport,
 
45
    )
32
46
from bzrlib.workingtree import WorkingTree
33
47
 
34
48
try:
40
54
 
41
55
def set_test_transport_to_sftp(testcase):
42
56
    """A helper to set transports on test case instances."""
43
 
    from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
44
57
    if getattr(testcase, '_get_remote_is_absolute', None) is None:
45
58
        testcase._get_remote_is_absolute = True
46
59
    if testcase._get_remote_is_absolute:
87
100
        l.unlock()
88
101
        l2.unlock()
89
102
 
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
103
 
101
104
class SFTPTransportTestRelative(TestCaseWithSFTPServer):
102
105
    """Test the SFTP transport with homedir based relative paths."""
126
129
    """Test the SFTP transport with homedir based relative paths."""
127
130
 
128
131
    def setUp(self):
 
132
        # Only SFTPHomeDirServer is tested here
129
133
        self._get_remote_is_absolute = False
130
134
        super(SFTPTransportTestRelativeRoot, self).setUp()
131
135
 
132
136
    def test__remote_path_relative_root(self):
133
137
        # relative paths are preserved
134
138
        t = self.get_transport('')
135
 
        # the remote path should be ''
136
 
        self.assertEqual('', t._path)
 
139
        self.assertEqual('/~/', t._path)
 
140
        # the remote path should be relative to home dir
 
141
        # (i.e. not begining with a '/')
137
142
        self.assertEqual('a', t._remote_path('a'))
138
143
 
139
144
 
140
 
class FakeSFTPTransport (object):
141
 
    _sftp = object()
142
 
fake = FakeSFTPTransport()
143
 
 
144
 
 
145
145
class SFTPNonServerTest(TestCase):
146
146
    def setUp(self):
147
147
        TestCase.setUp(self)
148
148
        if not paramiko_loaded:
149
149
            raise TestSkipped('you must have paramiko to run this test')
150
150
 
151
 
    def test_parse_url(self):
152
 
        from bzrlib.transport.sftp import SFTPTransport
153
 
        s = SFTPTransport('sftp://simple.example.com/home/source', clone_from=fake)
154
 
        self.assertEquals(s._host, 'simple.example.com')
155
 
        self.assertEquals(s._port, None)
156
 
        self.assertEquals(s._path, '/home/source')
157
 
        self.failUnless(s._password is None)
158
 
 
159
 
        self.assertEquals(s.base, 'sftp://simple.example.com/home/source/')
160
 
 
161
 
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative', clone_from=fake)
 
151
    def test_parse_url_with_home_dir(self):
 
152
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative')
162
153
        self.assertEquals(s._host, 'example.com')
163
154
        self.assertEquals(s._port, 2222)
164
 
        self.assertEquals(s._username, 'robey')
 
155
        self.assertEquals(s._user, 'robey')
165
156
        self.assertEquals(s._password, 'h@t')
166
 
        self.assertEquals(s._path, 'relative')
167
 
 
168
 
        # Base should not keep track of the password
169
 
        self.assertEquals(s.base, 'sftp://robey@example.com:2222/~/relative/')
 
157
        self.assertEquals(s._path, '/~/relative/')
170
158
 
171
159
    def test_relpath(self):
172
 
        from bzrlib.transport.sftp import SFTPTransport
173
 
        from bzrlib.errors import PathNotChild
174
 
 
175
 
        s = SFTPTransport('sftp://user@host.com/abs/path', clone_from=fake)
176
 
        self.assertEquals(s.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
177
 
        # Can't test this one, because we actually get an AssertionError
178
 
        # TODO: Consider raising an exception rather than an assert
179
 
        #self.assertRaises(PathNotChild, s.relpath, 'http://user@host.com/abs/path/sub')
180
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user2@host.com/abs/path/sub')
181
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@otherhost.com/abs/path/sub')
182
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com:33/abs/path/sub')
183
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com/~/rel/path/sub')
184
 
 
185
 
        # Make sure it works when we don't supply a username
186
 
        s = SFTPTransport('sftp://host.com/abs/path', clone_from=fake)
187
 
        self.assertEquals(s.relpath('sftp://host.com/abs/path/sub'), 'sub')
188
 
 
189
 
        # Make sure it works when parts of the path will be url encoded
190
 
        # TODO: These may be incorrect, we might need to urllib.urlencode() before
191
 
        # we pass the paths into the SFTPTransport constructor
192
 
        s = SFTPTransport('sftp://host.com/dev/,path', clone_from=fake)
193
 
        self.assertEquals(s.relpath('sftp://host.com/dev/,path/sub'), 'sub')
194
 
        s = SFTPTransport('sftp://host.com/dev/%path', clone_from=fake)
195
 
        self.assertEquals(s.relpath('sftp://host.com/dev/%path/sub'), 'sub')
196
 
 
197
 
    def test_parse_invalid_url(self):
198
 
        from bzrlib.transport.sftp import SFTPTransport, TransportError
199
 
        try:
200
 
            s = SFTPTransport('sftp://lilypond.org:~janneke/public_html/bzr/gub',
201
 
                              clone_from=fake)
202
 
            self.fail('expected exception not raised')
203
 
        except TransportError, e:
204
 
            self.assertEquals(str(e),
205
 
                    'Transport error: '
206
 
                    'invalid port number ~janneke in url:\n'
207
 
                    'sftp://lilypond.org:~janneke/public_html/bzr/gub ')
 
160
        s = SFTPTransport('sftp://user@host.com/abs/path')
 
161
        self.assertRaises(errors.PathNotChild, s.relpath,
 
162
                          'sftp://user@host.com/~/rel/path/sub')
208
163
 
209
164
    def test_get_paramiko_vendor(self):
210
165
        """Test that if no 'ssh' is available we get builtin paramiko"""
357
312
        """Test that a real connection attempt raises the right error"""
358
313
        from bzrlib.transport import ssh
359
314
        self.set_vendor(ssh.ParamikoVendor())
360
 
        self.assertRaises(errors.ConnectionError,
361
 
                          bzrlib.transport.get_transport, self.bogus_url)
 
315
        t = bzrlib.transport.get_transport(self.bogus_url)
 
316
        self.assertRaises(errors.ConnectionError, t.get, 'foobar')
362
317
 
363
318
    def test_bad_connection_ssh(self):
364
319
        """None => auto-detect vendor"""
399
354
        start_time = time.time()
400
355
        self.get_server().add_latency = 0.5
401
356
        transport = self.get_transport()
 
357
        transport.has('not me') # Force connection by issuing a request
402
358
        with_latency_knob_time = time.time() - start_time
403
359
        self.assertTrue(with_latency_knob_time > 0.4)
404
360
 
405
361
    def test_default(self):
406
362
        # This test is potentially brittle: under extremely high machine load
407
363
        # it could fail, but that is quite unlikely
 
364
        raise TestSkipped('Timing-sensitive test')
408
365
        start_time = time.time()
409
366
        transport = self.get_transport()
 
367
        transport.has('not me') # Force connection by issuing a request
410
368
        regular_time = time.time() - start_time
411
369
        self.assertTrue(regular_time < 0.5)
412
370