← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp.py

Merge from mbp.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp.py
17
17
import os
18
18
import socket
19
19
import threading
20
 
import unittest
21
20
 
22
 
from bzrlib.tests import TestCaseInTempDir
 
21
from bzrlib.tests import TestCaseInTempDir, TestCase
23
22
from bzrlib.tests.test_transport import TestTransportMixIn
24
23
 
25
24
try:
29
28
except ImportError:
30
29
    paramiko_loaded = False
31
30
 
 
31
# XXX: 20051124 jamesh
 
32
# The tests currently pop up a password prompt when an external ssh
 
33
# is used.  This forces the use of the paramiko implementation.
 
34
if paramiko_loaded:
 
35
    import bzrlib.transport.sftp
 
36
    bzrlib.transport.sftp._ssh_vendor = 'none'
 
37
 
32
38
 
33
39
STUB_SERVER_KEY = """
34
40
-----BEGIN RSA PRIVATE KEY-----
68
74
    
69
75
    def stop(self):
70
76
        self.stop_event.set()
 
77
        # We should consider waiting for the other thread
 
78
        # to stop, because otherwise we get spurious
 
79
        #   bzr: ERROR: Socket exception: Connection reset by peer (54)
 
80
        # because the test suite finishes before the thread has a chance
 
81
        # to close. (Especially when only running a few tests)
71
82
        
72
83
        
73
84
class TestCaseWithSFTPServer (TestCaseInTempDir):
82
93
        file(key_file, 'w').write(STUB_SERVER_KEY)
83
94
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
84
95
        ssh_server.add_server_key(host_key)
85
 
        server = StubServer()
 
96
        server = StubServer(self)
86
97
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer, StubSFTPServer, root=self._root)
87
98
        event = threading.Event()
88
99
        ssh_server.start_server(event, server)
92
103
    def setUp(self):
93
104
        TestCaseInTempDir.setUp(self)
94
105
        self._root = self.test_dir
 
106
        self._is_setup = False
95
107
 
96
108
    def delayed_setup(self):
97
109
        # some tests are just stubs that call setUp and then immediately call
98
110
        # tearDwon.  so don't create the port listener until get_transport is
99
111
        # called and we know we're in an actual test.
 
112
        if self._is_setup:
 
113
            return
100
114
        self._listener = SingleListener(self._run_server)
101
115
        self._listener.setDaemon(True)
102
116
        self._listener.start()        
103
117
        self._sftp_url = 'sftp://foo:bar@localhost:%d/' % (self._listener.port,)
 
118
        self._is_setup = True
104
119
        
105
120
    def tearDown(self):
106
121
        try:
112
127
        
113
128
class SFTPTransportTest (TestCaseWithSFTPServer, TestTransportMixIn):
114
129
    readonly = False
115
 
    setup = True
 
130
 
 
131
    def setUp(self):
 
132
        TestCaseWithSFTPServer.setUp(self)
 
133
        self.sftplogs = []
 
134
 
 
135
    def log(self, *args):
 
136
        """Override the default log to grab sftp server messages"""
 
137
        TestCaseWithSFTPServer.log(self, *args)
 
138
        if args and args[0].startswith('sftpserver'):
 
139
            self.sftplogs.append(args[0])
116
140
 
117
141
    def get_transport(self):
118
 
        if self.setup:
119
 
            self.delayed_setup()
120
 
            self.setup = False
 
142
        self.delayed_setup()
121
143
        from bzrlib.transport.sftp import SFTPTransport
122
144
        url = self._sftp_url
123
145
        return SFTPTransport(url)
147
169
        l.unlock()
148
170
        l2.unlock()
149
171
 
 
172
    def test_multiple_connections(self):
 
173
        t = self.get_transport()
 
174
        self.assertEquals(self.sftplogs, 
 
175
                ['sftpserver - authorizing: foo'
 
176
               , 'sftpserver - channel request: session, 1'])
 
177
        self.sftplogs = []
 
178
        # The second request should reuse the first connection
 
179
        # SingleListener only allows for a single connection,
 
180
        # So the next line fails unless the connection is reused
 
181
        t2 = self.get_transport()
 
182
        self.assertEquals(self.sftplogs, [])
 
183
 
150
184
 
151
185
class FakeSFTPTransport (object):
152
186
    _sftp = object()
153
187
fake = FakeSFTPTransport()
154
188
 
155
189
 
156
 
class SFTPNonServerTest (unittest.TestCase):
 
190
class SFTPNonServerTest(TestCase):
157
191
    def test_parse_url(self):
158
192
        from bzrlib.transport.sftp import SFTPTransport
159
193
        s = SFTPTransport('sftp://simple.example.com/%2fhome/source', clone_from=fake)
160
194
        self.assertEquals(s._host, 'simple.example.com')
161
 
        self.assertEquals(s._port, 22)
 
195
        self.assertEquals(s._port, None)
162
196
        self.assertEquals(s._path, '/home/source')
163
 
        self.assert_(s._password is None)
 
197
        self.failUnless(s._password is None)
 
198
 
 
199
        self.assertEquals(s.base, 'sftp://simple.example.com/%2Fhome/source')
164
200
        
165
201
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/relative', clone_from=fake)
166
202
        self.assertEquals(s._host, 'example.com')
168
204
        self.assertEquals(s._username, 'robey')
169
205
        self.assertEquals(s._password, 'h@t')
170
206
        self.assertEquals(s._path, 'relative')
171
 
        
 
207
 
 
208
        # Base should not keep track of the password
 
209
        self.assertEquals(s.base, 'sftp://robey@example.com:2222/relative')
 
210
 
 
211
        # Double slash should be accepted instead of using %2F
 
212
        s = SFTPTransport('sftp://user@example.com:22//absolute/path', clone_from=fake)
 
213
        self.assertEquals(s._host, 'example.com')
 
214
        self.assertEquals(s._port, 22)
 
215
        self.assertEquals(s._username, 'user')
 
216
        self.assertEquals(s._password, None)
 
217
        self.assertEquals(s._path, '/absolute/path')
 
218
 
 
219
        # Also, don't show the port if it is the default 22
 
220
        self.assertEquals(s.base, 'sftp://user@example.com:22/%2Fabsolute/path')
 
221
 
 
222
    def test_relpath(self):
 
223
        from bzrlib.transport.sftp import SFTPTransport
 
224
        from bzrlib.errors import NonRelativePath
 
225
 
 
226
        s = SFTPTransport('sftp://user@host.com//abs/path', clone_from=fake)
 
227
        self.assertEquals(s.relpath('sftp://user@host.com//abs/path/sub'), 'sub')
 
228
        # Can't test this one, because we actually get an AssertionError
 
229
        # TODO: Consider raising an exception rather than an assert
 
230
        #self.assertRaises(NonRelativePath, s.relpath, 'http://user@host.com//abs/path/sub')
 
231
        self.assertRaises(NonRelativePath, s.relpath, 'sftp://user2@host.com//abs/path/sub')
 
232
        self.assertRaises(NonRelativePath, s.relpath, 'sftp://user@otherhost.com//abs/path/sub')
 
233
        self.assertRaises(NonRelativePath, s.relpath, 'sftp://user@host.com:33//abs/path/sub')
 
234
        self.assertRaises(NonRelativePath, s.relpath, 'sftp://user@host.com/abs/path/sub')
 
235
 
 
236
        # Make sure it works when we don't supply a username
 
237
        s = SFTPTransport('sftp://host.com//abs/path', clone_from=fake)
 
238
        self.assertEquals(s.relpath('sftp://host.com//abs/path/sub'), 'sub')
 
239
 
 
240
        # Make sure it works when parts of the path will be url encoded
 
241
        # TODO: These may be incorrect, we might need to urllib.urlencode() before
 
242
        # we pass the paths into the SFTPTransport constructor
 
243
        s = SFTPTransport('sftp://host.com/dev/,path', clone_from=fake)
 
244
        self.assertEquals(s.relpath('sftp://host.com/dev/,path/sub'), 'sub')
 
245
        s = SFTPTransport('sftp://host.com/dev/%path', clone_from=fake)
 
246
        self.assertEquals(s.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
247
 
 
248
    def test_parse_invalid_url(self):
 
249
        from bzrlib.transport.sftp import SFTPTransport, SFTPTransportError
 
250
        try:
 
251
            s = SFTPTransport('sftp://lilypond.org:~janneke/public_html/bzr/gub',
 
252
                              clone_from=fake)
 
253
            self.fail('expected exception not raised')
 
254
        except SFTPTransportError, e:
 
255
            self.assertEquals(str(e), 
 
256
                    '~janneke: invalid port number')
 
257
 
172
258
 
173
259
class SFTPBranchTest(TestCaseWithSFTPServer):
174
260
    """Test some stuff when accessing a bzr Branch over sftp"""