From f6ed58252043dc22ede32a40d886daa3b996da15 Mon Sep 17 00:00:00 2001 From: Shashank Gopikrishna Date: Fri, 7 Aug 2026 03:59:43 -0400 Subject: [PATCH 01/20] fix wrong error raised when private key is missing or invalid --- sftpretty/__init__.py | 2 +- tests/test_connection.py | 21 +++++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 2e34ae25..01f06f79 100644 --- a/sftpretty/__init__.py +++ b/sftpretty/__init__.py @@ -238,7 +238,7 @@ def _set_authentication(self, password, private_key, private_key_pass): 'directory or does not exist, please revise ' 'and provide a path to a valid private key.')) raise err - finally: + else: private_key = key.from_private_key_file( key_file, password=private_key_pass) self._transport.auth_publickey(self._username, private_key) diff --git a/tests/test_connection.py b/tests/test_connection.py index 1c0feb64..d8ec1e6b 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -78,6 +78,27 @@ def test_connection_bad_credentials(): sftp.listdir() +def test_connection_missing_private_key_file(): + '''missing private-key file should raise the original parse error''' + copts = LOCAL.copy() + copts['private_key'] = 'id_doesnt_exist' + with pytest.raises(FileNotFoundError): + with Connection(**copts) as sftp: + sftp.close() + + +def test_connection_invalid_private_key_type(tmp_path): + '''invalid private-key content should raise the original parse error''' + key_path = Path(tmp_path / 'id_sftpretty.bad') + key_path.write_text('not a real key\n', encoding='utf-8') + + copts = LOCAL.copy() + copts['private_key'] = str(key_path) + with pytest.raises(KeyError): + with Connection(**copts) as sftp: + sftp.close() + + def test_connection_bad_host(): '''attempt connection to a non-existing server''' knownhosts = Path('~/.ssh/known_hosts').expanduser() From d553684c9a922b7ba99c24422f2325efa9db7e97 Mon Sep 17 00:00:00 2001 From: Shashank Gopikrishna Date: Fri, 7 Aug 2026 12:04:05 -0400 Subject: [PATCH 02/20] make test_connection_invalid_private_key_type file consistent in naming --- tests/test_connection.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_connection.py b/tests/test_connection.py index d8ec1e6b..5edbe908 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -81,7 +81,7 @@ def test_connection_bad_credentials(): def test_connection_missing_private_key_file(): '''missing private-key file should raise the original parse error''' copts = LOCAL.copy() - copts['private_key'] = 'id_doesnt_exist' + copts['private_key'] = 'id_sftpretty_missing' with pytest.raises(FileNotFoundError): with Connection(**copts) as sftp: sftp.close() @@ -89,7 +89,7 @@ def test_connection_missing_private_key_file(): def test_connection_invalid_private_key_type(tmp_path): '''invalid private-key content should raise the original parse error''' - key_path = Path(tmp_path / 'id_sftpretty.bad') + key_path = tmp_path / 'id_sftpretty_bad' key_path.write_text('not a real key\n', encoding='utf-8') copts = LOCAL.copy() From 431ac46bdaaf210be40133023ff7d0fcc85d0127 Mon Sep 17 00:00:00 2001 From: Shashank Gopikrishna Date: Fri, 7 Aug 2026 18:01:58 -0400 Subject: [PATCH 03/20] raise CredentialException for invalid key and explicitly handle FileNotFoundError logging --- sftpretty/__init__.py | 13 ++++++++++--- tests/test_connection.py | 12 ++++++------ 2 files changed, 16 insertions(+), 9 deletions(-) diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 01f06f79..0f1586b7 100644 --- a/sftpretty/__init__.py +++ b/sftpretty/__init__.py @@ -190,6 +190,7 @@ class Connection(object): :raises LoggingException: :raises PasswordRequiredException: :raises SSHException: + :raises FileNotFoundError: ''' def __init__(self, host, cnopts=None, default_path=None, password=None, port=22, private_key=None, private_key_pass=None, @@ -220,10 +221,16 @@ def _set_authentication(self, password, private_key, private_key_pass): with open(key_file, 'r', encoding='utf-8') as head: key_id = head.readline()[11:][:-18] log.debug(f'Key ID: [{key_id}]') + + if key_id.strip() not in key_types: + error_msg = f'Unable to identity key type from file provided: \n[{key_file}]' + log.error(error_msg) + raise CredentialException(error_msg) + key = key_types[key_id.strip()] - except KeyError as err: - log.error(('Unable to identify key type from file provided' - f': \n[{key_file}]')) + + except FileNotFoundError as err: + log.error(f'identity key file not found: \n[{key_file}]') raise err except PasswordRequiredException as err: log.error(('No password provided for encrypted private ' diff --git a/tests/test_connection.py b/tests/test_connection.py index 5edbe908..cfcdd687 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -7,7 +7,7 @@ from common import conn, LOCAL, VFS from pathlib import Path -from sftpretty import (CnOpts, Connection, ConnectionException, +from sftpretty import (CnOpts, Connection, ConnectionException, CredentialException, HostKeysException, SSHException) @@ -83,8 +83,8 @@ def test_connection_missing_private_key_file(): copts = LOCAL.copy() copts['private_key'] = 'id_sftpretty_missing' with pytest.raises(FileNotFoundError): - with Connection(**copts) as sftp: - sftp.close() + with Connection(**copts): + pass def test_connection_invalid_private_key_type(tmp_path): @@ -94,9 +94,9 @@ def test_connection_invalid_private_key_type(tmp_path): copts = LOCAL.copy() copts['private_key'] = str(key_path) - with pytest.raises(KeyError): - with Connection(**copts) as sftp: - sftp.close() + with pytest.raises(CredentialException): + with Connection(**copts): + pass def test_connection_bad_host(): From dc1baf9e599d82225ac690926f5f5c46c89535c7 Mon Sep 17 00:00:00 2001 From: Shashank Gopikrishna Date: Fri, 7 Aug 2026 18:23:13 -0400 Subject: [PATCH 04/20] handle bad password and private key input datatypes --- sftpretty/__init__.py | 71 ++++++++++++++++++++-------------------- tests/test_connection.py | 25 ++++++++++++-- 2 files changed, 57 insertions(+), 39 deletions(-) diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 0f1586b7..a03cbe14 100644 --- a/sftpretty/__init__.py +++ b/sftpretty/__init__.py @@ -212,44 +212,43 @@ def _set_authentication(self, password, private_key, private_key_pass): '''Authenticate transport. Prefer private key over password.''' if self._config.get('identityfile'): private_key = self._config['identityfile'][0] - if private_key is not None: + if private_key is not None and isinstance(private_key, str): # Use key path or provided key object key_types = {'EC': ECDSAKey, 'OPENSSH': Ed25519Key, 'RSA': RSAKey} - if isinstance(private_key, str): - key_file = Path(private_key).expanduser().absolute().as_posix() - try: - with open(key_file, 'r', encoding='utf-8') as head: - key_id = head.readline()[11:][:-18] - log.debug(f'Key ID: [{key_id}]') - - if key_id.strip() not in key_types: - error_msg = f'Unable to identity key type from file provided: \n[{key_file}]' - log.error(error_msg) - raise CredentialException(error_msg) - - key = key_types[key_id.strip()] - - except FileNotFoundError as err: - log.error(f'identity key file not found: \n[{key_file}]') - raise err - except PasswordRequiredException as err: - log.error(('No password provided for encrypted private ' - 'key encrypted private key.')) - raise err - except PermissionError as err: - log.error(('File permission preventing user access to:\n' - f'[{key_file}]')) - raise err - except SSHException as err: - log.error(('Path provided is an invalid key file, a ' - 'directory or does not exist, please revise ' - 'and provide a path to a valid private key.')) - raise err - else: - private_key = key.from_private_key_file( - key_file, password=private_key_pass) - self._transport.auth_publickey(self._username, private_key) - elif password is not None: + key_file = Path(private_key).expanduser().absolute().as_posix() + try: + with open(key_file, 'r', encoding='utf-8') as head: + key_id = head.readline()[11:][:-18] + log.debug(f'Key ID: [{key_id}]') + + if key_id.strip() not in key_types: + error_msg = f'Unable to identity key type from file provided: \n[{key_file}]' + log.error(error_msg) + raise CredentialException(error_msg) + + key = key_types[key_id.strip()] + + except FileNotFoundError as err: + log.error(f'identity key file not found: \n[{key_file}]') + raise err + except PasswordRequiredException as err: + log.error(('No password provided for encrypted private ' + 'key encrypted private key.')) + raise err + except PermissionError as err: + log.error(('File permission preventing user access to:\n' + f'[{key_file}]')) + raise err + except SSHException as err: + log.error(('Path provided is an invalid key file, a ' + 'directory or does not exist, please revise ' + 'and provide a path to a valid private key.')) + raise err + else: + private_key = key.from_private_key_file( + key_file, password=private_key_pass) + self._transport.auth_publickey(self._username, private_key) + elif password is not None and isinstance(password, str): self._transport.auth_password(self._username, password) else: raise CredentialException('No password or private key provided.') diff --git a/tests/test_connection.py b/tests/test_connection.py index cfcdd687..2b64735c 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -68,14 +68,24 @@ def test_cnopts_none_knownhosts(): assert cnopts.hostkeys is None -def test_connection_bad_credentials(): +def test_connection_wrong_password(): '''attempt connection to a non-existing server''' copts = LOCAL.copy() copts['password'] = 'badword' del copts['private_key'], copts['private_key_pass'] with pytest.raises(SSHException): - with Connection(**copts) as sftp: - sftp.listdir() + with Connection(**copts): + pass + + +def test_connection_bad_password_datatype(): + '''attempt connection to a non-existing server''' + copts = LOCAL.copy() + copts['password'] = True + del copts['private_key'], copts['private_key_pass'] + with pytest.raises(CredentialException): + with Connection(**copts): + pass def test_connection_missing_private_key_file(): @@ -87,6 +97,15 @@ def test_connection_missing_private_key_file(): pass +def test_connection_bad_private_key_datatype(): + '''missing private-key file should raise the original parse error''' + copts = LOCAL.copy() + copts['private_key'] = True + with pytest.raises(CredentialException): + with Connection(**copts): + pass + + def test_connection_invalid_private_key_type(tmp_path): '''invalid private-key content should raise the original parse error''' key_path = tmp_path / 'id_sftpretty_bad' From 92a7d607bef5bfc557425b44fefc5de86f1ad5c0 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Thu, 20 Aug 2026 04:30:44 +0000 Subject: [PATCH 05/20] fix for UnboundLocalError on key_type not found or file path doesn't exist, make key type parsing a bit more robust --- sftpretty/__init__.py | 49 +++++++++++++++++++++++++------------------ 1 file changed, 29 insertions(+), 20 deletions(-) diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 2e34ae25..47b3d94b 100644 --- a/sftpretty/__init__.py +++ b/sftpretty/__init__.py @@ -187,8 +187,11 @@ class Connection(object): :raises ConnectionException: :raises CredentialException: :raises HostKeysException: + :raises KeyError: :raises LoggingException: + :raises OSError: :raises PasswordRequiredException: + :raises PermissionError: :raises SSHException: ''' def __init__(self, host, cnopts=None, default_path=None, password=None, @@ -217,30 +220,36 @@ def _set_authentication(self, password, private_key, private_key_pass): if isinstance(private_key, str): key_file = Path(private_key).expanduser().absolute().as_posix() try: - with open(key_file, 'r', encoding='utf-8') as head: - key_id = head.readline()[11:][:-18] + with open(key_file, 'rb') as head: + header = head.readline(64).decode('ascii', 'replace') + key_id = header.rpartition(' PRIVATE KEY-----')[0][11:] log.debug(f'Key ID: [{key_id}]') key = key_types[key_id.strip()] - except KeyError as err: - log.error(('Unable to identify key type from file provided' - f': \n[{key_file}]')) - raise err - except PasswordRequiredException as err: - log.error(('No password provided for encrypted private ' - 'key encrypted private key.')) - raise err - except PermissionError as err: - log.error(('File permission preventing user access to:\n' - f'[{key_file}]')) - raise err - except SSHException as err: - log.error(('Path provided is an invalid key file, a ' - 'directory or does not exist, please revise ' - 'and provide a path to a valid private key.')) - raise err - finally: private_key = key.from_private_key_file( key_file, password=private_key_pass) + except KeyError: + log.error(('Unsupported key format, paramiko only reads ' + 'EC, OPENSSH and RSA PEM keys. Re-encode with ' + f'ssh-keygen -p -f :\n[{key_file}]')) + raise + except PermissionError: + log.error(('File permission preventing user access to:\n' + f'[{key_file}]')) + raise + except OSError: + log.error(('Path provided is a directory or does not ' + 'exist, please revise and provide a path to a ' + f'readable private key:\n[{key_file}]')) + raise + except PasswordRequiredException: + log.error(('No password provided for encrypted private ' + f'key:\n[{key_file}]')) + raise + except SSHException: + log.error(('Path provided is an invalid or corrupt key ' + 'file, please revise and provide a path to a ' + 'valid private key.')) + raise self._transport.auth_publickey(self._username, private_key) elif password is not None: self._transport.auth_password(self._username, password) From 3f3e1173a08a1edfdbd94a74f489002db482eca9 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Mon, 24 Aug 2026 06:52:56 +0000 Subject: [PATCH 06/20] adding tests for unsupported private key formats and missing or non-file type private key path. Adding Path to private key file resolution if block --- sftpretty/__init__.py | 2 +- tests/test_connection.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 47b3d94b..5dfb1b97 100644 --- a/sftpretty/__init__.py +++ b/sftpretty/__init__.py @@ -217,7 +217,7 @@ def _set_authentication(self, password, private_key, private_key_pass): if private_key is not None: # Use key path or provided key object key_types = {'EC': ECDSAKey, 'OPENSSH': Ed25519Key, 'RSA': RSAKey} - if isinstance(private_key, str): + if isinstance(private_key, (str, Path)): key_file = Path(private_key).expanduser().absolute().as_posix() try: with open(key_file, 'rb') as head: diff --git a/tests/test_connection.py b/tests/test_connection.py index 1c0feb64..85f6850f 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -94,6 +94,41 @@ def test_connection_bad_host(): sftp.listdir() +@pytest.mark.parametrize('blob', ( + b'\x30\x82\x04\xbe\x02\x01\x00', # binary DER, undecodable + b'-----BEGIN DSA PRIVATE KEY-----\n', # deprecated algorithm + b'-----BEGIN ENCRYPTED PRIVATE KEY-----\n', # PKCS#8, encrypted + b'-----BEGIN PRIVATE KEY-----\n', # PKCS#8 + b'' # empty file +)) +def test_connection_bad_private_key_format(blob, tmp_path): + '''deprecated or unsupported key formats must raise, not fail''' + key = tmp_path.joinpath('id_sftpretty_unsupported') + key.write_bytes(blob) + + copts = LOCAL.copy() + copts['private_key'] = key.as_posix() + with pytest.raises(KeyError): + Connection(**copts) as sftp: + sftp.listdir() + + +@pytest.mark.parametrize('kind', ('missing', 'directory')) +def test_connection_bad_private_key_path(kind, tmp_path): + '''private-key path pointing to missing or non-file type''' + key = tmp_path.joinpath(f'id_sftpretty_{kind}') + + if kind == 'directory': + key.mkdir() + + copts = LOCAL.copy() + key = tmp_path.joinpath(f'id_sftpretty_{kind}') + + with pytest.raises(OSError, match=key.name): + with Connection(**copts) as sftp: + sftp.listdir() + + def test_connection_good(sftpserver): '''connect to a public sftp server''' with sftpserver.serve_content(VFS): From 6ebc16bbb203cece40ee2f0319d7ef89af3a5a21 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Tue, 25 Aug 2026 15:59:21 +0000 Subject: [PATCH 07/20] addressing issue #82, reworking tests that produce artifacts in the Path.home() or the user running tests that weren't been cleaned either up after. Reworking the symlink normalization test and adding one for a dangling target as well. To these means adding a remote_rmdir and new fixture function to handle remote tmp directory creation and clean-up. --- tests/common.py | 19 +++++++++++++ tests/conftest.py | 16 ++++++++++- tests/test_connection.py | 2 +- tests/test_normalize.py | 28 ++++++++++++++----- tests/test_put.py | 8 ++++-- tests/test_put_d.py | 28 ++++++++----------- tests/test_put_r.py | 27 +++++++------------ tests/test_readlink.py | 17 +++++------- tests/test_remove.py | 24 ++++++++--------- tests/test_rmdir.py | 30 +++++++++++++-------- tests/test_sftp.py | 56 +++++++++++++++++++------------------- tests/test_truncate.py | 58 ++++++++-------------------------------- 12 files changed, 158 insertions(+), 155 deletions(-) diff --git a/tests/common.py b/tests/common.py index 7317f273..db5d35ad 100644 --- a/tests/common.py +++ b/tests/common.py @@ -6,6 +6,7 @@ from os import close, environ from pathlib import Path from sftpretty import CnOpts +from stat import S_ISDIR from tempfile import mkstemp @@ -36,7 +37,25 @@ def conn(sftpsrv): 'username': USER} +def remote_rmdir(sftp, dir): + '''recursively remove a remote directory tree''' + try: + listing = sftp.listdir_attr(dir) + except FileNotFoundError: + return + + for attr in listing: + remotepath = Path(dir).joinpath(attr.filename).as_posix() + if S_ISDIR(attr.st_mode): + remote_rmdir(sftp, remotepath) + else: + sftp.remove(remotepath) + + sftp.rmdir(dir) + + def rmdir(dir): + '''recursively remove a directory tree''' dir = Path(dir) for item in dir.iterdir(): if item.is_dir(): diff --git a/tests/conftest.py b/tests/conftest.py index eea15e61..10f44b1c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,8 +4,9 @@ from paramiko.hostkeys import HostKeys from pathlib import Path +from uuid import uuid4 -from common import LOCAL +from common import LOCAL, remote_rmdir, USER_HOME from sftpretty import CnOpts, Connection @@ -16,6 +17,7 @@ def lsftp(request): LOCAL['cnopts'] = cnopts lsftp = Connection(**LOCAL) request.addfinalizer(lsftp.close) + return lsftp @@ -36,3 +38,15 @@ def knownhosts(sftpserver, key_type='ssh-ed25519'): knownhosts.write_bytes(bytes(hostkeys, 'utf-8')) return + + +@pytest.fixture +def remote_tmpdir(lsftp): + '''setup unique remote temporary directory''' + remotedir = Path(USER_HOME).joinpath(f'sftpretty-{uuid4().hex[:8]}') + lsftp.mkdir_p(remotedir.as_posix()) + + try: + yield remotedir.as_posix() + finally: + remote_rmdir(lsftp, remotedir.as_posix()) diff --git a/tests/test_connection.py b/tests/test_connection.py index 85f6850f..c40dbf9c 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -109,7 +109,7 @@ def test_connection_bad_private_key_format(blob, tmp_path): copts = LOCAL.copy() copts['private_key'] = key.as_posix() with pytest.raises(KeyError): - Connection(**copts) as sftp: + with Connection(**copts) as sftp: sftp.listdir() diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 8f9972e3..0b45e89f 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -18,13 +18,27 @@ def test_normalize(sftpserver): assert sftp.normalize('.') == pubpath.as_posix() -# TODO -# def test_normalize_symlink(sftp): -# '''test normalize against a symlink''' -# home = Path.home() -# sftp.chdir(home.as_posix()) -# rsym = 'readme.sym' -# assert sftp.normalize(rsym) == home.joinpath(rsym).as_posix() +def test_normalize_dangling_symlink(lsftp, remote_tmpdir): + '''test normalize against a symlink whose target is missing''' + missing = Path(remote_tmpdir).joinpath('gone.txt').as_posix() + rsym = Path(remote_tmpdir).joinpath('dangling.sym').as_posix() + lsftp.symlink(missing, rsym) + + assert lsftp.lexists(rsym) + assert lsftp.exists(rsym) is False + assert lsftp.normalize(rsym) == missing + + +def test_normalize_symlink(lsftp, remote_tmpdir): + '''test normalize against a symlink''' + rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() + rsym = Path(remote_tmpdir).joinpath('readme.sym').as_posix() + lsftp.putfo(BytesIO(b'My hovercraft is full of eels.'), rfile) + lsftp.symlink(rfile, rsym) + + assert S_ISLNK(lsftp.lstat(rsym).st_mode) + assert lsftp.normalize(rsym) == lsftp.normalize(rfile) + assert lsftp.normalize(rsym) != rsym def test_pwd(sftpserver): diff --git a/tests/test_put.py b/tests/test_put.py index b5c35777..afdae842 100644 --- a/tests/test_put.py +++ b/tests/test_put.py @@ -46,6 +46,7 @@ def test_put_callback(lsftp): lsftp.put(fname, callback=cback) # clean up lsftp.remove(base_fname) + # verify callback was called assert cback.call_count @@ -58,6 +59,7 @@ def test_put_confirm(lsftp): result = lsftp.put(fname) # clean up lsftp.remove(base_fname) + # verify that an SFTPAttribute like Path.stat() was returned assert result.st_size == 8192 assert result.st_uid is not None @@ -67,11 +69,11 @@ def test_put_confirm(lsftp): # TODO -# def test_put_not_allowed(psftp): +# 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): -# psftp.put(fname) +# lsftp.put(fname) def test_put_preserve_mtime(lsftp): @@ -85,6 +87,7 @@ def test_put_preserve_mtime(lsftp): result2 = lsftp.put(fname, preserve_mtime=True) # clean up lsftp.remove(base_fname) + # see if times are modified # assert base.st_atime == result1.st_atime assert int(base.st_mtime) == result1.st_mtime @@ -101,5 +104,6 @@ def test_put_resume(lsftp): with open(fname, 'ab') as fh: fh.write('this...'.encode('utf-8')) result = lsftp.put(fname, preserve_mtime=True, resume=True) + assert base.st_size == result.st_size assert partial.st_mtime == result.st_mtime diff --git a/tests/test_put_d.py b/tests/test_put_d.py index a48852ae..883e6433 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -4,31 +4,25 @@ from blddirs import build_dir_struct from common import rmdir -from pathlib import Path -from tempfile import mkdtemp -def test_put_d(lsftp): +def test_put_d(lsftp, remote_tmpdir, tmp_path): '''test put_d''' - localpath = Path(mkdtemp()).as_posix() - remote = Path.home() - build_dir_struct(localpath) - local = Path(localpath).joinpath('pub') - lsftp.put_d(local.as_posix(), remote.as_posix()) - - rmdir(localpath) + build_dir_struct(tmp_path.as_posix()) + local = temp_path.joinpath('pub').as_posix() + lsftp.put_d(local, remote_tempdir) + remote = Path(remote_tmpdir).joinpath('pub').as_posix() + assert lsftp.listdir(remote) == ['make.txt'] # TODO -# def test_put_d_ro(psftp): -# '''test put_d failure on remote read-only srvr''' -# # run the op +# def test_put_d_ro(lsftp): +# '''test put_d failure on remote read-only server''' # with pytest.raises(IOError): -# psftp.put_d('.', '.') +# lsftp.put_d('.', '.') -def test_put_d_bad_local(lsftp): +def test_put_d_bad_local(lsftp, remote_tmpdir): '''test put_d failure on non-existing local directory''' - # run the op with pytest.raises(OSError): - lsftp.put_d('/non-existing', '.') + lsftp.put_d('/non-existing', remote_tmpdir) diff --git a/tests/test_put_r.py b/tests/test_put_r.py index a1cdbb4f..a3e02a7e 100644 --- a/tests/test_put_r.py +++ b/tests/test_put_r.py @@ -3,32 +3,25 @@ import pytest from blddirs import build_dir_struct -from common import rmdir -from pathlib import Path -from tempfile import mkdtemp -def test_put_r(lsftp): +def test_put_r(lsftp, remote_tmpdir, tmp_path): '''test put_r''' - localpath = Path(mkdtemp()).as_posix() - remote = Path.home() - build_dir_struct(localpath) - local = Path(localpath).joinpath('pub') - lsftp.put_r(local.as_posix(), remote.as_posix()) + build_dir_struct(tmp_path.as_posix()) + local = tmp_path.joinpath('pub').as_posix() + lsftp.put_r(local, remote_tmpdir) - rmdir(localpath) + assert lsftp.listdir(remote_tmpdir) != [] # TODO -# def test_put_r_ro(psftp): -# '''test put_r failure on remote read-only srvr''' -# # run the op +# def test_put_r_ro(lsftp): +# '''test put_r failure on remote read-only server''' # with pytest.raises(IOError): -# psftp.put_r('.', '.') +# lsftp.put_r('.', '.') -def test_put_r_bad_local(lsftp): +def test_put_r_bad_local(lsftp, remote_tmpdir): '''test put_r failure on non-existing local directory''' - # run the op with pytest.raises(OSError): - lsftp.put_r('/non-existing', '.') + lsftp.put_r('/non-existing', remote_tmpdir) diff --git a/tests/test_readlink.py b/tests/test_readlink.py index 8378266f..f7973d66 100644 --- a/tests/test_readlink.py +++ b/tests/test_readlink.py @@ -4,19 +4,14 @@ from pathlib import Path -def test_readlink(lsftp): +def test_readlink(lsftp, remote_tmpdir): '''test the readlink method''' - buf = b'I will not buy this record, it is scratched\nMy hovercraft'\ - b' is full of eels.' + buf = b'I will not buy this record, it is scratched.\nMy hovercraft '\ + b'is full of eels.' flo = BytesIO(buf) - rfile = 'readme.txt' - rlink = 'readme.sym' - rpath = Path.home().joinpath(rfile).as_posix() + rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() + rlink = Path(remote_tmpdir).joinpath('readme.sym').as_posix() lsftp.putfo(flo, rfile) lsftp.symlink(rfile, rlink) - result = lsftp.readlink(rlink).endswith(rpath) - lsftp.remove(rlink) - lsftp.remove(rfile) - # test assert after cleanup - assert result + assert lsftp.readlink(rlink).endswith(rfile) diff --git a/tests/test_remove.py b/tests/test_remove.py index c47189f4..f87fc4a9 100644 --- a/tests/test_remove.py +++ b/tests/test_remove.py @@ -6,30 +6,30 @@ from pathlib import Path -def test_remove(lsftp): +def test_remove(lsftp, remote_tmpdir): '''test the remove method''' with tempfile_containing() as fname: base_fname = Path(fname).name - lsftp.chdir(Path.home().as_posix()) - lsftp.put(fname) - is_there = base_fname in lsftp.listdir() - lsftp.remove(base_fname) - not_there = base_fname not in lsftp.listdir() + 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) assert is_there assert not_there # TODO -# def test_remove_roserver(psftp): +# def test_remove_roserver(lsftp, remote_tmpdir): # '''test reaction of attempting remove on read-only server''' -# psftp.chdir(Path.home().as_posix()) +# rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() # with pytest.raises(IOError): -# psftp.remove('readme.txt') +# lsftp.remove(rfile) -def test_remove_does_not_exist(lsftp): +def test_remove_does_not_exist(lsftp, remote_tmpdir): '''test remove against a non-existant file''' - lsftp.chdir(Path.home().as_posix()) + rfile = Path(remote_tmpdir).joinpath('i-am-not-here.txt').as_posix() with pytest.raises(IOError): - lsftp.remove('i-am-not-here.txt') + lsftp.remove(rfile) diff --git a/tests/test_rmdir.py b/tests/test_rmdir.py index bf22c0be..0cd1e358 100644 --- a/tests/test_rmdir.py +++ b/tests/test_rmdir.py @@ -1,17 +1,25 @@ '''test sftpretty.rmdir''' -def test_rmdir(lsftp): +def test_rmdir(lsftp, remote_tmpdir): '''test mkdir''' dirname = 'test-rm' - lsftp.mkdir(dirname) - assert dirname in lsftp.listdir() - lsftp.rmdir(dirname) - assert dirname not in lsftp.listdir() + remotedir = Path(remote_tmpdir).joinpath(dirname).as_posix() + lsftp.mkdir(remotedir) + assert dirname in lsftp.listdir(remote_tmpdir) + lsftp.rmdir(remotedir) + assert dirname not in lsftp.listdir(remote_tmpdir) -# TODO -# def test_rmdir_ro(psftp): -# '''test rmdir against read-only server''' -# psftp.chdir(Path.home().as_posix()) -# with pytest.raises(IOError): -# psftp.rmdir('pub') + +@SKIP_IF_ROOT +def test_rmdir_ro(lsftp, remote_tmpdir): + '''test rmdir against read-only server''' + parent = Path(remote_tmpdir).joinpath('readonly') + remotedir = parent.joinpath('test-rm') + lsftp.mkdir_p(remotedir.as_posix()) + lsftp.chmod(parent.as_posix(), 500) + try: + with pytest.raises(PermissionError): + lsftp.rmdir(remotedir.as_posix()) + finally: + lsftp.chmod(parent.as_posix(), 700) diff --git a/tests/test_sftp.py b/tests/test_sftp.py index f962c4cc..49bddc79 100644 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -7,7 +7,7 @@ def test_sftp_client(lsftp): - '''test for access to the underlying, active sftpclient''' + '''test for access to the underlying active sftpclient''' with Connection(**LOCAL) as sftp: assert 'normalize' in dir(sftp.sftp_client) assert 'readlink' in dir(sftp.sftp_client) @@ -16,10 +16,10 @@ def test_sftp_client(lsftp): assert 'readlink' in dir(lsftp.sftp_client) -def test_mkdir_p(lsftp): +def test_mkdir_p(lsftp, remote_tmpdir): '''test mkdir_p simple, testing 2 things, oh well''' - rdir = 'foo/bar/baz' - rdir2 = 'foo/bar' + rdir = Path(remote_tmpdir).joinpath('foo/bar/baz').as_posix() + rdir2 = Path(remote_tmpdir).joinpath('foo/bar').as_posix() assert lsftp.exists(rdir) is False lsftp.mkdir_p(rdir) is_dir = lsftp.isdir(rdir) @@ -27,34 +27,30 @@ def test_mkdir_p(lsftp): lsftp.rmdir(rdir2) lsftp.mkdir_p(rdir) is_dir_partial = lsftp.isdir(rdir) - lsftp.rmdir(rdir) - lsftp.rmdir(rdir2) - lsftp.rmdir('foo') + assert is_dir assert is_dir_partial -# def test_lexists_symbolic(psftp): -# '''test .lexists() vs. symbolic link''' -# rsym = 'readme.sym' -# assert psftp.lexists(rsym) +# def test_lexists_symbolic(lsftp, remote_tmpdir): +# '''test lexists vs symbolic link''' +# rsym = Path(remote_tmpdir).joinpath('readme.sym').as_posix() +# assert lsftp.lexists(rsym) -def test_symlink(lsftp): +def test_symlink(lsftp, remote_tmpdir): '''test symlink creation''' - rdest = Path.home().joinpath('honey-boo-boo') + rdest = Path(remote_tmpdir).joinpath('honey-boo-boo').as_posix() with tempfile_containing() as fname: - lsftp.put(fname) - lsftp.symlink(fname, rdest.as_posix()) - rslt = lsftp.lstat(rdest.as_posix()) - is_link = S_ISLNK(rslt.st_mode) - lsftp.remove(rdest.as_posix()) - lsftp.remove(Path(fname).name) - assert is_link + rfile = Path(remote_tmpdir).joinpath(Path(fname).name).as_posix() + lsftp.put(fname, rfile) + lsftp.symlink(rfile, rdest) + + assert S_ISLNK(lsftp.lstat(rdest).st_mode) def test_exists(sftpserver): - '''test .exists() fuctionality''' + '''test exists fuctionality''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: rfile = 'pub/foo2/bar1/bar1.txt' @@ -64,12 +60,14 @@ def test_exists(sftpserver): assert sftp.exists('pub') -def test_lexists(lsftp): - '''test .lexists() functionality''' +def test_lexists(lsftp, remote_tmpdir): + '''test lexists functionality''' with tempfile_containing() as fname: - base_fname = Path(fname).name - lsftp.put(fname) - rbad = Path.home().joinpath('peek-a-boo.txt') - assert lsftp.lexists(fname) - lsftp.remove(base_fname) - assert lsftp.lexists(rbad.as_posix()) is False + 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 diff --git a/tests/test_truncate.py b/tests/test_truncate.py index b88b0c1b..edf01cbd 100644 --- a/tests/test_truncate.py +++ b/tests/test_truncate.py @@ -2,59 +2,23 @@ from common import STARS8192 from io import BytesIO +from path import Path -def test_truncate_smaller(lsftp): - '''test truncate, make file smaller''' +@pytest.mark.parametrize('size', (2 * 8192, 8192, 4096), + ids=('larger', 'same', 'smaller')) +def test_truncate(lsftp, remote_tmpdir, size): + '''test truncate to a larger, same and smaller size''' flo = BytesIO(bytes(STARS8192, 'UTF-8')) - rname = 'truncate.txt' - - try: - lsftp.remove(rname) - except IOError: - pass - + rname = Path(remote_tmpdir).joinpath('truncate.txt').as_posix() lsftp.putfo(flo, rname) - new_size = lsftp.truncate(rname, 4096) - assert new_size == 4096 - lsftp.remove(rname) - -def test_truncate_larger(lsftp): - '''test truncate, make file larger''' - flo = BytesIO(bytes(STARS8192, 'UTF-8')) - rname = 'truncate.txt' - - try: - lsftp.remove(rname) - except IOError: - pass - - lsftp.putfo(flo, rname) - new_size = lsftp.truncate(rname, 2 * 8192) - assert new_size == 2 * 8192 - lsftp.remove(rname) - - -def test_truncate_same(lsftp): - '''test truncate, make file same size''' - flo = BytesIO(bytes(STARS8192, 'UTF-8')) - rname = 'truncate.txt' - - try: - lsftp.remove(rname) - except IOError: - pass - - lsftp.putfo(flo, rname) - new_size = lsftp.truncate(rname, 8192) - assert new_size == 8192 - lsftp.remove(rname) + assert lsftp.truncate(rname, size) == size # TODO -# def test_truncate_ro(psftp): -# '''test truncate, against read-only server''' -# rname = Path.home().joinpath('readme.txt').as_posix() +# def test_truncate_ro(lsftp,, remote_tmpdir): +# '''test truncate against read-only server''' +# rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() # with pytest.raises(IOError): -# _ = psftp.truncate(rname, 8192) +# _ = lsftp.truncate(rfile, 8192) From 4539b2853a120824d5901ec0398c7ef273fa8e42 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Wed, 26 Aug 2026 00:13:33 +0000 Subject: [PATCH 08/20] missing imports and a few typos in tests --- tests/common.py | 2 ++ tests/test_normalize.py | 2 ++ tests/test_put_d.py | 6 +++--- tests/test_rmdir.py | 5 +++++ tests/test_truncate.py | 2 ++ 5 files changed, 14 insertions(+), 3 deletions(-) diff --git a/tests/common.py b/tests/common.py index db5d35ad..84e3348e 100644 --- a/tests/common.py +++ b/tests/common.py @@ -14,6 +14,8 @@ SKIP_IF_CI = pytest.mark.skipif(environ.get('CI', '') > '', reason='Not Local') SKIP_IF_MAC = pytest.mark.skipif(environ.get('RUNNER_OS', '') == 'macOS', reason='WhackMac') +SKIP_IF_ROOT = pytest.mark.skipif(environ.get('USER', '') == 'root', + reason='RootRules') SKIP_IF_WIN = pytest.mark.skipif(environ.get('RUNNER_OS', '') == 'Windows', reason='NoWinZone') STARS8192 = '*' * 8192 diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 0b45e89f..2ddb815b 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -1,8 +1,10 @@ '''test sftpretty.normalize''' from common import VFS, conn +from io import BytesIO from pathlib import Path from sftpretty import Connection +from stat import S_ISLNK def test_normalize(sftpserver): diff --git a/tests/test_put_d.py b/tests/test_put_d.py index 883e6433..68302727 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -3,14 +3,14 @@ import pytest from blddirs import build_dir_struct -from common import rmdir +from pathlib import Path def test_put_d(lsftp, remote_tmpdir, tmp_path): '''test put_d''' build_dir_struct(tmp_path.as_posix()) - local = temp_path.joinpath('pub').as_posix() - lsftp.put_d(local, remote_tempdir) + local = tmp_path.joinpath('pub').as_posix() + lsftp.put_d(local, remote_tmpdir) remote = Path(remote_tmpdir).joinpath('pub').as_posix() assert lsftp.listdir(remote) == ['make.txt'] diff --git a/tests/test_rmdir.py b/tests/test_rmdir.py index 0cd1e358..96b0d0c0 100644 --- a/tests/test_rmdir.py +++ b/tests/test_rmdir.py @@ -1,5 +1,10 @@ '''test sftpretty.rmdir''' +import pytest + +from common import SKIP_IF_ROOT +from pathlib import Path + def test_rmdir(lsftp, remote_tmpdir): '''test mkdir''' diff --git a/tests/test_truncate.py b/tests/test_truncate.py index edf01cbd..3d933c61 100644 --- a/tests/test_truncate.py +++ b/tests/test_truncate.py @@ -1,5 +1,7 @@ '''test sftpretty.listdir''' +import pytest + from common import STARS8192 from io import BytesIO from path import Path From 839b53d2536dbf65620338529a28881d8acf7d79 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Wed, 26 Aug 2026 04:09:20 +0000 Subject: [PATCH 09/20] fixing up another read-only test in this time for put_d, fix typo in import in truncate test file --- tests/test_put_d.py | 26 +++++++++++++++++++++----- tests/test_truncate.py | 2 +- 2 files changed, 22 insertions(+), 6 deletions(-) diff --git a/tests/test_put_d.py b/tests/test_put_d.py index 68302727..a1f37c1e 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -3,6 +3,7 @@ import pytest from blddirs import build_dir_struct +from common import SKIP_IF_ROOT from pathlib import Path @@ -15,11 +16,26 @@ def test_put_d(lsftp, remote_tmpdir, tmp_path): assert lsftp.listdir(remote) == ['make.txt'] -# TODO -# def test_put_d_ro(lsftp): -# '''test put_d failure on remote read-only server''' -# with pytest.raises(IOError): -# lsftp.put_d('.', '.') + +@SKIP_IF_ROOT +@pytest.mark.parametrize('refuse', ('mkdir', 'write')) +def test_put_d_ro(lsftp, refuse, remote_tmpdir, tmp_path): + '''test put_d failure on remote read-only server''' + build_dir_struct(tmp_path.as_posix()) + local = tmp_path.joinpath('pub').as_posix() + + if refuse == 'mkdir': + remote = remote_tmpdir + else: + remote = Path(remote_tmpdir).joinpath('pub').as_posix() + lsftp.mkdir_p(remote) + + lsftp.chmod(remote, 500) + try: + with pytest.raises(PermissionError): + lsftp.put_d(local, remote_tmpdir) + finally: + lsftp.chmod(remote, 700) def test_put_d_bad_local(lsftp, remote_tmpdir): diff --git a/tests/test_truncate.py b/tests/test_truncate.py index 3d933c61..07cf0bd9 100644 --- a/tests/test_truncate.py +++ b/tests/test_truncate.py @@ -4,7 +4,7 @@ from common import STARS8192 from io import BytesIO -from path import Path +from pathlib import Path @pytest.mark.parametrize('size', (2 * 8192, 8192, 4096), From b11270e2829ff05c3e42947e6ea326c131b04073 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Wed, 26 Aug 2026 04:41:47 +0000 Subject: [PATCH 10/20] lint trapped for spaces before my comments, duplicate key line in private key path tests meant the private key value in copts wasn't being set correctly, doh --- tests/test_connection.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/test_connection.py b/tests/test_connection.py index c40dbf9c..74551d45 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -95,11 +95,11 @@ def test_connection_bad_host(): @pytest.mark.parametrize('blob', ( - b'\x30\x82\x04\xbe\x02\x01\x00', # binary DER, undecodable - b'-----BEGIN DSA PRIVATE KEY-----\n', # deprecated algorithm - b'-----BEGIN ENCRYPTED PRIVATE KEY-----\n', # PKCS#8, encrypted - b'-----BEGIN PRIVATE KEY-----\n', # PKCS#8 - b'' # empty file + b'\x30\x82\x04\xbe\x02\x01\x00', # binary DER, undecodable + b'-----BEGIN DSA PRIVATE KEY-----\n', # deprecated algorithm + b'-----BEGIN ENCRYPTED PRIVATE KEY-----\n', # PKCS#8, encrypted + b'-----BEGIN PRIVATE KEY-----\n', # PKCS#8 + b'' # empty file )) def test_connection_bad_private_key_format(blob, tmp_path): '''deprecated or unsupported key formats must raise, not fail''' @@ -122,7 +122,7 @@ def test_connection_bad_private_key_path(kind, tmp_path): key.mkdir() copts = LOCAL.copy() - key = tmp_path.joinpath(f'id_sftpretty_{kind}') + copts['private_key'] = key.as_posix() with pytest.raises(OSError, match=key.name): with Connection(**copts) as sftp: From 40c15a96a5440079812ef0419e07a3870675fa78 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Sun, 30 Aug 2026 06:52:52 +0000 Subject: [PATCH 11/20] fix for usage of secrets in tests that was preventing tests from running for Linux & Windows with PR's opened from forks. --- .github/workflows/test.yml | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 633d362d..5f7c640f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -27,6 +27,18 @@ jobs: uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} + - name: Runner Environment + run: | + PASS=$(grep -m1 '^PASS = ' tests/common.py | cut -d\' -f2) + if [ -z "$PASS" ]; then + LINE=$(awk '/^PASS/{print NR; exit}' tests/common.py) + echo "::error file=tests/common.py,line=${LINE:-1},title=Test passphrase unreadable::No single-quoted value found for the PASS assignment." + exit 1 + fi + PASSWORD="$(openssl rand -base64 33)" + echo "::add-mask::$PASSWORD" + echo "SFTPRETTY_KEY_PASS=$PASS" >> $GITHUB_ENV + echo "PASSWORD=$PASSWORD" >> $GITHUB_ENV - name: ${{ matrix.os }} SSH if: startsWith(matrix.os, 'macos') run: | @@ -35,12 +47,12 @@ jobs: - name: ${{ matrix.os }} SSH if: startsWith(matrix.os, 'ubuntu') run: | - (echo ${{ secrets.PASSWORD }}; echo ${{ secrets.PASSWORD }}) | sudo passwd $USER + (echo "$PASSWORD"; echo "$PASSWORD") | sudo passwd $USER - name: ${{ matrix.os }} SSH if: startsWith(matrix.os, 'windows') run: | $authorizedKey = Get-Content -Path id_sftpretty.pub - $pass = ConvertTo-SecureString -AsPlainText -Force -String ${{ secrets.PASSWORD }} + $pass = ConvertTo-SecureString -AsPlainText -Force -String $env:PASSWORD $privateKey = Get-Content -Path id_sftpretty $user = Get-LocalUser -Name (([System.Environment]::UserName)) $user | Set-LocalUser -Password $pass @@ -58,7 +70,7 @@ jobs: Get-WindowsCapability -Online | Where-Object Name -like 'OpenSSH*' Set-Service -Name sshd -StartupType Automatic -Status Running Set-Service -Name ssh-agent -StartupType Automatic -Status Running - ssh-keygen -f $Key --% -N "" -p -P ${{ secrets.PRIVATE_KEY }} + ssh-keygen -f $Key --% -N "" -p -P %SFTPRETTY_KEY_PASS% if (!(Get-NetFirewallRule -Name 'OpenSSH-Server-In-TCP' -ErrorAction SilentlyContinue | Select-Object Name, Enabled)) { Write-Output "Firewall Rule 'OpenSSH-Server-In-TCP' does not exist, creating it..." New-NetFirewallRule -Name 'OpenSSH-Server-In-TCP' -DisplayName 'OpenSSH Server (sshd)' -Enabled True -Direction Inbound -Protocol TCP -Action Allow -LocalPort 22 From 0e44bb702498b972712a7d180cdc6ce02a5478bb Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Sun, 30 Aug 2026 07:01:38 +0000 Subject: [PATCH 12/20] be explicit about the shell for new runner env section of test workflow --- .github/workflows/test.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5f7c640f..089336b3 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -39,6 +39,7 @@ jobs: echo "::add-mask::$PASSWORD" echo "SFTPRETTY_KEY_PASS=$PASS" >> $GITHUB_ENV echo "PASSWORD=$PASSWORD" >> $GITHUB_ENV + shell: bash - name: ${{ matrix.os }} SSH if: startsWith(matrix.os, 'macos') run: | From ff90d0a616dea2aa7c6e5f37a054bae2afc3e88b Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Thu, 3 Sep 2026 05:28:40 +0000 Subject: [PATCH 13/20] fell down a rabbit hole fixing gaps in old drivedrop helper function. Normalize everything to posix syntax, always keep drive and improve unc path handling. Improved utf8 character support to prevent path mangaling during normalization. Added test for new drivepath function. Added remotepath call to get, getfo, putfo, exists, isdir, isfile. --- sftpretty/__init__.py | 66 ++++++++++++++++++++++------------------- sftpretty/helpers.py | 35 +++++++++++++++++----- tests/conftest.py | 2 +- tests/test_helpers.py | 47 +++++++++++++++++++++++++++++ tests/test_normalize.py | 3 +- tests/test_put_d.py | 3 +- tests/test_readlink.py | 3 +- tests/test_rmdir.py | 3 +- 8 files changed, 119 insertions(+), 43 deletions(-) create mode 100644 tests/test_helpers.py diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 5dfb1b97..c1578901 100644 --- a/sftpretty/__init__.py +++ b/sftpretty/__init__.py @@ -13,7 +13,7 @@ from pathlib import Path from sftpretty.exceptions import (CredentialException, ConnectionException, HostKeysException, LoggingException) -from sftpretty.helpers import _callback, drivedrop, hash, localtree, retry +from sftpretty.helpers import _callback, drivepath, hash, localtree, retry from socket import gaierror, timeout from stat import S_ISDIR, S_ISREG from tempfile import mkstemp @@ -340,9 +340,9 @@ def _sftp_channel(self): meta.settimeout(self._timeout) if self._cache.cwd is None: - self._cache.cwd = drivedrop(channel.normalize('.')) + self._cache.cwd = drivepath(channel.normalize('.')) - channel.chdir(drivedrop(self._cache.cwd)) + channel.chdir(drivepath(self._cache.cwd)) log.info(f'Current Working Directory: [{self._cache.cwd}]') yield channel @@ -565,6 +565,8 @@ def _get(self, remotefile, localpath=None, callback=None, max_concurrent_prefetch_requests=None, prefetch=True, preserve_mtime=False, resume=False): + remotefile = drivepath(remotefile) + if localpath is None: localpath = Path(remotefile).name @@ -816,6 +818,8 @@ def getfo(self, remotefile, flo, callback=None, def _getfo(self, remotefile, flo, callback=None, max_concurrent_prefetch_requests=None, prefetch=True): + remotefile = drivepath(remotefile) + if callback is None: callback = partial(_callback, remotefile, logger=logger) @@ -885,7 +889,7 @@ def _put(self, localfile, remotepath=None, callback=None, local_attributes.st_mtime) with self._sftp_channel() as channel: - remotepath = drivedrop(remotepath) + remotepath = drivepath(remotepath) if resume: remote = channel.stat(remotepath) if S_ISREG(remote.st_mode): @@ -1117,6 +1121,8 @@ def _putfo(self, flo, remotepath=None, file_size=None, callback=None, if remotepath is None: remotepath = uuid4().hex + else: + remotepath = drivepath(remotepath) with self._sftp_channel() as channel: attributes = channel.putfo(flo, remotepath=remotepath, @@ -1201,8 +1207,8 @@ def chdir(self, remotepath): :raises: IOError, if path does not exist ''' with self._sftp_channel() as channel: - channel.chdir(drivedrop(remotepath)) - self._cache.cwd = drivedrop(channel.normalize('.')) + channel.chdir(drivepath(remotepath)) + self._cache.cwd = drivepath(channel.normalize('.')) def chmod(self, remotepath, mode=700): '''Set the permission mode of a remotepath, where mode is an octal. @@ -1215,7 +1221,7 @@ def chmod(self, remotepath, mode=700): :raises: IOError, if the file doesn't exist ''' with self._sftp_channel() as channel: - channel.chmod(drivedrop(remotepath), mode=int(str(mode), 8)) + channel.chmod(drivepath(remotepath), mode=int(str(mode), 8)) def chown(self, remotepath, uid=None, gid=None): '''Set uid/gid on remotepath, you may specify either or both. @@ -1229,7 +1235,7 @@ def chown(self, remotepath, uid=None, gid=None): :raises: IOError, if user lacks permission or if the file doesn't exist ''' with self._sftp_channel() as channel: - remotepath = drivedrop(remotepath) + remotepath = drivepath(remotepath) if uid is None or gid is None: if uid is None and gid is None: return @@ -1275,7 +1281,7 @@ def exists(self, remotepath): ''' with self._sftp_channel() as channel: try: - channel.stat(remotepath) + channel.stat(drivepath(remotepath)) except IOError as err: if err.errno == 2: return False @@ -1290,7 +1296,7 @@ def getcwd(self): :returns: (str) Remote current working directory. None, if not set. ''' with self._sftp_channel() as channel: - cwd = drivedrop(channel.getcwd()) + cwd = drivepath(channel.getcwd()) return cwd @@ -1303,7 +1309,7 @@ def isdir(self, remotepath): ''' with self._sftp_channel() as channel: try: - result = S_ISDIR(channel.stat(remotepath).st_mode) + result = S_ISDIR(channel.stat(drivepath(remotepath)).st_mode) except IOError: # No such directory result = False @@ -1319,7 +1325,7 @@ def isfile(self, remotepath): ''' with self._sftp_channel() as channel: try: - result = S_ISREG(channel.stat(remotepath).st_mode) + result = S_ISREG(channel.stat(drivepath(remotepath)).st_mode) except IOError: # No such file result = False @@ -1335,7 +1341,7 @@ def lexists(self, remotepath): ''' with self._sftp_channel() as channel: try: - channel.lstat(drivedrop(remotepath)) + channel.lstat(drivepath(remotepath)) except IOError: return False @@ -1350,7 +1356,7 @@ def listdir(self, remotepath='.'): ''' with self._sftp_channel() as channel: - directory = sorted(channel.listdir(drivedrop(remotepath))) + directory = sorted(channel.listdir(drivepath(remotepath))) return directory @@ -1368,7 +1374,7 @@ def listdir_attr(self, remotepath='.'): :returns: (list of SFTPAttributes) Sorted directory content as objects. ''' with self._sftp_channel() as channel: - directory = sorted(channel.listdir_attr(drivedrop(remotepath)), + directory = sorted(channel.listdir_attr(drivepath(remotepath)), key=lambda attribute: attribute.filename) return directory @@ -1382,7 +1388,7 @@ def lstat(self, remotepath): :returns: (obj) SFTPAttributes object ''' with self._sftp_channel() as channel: - lstat = channel.lstat(drivedrop(remotepath)) + lstat = channel.lstat(drivepath(remotepath)) return lstat @@ -1396,7 +1402,7 @@ def mkdir(self, remotedir, mode=700): :returns: None ''' with self._sftp_channel() as channel: - channel.mkdir(drivedrop(remotedir), mode=int(str(mode), 8)) + channel.mkdir(drivepath(remotedir), mode=int(str(mode), 8)) def mkdir_p(self, remotedir, mode=700): '''Create a directory and any missing parent locations as needed. Set @@ -1411,7 +1417,7 @@ def mkdir_p(self, remotedir, mode=700): :raises: OSError ''' try: - remotedir = drivedrop(remotedir) + remotedir = drivepath(remotedir) if self.isdir(remotedir): return elif self.isfile(remotedir): @@ -1440,9 +1446,9 @@ def normalize(self, remotepath): :raises: IOError, if remotepath can't be resolved ''' with self._sftp_channel() as channel: - absolute = channel.normalize(drivedrop(remotepath)) + absolute = channel.normalize(drivepath(remotepath)) - return drivedrop(absolute) + return drivepath(absolute) def open(self, remotefile, bufsize=-1, mode='r'): '''Open a file on the remote server. @@ -1456,7 +1462,7 @@ def open(self, remotefile, bufsize=-1, mode='r'): :raises: IOError, if the file could not be opened. ''' with self._sftp_channel() as channel: - remotefile = drivedrop(remotefile) + remotefile = drivepath(remotefile) flo = channel.open(remotefile, bufsize=bufsize, mode=mode) return flo @@ -1469,10 +1475,10 @@ def readlink(self, remotelink): :return: (str) Absolute path to target. ''' with self._sftp_channel() as channel: - remotelink = drivedrop(remotelink) + remotelink = drivepath(remotelink) link_destination = channel.normalize(channel.readlink(remotelink)) - return drivedrop(link_destination) + return drivepath(link_destination) def remotetree(self, container, remotedir, localdir, recurse=True): '''Recursively map remote directory tree to a dictionary container. @@ -1519,7 +1525,7 @@ def remove(self, remotefile): :raises: IOError ''' with self._sftp_channel() as channel: - channel.remove(drivedrop(remotefile)) + channel.remove(drivepath(remotefile)) def rename(self, remotepath, newpath, posix=True): '''Rename a path on the remote host. @@ -1536,7 +1542,7 @@ def rename(self, remotepath, newpath, posix=True): ''' with self._sftp_channel() as channel: renamer = channel.posix_rename if posix else channel.rename - renamer(drivedrop(remotepath), drivedrop(newpath)) + renamer(drivepath(remotepath), drivepath(newpath)) def rmdir(self, remotedir): '''Delete remote directory. @@ -1546,7 +1552,7 @@ def rmdir(self, remotedir): :returns: None ''' with self._sftp_channel() as channel: - channel.rmdir(drivedrop(remotedir)) + channel.rmdir(drivepath(remotedir)) def stat(self, remotepath): '''Return information about remote location. @@ -1556,7 +1562,7 @@ def stat(self, remotepath): :returns: (obj) SFTPAttributes ''' with self._sftp_channel() as channel: - stat = channel.stat(drivedrop(remotepath)) + stat = channel.stat(drivepath(remotepath)) return stat @@ -1571,7 +1577,7 @@ def symlink(self, remote_src, remote_dest): :raises: any underlying error, IOError if remote_dest already exists ''' with self._sftp_channel() as channel: - channel.symlink(remote_src, drivedrop(remote_dest)) + channel.symlink(drivepath(remote_src), drivepath(remote_dest)) def truncate(self, remotepath, size): '''Change the size of the file specified by path. Used to modify the @@ -1586,7 +1592,7 @@ def truncate(self, remotepath, size): :raises: IOError, if file does not exist ''' with self._sftp_channel() as channel: - remotepath = drivedrop(remotepath) + remotepath = drivepath(remotepath) channel.truncate(remotepath, size) size = channel.stat(remotepath).st_size @@ -1629,7 +1635,7 @@ def pwd(self): :returns: (str) Current working directory. ''' with self._sftp_channel() as channel: - self._cache.cwd = drivedrop(channel.normalize('.')) + self._cache.cwd = drivepath(channel.normalize('.')) return self._cache.cwd diff --git a/sftpretty/helpers.py b/sftpretty/helpers.py index 8233c246..fd743be3 100644 --- a/sftpretty/helpers.py +++ b/sftpretty/helpers.py @@ -2,6 +2,7 @@ from hashlib import new, sha3_512 from io import BytesIO, IOBase from pathlib import Path, PurePosixPath, PureWindowsPath +from re import sub from stat import S_IMODE from time import sleep @@ -16,15 +17,33 @@ def _callback(filename, bytes_so_far, bytes_total, logger=None): print(message) -def drivedrop(filepath): +def drivepath(filepath): + '''Normalize a filepath to POSIX form, retaining any drive letter + + :param str filename: + path to file or string to process + + :returns str: normalized POSIX path + ''' if filepath: - if PureWindowsPath(filepath).drive and not filepath.startswith('//'): - filepath = PurePosixPath('/').joinpath( - *PureWindowsPath(filepath).parts[1:]).as_posix() - filepath = filepath.encode('unicode_escape').decode() - filepath = filepath.replace('\\', '/').replace('//', '/') - elif filepath.startswith('//'): - filepath = PurePosixPath(filepath.replace('//', '/')).as_posix() + if '\\' in filepath or PureWindowsPath(filepath).drive: + unc = filepath[:1] == '\\' or filepath[:2] == '//' + utf = filepath.encode('unicode_escape').decode() + utf = utf.replace('\\\\', '/') + utf = sub(r'\\([^xuU])', r'/\1', utf) + filepath = sub('/{2,}', '/', + utf.encode('ascii').decode('unicode_escape')) + winpath = PureWindowsPath(filepath) + drive = winpath.drive + filepath = winpath.as_posix() + if unc: + filepath = f'/{filepath}' + elif drive: + if not winpath.root: + filepath = f'{drive}/{filepath[len(drive):]}' + filepath = f'/{filepath}' + if filepath.endswith(':'): + filepath += '/' return filepath diff --git a/tests/conftest.py b/tests/conftest.py index 10f44b1c..d9204359 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -47,6 +47,6 @@ def remote_tmpdir(lsftp): lsftp.mkdir_p(remotedir.as_posix()) try: - yield remotedir.as_posix() + yield lsftp.normalize(remotedir.as_posix()) finally: remote_rmdir(lsftp, remotedir.as_posix()) diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 00000000..c6f165ec --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,47 @@ +'''test sftpretty.helpers''' + +import pytest + +from sftpretty.helpers import drivepath + + +@pytest.mark.parametrize('path,expected', ( + # drive qualified + ('C:\tmp\test.txt', '/C:/tmp/test.txt'), + ('C:\\tmp\test.txt', '/C:/tmp/test.txt'), + ('C:\notes\run.txt', '/C:/notes/run.txt'), + ('C:\\Users\\nick\\file.txt', '/C:/Users/nick/file.txt'), + ('C:/tmp/test.txt', '/C:/tmp/test.txt'), + ('C:\\tmp/mixed\\sep.txt', '/C:/tmp/mixed/sep.txt'), + ('D:\\data\\x.txt', '/D:/data/x.txt'), + ('c:\\lower\\case.txt', '/c:/lower/case.txt'), + # \t \n \r recover, \a \b \f \v ride through as the chars Python made + ('C:\bin\app.exe', '/C:/\bin\app.exe'), + # drive relative and drive roots + ('C:tmp\\test.txt', '/C:/tmp/test.txt'), + ('C:', '/C:/'), ('C:/', '/C:/'), ('C:\\', '/C:/'), + ('/C:', '/C:/'), ('/C:/', '/C:/'), + # leading backslash is UNC, typed pairs arrive collapsed + ('\\\\server\\share\\file.txt', '//server/share/file.txt'), + ('\\server\share\file.txt', '//server/share\file.txt'), # \s survives + ('\\tmp\test.txt', '//tmp/test.txt'), + ('//tmp/test.txt', '//tmp/test.txt'), + ('//server/share//dbl/f.txt', '//server/share/dbl/f.txt'), + # relative + ('tmp\\test.txt', 'tmp/test.txt'), + ('relative/file.txt', 'relative/file.txt'), + # canonical and posix forms pass through untouched + ('/C:/Users/x', '/C:/Users/x'), + ('/cygdrive/c/Users/x', '/cygdrive/c/Users/x'), + ('/home/user/file.txt', '/home/user/file.txt'), + ('/home/user/we\tird.txt', '/home/user/we\tird.txt'), + # data survives the conversion + ('C:/tmp/café.txt', '/C:/tmp/café.txt'), + ('C:/tmp/日本語.txt', '/C:/tmp/日本語.txt'), + ('C:/a//b/c.txt', '/C:/a/b/c.txt'), + # degenerate + ('', ''), (None, None) +)) +def test_drivepath(path, expected): + assert drivepath(path) == expected + assert drivepath(expected) == expected diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 2ddb815b..93be9cfc 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -1,6 +1,6 @@ '''test sftpretty.normalize''' -from common import VFS, conn +from common import conn, SKIP_IF_WIN, VFS from io import BytesIO from pathlib import Path from sftpretty import Connection @@ -20,6 +20,7 @@ def test_normalize(sftpserver): assert sftp.normalize('.') == pubpath.as_posix() +@SKIP_IF_WIN def test_normalize_dangling_symlink(lsftp, remote_tmpdir): '''test normalize against a symlink whose target is missing''' missing = Path(remote_tmpdir).joinpath('gone.txt').as_posix() diff --git a/tests/test_put_d.py b/tests/test_put_d.py index a1f37c1e..93c63be4 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -3,7 +3,7 @@ import pytest from blddirs import build_dir_struct -from common import SKIP_IF_ROOT +from common import SKIP_IF_ROOT, SKIP_IF_WIN from pathlib import Path @@ -18,6 +18,7 @@ def test_put_d(lsftp, remote_tmpdir, tmp_path): @SKIP_IF_ROOT +@SKIP_IF_WIN @pytest.mark.parametrize('refuse', ('mkdir', 'write')) def test_put_d_ro(lsftp, refuse, remote_tmpdir, tmp_path): '''test put_d failure on remote read-only server''' diff --git a/tests/test_readlink.py b/tests/test_readlink.py index f7973d66..b7c88a96 100644 --- a/tests/test_readlink.py +++ b/tests/test_readlink.py @@ -2,6 +2,7 @@ from io import BytesIO from pathlib import Path +from sftpretty.helpers import drivepath def test_readlink(lsftp, remote_tmpdir): @@ -14,4 +15,4 @@ def test_readlink(lsftp, remote_tmpdir): lsftp.putfo(flo, rfile) lsftp.symlink(rfile, rlink) - assert lsftp.readlink(rlink).endswith(rfile) + assert lsftp.readlink(rlink).endswith(drivepath(rfile)) diff --git a/tests/test_rmdir.py b/tests/test_rmdir.py index 96b0d0c0..3c9eb51a 100644 --- a/tests/test_rmdir.py +++ b/tests/test_rmdir.py @@ -2,7 +2,7 @@ import pytest -from common import SKIP_IF_ROOT +from common import SKIP_IF_ROOT, SKIP_IF_WIN from pathlib import Path @@ -17,6 +17,7 @@ def test_rmdir(lsftp, remote_tmpdir): @SKIP_IF_ROOT +@SKIP_IF_WIN def test_rmdir_ro(lsftp, remote_tmpdir): '''test rmdir against read-only server''' parent = Path(remote_tmpdir).joinpath('readonly') From d4cdbe3342491c483df10af07f666bd641ed4f00 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Thu, 3 Sep 2026 07:49:26 +0000 Subject: [PATCH 14/20] symlink normalization test can't run on Windows runners due to no realpath in Windows C runtime, implementation uses lexical normalization that doesn't dereference on disk. Adding read only tests for chmod and chown for non-Windows runners. --- tests/test_chmod.py | 28 ++++++++++++++++++---------- tests/test_chown.py | 21 ++++++++++++++------- tests/test_helpers.py | 2 +- tests/test_normalize.py | 3 ++- tests/test_put_d.py | 2 +- tests/test_rmdir.py | 2 +- 6 files changed, 37 insertions(+), 21 deletions(-) diff --git a/tests/test_chmod.py b/tests/test_chmod.py index cc7d03f6..e79239a0 100644 --- a/tests/test_chmod.py +++ b/tests/test_chmod.py @@ -16,7 +16,24 @@ def test_chmod_not_exist(sftpserver): sftp.chmod('i-do-not-exist.txt', 666) -@SKIP_IF_WIN +@SKIP_IF_ROOT +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +def test_chmod_ro(lsftp, remote_tmpdir): + '''test chmod against read-only path''' + 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()) + lsftp.chmod(parent.as_posix(), 400) # no search bit, 500 fails + try: + with pytest.raises(PermissionError): + lsftp.chmod(rfile.as_posix(), 440) + finally: + lsftp.chmod(parent.as_posix(), 700) + + +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs def test_chmod_simple(lsftp): '''test basic chmod with octal mode represented by an int''' new_mode = 711 @@ -29,12 +46,3 @@ def test_chmod_simple(lsftp): assert st_mode_to_int(new_attrs.st_mode) == new_mode assert new_attrs.st_mode != org_attrs.st_mode - - -# TODO -# def test_chmod_fail_ro(psftp): -# '''test chmod against read-only server''' -# new_mode = 440 -# fname = 'readme.txt' -# with pytest.raises(IOError): -# psftp.chmod(fname, new_mode) diff --git a/tests/test_chown.py b/tests/test_chown.py index 5035aa46..50a1586f 100644 --- a/tests/test_chown.py +++ b/tests/test_chown.py @@ -35,7 +35,7 @@ def test_chown_gid(lsftp): def test_chown_none(lsftp): - '''call .chown with no gid or uid specified''' + '''call chown with no gid or uid specified''' with tempfile_containing() as fname: base_fname = Path(fname).name org_attrs = lsftp.put(fname) @@ -47,13 +47,20 @@ def test_chown_none(lsftp): def test_chown_not_exist(lsftp): - '''call .chown on a non-existing path''' + '''call chown on a non-existing path''' with pytest.raises(IOError): lsftp.chown('i-do-not-exist.txt', 666) -# TODO -# def test_chown_ro_server(psftp): -# '''call .chown against path on read-only server''' -# with pytest.raises(IOError): -# psftp.chown('readme.txt', gid=1000, uid=1000) +@SKIP_IF_ROOT +@SKIP_IF_WIN # ownership ids are synthetic, cannot be set +def test_chown_ro(lsftp): + '''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) diff --git a/tests/test_helpers.py b/tests/test_helpers.py index c6f165ec..cd1c31c6 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -23,7 +23,7 @@ ('/C:', '/C:/'), ('/C:/', '/C:/'), # leading backslash is UNC, typed pairs arrive collapsed ('\\\\server\\share\\file.txt', '//server/share/file.txt'), - ('\\server\share\file.txt', '//server/share\file.txt'), # \s survives + ('\\server\share\file.txt', '//server/share\file.txt'), ('\\tmp\test.txt', '//tmp/test.txt'), ('//tmp/test.txt', '//tmp/test.txt'), ('//server/share//dbl/f.txt', '//server/share/dbl/f.txt'), diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 93be9cfc..6221ac5f 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -20,7 +20,7 @@ def test_normalize(sftpserver): assert sftp.normalize('.') == pubpath.as_posix() -@SKIP_IF_WIN +@SKIP_IF_WIN # CreateSymbolicLinkW stats target on creation, returns ENOENT def test_normalize_dangling_symlink(lsftp, remote_tmpdir): '''test normalize against a symlink whose target is missing''' missing = Path(remote_tmpdir).joinpath('gone.txt').as_posix() @@ -32,6 +32,7 @@ def test_normalize_dangling_symlink(lsftp, remote_tmpdir): assert lsftp.normalize(rsym) == missing +@SKIP_IF_WIN # uses lexical _wfullpath instead of realpath, undereferenced def test_normalize_symlink(lsftp, remote_tmpdir): '''test normalize against a symlink''' rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() diff --git a/tests/test_put_d.py b/tests/test_put_d.py index 93c63be4..8615caa1 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -18,7 +18,7 @@ def test_put_d(lsftp, remote_tmpdir, tmp_path): @SKIP_IF_ROOT -@SKIP_IF_WIN +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs @pytest.mark.parametrize('refuse', ('mkdir', 'write')) def test_put_d_ro(lsftp, refuse, remote_tmpdir, tmp_path): '''test put_d failure on remote read-only server''' diff --git a/tests/test_rmdir.py b/tests/test_rmdir.py index 3c9eb51a..5cb64c2c 100644 --- a/tests/test_rmdir.py +++ b/tests/test_rmdir.py @@ -17,7 +17,7 @@ def test_rmdir(lsftp, remote_tmpdir): @SKIP_IF_ROOT -@SKIP_IF_WIN +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs def test_rmdir_ro(lsftp, remote_tmpdir): '''test rmdir against read-only server''' parent = Path(remote_tmpdir).joinpath('readonly') From 62ef0b201d138c65243c9bc369ab2fe92a0c879d Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Thu, 3 Sep 2026 08:04:01 +0000 Subject: [PATCH 15/20] missed an import for read-only tests and adding explicit pkey object test to connection suite --- tests/test_chmod.py | 2 +- tests/test_chown.py | 2 +- tests/test_connection.py | 16 +++++++++++++--- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/tests/test_chmod.py b/tests/test_chmod.py index e79239a0..8ae3bb21 100644 --- a/tests/test_chmod.py +++ b/tests/test_chmod.py @@ -2,7 +2,7 @@ import pytest -from common import conn, SKIP_IF_WIN, tempfile_containing, VFS +from common import conn, SKIP_IF_ROOT, SKIP_IF_WIN, tempfile_containing, VFS from pathlib import Path from sftpretty import Connection from sftpretty.helpers import st_mode_to_int diff --git a/tests/test_chown.py b/tests/test_chown.py index 50a1586f..6d0c2a80 100644 --- a/tests/test_chown.py +++ b/tests/test_chown.py @@ -2,7 +2,7 @@ import pytest -from common import SKIP_IF_WIN, tempfile_containing +from common import SKIP_IF_ROOT, SKIP_IF_WIN, tempfile_containing from pathlib import Path diff --git a/tests/test_connection.py b/tests/test_connection.py index 74551d45..254f4abd 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -129,10 +129,20 @@ def test_connection_bad_private_key_path(kind, tmp_path): sftp.listdir() -def test_connection_good(sftpserver): - '''connect to a public sftp server''' +@pytest.mark.parametrize('kind', ('path', 'pkey')) +def test_connection_good(kind, sftpserver): + '''connect to a public sftp server with key given as path or object''' + copts = conn(sftpserver) + + if kind == 'pkey': + copts['private_key'] = Ed25519Key( + filename=copts['private_key'], + password=copts['private_key_pass']) + del copts['private_key_pass'] + with sftpserver.serve_content(VFS): - sftp = Connection(**conn(sftpserver)) + sftp = Connection(**copts) + assert sftp.listdir() == ['pub', 'read.me'] sftp.close() From 9a9731e70ae7fb9b5189ce79a2dee434a77b99c4 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Thu, 3 Sep 2026 08:12:47 +0000 Subject: [PATCH 16/20] lint trapped, invalid escape sequence and spaces before comment --- sftpretty/helpers.py | 2 +- tests/test_chmod.py | 4 ++-- tests/test_helpers.py | 2 +- tests/test_normalize.py | 4 ++-- tests/test_put_d.py | 2 +- tests/test_rmdir.py | 2 +- 6 files changed, 8 insertions(+), 8 deletions(-) diff --git a/sftpretty/helpers.py b/sftpretty/helpers.py index fd743be3..e9b273b9 100644 --- a/sftpretty/helpers.py +++ b/sftpretty/helpers.py @@ -1,7 +1,7 @@ from functools import wraps from hashlib import new, sha3_512 from io import BytesIO, IOBase -from pathlib import Path, PurePosixPath, PureWindowsPath +from pathlib import Path, PureWindowsPath from re import sub from stat import S_IMODE from time import sleep diff --git a/tests/test_chmod.py b/tests/test_chmod.py index 8ae3bb21..a80f879e 100644 --- a/tests/test_chmod.py +++ b/tests/test_chmod.py @@ -17,7 +17,7 @@ def test_chmod_not_exist(sftpserver): @SKIP_IF_ROOT -@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs def test_chmod_ro(lsftp, remote_tmpdir): '''test chmod against read-only path''' parent = Path(remote_tmpdir).joinpath('readonly') @@ -33,7 +33,7 @@ def test_chmod_ro(lsftp, remote_tmpdir): lsftp.chmod(parent.as_posix(), 700) -@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs def test_chmod_simple(lsftp): '''test basic chmod with octal mode represented by an int''' new_mode = 711 diff --git a/tests/test_helpers.py b/tests/test_helpers.py index cd1c31c6..1bf3f65f 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -23,7 +23,7 @@ ('/C:', '/C:/'), ('/C:/', '/C:/'), # leading backslash is UNC, typed pairs arrive collapsed ('\\\\server\\share\\file.txt', '//server/share/file.txt'), - ('\\server\share\file.txt', '//server/share\file.txt'), + (r'\\server\share\file.txt', '//server/share\file.txt'), ('\\tmp\test.txt', '//tmp/test.txt'), ('//tmp/test.txt', '//tmp/test.txt'), ('//server/share//dbl/f.txt', '//server/share/dbl/f.txt'), diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 6221ac5f..88ee68ef 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -20,7 +20,7 @@ def test_normalize(sftpserver): assert sftp.normalize('.') == pubpath.as_posix() -@SKIP_IF_WIN # CreateSymbolicLinkW stats target on creation, returns ENOENT +@SKIP_IF_WIN # CreateSymbolicLinkW stats target on creation, returns ENOENT def test_normalize_dangling_symlink(lsftp, remote_tmpdir): '''test normalize against a symlink whose target is missing''' missing = Path(remote_tmpdir).joinpath('gone.txt').as_posix() @@ -32,7 +32,7 @@ def test_normalize_dangling_symlink(lsftp, remote_tmpdir): assert lsftp.normalize(rsym) == missing -@SKIP_IF_WIN # uses lexical _wfullpath instead of realpath, undereferenced +@SKIP_IF_WIN # uses lexical _wfullpath instead of realpath, undereferenced def test_normalize_symlink(lsftp, remote_tmpdir): '''test normalize against a symlink''' rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix() diff --git a/tests/test_put_d.py b/tests/test_put_d.py index 8615caa1..2ba52c85 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -18,7 +18,7 @@ def test_put_d(lsftp, remote_tmpdir, tmp_path): @SKIP_IF_ROOT -@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs @pytest.mark.parametrize('refuse', ('mkdir', 'write')) def test_put_d_ro(lsftp, refuse, remote_tmpdir, tmp_path): '''test put_d failure on remote read-only server''' diff --git a/tests/test_rmdir.py b/tests/test_rmdir.py index 5cb64c2c..224bfada 100644 --- a/tests/test_rmdir.py +++ b/tests/test_rmdir.py @@ -17,7 +17,7 @@ def test_rmdir(lsftp, remote_tmpdir): @SKIP_IF_ROOT -@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs +@SKIP_IF_WIN # Win32-OpenSSH doesn't translate mode bits into ACLs def test_rmdir_ro(lsftp, remote_tmpdir): '''test rmdir against read-only server''' parent = Path(remote_tmpdir).joinpath('readonly') From 9fc0f38e39d5d46515e16a5303d953fc42214b8f Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Thu, 3 Sep 2026 08:22:56 +0000 Subject: [PATCH 17/20] suppress W605 warning on drivepath test line --- tests/test_helpers.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 1bf3f65f..c5f5feb3 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -23,7 +23,7 @@ ('/C:', '/C:/'), ('/C:/', '/C:/'), # leading backslash is UNC, typed pairs arrive collapsed ('\\\\server\\share\\file.txt', '//server/share/file.txt'), - (r'\\server\share\file.txt', '//server/share\file.txt'), + ('\\server\share\file.txt', '//server/share\file.txt'), # noqa: W605 ('\\tmp\test.txt', '//tmp/test.txt'), ('//tmp/test.txt', '//tmp/test.txt'), ('//server/share//dbl/f.txt', '//server/share/dbl/f.txt'), From 1e2974b5b3de4aff31256a3f055aef6b2bd9b55b Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Sun, 6 Sep 2026 12:42:22 +0000 Subject: [PATCH 18/20] switching to an os derived canonical test home for the VFS. Updating drivepath to better distinguish real unc input from os.path.isabs behavior introduced in python >= 3.13 on windows hosts, updating tests affected by the VFS change. --- sftpretty/helpers.py | 4 +++- tests/common.py | 41 +++++++++++++++++----------------------- tests/test_cd.py | 29 +++++++++++++++------------- tests/test_connection.py | 4 ++-- tests/test_getcwd.py | 9 +++++---- tests/test_issue_65.py | 6 +++--- tests/test_issue_xx.py | 15 ++++++++------- tests/test_normalize.py | 23 ++++++++++++---------- tests/test_remotetree.py | 23 ++++++++++++---------- 9 files changed, 80 insertions(+), 74 deletions(-) diff --git a/sftpretty/helpers.py b/sftpretty/helpers.py index e9b273b9..9fa45b5d 100644 --- a/sftpretty/helpers.py +++ b/sftpretty/helpers.py @@ -27,7 +27,9 @@ def drivepath(filepath): ''' if filepath: if '\\' in filepath or PureWindowsPath(filepath).drive: - unc = filepath[:1] == '\\' or filepath[:2] == '//' + host = filepath.lstrip('\\/') + unc = ((filepath[:1] == '\\' or filepath[:2] == '//') + and host != '' and host[1:2] != ':') utf = filepath.encode('unicode_escape').decode() utf = utf.replace('\\\\', '/') utf = sub(r'\\([^xuU])', r'/\1', utf) diff --git a/tests/common.py b/tests/common.py index 84e3348e..8b0bc5e4 100644 --- a/tests/common.py +++ b/tests/common.py @@ -22,6 +22,22 @@ USER = environ.get('USER', environ.get('USERNAME')) USER_HOME = Path.home().as_posix() USER_HOME_PARENT = Path(USER_HOME).parent.as_posix() +VFS = { + 'pub': { + 'foo1': {'foo1.txt': 'content of foo1.txt', + 'image01.jpg': 'data for image01.jpg'}, + 'make.txt': 'content of make.txt', + 'foo2': {'bar1': {'bar1.txt': 'contents bar1.txt'}, + 'foo2.txt': 'content of foo2.txt'} + }, + 'read.me': 'contents of read.me' +} +VFS_HOME = Path(USER_HOME_PARENT).joinpath('test').as_posix() + + +# filesystem served by pytest-sftpserver plugin +for node in reversed(VFS_HOME.strip('/').split('/')): + VFS = {node: VFS} LOCAL = {'default_path': USER_HOME, 'host': 'localhost', @@ -33,7 +49,7 @@ def conn(sftpsrv): '''return a dictionary holding argument info for the sftpretty client''' cnopts = CnOpts(knownhosts='sftpserver.pub') cnopts.log_level = 'debug' - return {'cnopts': cnopts, 'default_path': '/home/test', + return {'cnopts': cnopts, 'default_path': VFS_HOME, 'host': sftpsrv.host, 'port': sftpsrv.port, 'private_key': 'id_sftpretty', 'private_key_pass': PASS, 'username': USER} @@ -82,26 +98,3 @@ def tempfile_containing(contents=STARS8192, suffix=''): yield Path(temp_path).as_posix() finally: Path(temp_path).unlink() - - -# filesystem served by pytest-sftpserver plugin -VFS = { - 'home': { - 'test': { - 'pub': { - 'foo1': { - 'foo1.txt': 'content of foo1.txt', - 'image01.jpg': 'data for image01.jpg' - }, - 'make.txt': 'content of make.txt', - 'foo2': { - 'bar1': { - 'bar1.txt': 'contents bar1.txt' - }, - 'foo2.txt': 'content of foo2.txt' - } - }, - 'read.me': 'contents of read.me' - } - } -} diff --git a/tests/test_cd.py b/tests/test_cd.py index 10188561..05179921 100644 --- a/tests/test_cd.py +++ b/tests/test_cd.py @@ -2,46 +2,49 @@ import pytest -from common import conn, VFS +from common import conn, VFS, VFS_HOME from pathlib import PurePosixPath from sftpretty import Connection +from sftpretty.helpers import drivepath def test_cd_none(sftpserver): '''test sftpretty.cd with None''' - pubpath = PurePosixPath('/home/test').joinpath('pub') + pubpath = PurePosixPath(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd(): sftp.chdir('pub') - assert sftp.pwd == pubpath.as_posix() - assert home == pubpath.parent.as_posix() + assert sftp.pwd == drivepath(pubpath.as_posix()) + assert home == drivepath(pubpath.parent.as_posix()) def test_cd_path(sftpserver): '''test sftpretty.cd with a path''' - pubpath = PurePosixPath('/home/test').joinpath('pub') + pubpath = PurePosixPath(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd('pub'): - assert sftp.pwd == pubpath.as_posix() - assert home == pubpath.parent.as_posix() + assert sftp.pwd == drivepath(pubpath.as_posix()) + assert home == drivepath(pubpath.parent.as_posix()) def test_cd_nested(sftpserver): '''test nested cd's''' - pubpath = PurePosixPath('/home/test').joinpath('pub') + pubpath = PurePosixPath(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd('pub'): - assert sftp.pwd == pubpath.as_posix() + assert sftp.pwd == drivepath(pubpath.as_posix()) with sftp.cd('foo1'): - assert sftp.pwd == pubpath.joinpath('foo1').as_posix() - assert sftp.pwd == pubpath.as_posix() - assert home == pubpath.parent.as_posix() + assert sftp.pwd == drivepath( + pubpath.joinpath('foo1').as_posix() + ) + assert sftp.pwd == drivepath(pubpath.as_posix()) + assert home == drivepath(pubpath.parent.as_posix()) def test_cd_bad_path(sftpserver): @@ -52,4 +55,4 @@ def test_cd_bad_path(sftpserver): with pytest.raises(IOError): with sftp.cd('not-there'): pass - assert home == '/home/test' + assert home == VFS_HOME diff --git a/tests/test_connection.py b/tests/test_connection.py index 254f4abd..2f9cbc64 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -5,7 +5,7 @@ from paramiko import SFTPError from paramiko.ed25519key import Ed25519Key -from common import conn, LOCAL, VFS +from common import conn, LOCAL, VFS, VFS_HOME from pathlib import Path from sftpretty import (CnOpts, Connection, ConnectionException, HostKeysException, SSHException) @@ -27,7 +27,7 @@ def test_channel_exception(sftpserver): with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: with pytest.raises(SFTPError): - sftp.chdir('/home/test/read.me') + sftp.chdir(f'{VFS_HOME}/read.me') with sftpserver.serve_content(VFS): sftp = Connection(**conn(sftpserver)) diff --git a/tests/test_getcwd.py b/tests/test_getcwd.py index 4e922e69..26fb94b3 100644 --- a/tests/test_getcwd.py +++ b/tests/test_getcwd.py @@ -1,8 +1,9 @@ '''test sftpretty.getcwd''' -from common import conn, VFS +from common import conn, VFS, VFS_HOME from pathlib import Path from sftpretty import Connection +from sftpretty.helpers import drivepath def test_getcwd_none(sftpserver): @@ -18,15 +19,15 @@ def test_getcwd_default_path(sftpserver): '''test .getcwd when using default_path arg''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: - assert sftp.getcwd() == '/home/test' + assert sftp.getcwd() == VFS_HOME def test_getcwd_after_chdir(sftpserver): '''test getcwd after a chdir operation''' - pubpath = Path('/home/test').joinpath('pub/foo1') + pubpath = Path(VFS_HOME).joinpath('pub/foo1') with sftpserver.serve_content(VFS): cnn = conn(sftpserver) cnn['default_path'] = None with Connection(**cnn) as sftp: sftp.chdir(pubpath.as_posix()) - assert sftp.getcwd() == pubpath.as_posix() + assert sftp.getcwd() == drivepath(pubpath.as_posix()) diff --git a/tests/test_issue_65.py b/tests/test_issue_65.py index 99d86654..5fb60dcd 100644 --- a/tests/test_issue_65.py +++ b/tests/test_issue_65.py @@ -1,7 +1,7 @@ '''use the cd contextmanager prior to paramiko establishing a directory location''' -from common import conn, VFS +from common import conn, VFS, VFS_HOME from pathlib import PurePosixPath from sftpretty import Connection @@ -9,12 +9,12 @@ def test_issue_65(sftpserver): '''using the .cd() context manager prior to setting a directory via chdir causes an error''' - pubpath = PurePosixPath('/home/test').joinpath('pub') + pubpath = PurePosixPath(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): cnn = conn(sftpserver) cnn['default_path'] = None with Connection(**cnn) as sftp: - assert sftp.getcwd() == '/' + assert sftp.getcwd() == pubpath.root with sftp.cd(pubpath.as_posix()): pass diff --git a/tests/test_issue_xx.py b/tests/test_issue_xx.py index 122d4cd3..392724fc 100644 --- a/tests/test_issue_xx.py +++ b/tests/test_issue_xx.py @@ -1,38 +1,39 @@ '''a template for creating tests that display or duplicate issues''' -from common import conn, USER, USER_HOME_PARENT, VFS +from common import conn, VFS, VFS_HOME from pathlib import Path from sftpretty import Connection +from sftpretty.helpers import drivepath # this is the preferred test type as it can be run on the CI server and -# requires no configuarion of a real sftp server. However issues that involve +# requires no configuarion of a real sftp server. However issues that involve # authentication and/or authorization currently have to use a real sftp # server (see test_issue_xx_lsftp) def test_issue_xx_sftpserver_plugin(sftpserver): '''an example showing how to use the sftpserver plugin in a test''' - testpath = Path('/home/test').joinpath('pub') + testpath = Path(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd(): sftp.chdir('pub') - assert sftp.pwd == testpath.as_posix() - assert home == testpath.parent.as_posix() + assert sftp.pwd == drivepath(testpath.as_posix()) + assert home == drivepath(testpath.parent.as_posix()) def test_issue_xx_local_sftpserver(lsftp): '''same as test_issue_xx_sftpserver_plugin but written with the local sfptserver mechanism, lsftp''' home = lsftp.pwd - testpath = Path(f'{USER_HOME_PARENT}/{USER}').joinpath('pub') + testpath = Path(VFS_HOME).joinpath('pub') # starting condition of default directory should be empty, so we need to # construct whatever structure we need prior to peforming the test lsftp.mkdir('pub') with lsftp.cd(): lsftp.chdir('pub') - pubdir = lsftp.pwd.endswith(testpath.as_posix()) + pubdir = lsftp.pwd.endswith(drivepath(testpath.as_posix())) homedir = home == lsftp.pwd lsftp.rmdir('pub') diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 88ee68ef..d83afc91 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -1,23 +1,24 @@ '''test sftpretty.normalize''' -from common import conn, SKIP_IF_WIN, VFS +from common import conn, SKIP_IF_WIN, VFS, VFS_HOME from io import BytesIO from pathlib import Path from sftpretty import Connection +from sftpretty.helpers import drivepath from stat import S_ISLNK def test_normalize(sftpserver): '''test the normalize function''' - pubpath = Path('/home/test').joinpath('pub') + pubpath = Path(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: makepath = pubpath.parent.joinpath('make.txt').as_posix() - assert sftp.normalize('make.txt') == makepath - assert sftp.normalize('.') == pubpath.parent.as_posix() - assert sftp.normalize('pub') == pubpath.as_posix() + assert sftp.normalize('make.txt') == drivepath(makepath) + assert sftp.normalize('.') == drivepath(pubpath.parent.as_posix()) + assert sftp.normalize('pub') == drivepath(pubpath.as_posix()) sftp.chdir('pub') - assert sftp.normalize('.') == pubpath.as_posix() + assert sftp.normalize('.') == drivepath(pubpath.as_posix()) @SKIP_IF_WIN # CreateSymbolicLinkW stats target on creation, returns ENOENT @@ -47,12 +48,14 @@ def test_normalize_symlink(lsftp, remote_tmpdir): def test_pwd(sftpserver): '''test the pwd property''' - pubpath = Path('/home/test').joinpath('pub') + pubpath = Path(VFS_HOME).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo2') - assert sftp.pwd == pubpath.joinpath('foo2').as_posix() + assert sftp.pwd == drivepath(pubpath.joinpath('foo2').as_posix()) sftp.chdir('bar1') - assert sftp.pwd == pubpath.joinpath('foo2/bar1').as_posix() + assert sftp.pwd == drivepath( + pubpath.joinpath('foo2/bar1').as_posix() + ) sftp.chdir('../../foo1') - assert sftp.pwd == pubpath.joinpath('foo1').as_posix() + assert sftp.pwd == drivepath(pubpath.joinpath('foo1').as_posix()) diff --git a/tests/test_remotetree.py b/tests/test_remotetree.py index cbf2c2ea..681e4e0e 100644 --- a/tests/test_remotetree.py +++ b/tests/test_remotetree.py @@ -1,8 +1,9 @@ '''test sftpretty.remotetree''' -from common import conn, VFS +from common import conn, VFS, VFS_HOME from pathlib import Path from sftpretty import Connection +from sftpretty.helpers import drivepath from tempfile import mkdtemp @@ -12,22 +13,23 @@ def test_remotetree(sftpserver): with Connection(**conn(sftpserver)) as sftp: cwd = sftp.pwd localpath = Path(mkdtemp()).as_posix() + testpath = drivepath(VFS_HOME) tree = {} sftp.remotetree(tree, cwd, localpath) remote = { - '/home/test': [ - ('/home/test/pub', f'{localpath}/pub') + f'{testpath}': [ + (f'{testpath}/pub', f'{localpath}/pub') ], - '/home/test/pub': [ - ('/home/test/pub/foo1', + f'{testpath}/pub': [ + (f'{testpath}/pub/foo1', f'{localpath}/pub/foo1'), - ('/home/test/pub/foo2', + (f'{testpath}/pub/foo2', f'{localpath}/pub/foo2') ], - '/home/test/pub/foo2': [ - ('/home/test/pub/foo2/bar1', + f'{testpath}/pub/foo2': [ + (f'{testpath}/pub/foo2/bar1', f'{localpath}/pub/foo2/bar1') ] } @@ -44,13 +46,14 @@ def test_remotetree_no_recurse(sftpserver): with Connection(**conn(sftpserver)) as sftp: cwd = sftp.pwd localpath = Path(mkdtemp()).as_posix() + testpath = drivepath(VFS_HOME) tree = {} sftp.remotetree(tree, cwd, localpath, recurse=False) remote = { - '/home/test': [ - ('/home/test/pub', f'{localpath}/pub') + f'{testpath}': [ + (f'{testpath}/pub', f'{localpath}/pub') ] } From 1e7d44c5d85ce47f8a23dfab4636bdaaebf3d950 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Sun, 6 Sep 2026 20:12:03 +0000 Subject: [PATCH 19/20] fixing lint snafu's, fixing additional tests that need drivepath and one that wasn't against the mock server and needs USER_HOME as was originally used. --- sftpretty/helpers.py | 4 ++-- tests/test_cd.py | 26 ++++++++++++-------------- tests/test_getcwd.py | 2 +- tests/test_issue_65.py | 3 ++- tests/test_issue_xx.py | 12 ++++++------ tests/test_normalize.py | 20 +++++++++----------- 6 files changed, 32 insertions(+), 35 deletions(-) diff --git a/sftpretty/helpers.py b/sftpretty/helpers.py index 9fa45b5d..fbd0aeb3 100644 --- a/sftpretty/helpers.py +++ b/sftpretty/helpers.py @@ -28,8 +28,8 @@ def drivepath(filepath): if filepath: if '\\' in filepath or PureWindowsPath(filepath).drive: host = filepath.lstrip('\\/') - unc = ((filepath[:1] == '\\' or filepath[:2] == '//') - and host != '' and host[1:2] != ':') + unc = ((filepath[:1] == '\\' or filepath[:2] == '//') and + host != '' and host[1:2] != ':') utf = filepath.encode('unicode_escape').decode() utf = utf.replace('\\\\', '/') utf = sub(r'\\([^xuU])', r'/\1', utf) diff --git a/tests/test_cd.py b/tests/test_cd.py index 05179921..6af52a31 100644 --- a/tests/test_cd.py +++ b/tests/test_cd.py @@ -10,41 +10,39 @@ def test_cd_none(sftpserver): '''test sftpretty.cd with None''' - pubpath = PurePosixPath(VFS_HOME).joinpath('pub') + pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd(): sftp.chdir('pub') - assert sftp.pwd == drivepath(pubpath.as_posix()) - assert home == drivepath(pubpath.parent.as_posix()) + assert sftp.pwd == pubpath.as_posix() + assert home == pubpath.parent.as_posix() def test_cd_path(sftpserver): '''test sftpretty.cd with a path''' - pubpath = PurePosixPath(VFS_HOME).joinpath('pub') + pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd('pub'): - assert sftp.pwd == drivepath(pubpath.as_posix()) - assert home == drivepath(pubpath.parent.as_posix()) + assert sftp.pwd == pubpath.as_posix() + assert home == pubpath.parent.as_posix() def test_cd_nested(sftpserver): '''test nested cd's''' - pubpath = PurePosixPath(VFS_HOME).joinpath('pub') + pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd('pub'): - assert sftp.pwd == drivepath(pubpath.as_posix()) + assert sftp.pwd == pubpath.as_posix() with sftp.cd('foo1'): - assert sftp.pwd == drivepath( - pubpath.joinpath('foo1').as_posix() - ) - assert sftp.pwd == drivepath(pubpath.as_posix()) - assert home == drivepath(pubpath.parent.as_posix()) + assert sftp.pwd == pubpath.joinpath('foo1').as_posix() + assert sftp.pwd == pubpath.as_posix() + assert home == pubpath.parent.as_posix() def test_cd_bad_path(sftpserver): @@ -55,4 +53,4 @@ def test_cd_bad_path(sftpserver): with pytest.raises(IOError): with sftp.cd('not-there'): pass - assert home == VFS_HOME + assert home == drivepath(VFS_HOME) diff --git a/tests/test_getcwd.py b/tests/test_getcwd.py index 26fb94b3..1ea2e0b1 100644 --- a/tests/test_getcwd.py +++ b/tests/test_getcwd.py @@ -19,7 +19,7 @@ def test_getcwd_default_path(sftpserver): '''test .getcwd when using default_path arg''' with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: - assert sftp.getcwd() == VFS_HOME + assert sftp.getcwd() == drivepath(VFS_HOME) def test_getcwd_after_chdir(sftpserver): diff --git a/tests/test_issue_65.py b/tests/test_issue_65.py index 5fb60dcd..24207443 100644 --- a/tests/test_issue_65.py +++ b/tests/test_issue_65.py @@ -4,12 +4,13 @@ from common import conn, VFS, VFS_HOME from pathlib import PurePosixPath from sftpretty import Connection +from sftpretty.helpers import drivepath def test_issue_65(sftpserver): '''using the .cd() context manager prior to setting a directory via chdir causes an error''' - pubpath = PurePosixPath(VFS_HOME).joinpath('pub') + pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): cnn = conn(sftpserver) cnn['default_path'] = None diff --git a/tests/test_issue_xx.py b/tests/test_issue_xx.py index 392724fc..6d5f58b2 100644 --- a/tests/test_issue_xx.py +++ b/tests/test_issue_xx.py @@ -1,7 +1,7 @@ '''a template for creating tests that display or duplicate issues''' -from common import conn, VFS, VFS_HOME +from common import conn, USER_HOME, VFS, VFS_HOME from pathlib import Path from sftpretty import Connection from sftpretty.helpers import drivepath @@ -13,27 +13,27 @@ # server (see test_issue_xx_lsftp) def test_issue_xx_sftpserver_plugin(sftpserver): '''an example showing how to use the sftpserver plugin in a test''' - testpath = Path(VFS_HOME).joinpath('pub') + testpath = Path(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd with sftp.cd(): sftp.chdir('pub') - assert sftp.pwd == drivepath(testpath.as_posix()) - assert home == drivepath(testpath.parent.as_posix()) + assert sftp.pwd == testpath.as_posix() + assert home == testpath.parent.as_posix() def test_issue_xx_local_sftpserver(lsftp): '''same as test_issue_xx_sftpserver_plugin but written with the local sfptserver mechanism, lsftp''' home = lsftp.pwd - testpath = Path(VFS_HOME).joinpath('pub') + testpath = Path(drivepath(USER_HOME)).joinpath('pub') # starting condition of default directory should be empty, so we need to # construct whatever structure we need prior to peforming the test lsftp.mkdir('pub') with lsftp.cd(): lsftp.chdir('pub') - pubdir = lsftp.pwd.endswith(drivepath(testpath.as_posix())) + pubdir = lsftp.pwd.endswith(testpath.as_posix()) homedir = home == lsftp.pwd lsftp.rmdir('pub') diff --git a/tests/test_normalize.py b/tests/test_normalize.py index d83afc91..d9eb7499 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -10,15 +10,15 @@ def test_normalize(sftpserver): '''test the normalize function''' - pubpath = Path(VFS_HOME).joinpath('pub') + pubpath = Path(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: makepath = pubpath.parent.joinpath('make.txt').as_posix() - assert sftp.normalize('make.txt') == drivepath(makepath) - assert sftp.normalize('.') == drivepath(pubpath.parent.as_posix()) - assert sftp.normalize('pub') == drivepath(pubpath.as_posix()) + assert sftp.normalize('make.txt') == makepath + assert sftp.normalize('.') == pubpath.parent.as_posix() + assert sftp.normalize('pub') == pubpath.as_posix() sftp.chdir('pub') - assert sftp.normalize('.') == drivepath(pubpath.as_posix()) + assert sftp.normalize('.') == pubpath.as_posix() @SKIP_IF_WIN # CreateSymbolicLinkW stats target on creation, returns ENOENT @@ -48,14 +48,12 @@ def test_normalize_symlink(lsftp, remote_tmpdir): def test_pwd(sftpserver): '''test the pwd property''' - pubpath = Path(VFS_HOME).joinpath('pub') + pubpath = Path(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo2') - assert sftp.pwd == drivepath(pubpath.joinpath('foo2').as_posix()) + assert sftp.pwd == pubpath.joinpath('foo2').as_posix() sftp.chdir('bar1') - assert sftp.pwd == drivepath( - pubpath.joinpath('foo2/bar1').as_posix() - ) + assert sftp.pwd == pubpath.joinpath('foo2/bar1').as_posix() sftp.chdir('../../foo1') - assert sftp.pwd == drivepath(pubpath.joinpath('foo1').as_posix()) + assert sftp.pwd == pubpath.joinpath('foo1').as_posix() From d6dc5887b4a88059d90f3b1fca4624693f98bac8 Mon Sep 17 00:00:00 2001 From: byteskeptical <40208858+byteskeptical@users.noreply.github.com> Date: Sun, 6 Sep 2026 22:10:14 +0000 Subject: [PATCH 20/20] check the returned localpath of a get instead of the remote location, duh --- tests/test_get_r.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_get_r.py b/tests/test_get_r.py index c4dfa0b3..d3c1fc86 100644 --- a/tests/test_get_r.py +++ b/tests/test_get_r.py @@ -73,7 +73,7 @@ def test_get_r_pathed(sftpserver): localtree(local_tree, localpath, remote_cwd) sftp.remotetree(remote_tree, remote_cwd, localpath) - actual = hash(remote_cwd + '/bar1.txt') + actual = hash(Path(localpath).joinpath('bar1.txt').as_posix()) expected = ('a69f73cca23a9ac5c8b567dc185a756e97c982164fe258' '59e0d1dcc1475c80a615b2123af1f5f94c11e3e9402c3a' 'c558f500199d95b6d3e301758586281dcd26')