← 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 integration.

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