=== modified file 'bzrlib/tests/test_sftp_transport.py' --- bzrlib/tests/test_sftp_transport.py +++ bzrlib/tests/test_sftp_transport.py @@ -38,16 +38,23 @@ if not paramiko_loaded: raise TestSkipped('you must have paramiko to run this test') super(TestCaseWithSFTPServer, self).setUp() - from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer - if getattr(self, '_get_remote_is_absolute', None) is None: - self._get_remote_is_absolute = True - if self._get_remote_is_absolute: - self.server = SFTPAbsoluteServer() + from bzrlib.transport.sftp import SFTPAbsoluteServer, SFTPHomeDirServer, SFTPServer + if getattr(self, '_full_handshake', False): + self.server = SFTPServer() else: - self.server = SFTPHomeDirServer() + self._full_handshake = False + if getattr(self, '_get_remote_is_absolute', None) is None: + self._get_remote_is_absolute = True + if self._get_remote_is_absolute: + self.server = SFTPAbsoluteServer() + else: + self.server = SFTPHomeDirServer() self.server.setUp() self.addCleanup(self.server.tearDown) - self._sftp_url = self.server.get_url() + if self._full_handshake: + self._sftp_url = self.server._get_sftp_url("") + else: + self._sftp_url = self.server.get_url() self._root = self.test_dir # Set to a string in setUp to give sftp server a new homedir. self._override_home = None @@ -255,3 +262,9 @@ self.assertEquals(b2.revision_history(), ['a1', 'a2']) +class SFTPFullHandshakingTest(TestCaseWithSFTPServer): + """Verify that a full-handshake (SSH over loopback TCP) sftp connection works.""" + _full_handshake = True + + def test_connection(self): + self.get_transport() === modified file 'bzrlib/transport/sftp.py' --- bzrlib/transport/sftp.py +++ bzrlib/transport/sftp.py @@ -156,11 +156,11 @@ def recv(self, n): return self.__socket.recv(n) + def recv_ready(self): + return True + def close(self): self.__socket.close() - - def recv_ready(self): - return True SYSTEM_HOSTKEYS = {} @@ -230,7 +230,7 @@ def __del__(self): """Should this warn, or actually try to cleanup?""" if self.lock_file: - warn("SFTPLock %r not explicitly unlocked" % (self.path,)) + warning("SFTPLock %r not explicitly unlocked" % (self.path,)) self.unlock() def unlock(self):