← 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] John, sftp and others

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:
32
31
# XXX: 20051124 jamesh
33
32
# The tests currently pop up a password prompt when an external ssh
34
33
# is used.  This forces the use of the paramiko implementation.
35
 
import bzrlib.transport.sftp
36
 
bzrlib.transport.sftp._ssh_vendor = 'none'
 
34
if paramiko_loaded:
 
35
    import bzrlib.transport.sftp
 
36
    bzrlib.transport.sftp._ssh_vendor = 'none'
37
37
 
38
38
 
39
39
STUB_SERVER_KEY = """
74
74
    
75
75
    def stop(self):
76
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)
77
82
        
78
83
        
79
84
class TestCaseWithSFTPServer (TestCaseInTempDir):
88
93
        file(key_file, 'w').write(STUB_SERVER_KEY)
89
94
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
90
95
        ssh_server.add_server_key(host_key)
91
 
        server = StubServer()
 
96
        server = StubServer(self)
92
97
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer, StubSFTPServer, root=self._root)
93
98
        event = threading.Event()
94
99
        ssh_server.start_server(event, server)
98
103
    def setUp(self):
99
104
        TestCaseInTempDir.setUp(self)
100
105
        self._root = self.test_dir
 
106
        self._is_setup = False
101
107
 
102
108
    def delayed_setup(self):
103
109
        # some tests are just stubs that call setUp and then immediately call
104
110
        # tearDwon.  so don't create the port listener until get_transport is
105
111
        # called and we know we're in an actual test.
 
112
        if self._is_setup:
 
113
            return
106
114
        self._listener = SingleListener(self._run_server)
107
115
        self._listener.setDaemon(True)
108
116
        self._listener.start()        
109
117
        self._sftp_url = 'sftp://foo:bar@localhost:%d/' % (self._listener.port,)
 
118
        self._is_setup = True
110
119
        
111
120
    def tearDown(self):
112
121
        try:
118
127
        
119
128
class SFTPTransportTest (TestCaseWithSFTPServer, TestTransportMixIn):
120
129
    readonly = False
121
 
    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])
122
140
 
123
141
    def get_transport(self):
124
 
        if self.setup:
125
 
            self.delayed_setup()
126
 
            self.setup = False
 
142
        self.delayed_setup()
127
143
        from bzrlib.transport.sftp import SFTPTransport
128
144
        url = self._sftp_url
129
145
        return SFTPTransport(url)
153
169
        l.unlock()
154
170
        l2.unlock()
155
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
 
156
184
 
157
185
class FakeSFTPTransport (object):
158
186
    _sftp = object()
159
187
fake = FakeSFTPTransport()
160
188
 
161
189
 
162
 
class SFTPNonServerTest(unittest.TestCase):
 
190
class SFTPNonServerTest(TestCase):
163
191
    def test_parse_url(self):
164
192
        from bzrlib.transport.sftp import SFTPTransport
165
193
        s = SFTPTransport('sftp://simple.example.com/%2fhome/source', clone_from=fake)
166
194
        self.assertEquals(s._host, 'simple.example.com')
167
195
        self.assertEquals(s._port, None)
168
196
        self.assertEquals(s._path, '/home/source')
169
 
        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')
170
200
        
171
201
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/relative', clone_from=fake)
172
202
        self.assertEquals(s._host, 'example.com')
175
205
        self.assertEquals(s._password, 'h@t')
176
206
        self.assertEquals(s._path, 'relative')
177
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
 
178
248
    def test_parse_invalid_url(self):
179
249
        from bzrlib.transport.sftp import SFTPTransport, SFTPTransportError
180
250
        try:
185
255
            self.assertEquals(str(e), 
186
256
                    '~janneke: invalid port number')
187
257
 
188
 
        
189
258
 
190
259
class SFTPBranchTest(TestCaseWithSFTPServer):
191
260
    """Test some stuff when accessing a bzr Branch over sftp"""