test_sftp.py 文件源码

def test_sftp_read(self):
        self.assertEqual(self._auth(), 0)
        sftp = self.session.sftp_init()
        self.assertTrue(sftp is not None)
        if int(platform.python_version_tuple()[0]) >= 3:
            test_file_data = b'test' + bytes(os.linesep, 'utf-8')
        else:
            test_file_data = b'test' + os.linesep
        remote_filename = os.sep.join([os.path.dirname(__file__),
                                       'remote_test_file'])
        with open(remote_filename, 'wb') as test_fh:
            test_fh.write(test_file_data)
        with sftp.open(remote_filename, 0, 0) as remote_fh:
            try:
                self.assertTrue(remote_fh is not None)
                remote_data = b""
                for rc, data in remote_fh:
                    remote_data += data
                self.assertEqual(remote_fh.close(), 0)
                self.assertEqual(remote_data, test_file_data)
            finally:
                os.unlink(remote_filename)