← 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] sftp fixes

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:
74
73
    
75
74
    def stop(self):
76
75
        self.stop_event.set()
 
76
        # We should consider waiting for the other thread
 
77
        # to stop, because otherwise we get spurious
 
78
        #   bzr: ERROR: Socket exception: Connection reset by peer (54)
 
79
        # because the test suite finishes before the thread has a chance
 
80
        # to close. (Especially when only running a few tests)
77
81
        
78
82
        
79
83
class TestCaseWithSFTPServer (TestCaseInTempDir):
88
92
        file(key_file, 'w').write(STUB_SERVER_KEY)
89
93
        host_key = paramiko.RSAKey.from_private_key_file(key_file)
90
94
        ssh_server.add_server_key(host_key)
91
 
        server = StubServer()
 
95
        server = StubServer(self)
92
96
        ssh_server.set_subsystem_handler('sftp', paramiko.SFTPServer, StubSFTPServer, root=self._root)
93
97
        event = threading.Event()
94
98
        ssh_server.start_server(event, server)
98
102
    def setUp(self):
99
103
        TestCaseInTempDir.setUp(self)
100
104
        self._root = self.test_dir
 
105
        self._is_setup = False
101
106
 
102
107
    def delayed_setup(self):
103
108
        # some tests are just stubs that call setUp and then immediately call
104
109
        # tearDwon.  so don't create the port listener until get_transport is
105
110
        # called and we know we're in an actual test.
 
111
        if self._is_setup:
 
112
            return
106
113
        self._listener = SingleListener(self._run_server)
107
114
        self._listener.setDaemon(True)
108
115
        self._listener.start()        
109
116
        self._sftp_url = 'sftp://foo:bar@localhost:%d/' % (self._listener.port,)
 
117
        self._is_setup = True
110
118
        
111
119
    def tearDown(self):
112
120
        try:
118
126
        
119
127
class SFTPTransportTest (TestCaseWithSFTPServer, TestTransportMixIn):
120
128
    readonly = False
121
 
    setup = True
 
129
 
 
130
    def setUp(self):
 
131
        TestCaseWithSFTPServer.setUp(self)
 
132
        self.sftplogs = []
 
133
 
 
134
    def log(self, *args):
 
135
        """Override the default log to grab sftp server messages"""
 
136
        TestCaseWithSFTPServer.log(self, *args)
 
137
        if args and args[0].startswith('sftpserver'):
 
138
            self.sftplogs.append(args[0])
122
139
 
123
140
    def get_transport(self):
124
 
        if self.setup:
125
 
            self.delayed_setup()
126
 
            self.setup = False
 
141
        self.delayed_setup()
127
142
        from bzrlib.transport.sftp import SFTPTransport
128
143
        url = self._sftp_url
129
144
        return SFTPTransport(url)
153
168
        l.unlock()
154
169
        l2.unlock()
155
170
 
 
171
    def test_multiple_connections(self):
 
172
        t = self.get_transport()
 
173
        self.assertEquals(self.sftplogs, 
 
174
                ['sftpserver - authorizing: foo'
 
175
               , 'sftpserver - channel request: session, 1'])
 
176
        self.sftplogs = []
 
177
        # The second request should reuse the first connection
 
178
        # SingleListener only allows for a single connection,
 
179
        # So the next line fails unless the connection is reused
 
180
        t2 = self.get_transport()
 
181
        self.assertEquals(self.sftplogs, [])
 
182
 
156
183
 
157
184
class FakeSFTPTransport (object):
158
185
    _sftp = object()
159
186
fake = FakeSFTPTransport()
160
187
 
161
188
 
162
 
class SFTPNonServerTest(unittest.TestCase):
 
189
class SFTPNonServerTest(TestCase):
163
190
    def test_parse_url(self):
164
191
        from bzrlib.transport.sftp import SFTPTransport
165
192
        s = SFTPTransport('sftp://simple.example.com/%2fhome/source', clone_from=fake)
166
193
        self.assertEquals(s._host, 'simple.example.com')
167
 
        self.assertEquals(s._port, None)
 
194
        self.assertEquals(s._port, 22)
168
195
        self.assertEquals(s._path, '/home/source')
169
 
        self.assert_(s._password is None)
 
196
        self.failUnless(s._password is None)
 
197
 
 
198
        self.assertEquals(s.base, 'sftp://simple.example.com/%2Fhome/source')
170
199
        
171
200
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/relative', clone_from=fake)
172
201
        self.assertEquals(s._host, 'example.com')
175
204
        self.assertEquals(s._password, 'h@t')
176
205
        self.assertEquals(s._path, 'relative')
177
206
 
 
207
        # Base should not keep track of the password
 
208
        self.assertEquals(s.base, 'sftp://robey@example.com:2222/relative')
 
209
 
 
210
        # Double slash should be accepted instead of using %2F
 
211
        s = SFTPTransport('sftp://user@example.com:22//absolute/path', clone_from=fake)
 
212
        self.assertEquals(s._host, 'example.com')
 
213
        self.assertEquals(s._port, 22)
 
214
        self.assertEquals(s._username, 'user')
 
215
        self.assertEquals(s._password, None)
 
216
        self.assertEquals(s._path, '/absolute/path')
 
217
 
 
218
        # Also, don't show the port if it is the default 22
 
219
        self.assertEquals(s.base, 'sftp://user@example.com/%2Fabsolute/path')
 
220
 
178
221
    def test_parse_invalid_url(self):
179
222
        from bzrlib.transport.sftp import SFTPTransport, SFTPTransportError
180
223
        try:
185
228
            self.assertEquals(str(e), 
186
229
                    '~janneke: invalid port number')
187
230
 
188
 
        
189
231
 
190
232
class SFTPBranchTest(TestCaseWithSFTPServer):
191
233
    """Test some stuff when accessing a bzr Branch over sftp"""