summaryrefslogtreecommitdiff
path: root/tests/test_sftp.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_sftp.py')
-rw-r--r--tests/test_sftp.py126
1 files changed, 59 insertions, 67 deletions
diff --git a/tests/test_sftp.py b/tests/test_sftp.py
index 87c57340..e4e18e5a 100644
--- a/tests/test_sftp.py
+++ b/tests/test_sftp.py
@@ -26,23 +26,18 @@ do test file operations in (so no existing files will be harmed).
import os
import socket
import sys
-import threading
-import unittest
import warnings
from binascii import hexlify
from tempfile import mkstemp
import pytest
-import paramiko
-import paramiko.util
from paramiko.py3compat import PY2, b, u, StringIO
from paramiko.common import o777, o600, o666, o644
from paramiko.sftp_attr import SFTPAttributes
from .util import needs_builtin
-from .stub_sftp import StubServer, StubSFTPServer
-from .util import _support, slow
+from .util import slow
ARTICLE = """
@@ -74,14 +69,21 @@ decreased compared with chicken.
# Here is how unicode characters are encoded over 1 to 6 bytes in utf-8
-# U-00000000 - U-0000007F: 0xxxxxxx
-# U-00000080 - U-000007FF: 110xxxxx 10xxxxxx
-# U-00000800 - U-0000FFFF: 1110xxxx 10xxxxxx 10xxxxxx
-# U-00010000 - U-001FFFFF: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx
-# U-00200000 - U-03FFFFFF: 111110xx 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx
-# U-04000000 - U-7FFFFFFF: 1111110x 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx
+# U-00000000 - U-0000007F:
+# 0xxxxxxx
+# U-00000080 - U-000007FF:
+# 110xxxxx 10xxxxxx
+# U-00000800 - U-0000FFFF:
+# 1110xxxx 10xxxxxx 10xxxxxx
+# U-00010000 - U-001FFFFF:
+# 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx
+# U-00200000 - U-03FFFFFF:
+# 111110xx 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx
+# U-04000000 - U-7FFFFFFF:
+# 1111110x 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx 10xxxxxx
# Note that: hex(int('11000011',2)) == '0xc3'
-# Thus, the following 2-bytes sequence is not valid utf8: "invalid continuation byte"
+# Thus, the following 2-bytes sequence is not valid utf8: "invalid continuation
+# byte"
NON_UTF8_DATA = b"\xC3\xC3"
unicode_folder = u"\u00fcnic\u00f8de" if PY2 else "\u00fcnic\u00f8de"
@@ -90,7 +92,7 @@ utf8_folder = b"/\xc3\xbcnic\xc3\xb8\x64\x65"
@slow
class TestSFTP(object):
- def test_1_file(self, sftp):
+ def test_file(self, sftp):
"""
verify that we can create a file.
"""
@@ -101,7 +103,7 @@ class TestSFTP(object):
f.close()
sftp.remove(sftp.FOLDER + "/test")
- def test_2_close(self, sftp):
+ def test_close(self, sftp):
"""
Verify that SFTP session close() causes a socket error on next action.
"""
@@ -109,7 +111,7 @@ class TestSFTP(object):
with pytest.raises(socket.error, match="Socket is closed"):
sftp.open(sftp.FOLDER + "/test2", "w")
- def test_2_sftp_can_be_used_as_context_manager(self, sftp):
+ def test_sftp_can_be_used_as_context_manager(self, sftp):
"""
verify that the sftp session is closed when exiting the context manager
"""
@@ -118,7 +120,7 @@ class TestSFTP(object):
with pytest.raises(socket.error, match="Socket is closed"):
sftp.open(sftp.FOLDER + "/test2", "w")
- def test_3_write(self, sftp):
+ def test_write(self, sftp):
"""
verify that a file can be created and written, and the size is correct.
"""
@@ -129,7 +131,7 @@ class TestSFTP(object):
finally:
sftp.remove(sftp.FOLDER + "/duck.txt")
- def test_3_sftp_file_can_be_used_as_context_manager(self, sftp):
+ def test_sftp_file_can_be_used_as_context_manager(self, sftp):
"""
verify that an opened file can be used as a context manager
"""
@@ -140,7 +142,7 @@ class TestSFTP(object):
finally:
sftp.remove(sftp.FOLDER + "/duck.txt")
- def test_4_append(self, sftp):
+ def test_append(self, sftp):
"""
verify that a file can be opened for append, and tell() still works.
"""
@@ -158,7 +160,7 @@ class TestSFTP(object):
finally:
sftp.remove(sftp.FOLDER + "/append.txt")
- def test_5_rename(self, sftp):
+ def test_rename(self, sftp):
"""
verify that renaming a file works.
"""
@@ -185,7 +187,7 @@ class TestSFTP(object):
except:
pass
- def test_5a_posix_rename(self, sftp):
+ def testa_posix_rename(self, sftp):
"""Test posix-rename@openssh.com protocol extension."""
try:
# first check that the normal rename works as specified
@@ -194,20 +196,15 @@ class TestSFTP(object):
sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b")
with sftp.open(sftp.FOLDER + "/a", "w") as f:
f.write("two")
- try:
+ with pytest.raises(IOError): # actual message seems generic
sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b")
- self.assertTrue(
- False, "no exception when rename-ing onto existing file"
- )
- except (OSError, IOError):
- pass
# now check with the posix_rename
sftp.posix_rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b")
with sftp.open(sftp.FOLDER + "/b", "r") as f:
data = u(f.read())
err = "Contents of renamed file not the same as original file"
- assert data == "two", err
+ assert "two" == data, err
finally:
try:
@@ -219,7 +216,7 @@ class TestSFTP(object):
except:
pass
- def test_6_folder(self, sftp):
+ def test_folder(self, sftp):
"""
create a temporary folder, verify that we can create a file in it, then
remove the folder and verify that we can't create a file in it anymore.
@@ -232,7 +229,7 @@ class TestSFTP(object):
with pytest.raises(IOError, match="No such file"):
sftp.open(sftp.FOLDER + "/subfolder/test")
- def test_7_listdir(self, sftp):
+ def test_listdir(self, sftp):
"""
verify that a folder can be created, a bunch of files can be placed in
it, and those files show up in sftp.listdir.
@@ -253,7 +250,7 @@ class TestSFTP(object):
sftp.remove(sftp.FOLDER + "/fish.txt")
sftp.remove(sftp.FOLDER + "/tertiary.py")
- def test_7_5_listdir_iter(self, sftp):
+ def test_listdir_iter(self, sftp):
"""
listdir_iter version of above test
"""
@@ -273,7 +270,7 @@ class TestSFTP(object):
sftp.remove(sftp.FOLDER + "/fish.txt")
sftp.remove(sftp.FOLDER + "/tertiary.py")
- def test_8_setstat(self, sftp):
+ def test_setstat(self, sftp):
"""
verify that the setstat functions (chown, chmod, utime, truncate) work.
"""
@@ -310,7 +307,7 @@ class TestSFTP(object):
finally:
sftp.remove(sftp.FOLDER + "/special")
- def test_9_fsetstat(self, sftp):
+ def test_fsetstat(self, sftp):
"""
verify that the fsetstat functions (chown, chmod, utime, truncate)
work on open files.
@@ -350,12 +347,12 @@ class TestSFTP(object):
finally:
sftp.remove(sftp.FOLDER + "/special")
- def test_A_readline_seek(self, sftp):
+ def test_readline_seek(self, sftp):
"""
- create a text file and write a bunch of text into it. then count the lines
- in the file, and seek around to retrieve particular lines. this should
- verify that read buffering and 'tell' work well together, and that read
- buffering is reset on 'seek'.
+ create a text file and write a bunch of text into it. then count the
+ lines in the file, and seek around to retrieve particular lines. this
+ should verify that read buffering and 'tell' work well together, and
+ that read buffering is reset on 'seek'.
"""
try:
with sftp.open(sftp.FOLDER + "/duck.txt", "w") as f:
@@ -375,17 +372,14 @@ class TestSFTP(object):
f.seek(pos_list[17], f.SEEK_SET)
assert f.readline()[:4] == "duck"
f.seek(pos_list[10], f.SEEK_SET)
- assert (
- f.readline()
- == "duck types were equally resistant to exogenous insulin compared with chicken.\n"
- )
+ expected = "duck types were equally resistant to exogenous insulin compared with chicken.\n" # noqa
+ assert f.readline() == expected
finally:
sftp.remove(sftp.FOLDER + "/duck.txt")
- def test_B_write_seek(self, sftp):
+ def test_write_seek(self, sftp):
"""
- create a text file, seek back and change part of it, and verify that the
- changes worked.
+ Create a text file, seek back, change it, and verify.
"""
try:
with sftp.open(sftp.FOLDER + "/testing.txt", "w") as f:
@@ -400,7 +394,7 @@ class TestSFTP(object):
finally:
sftp.remove(sftp.FOLDER + "/testing.txt")
- def test_C_symlink(self, sftp):
+ def test_symlink(self, sftp):
"""
create a symlink and then check that lstat doesn't follow it.
"""
@@ -447,7 +441,7 @@ class TestSFTP(object):
except:
pass
- def test_D_flush_seek(self, sftp):
+ def test_flush_seek(self, sftp):
"""
verify that buffered writes are automatically flushed on seek.
"""
@@ -467,7 +461,7 @@ class TestSFTP(object):
except:
pass
- def test_E_realpath(self, sftp):
+ def test_realpath(self, sftp):
"""
test that realpath is returning something non-empty and not an
error.
@@ -478,7 +472,7 @@ class TestSFTP(object):
assert len(f) > 0
assert os.path.join(pwd, sftp.FOLDER) == f
- def test_F_mkdir(self, sftp):
+ def test_mkdir(self, sftp):
"""
verify that mkdir/rmdir work.
"""
@@ -489,7 +483,7 @@ class TestSFTP(object):
with pytest.raises(IOError, match="No such file"):
sftp.rmdir(sftp.FOLDER + "/subfolder")
- def test_G_chdir(self, sftp):
+ def test_chdir(self, sftp):
"""
verify that chdir/getcwd work.
"""
@@ -525,7 +519,7 @@ class TestSFTP(object):
except:
pass
- def test_H_get_put(self, sftp):
+ def test_get_put(self, sftp):
"""
verify that get/put work.
"""
@@ -560,7 +554,7 @@ class TestSFTP(object):
os.unlink(localname)
sftp.unlink(sftp.FOLDER + "/bunny.txt")
- def test_I_check(self, sftp):
+ def test_check(self, sftp):
"""
verify that file.check() works against our own server.
(it's an sftp extension that we support, and may be the only ones who
@@ -582,14 +576,12 @@ class TestSFTP(object):
== u(hexlify(sum)).upper()
)
sum = f.check("md5", 0, 0, 510)
- assert (
- u(hexlify(sum)).upper()
- == "EB3B45B8CD55A0707D99B177544A319F373183D241432BB2157AB9E46358C4AC90370B5CADE5D90336FC1716F90B36D6"
- ) # noqa
+ expected = "EB3B45B8CD55A0707D99B177544A319F373183D241432BB2157AB9E46358C4AC90370B5CADE5D90336FC1716F90B36D6" # noqa
+ assert u(hexlify(sum)).upper() == expected
finally:
sftp.unlink(sftp.FOLDER + "/kitty.txt")
- def test_J_x_flag(self, sftp):
+ def test_x_flag(self, sftp):
"""
verify that the 'x' flag works when opening a file.
"""
@@ -604,7 +596,7 @@ class TestSFTP(object):
finally:
sftp.unlink(sftp.FOLDER + "/unusual.txt")
- def test_K_utf8(self, sftp):
+ def test_utf8(self, sftp):
"""
verify that unicode strings are encoded into utf8 correctly.
"""
@@ -620,7 +612,7 @@ class TestSFTP(object):
self.fail("exception " + str(e))
sftp.unlink(b(sftp.FOLDER) + utf8_folder)
- def test_L_utf8_chdir(self, sftp):
+ def test_utf8_chdir(self, sftp):
sftp.mkdir(sftp.FOLDER + "/" + unicode_folder)
try:
sftp.chdir(sftp.FOLDER + "/" + unicode_folder)
@@ -631,7 +623,7 @@ class TestSFTP(object):
sftp.chdir()
sftp.rmdir(sftp.FOLDER + "/" + unicode_folder)
- def test_M_bad_readv(self, sftp):
+ def test_bad_readv(self, sftp):
"""
verify that readv at the end of the file doesn't essplode.
"""
@@ -647,7 +639,7 @@ class TestSFTP(object):
finally:
sftp.unlink(sftp.FOLDER + "/zero")
- def test_N_put_without_confirm(self, sftp):
+ def test_put_without_confirm(self, sftp):
"""
verify that get/put work without confirmation.
"""
@@ -676,11 +668,11 @@ class TestSFTP(object):
os.unlink(localname)
sftp.unlink(sftp.FOLDER + "/bunny.txt")
- def test_O_getcwd(self, sftp):
+ def test_getcwd(self, sftp):
"""
verify that chdir/getcwd work.
"""
- assert sftp.getcwd() == None
+ assert sftp.getcwd() is None
root = sftp.normalize(".")
if root[-1] != "/":
root += "/"
@@ -695,7 +687,7 @@ class TestSFTP(object):
except:
pass
- def XXX_test_M_seek_append(self, sftp):
+ def test_seek_append(self, sftp):
"""
verify that seek does't affect writes during append.
@@ -733,7 +725,7 @@ class TestSFTP(object):
# appear but they're clearly emitted from subthreads that have no error
# handling. No point running it until that is fixed somehow.
@pytest.mark.skip("Doesn't prove anything right now")
- def test_N_file_with_percent(self, sftp):
+ def test_file_with_percent(self, sftp):
"""
verify that we can create a file with a '%' in the filename.
( it needs to be properly escaped by _log() )
@@ -745,7 +737,7 @@ class TestSFTP(object):
f.close()
sftp.remove(sftp.FOLDER + "/test%file")
- def test_O_non_utf8_data(self, sftp):
+ def test_non_utf8_data(self, sftp):
"""Test write() and read() of non utf8 data"""
try:
with sftp.open("%s/nonutf8data" % sftp.FOLDER, "w") as f:
@@ -775,7 +767,7 @@ class TestSFTP(object):
try:
with sftp.open("%s/write_buffer" % sftp.FOLDER, "wb") as f:
for offset in range(0, len(data), 8):
- f.write(buffer(data, offset, 8))
+ f.write(buffer(data, offset, 8)) # noqa
with sftp.open("%s/write_buffer" % sftp.FOLDER, "rb") as f:
assert f.read() == data