← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

first cut at merge from integration.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
39
39
        if not paramiko_loaded:
40
40
            raise TestSkipped('you must have paramiko to run this test')
41
41
        super(TestCaseWithSFTPServer, self).setUp()
42
 
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer
43
 
        if getattr(self, '_get_remote_is_absolute', None) is None:
44
 
            self._get_remote_is_absolute = True
45
 
        if self._get_remote_is_absolute:
46
 
            self.server = SFTPAbsoluteServer()
 
42
        from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer, SFTPServer
 
43
        if getattr(self, '_full_handshake', False):
 
44
            self.server = SFTPServer()
47
45
        else:
48
 
            self.server = SFTPHomeDirServer()
 
46
            self._full_handshake = False
 
47
            if getattr(self, '_get_remote_is_absolute', None) is None:
 
48
                self._get_remote_is_absolute = True
 
49
            if self._get_remote_is_absolute:
 
50
                self.server = SFTPAbsoluteServer()
 
51
            else:
 
52
                self.server = SFTPHomeDirServer()
49
53
        self.server.setUp()
50
54
        self.addCleanup(self.server.tearDown)
51
 
        self._sftp_url = self.server.get_url()
 
55
        if self._full_handshake:
 
56
            self._sftp_url = self.server._get_sftp_url("")
 
57
        else:
 
58
            self._sftp_url = self.server.get_url()
52
59
        self._root = self.test_dir
53
60
        # Set to a string in setUp to give sftp server a new homedir.
54
61
        self._override_home = None
98
105
 
99
106
    def test_multiple_connections(self):
100
107
        t = self.get_transport()
101
 
        self.assertEquals(self.server.logs, 
102
 
                ['sftpserver - authorizing: foo'
103
 
               , 'sftpserver - channel request: session, 1'])
 
108
        self.assertTrue('sftpserver - new connection' in self.server.logs)
104
109
        self.server.logs = []
105
110
        # The second request should reuse the first connection
106
111
        # SingleListener only allows for a single connection,
247
252
        self.assertEquals(b2.revision_history(), ['a1', 'a2'])
248
253
 
249
254
 
 
255
class SFTPFullHandshakingTest(TestCaseWithSFTPServer):
 
256
    """Verify that a full-handshake (SSH over loopback TCP) sftp connection works."""
 
257
    _full_handshake = True
 
258
    
 
259
    def test_connection(self):
 
260
        self.get_transport()