diff --git a/docs/authors.rst b/docs/authors.rst index e37a2988..4c5240e3 100644 --- a/docs/authors.rst +++ b/docs/authors.rst @@ -5,6 +5,8 @@ Contributors of code, tests and documentation to the project who have agreed to Acknowledgment -------------- + * PR #84 by miraz12 + * PR #80 & #81 by shashfrankenstien * PR #46 by ofiliojo * PR #19 by coreyhartley * PR #15 by RobertCochran diff --git a/docs/changes.rst b/docs/changes.rst index ddcc6436..7f6936f7 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -1,5 +1,14 @@ -1.2.2 (current, released 2026-5-10) +1.2.3 (current, released 2026-9-08) ----------------------------------- + * adding new tests for connections, hash, drivepath and read-only servers. + * reworking tests to incorporate self-cleaning of all artifacts. + * fix for error handling catches in _set_authentication function. + * fix for github workflow so all tests can run from fork PRs. + * fix for hash function return that was breaking host key verification. + * use paramiko's compression algorithm list by default instead of None. + +1.2.2 (released 2026-5-10) +-------------------------- * adding new test for _sftp_channel exception handling. * adding curve25519-sha256@libssh.org to kex list. * fix for UnboundLocalError on a certain exception in _sftp_channel. diff --git a/docs/conf.py b/docs/conf.py index 3f918a10..d9d882cd 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -54,9 +54,9 @@ # built documents. # # The short X.Y version. -version = '1.2.2' +version = '1.2.3' # The full version, including alpha/beta/rc tags. -release = '1.2.2' +release = '1.2.3' # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. diff --git a/pyproject.toml b/pyproject.toml index 5e7202a9..6db4bed8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,7 @@ keywords = [ name = 'sftpretty' readme = 'README.rst' requires-python = '>=3.6' -version = '1.2.2' +version = '1.2.3' [project.scripts] sftpretty = 'sftpretty:Connection' diff --git a/sftpretty/helpers.py b/sftpretty/helpers.py index fbd0aeb3..719eada9 100644 --- a/sftpretty/helpers.py +++ b/sftpretty/helpers.py @@ -71,16 +71,16 @@ def hash(filename, algorithm=sha3_512(), blocksize=65536): with open(filename, 'rb') as filestream: for chunk in iter(lambda: filestream.read(blocksize), b''): buffer.update(chunk) - except FileNotFoundError: + except OSError: buffer.update(bytes(filename.encode('utf-8'))) elif isinstance(filename, BytesIO): - for chunk in iter(lambda: filestream.read1(blocksize), b''): + for chunk in iter(lambda: filename.read1(blocksize), b''): buffer.update(chunk) elif isinstance(filename, IOBase): - for chunk in iter(lambda: filestream.read(blocksize), b''): + for chunk in iter(lambda: filename.read(blocksize), b''): buffer.update(chunk) - return algorithm.hexdigest() + return buffer.hexdigest() def localtree(container, localdir, remotedir, recurse=True): diff --git a/tests/common.py b/tests/common.py index 8b0bc5e4..8508863a 100644 --- a/tests/common.py +++ b/tests/common.py @@ -3,11 +3,10 @@ import pytest from contextlib import contextmanager -from os import close, environ +from os import environ from pathlib import Path from sftpretty import CnOpts from stat import S_ISDIR -from tempfile import mkstemp PASS = 'tEst@!357' @@ -81,20 +80,3 @@ def rmdir(dir): else: item.unlink() dir.rmdir() - - -@contextmanager -def tempfile_containing(contents=STARS8192, suffix=''): - '''create a temporary file, with optional suffix and return the filename, - cleanup when finished''' - - fd, temp_path = mkstemp(suffix=suffix) - close(fd) - - with open(temp_path, 'wb') as fh: - fh.write(contents.encode('utf-8')) - - try: - yield Path(temp_path).as_posix() - finally: - Path(temp_path).unlink() diff --git a/tests/conftest.py b/tests/conftest.py index d9204359..3545d759 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,12 +2,13 @@ import pytest +from common import LOCAL, remote_rmdir, STARS8192, USER_HOME +from os import close from paramiko.hostkeys import HostKeys from pathlib import Path -from uuid import uuid4 - -from common import LOCAL, remote_rmdir, USER_HOME from sftpretty import CnOpts, Connection +from tempfile import mkstemp +from uuid import uuid4 @pytest.fixture(scope='session') @@ -50,3 +51,19 @@ def remote_tmpdir(lsftp): yield lsftp.normalize(remotedir.as_posix()) finally: remote_rmdir(lsftp, remotedir.as_posix()) + + +@pytest.fixture +def tempfile_containing(tmp_path): + '''create a temporary file, with optional suffix, holding content and + return the filename''' + def contentfile(contents=STARS8192, suffix=''): + fd, temp_path = mkstemp(dir=tmp_path, suffix=suffix) + close(fd) + + with open(temp_path, 'wb') as tempfile: + tempfile.write(contents.encode('utf-8')) + + return Path(temp_path).as_posix() + + return contentfile diff --git a/tests/test_cd.py b/tests/test_cd.py index 6af52a31..4040b8ee 100644 --- a/tests/test_cd.py +++ b/tests/test_cd.py @@ -9,7 +9,7 @@ def test_cd_none(sftpserver): - '''test sftpretty.cd with None''' + '''test cd with None''' pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: @@ -21,7 +21,7 @@ def test_cd_none(sftpserver): def test_cd_path(sftpserver): - '''test sftpretty.cd with a path''' + '''test cd with a path''' pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: @@ -46,7 +46,7 @@ def test_cd_nested(sftpserver): def test_cd_bad_path(sftpserver): - '''test sftpretty.cd with a bad path''' + '''test cd with a bad path''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd diff --git a/tests/test_chmod.py b/tests/test_chmod.py index a80f879e..161d848b 100644 --- a/tests/test_chmod.py +++ b/tests/test_chmod.py @@ -2,7 +2,7 @@ import pytest -from common import conn, SKIP_IF_ROOT, SKIP_IF_WIN, tempfile_containing, VFS +from common import conn, SKIP_IF_ROOT, SKIP_IF_WIN, VFS from pathlib import Path from sftpretty import Connection from sftpretty.helpers import st_mode_to_int @@ -18,13 +18,14 @@ def test_chmod_not_exist(sftpserver): @SKIP_IF_ROOT @SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs -def test_chmod_ro(lsftp, remote_tmpdir): +def test_chmod_ro(lsftp, remote_tmpdir, tempfile_containing): '''test chmod against read-only path''' + content = 'You answer me, although I never ask you questions, what am I?' parent = Path(remote_tmpdir).joinpath('readonly') rfile = parent.joinpath('readme.txt') lsftp.mkdir_p(parent.as_posix()) - with tempfile_containing() as fname: - lsftp.put(fname, rfile.as_posix()) + localfile = tempfile_containing(contents=content) + lsftp.put(localfile, rfile.as_posix()) lsftp.chmod(parent.as_posix(), 400) # no search bit, 500 fails try: with pytest.raises(PermissionError): @@ -34,15 +35,16 @@ def test_chmod_ro(lsftp, remote_tmpdir): @SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs -def test_chmod_simple(lsftp): +def test_chmod_simple(lsftp, tempfile_containing): '''test basic chmod with octal mode represented by an int''' + content = 'Which three letters can frighten a thief away?' new_mode = 711 - with tempfile_containing(contents='') as fname: - base_fname = Path(fname).name - org_attrs = lsftp.put(fname) - lsftp.chmod(base_fname, new_mode) - new_attrs = lsftp.stat(base_fname) - lsftp.remove(base_fname) + localfile = tempfile_containing(contents=content) + base_fname = Path(localfile).name + org_attrs = lsftp.put(localfile) + lsftp.chmod(base_fname, new_mode) + new_attrs = lsftp.stat(base_fname) + lsftp.remove(base_fname) assert st_mode_to_int(new_attrs.st_mode) == new_mode assert new_attrs.st_mode != org_attrs.st_mode diff --git a/tests/test_chown.py b/tests/test_chown.py index 6d0c2a80..cf9a28d7 100644 --- a/tests/test_chown.py +++ b/tests/test_chown.py @@ -2,46 +2,52 @@ import pytest -from common import SKIP_IF_ROOT, SKIP_IF_WIN, tempfile_containing +from common import SKIP_IF_ROOT, SKIP_IF_WIN from pathlib import Path @SKIP_IF_WIN # uid comes through as 0, lacks support -def test_chown_uid(lsftp): +def test_chown_uid(lsftp, tempfile_containing): '''test changing just the uid''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - org_attrs = lsftp.put(fname) - uid = org_attrs.st_uid - lsftp.chown(base_fname, uid=uid) - new_attrs = lsftp.stat(base_fname) - lsftp.remove(base_fname) + content = 'What is the end of everything?' + localfile = tempfile_containing(contents=content) + base_fname = Path(localfile).name + org_attrs = lsftp.put(localfile) + uid = org_attrs.st_uid + lsftp.chown(base_fname, uid=uid) + new_attrs = lsftp.stat(base_fname) + lsftp.remove(base_fname) + assert new_attrs.st_gid == org_attrs.st_gid assert new_attrs.st_uid == uid @SKIP_IF_WIN # gid comes through as 0, lacks support -def test_chown_gid(lsftp): +def test_chown_gid(lsftp, tempfile_containing): '''test changing just the gid''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - org_attrs = lsftp.put(fname) - gid = org_attrs.st_gid - lsftp.chown(base_fname, gid=gid) - new_attrs = lsftp.stat(base_fname) - lsftp.remove(base_fname) + content = 'I am wet when drying. What am I?' + localfile = tempfile_containing(contents=content) + base_fname = Path(localfile).name + org_attrs = lsftp.put(localfile) + gid = org_attrs.st_gid + lsftp.chown(base_fname, gid=gid) + new_attrs = lsftp.stat(base_fname) + lsftp.remove(base_fname) + assert new_attrs.st_gid == gid assert new_attrs.st_uid == org_attrs.st_uid -def test_chown_none(lsftp): +def test_chown_none(lsftp, tempfile_containing): '''call chown with no gid or uid specified''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - org_attrs = lsftp.put(fname) - lsftp.chown(base_fname) - new_attrs = lsftp.stat(base_fname) - lsftp.remove(base_fname) + content = 'What color is the wind?' + localfile = tempfile_containing(contents=content) + base_fname = Path(localfile).name + org_attrs = lsftp.put(localfile) + lsftp.chown(base_fname) + new_attrs = lsftp.stat(base_fname) + lsftp.remove(base_fname) + assert new_attrs.st_gid == org_attrs.st_gid assert new_attrs.st_uid == org_attrs.st_uid @@ -54,13 +60,14 @@ def test_chown_not_exist(lsftp): @SKIP_IF_ROOT @SKIP_IF_WIN # ownership ids are synthetic, cannot be set -def test_chown_ro(lsftp): +def test_chown_ro(lsftp, tempfile_containing): '''call chown against path on read-only server''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - lsftp.put(fname) - try: - with pytest.raises(PermissionError): - lsftp.chown(base_fname, gid=0, uid=0) - finally: - lsftp.remove(base_fname) + content = 'What only works the first time you use it?' + localfile = tempfile_containing(contents=content) + base_fname = Path(localfile).name + lsftp.put(localfile) + try: + with pytest.raises(PermissionError): + lsftp.chown(base_fname, gid=0, uid=0) + finally: + lsftp.remove(base_fname) diff --git a/tests/test_get.py b/tests/test_get.py index 467f6a9d..5605f877 100644 --- a/tests/test_get.py +++ b/tests/test_get.py @@ -2,79 +2,84 @@ import pytest -from common import conn, tempfile_containing, VFS +from common import conn, VFS from pathlib import Path from sftpretty import Connection from unittest.mock import Mock -def test_get(sftpserver): +def test_get(sftpserver, tempfile_containing): '''download a file''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo1') - with tempfile_containing(contents='') as fname: - sftp.get('foo1.txt', fname) - assert open(fname, 'rb').read() == b'content of foo1.txt' + localfile = tempfile_containing(contents='') + sftp.get('foo1.txt', localfile) + assert open(localfile, 'rb').read() == b'content of foo1.txt' -def test_get_bad_remote(sftpserver): + +def test_get_bad_remote(sftpserver, tempfile_containing): '''download a file but it does not exist''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo1') - with tempfile_containing(contents='') as fname: - with pytest.raises(IOError): - sftp.get('readme-not-there.txt', fname) - assert open(fname, 'rb').read()[0:7] != b'Welcome' + localfile = tempfile_containing(contents='') + with pytest.raises(IOError): + sftp.get('readme-not-there.txt', localfile) + + assert open(localfile, 'rb').read()[0:7] != b'Welcome' -def test_get_callback(sftpserver): +def test_get_callback(sftpserver, tempfile_containing): '''test .get callback''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo1') cback = Mock(return_value=None) - with tempfile_containing(contents='') as fname: - result = sftp.get('foo1.txt', fname, callback=cback) - assert open(fname, 'rb').read() == b'content of foo1.txt' + localfile = tempfile_containing(contents='') + result = sftp.get('foo1.txt', localfile, callback=cback) + + assert open(localfile, 'rb').read() == b'content of foo1.txt' # verify callback was called assert cback.call_count # unlike .put() nothing is returned from the operation assert result is None -def test_get_glob_fails(sftpserver): +def test_get_glob_fails(sftpserver, tempfile_containing): '''try and use get a file with a pattern - Fails''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo1') - with tempfile_containing(contents='') as fname: - with pytest.raises(IOError): - sftp.get('*', fname) + localfile = tempfile_containing(contents='') + with pytest.raises(IOError): + sftp.get('*', localfile) -def test_get_preserve_mtime(sftpserver): +def test_get_preserve_mtime(sftpserver, tempfile_containing): '''test that m_time is preserved from local to remote, when get''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo1') remotefile = 'foo1.txt' r_stat = sftp.stat(remotefile) - with tempfile_containing(contents='') as localfile: - sftp.get(remotefile, localfile, preserve_mtime=True) - assert r_stat.st_mtime == Path(localfile).stat().st_mtime + localfile = tempfile_containing(contents='') + sftp.get(remotefile, localfile, preserve_mtime=True) + assert r_stat.st_mtime == Path(localfile).stat().st_mtime -def test_get_resume(sftpserver): + +def test_get_resume(sftpserver, tempfile_containing): '''test that resume continues partial download when it exists''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo1') remotesize = sftp.stat('foo1.txt').st_size - with tempfile_containing(contents='content of') as fname: - localsize = Path(fname).stat().st_size - sftp.get('foo1.txt', fname, resume=True) - assert open(fname, 'rb').read() == b'content of foo1.txt' + localfile = tempfile_containing(contents='content of') + localsize = Path(localfile).stat().st_size + sftp.get('foo1.txt', localfile, resume=True) + + assert open(localfile, 'rb').read() == b'content of foo1.txt' # verify difference between remotesize and partial localsize assert 9 == (remotesize - localsize) diff --git a/tests/test_get_r.py b/tests/test_get_r.py index d3c1fc86..fc83a59e 100644 --- a/tests/test_get_r.py +++ b/tests/test_get_r.py @@ -74,9 +74,9 @@ def test_get_r_pathed(sftpserver): sftp.remotetree(remote_tree, remote_cwd, localpath) actual = hash(Path(localpath).joinpath('bar1.txt').as_posix()) - expected = ('a69f73cca23a9ac5c8b567dc185a756e97c982164fe258' - '59e0d1dcc1475c80a615b2123af1f5f94c11e3e9402c3a' - 'c558f500199d95b6d3e301758586281dcd26') + expected = ('126f175986225cdaa8c6eccd1ed9296b487cc799a982ff' + 'db33818806228c9efb8d4ca14c37641097eb367bdd2f4c' + 'a7916d63893af38da137251520fff41f3cba') assert local_tree.keys() == remote_tree.keys() assert actual == expected diff --git a/tests/test_helpers.py b/tests/test_helpers.py index c5f5feb3..b1cb5304 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -2,7 +2,9 @@ import pytest -from sftpretty.helpers import drivepath +from hashlib import md5, new, sha1, sha256, sha3_512 +from io import BytesIO +from sftpretty.helpers import drivepath, hash @pytest.mark.parametrize('path,expected', ( @@ -43,5 +45,91 @@ ('', ''), (None, None) )) def test_drivepath(path, expected): + '''test drivepath input variations across supported platforms''' assert drivepath(path) == expected assert drivepath(expected) == expected + + +@pytest.mark.parametrize('algorithm', (md5(), sha1(), sha256(), sha3_512()), + ids=('md5', 'sha1', 'sha256', 'sha3_512')) +def test_hash_algorithm(algorithm, tempfile_containing): + '''test file and string digest agree with hashlib, per algorithm''' + content = 'My hovercraft is full of eels.' + expected = new(algorithm.name, content.encode()).hexdigest() + localfile = tempfile_containing(contents=content) + + assert hash(localfile, algorithm=algorithm) == expected + assert hash(content, algorithm=algorithm) == expected + + +@pytest.mark.parametrize('blocksize', (1, 7, 65536), + ids=('byte', 'partial', 'default')) +def test_hash_blocksize(blocksize, tempfile_containing): + '''test chunked reads digest the same as a single read''' + content = 'My hovercraft is full of eels.' + expected = sha3_512(content.encode()).hexdigest() + localfile = tempfile_containing(contents=content) + + assert hash(localfile, blocksize=blocksize) == expected + assert hash(BytesIO(content.encode()), blocksize=blocksize) == expected + + +@pytest.mark.parametrize('left, right', ( + ('CASE', 'case'), + ('some content', 'some different content'), + ('trailing', 'trailing '))) +def test_hash_distinct(left, right): + '''test distinct payloads never share a digest''' + empty = sha3_512(b'').hexdigest() + + assert hash(left) != hash(right) + assert empty not in (hash(left), hash(right)) + + +@pytest.mark.parametrize('form', ('bytesio', 'path', 'string')) +def test_hash_empty(form, tempfile_containing): + '''test empty input digests to the empty digest, whatever the form''' + empty = sha3_512(b'').hexdigest() + + if form == 'path': + assert hash(tempfile_containing(contents='')) == empty + elif form == 'string': + assert hash('') == empty + else: + assert hash(BytesIO(b'')) == empty + + +@pytest.mark.parametrize('form', ('path', 'string', 'bytesio', 'fileobject')) +def test_hash_input(form, tempfile_containing): + '''test input form digests the same bytes alike''' + content = 'My hovercraft is full of eels.' + expected = sha3_512(content.encode()).hexdigest() + localfile = tempfile_containing(contents=content) + + if form == 'path': + assert hash(localfile) == expected + elif form == 'string': + assert hash(content) == expected + elif form == 'bytesio': + assert hash(BytesIO(content.encode())) == expected + else: + with open(localfile, 'rb') as filestream: + assert hash(filestream) == expected + + +def test_hash_repeatable(tempfile_containing): + '''test repeated calls do not accumulate state''' + content = 'My hovercraft is full of eels.' + localfile = tempfile_containing(contents=content) + + assert hash(localfile) == hash(localfile) + assert hash('some content') == hash('some content') + + +@pytest.mark.parametrize('unreadable', ('/C:/Users/test/pub/bar1.txt', + 'i-do-not-exist.txt', + 'some content', 'a' * 512), + ids=('drive', 'missing', 'spaces', 'toolong')) +def test_hash_unreadable(unreadable): + '''test a string that cannot be opened is digested as a string''' + assert hash(unreadable) == sha3_512(unreadable.encode()).hexdigest() diff --git a/tests/test_put.py b/tests/test_put.py index afdae842..74d0f09c 100644 --- a/tests/test_put.py +++ b/tests/test_put.py @@ -2,63 +2,67 @@ import pytest -from common import conn, tempfile_containing, VFS +from common import conn, SKIP_IF_ROOT, SKIP_IF_WIN, VFS from pathlib import Path from sftpretty import Connection from time import sleep from unittest.mock import Mock -def test_put(lsftp): +def test_put(lsftp, tempfile_containing): '''test upload to localhost''' - contents = 'now is the time\nfor all good...' - with tempfile_containing(contents=contents) as fname: - base_fname = Path(fname).name - if base_fname in lsftp.listdir(): - lsftp.remove(base_fname) - assert base_fname not in lsftp.listdir() - lsftp.put(fname) - assert base_fname in lsftp.listdir() - with tempfile_containing(contents='') as tfile: - lsftp.get(base_fname, tfile) - assert open(tfile).read() == contents - # clean up + content = 'now is the time\nfor all good...' + localfile = tempfile_containing(contents=content) + localfileZ = tempfile_containing(contents='') + base_fname = Path(localfile).name + if base_fname in lsftp.listdir(): lsftp.remove(base_fname) + assert base_fname not in lsftp.listdir() -def test_put_bad_local(sftpserver): + lsftp.put(localfile) + assert base_fname in lsftp.listdir() + + lsftp.get(base_fname, localfileZ) + assert open(localfileZ).read() == content + + # clean up + lsftp.remove(base_fname) + + +def test_put_bad_local(sftpserver, tempfile_containing): '''try to put a non-existing file to a read-only server''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: - with tempfile_containing() as fname: - pass + localfile = tempfile_containing() + Path(localfile).unlink() # tempfile has been removed with pytest.raises(OSError): - sftp.put(fname) + sftp.put(localfile) -def test_put_callback(lsftp): +def test_put_callback(lsftp, tempfile_containing): '''test the callback feature of put''' cback = Mock(return_value=None) - with tempfile_containing() as fname: - base_fname = Path(fname).name - lsftp.chdir(Path.home().as_posix()) - lsftp.put(fname, callback=cback) - # clean up - lsftp.remove(base_fname) + localfile = tempfile_containing() + base_fname = Path(localfile).name + lsftp.chdir(Path.home().as_posix()) + lsftp.put(localfile, callback=cback) + # clean up + lsftp.remove(base_fname) # verify callback was called assert cback.call_count -def test_put_confirm(lsftp): +def test_put_confirm(lsftp, tempfile_containing): '''test the confirm feature of put''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - lsftp.chdir(Path.home().as_posix()) - result = lsftp.put(fname) - # clean up - lsftp.remove(base_fname) + localfile = tempfile_containing() + base_fname = Path(localfile).name + lsftp.chdir(Path.home().as_posix()) + result = lsftp.put(localfile) + # clean up + lsftp.remove(base_fname) # verify that an SFTPAttribute like Path.stat() was returned assert result.st_size == 8192 @@ -68,25 +72,17 @@ def test_put_confirm(lsftp): assert result.st_mtime -# TODO -# def test_put_not_allowed(lsftp): -# '''try to put a file to a read-only server''' -# with tempfile_containing() as fname: -# with pytest.raises(IOError): -# lsftp.put(fname) - - -def test_put_preserve_mtime(lsftp): +def test_put_preserve_mtime(lsftp, tempfile_containing): '''test that m_time is preserved from local to remote, when put''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - base = Path(fname).stat() - # with Connection(**LOCAL) as sftp: - result1 = lsftp.put(fname, preserve_mtime=True) - sleep(2) - result2 = lsftp.put(fname, preserve_mtime=True) - # clean up - lsftp.remove(base_fname) + localfile = tempfile_containing() + base_fname = Path(localfile).name + base = Path(localfile).stat() + # with Connection(**LOCAL) as sftp: + result1 = lsftp.put(localfile, preserve_mtime=True) + sleep(2) + result2 = lsftp.put(localfile, preserve_mtime=True) + # clean up + lsftp.remove(base_fname) # see if times are modified # assert base.st_atime == result1.st_atime @@ -95,15 +91,31 @@ def test_put_preserve_mtime(lsftp): assert int(result1.st_mtime) == result2.st_mtime -def test_put_resume(lsftp): +def test_put_resume(lsftp, tempfile_containing): '''test upload resume feature''' - with tempfile_containing(contents='resume this...') as fname: - base = Path(fname).stat() - with tempfile_containing(contents='resume ') as fname: - partial = lsftp.put(fname) - with open(fname, 'ab') as fh: - fh.write('this...'.encode('utf-8')) - result = lsftp.put(fname, preserve_mtime=True, resume=True) + localfile = tempfile_containing(contents='resume this...') + localfileZ = tempfile_containing(contents='resume ') + base = Path(localfile).stat() + partial = lsftp.put(localfileZ) + with open(localfileZ, 'ab') as fh: + fh.write('this...'.encode('utf-8')) + result = lsftp.put(localfileZ, preserve_mtime=True, resume=True) assert base.st_size == result.st_size assert partial.st_mtime == result.st_mtime + + +@SKIP_IF_ROOT +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +def test_put_ro(lsftp, remote_tmpdir, tempfile_containing): + '''try to put a file on a read-only server''' + localfile = tempfile_containing() + remotedir = Path(remote_tmpdir).joinpath('readonly') + remotefile = remotedir.joinpath(Path(localfile).name) + lsftp.mkdir_p(remotedir.as_posix()) + lsftp.chmod(remotedir.as_posix(), 500) + try: + with pytest.raises(PermissionError): + lsftp.put(localfile, remotefile.as_posix()) + finally: + lsftp.chmod(remotedir.as_posix(), 700) diff --git a/tests/test_remove.py b/tests/test_remove.py index f87fc4a9..4dcd38b6 100644 --- a/tests/test_remove.py +++ b/tests/test_remove.py @@ -2,34 +2,43 @@ import pytest -from common import tempfile_containing +from common import SKIP_IF_ROOT, SKIP_IF_WIN from pathlib import Path -def test_remove(lsftp, remote_tmpdir): +def test_remove(lsftp, remote_tmpdir, tempfile_containing): '''test the remove method''' - with tempfile_containing() as fname: - base_fname = Path(fname).name - rfile = Path(remote_tmpdir).joinpath(base_fname).as_posix() - lsftp.put(fname, rfile) - is_there = base_fname in lsftp.listdir(remote_tmpdir) - lsftp.remove(rfile) - not_there = base_fname not in lsftp.listdir(remote_tmpdir) + localfile = tempfile_containing() + base_fname = Path(localfile).name + rfile = Path(remote_tmpdir).joinpath(base_fname).as_posix() + lsftp.put(localfile, rfile) + is_there = base_fname in lsftp.listdir(remote_tmpdir) + lsftp.remove(rfile) + not_there = base_fname not in lsftp.listdir(remote_tmpdir) assert is_there assert not_there -# TODO -# def test_remove_roserver(lsftp, remote_tmpdir): -# '''test reaction of attempting remove on read-only server''' -# rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() -# with pytest.raises(IOError): -# lsftp.remove(rfile) - - def test_remove_does_not_exist(lsftp, remote_tmpdir): '''test remove against a non-existant file''' rfile = Path(remote_tmpdir).joinpath('i-am-not-here.txt').as_posix() with pytest.raises(IOError): lsftp.remove(rfile) + + +@SKIP_IF_ROOT +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +def test_remove_ro(lsftp, remote_tmpdir, tempfile_containing): + '''test remove against read-only server''' + localfile = tempfile_containing() + remotedir = Path(remote_tmpdir).joinpath('readonly') + remotefile = remotedir.joinpath(Path(localfile).name) + lsftp.mkdir_p(remotedir.as_posix()) + lsftp.put(localfile, remotefile.as_posix()) + lsftp.chmod(remotedir.as_posix(), 500) + try: + with pytest.raises(PermissionError): + lsftp.remove(remotefile.as_posix()) + finally: + lsftp.chmod(remotedir.as_posix(), 700) diff --git a/tests/test_rename.py b/tests/test_rename.py index f559a5f1..37e863ff 100644 --- a/tests/test_rename.py +++ b/tests/test_rename.py @@ -1,31 +1,53 @@ '''test sftpretty.rename''' -from common import tempfile_containing +import pytest + +from common import SKIP_IF_ROOT, SKIP_IF_WIN from pathlib import Path -def test_rename(lsftp): +def test_rename(lsftp, remote_tmpdir, tempfile_containing): '''test rename on remote''' - contents = 'now is the time\nfor all good...' - with tempfile_containing(contents=contents) as fname: - base_fname = Path(fname).name - if base_fname in lsftp.listdir(): - lsftp.remove(base_fname) - assert base_fname not in lsftp.listdir() - lsftp.put(fname) - lsftp.rename(base_fname, 'alice') - rdirs = lsftp.listdir() - assert 'alice' in rdirs - assert base_fname not in rdirs - lsftp.rename('alice', 'bob', posix=False) - rdirs = lsftp.listdir() - assert 'alice' not in rdirs - assert 'bob' in rdirs - lsftp.remove('bob') - - -# TODO -# def test_rename_ro(psftp): -# '''test rename on a read-only server''' -# with pytest.raises(IOError): -# psftp.rename('readme.txt', 'bob') + content = 'now is the time\nfor all good...' + localfile = tempfile_containing(contents=content) + base_fname = Path(localfile).name + alice = Path(remote_tmpdir).joinpath('alice').as_posix() + bob = Path(remote_tmpdir).joinpath('bob').as_posix() + remotefile = Path(remote_tmpdir).joinpath(base_fname).as_posix() + + assert base_fname not in lsftp.listdir(remote_tmpdir) + + lsftp.put(localfile, remotefile) + lsftp.rename(remotefile, alice) + rdirs = lsftp.listdir(remote_tmpdir) + + assert 'alice' in rdirs + assert base_fname not in rdirs + + lsftp.rename(alice, bob, posix=False) + rdirs = lsftp.listdir(remote_tmpdir) + + assert 'alice' not in rdirs + assert 'bob' in rdirs + + with lsftp.open(bob) as remote: + assert remote.read().decode() == content + + +@SKIP_IF_ROOT +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +@pytest.mark.parametrize('posix', (True, False)) +def test_rename_ro(lsftp, posix, remote_tmpdir, tempfile_containing): + '''test rename on a read-only server''' + localfile = tempfile_containing() + remotedir = Path(remote_tmpdir).joinpath('readonly') + remotefile = remotedir.joinpath(Path(localfile).name) + lsftp.mkdir_p(remotedir.as_posix()) + lsftp.put(localfile, remotefile.as_posix()) + lsftp.chmod(remotedir.as_posix(), 500) + try: + with pytest.raises(PermissionError): + lsftp.rename(remotefile.as_posix(), + remotedir.joinpath('alice').as_posix(), posix=posix) + finally: + lsftp.chmod(remotedir.as_posix(), 700) diff --git a/tests/test_sftp.py b/tests/test_sftp.py index 49bddc79..a6a9933b 100644 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -1,6 +1,6 @@ '''test sftpretty module''' -from common import conn, LOCAL, tempfile_containing, VFS +from common import conn, LOCAL, VFS from pathlib import Path from sftpretty import Connection from stat import S_ISLNK @@ -38,13 +38,13 @@ def test_mkdir_p(lsftp, remote_tmpdir): # assert lsftp.lexists(rsym) -def test_symlink(lsftp, remote_tmpdir): +def test_symlink(lsftp, remote_tmpdir, tempfile_containing): '''test symlink creation''' rdest = Path(remote_tmpdir).joinpath('honey-boo-boo').as_posix() - with tempfile_containing() as fname: - rfile = Path(remote_tmpdir).joinpath(Path(fname).name).as_posix() - lsftp.put(fname, rfile) - lsftp.symlink(rfile, rdest) + localfile = tempfile_containing() + rfile = Path(remote_tmpdir).joinpath(Path(localfile).name).as_posix() + lsftp.put(localfile, rfile) + lsftp.symlink(rfile, rdest) assert S_ISLNK(lsftp.lstat(rdest).st_mode) @@ -60,14 +60,14 @@ def test_exists(sftpserver): assert sftp.exists('pub') -def test_lexists(lsftp, remote_tmpdir): +def test_lexists(lsftp, remote_tmpdir, tempfile_containing): '''test lexists functionality''' - with tempfile_containing() as fname: - rfile = Path(remote_tmpdir).joinpath(Path(fname).name).as_posix() - rbad = Path(remote_tmpdir).joinpath('peek-a-boo.txt').as_posix() - lsftp.put(fname, rfile) - - assert lsftp.lexists(rfile) - lsftp.remove(rfile) - assert lsftp.lexists(rfile) is False - assert lsftp.lexists(rbad) is False + localfile = tempfile_containing() + rfile = Path(remote_tmpdir).joinpath(Path(localfile).name).as_posix() + rbad = Path(remote_tmpdir).joinpath('peek-a-boo.txt').as_posix() + lsftp.put(localfile, rfile) + + assert lsftp.lexists(rfile) + lsftp.remove(rfile) + assert lsftp.lexists(rfile) is False + assert lsftp.lexists(rbad) is False