diff options
author | Martin Packman <gzlist@googlemail.com> | 2017-05-26 01:53:44 +0100 |
---|---|---|
committer | Jeff Forcier <jeff@bitprophet.org> | 2017-06-09 13:46:06 -0700 |
commit | 17231b9a7c56bf06f1a848fc52554541a09a2eba (patch) | |
tree | cb5c4c92974f6b4cbb791614c88d8be8fcf52665 /tests/test_sftp.py | |
parent | c2c402cb6e4a4f86ff3053389bb6300c4b4505f1 (diff) |
Allow any buffer type to be written to SFTPFile
Fixes #967 #968
Rollup of earlier branches proposed as #969 and #970 with
additional fix inside sftp_client.
Includes new tests for SFTPFile usage.
Change against the 1.17 branch.
Diffstat (limited to 'tests/test_sftp.py')
-rwxr-xr-x | tests/test_sftp.py | 30 |
1 files changed, 30 insertions, 0 deletions
diff --git a/tests/test_sftp.py b/tests/test_sftp.py index d3064fff..98a9cebb 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -35,6 +35,7 @@ from tempfile import mkstemp import paramiko from paramiko.py3compat import PY2, b, u, StringIO from paramiko.common import o777, o600, o666, o644 +from tests import skipUnlessBuiltin from tests.stub_sftp import StubServer, StubSFTPServer from tests.loop import LoopSocket from tests.util import test_path @@ -817,6 +818,35 @@ class SFTPTest (unittest.TestCase): sftp_attributes = SFTPAttributes() self.assertEqual(str(sftp_attributes), "?--------- 1 0 0 0 (unknown date) ?") + @skipUnlessBuiltin('buffer') + def test_write_buffer(self): + """Test write() using a buffer instance.""" + data = 3 * b'A potentially large block of data to chunk up.\n' + try: + with sftp.open('%s/write_buffer' % FOLDER, 'wb') as f: + for offset in range(0, len(data), 8): + f.write(buffer(data, offset, 8)) + + with sftp.open('%s/write_buffer' % FOLDER, 'rb') as f: + self.assertEqual(f.read(), data) + finally: + sftp.remove('%s/write_buffer' % FOLDER) + + @skipUnlessBuiltin('memoryview') + def test_write_memoryview(self): + """Test write() using a memoryview instance.""" + data = 3 * b'A potentially large block of data to chunk up.\n' + try: + with sftp.open('%s/write_memoryview' % FOLDER, 'wb') as f: + view = memoryview(data) + for offset in range(0, len(data), 8): + f.write(view[offset:offset+8]) + + with sftp.open('%s/write_memoryview' % FOLDER, 'rb') as f: + self.assertEqual(f.read(), data) + finally: + sftp.remove('%s/write_memoryview' % FOLDER) + if __name__ == '__main__': SFTPTest.init_loopback() |