~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

MergeĀ fromĀ mainine.

Show diffs side-by-side

added added

removed removed

Lines of Context:
38
38
        if not paramiko_loaded:
39
39
            raise TestSkipped('you must have paramiko to run this test')
40
40
        super(TestCaseWithSFTPServer, self).setUp()
41
 
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
42
 
        if getattr(self, '_get_remote_is_absolute', None) is None:
43
 
            self._get_remote_is_absolute = True
44
 
        if self._get_remote_is_absolute:
45
 
            self.server = SFTPAbsoluteServer()
 
41
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer, SFTPServer
 
42
        if getattr(self, '_full_handshake', False):
 
43
            self.server = SFTPServer()
46
44
        else:
47
 
            self.server = SFTPHomeDirServer()
 
45
            self._full_handshake = False
 
46
            if getattr(self, '_get_remote_is_absolute', None) is None:
 
47
                self._get_remote_is_absolute = True
 
48
            if self._get_remote_is_absolute:
 
49
                self.server = SFTPAbsoluteServer()
 
50
            else:
 
51
                self.server = SFTPHomeDirServer()
48
52
        self.server.setUp()
49
53
        self.addCleanup(self.server.tearDown)
50
 
        self._sftp_url = self.server.get_url()
 
54
        if self._full_handshake:
 
55
            self._sftp_url = self.server._get_sftp_url("")
 
56
        else:
 
57
            self._sftp_url = self.server.get_url()
51
58
        self._root = self.test_dir
52
59
        # Set to a string in setUp to give sftp server a new homedir.
53
60
        self._override_home = None
97
104
 
98
105
    def test_multiple_connections(self):
99
106
        t = self.get_transport()
100
 
        self.assertEquals(self.server.logs, 
101
 
                ['sftpserver - authorizing: foo'
102
 
               , 'sftpserver - channel request: session, 1'])
 
107
        self.assertTrue('sftpserver - new connection' in self.server.logs)
103
108
        self.server.logs = []
104
109
        # The second request should reuse the first connection
105
110
        # SingleListener only allows for a single connection,
151
156
 
152
157
    def test_parse_url(self):
153
158
        from bzrlib.transport.sftp import SFTPTransport
154
 
        s = SFTPTransport('sftp://simple.example.com/%2fhome/source', clone_from=fake)
 
159
        s = SFTPTransport('sftp://simple.example.com/home/source', clone_from=fake)
155
160
        self.assertEquals(s._host, 'simple.example.com')
156
161
        self.assertEquals(s._port, None)
157
162
        self.assertEquals(s._path, '/home/source')
158
163
        self.failUnless(s._password is None)
159
164
 
160
 
        self.assertEquals(s.base, 'sftp://simple.example.com/%2Fhome/source/')
161
 
        
162
 
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/relative', clone_from=fake)
 
165
        self.assertEquals(s.base, 'sftp://simple.example.com/home/source/')
 
166
 
 
167
        s = SFTPTransport('sftp://ro%62ey:h%40t@example.com:2222/~/relative', clone_from=fake)
163
168
        self.assertEquals(s._host, 'example.com')
164
169
        self.assertEquals(s._port, 2222)
165
170
        self.assertEquals(s._username, 'robey')
167
172
        self.assertEquals(s._path, 'relative')
168
173
 
169
174
        # Base should not keep track of the password
170
 
        self.assertEquals(s.base, 'sftp://robey@example.com:2222/relative/')
171
 
 
172
 
        # Double slash should be accepted instead of using %2F
173
 
        s = SFTPTransport('sftp://user@example.com:22//absolute/path/', clone_from=fake)
174
 
        self.assertEquals(s._host, 'example.com')
175
 
        self.assertEquals(s._port, 22)
176
 
        self.assertEquals(s._username, 'user')
177
 
        self.assertEquals(s._password, None)
178
 
        self.assertEquals(s._path, '/absolute/path/')
179
 
 
180
 
        # Also, don't show the port if it is the default 22
181
 
        self.assertEquals(s.base, 'sftp://user@example.com:22/%2Fabsolute/path/')
 
175
        self.assertEquals(s.base, 'sftp://robey@example.com:2222/~/relative/')
182
176
 
183
177
    def test_relpath(self):
184
178
        from bzrlib.transport.sftp import SFTPTransport
185
179
        from bzrlib.errors import PathNotChild
186
180
 
187
 
        s = SFTPTransport('sftp://user@host.com//abs/path', clone_from=fake)
188
 
        self.assertEquals(s.relpath('sftp://user@host.com//abs/path/sub'), 'sub')
 
181
        s = SFTPTransport('sftp://user@host.com/abs/path', clone_from=fake)
 
182
        self.assertEquals(s.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
189
183
        # Can't test this one, because we actually get an AssertionError
190
184
        # TODO: Consider raising an exception rather than an assert
191
 
        #self.assertRaises(PathNotChild, s.relpath, 'http://user@host.com//abs/path/sub')
192
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user2@host.com//abs/path/sub')
193
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@otherhost.com//abs/path/sub')
194
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com:33//abs/path/sub')
195
 
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com/abs/path/sub')
 
185
        #self.assertRaises(PathNotChild, s.relpath, 'http://user@host.com/abs/path/sub')
 
186
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user2@host.com/abs/path/sub')
 
187
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@otherhost.com/abs/path/sub')
 
188
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com:33/abs/path/sub')
 
189
        self.assertRaises(PathNotChild, s.relpath, 'sftp://user@host.com/~/rel/path/sub')
196
190
 
197
191
        # Make sure it works when we don't supply a username
198
 
        s = SFTPTransport('sftp://host.com//abs/path', clone_from=fake)
199
 
        self.assertEquals(s.relpath('sftp://host.com//abs/path/sub'), 'sub')
 
192
        s = SFTPTransport('sftp://host.com/abs/path', clone_from=fake)
 
193
        self.assertEquals(s.relpath('sftp://host.com/abs/path/sub'), 'sub')
200
194
 
201
195
        # Make sure it works when parts of the path will be url encoded
202
196
        # TODO: These may be incorrect, we might need to urllib.urlencode() before
257
251
        self.assertEquals(b2.revision_history(), ['a1', 'a2'])
258
252
 
259
253
 
 
254
class SFTPFullHandshakingTest(TestCaseWithSFTPServer):
 
255
    """Verify that a full-handshake (SSH over loopback TCP) sftp connection works."""
 
256
    _full_handshake = True
 
257
    
 
258
    def test_connection(self):
 
259
        self.get_transport()