← 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

  • Committer: Martin Pool
  • Date: 2006-01-30 06:23:50 UTC
  • mfrom: (1534.1.17 integration)
  • Revision ID: mbp@sourcefrog.net-20060130062350-d6f25277ddcdfd79
[merge] robert's integration of much recent work

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
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,
246
251
        self.assertEquals(b2.revision_history(), ['a1', 'a2'])
247
252
 
248
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()