diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 633d362d..089336b3 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -27,6 +27,19 @@ 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 + shell: bash - name: ${{ matrix.os }} SSH if: startsWith(matrix.os, 'macos') run: | @@ -35,12 +48,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 +71,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 diff --git a/sftpretty/__init__.py b/sftpretty/__init__.py index 2e34ae25..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 @@ -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, @@ -214,33 +217,39 @@ 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, '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) @@ -331,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 @@ -556,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 @@ -807,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) @@ -876,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): @@ -1108,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, @@ -1192,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. @@ -1206,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. @@ -1220,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 @@ -1266,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 @@ -1281,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 @@ -1294,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 @@ -1310,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 @@ -1326,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 @@ -1341,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 @@ -1359,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 @@ -1373,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 @@ -1387,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 @@ -1402,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): @@ -1431,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. @@ -1447,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 @@ -1460,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. @@ -1510,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. @@ -1527,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. @@ -1537,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. @@ -1547,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 @@ -1562,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 @@ -1577,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 @@ -1620,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..fbd0aeb3 100644 --- a/sftpretty/helpers.py +++ b/sftpretty/helpers.py @@ -1,7 +1,8 @@ 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 @@ -16,15 +17,35 @@ 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: + 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) + 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/common.py b/tests/common.py index 7317f273..8b0bc5e4 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 @@ -13,12 +14,30 @@ 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 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', @@ -30,13 +49,31 @@ 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} +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(): @@ -61,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/conftest.py b/tests/conftest.py index eea15e61..d9204359 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 lsftp.normalize(remotedir.as_posix()) + finally: + remote_rmdir(lsftp, remotedir.as_posix()) diff --git a/tests/test_cd.py b/tests/test_cd.py index 10188561..6af52a31 100644 --- a/tests/test_cd.py +++ b/tests/test_cd.py @@ -2,14 +2,15 @@ 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(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd @@ -21,7 +22,7 @@ def test_cd_none(sftpserver): def test_cd_path(sftpserver): '''test sftpretty.cd with a path''' - pubpath = PurePosixPath('/home/test').joinpath('pub') + pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd @@ -32,7 +33,7 @@ def test_cd_path(sftpserver): def test_cd_nested(sftpserver): '''test nested cd's''' - pubpath = PurePosixPath('/home/test').joinpath('pub') + pubpath = PurePosixPath(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd @@ -52,4 +53,4 @@ def test_cd_bad_path(sftpserver): with pytest.raises(IOError): with sftp.cd('not-there'): pass - assert home == '/home/test' + assert home == drivepath(VFS_HOME) diff --git a/tests/test_chmod.py b/tests/test_chmod.py index cc7d03f6..a80f879e 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 @@ -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..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 @@ -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_connection.py b/tests/test_connection.py index 1c0feb64..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)) @@ -94,10 +94,55 @@ def test_connection_bad_host(): sftp.listdir() -def test_connection_good(sftpserver): - '''connect to a public sftp server''' +@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): + with 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() + copts['private_key'] = key.as_posix() + + with pytest.raises(OSError, match=key.name): + with Connection(**copts) as sftp: + sftp.listdir() + + +@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() 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') diff --git a/tests/test_getcwd.py b/tests/test_getcwd.py index 4e922e69..1ea2e0b1 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() == drivepath(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_helpers.py b/tests/test_helpers.py new file mode 100644 index 00000000..c5f5feb3 --- /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'), # noqa: W605 + ('\\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_issue_65.py b/tests/test_issue_65.py index 99d86654..24207443 100644 --- a/tests/test_issue_65.py +++ b/tests/test_issue_65.py @@ -1,20 +1,21 @@ '''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 +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('/home/test').joinpath('pub') + pubpath = PurePosixPath(drivepath(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..6d5f58b2 100644 --- a/tests/test_issue_xx.py +++ b/tests/test_issue_xx.py @@ -1,18 +1,19 @@ '''a template for creating tests that display or duplicate issues''' -from common import conn, USER, USER_HOME_PARENT, VFS +from common import conn, USER_HOME, 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(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: home = sftp.pwd @@ -26,7 +27,7 @@ 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(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') diff --git a/tests/test_normalize.py b/tests/test_normalize.py index 8f9972e3..d9eb7499 100644 --- a/tests/test_normalize.py +++ b/tests/test_normalize.py @@ -1,13 +1,16 @@ '''test sftpretty.normalize''' -from common import VFS, conn +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(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: makepath = pubpath.parent.joinpath('make.txt').as_posix() @@ -18,18 +21,34 @@ 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() +@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() + 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 + + +@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() + 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): '''test the pwd property''' - pubpath = Path('/home/test').joinpath('pub') + pubpath = Path(drivepath(VFS_HOME)).joinpath('pub') with sftpserver.serve_content(VFS): with Connection(**conn(sftpserver)) as sftp: sftp.chdir('pub/foo2') 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..2ba52c85 100644 --- a/tests/test_put_d.py +++ b/tests/test_put_d.py @@ -3,32 +3,43 @@ import pytest from blddirs import build_dir_struct -from common import rmdir +from common import SKIP_IF_ROOT, SKIP_IF_WIN 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()) + build_dir_struct(tmp_path.as_posix()) + local = tmp_path.joinpath('pub').as_posix() + lsftp.put_d(local, remote_tmpdir) + remote = Path(remote_tmpdir).joinpath('pub').as_posix() - rmdir(localpath) + 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 -# with pytest.raises(IOError): -# psftp.put_d('.', '.') +@SKIP_IF_ROOT +@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''' + 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) -def test_put_d_bad_local(lsftp): + 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): '''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..b7c88a96 100644 --- a/tests/test_readlink.py +++ b/tests/test_readlink.py @@ -2,21 +2,17 @@ from io import BytesIO from pathlib import Path +from sftpretty.helpers import drivepath -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(drivepath(rfile)) 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') ] } 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..224bfada 100644 --- a/tests/test_rmdir.py +++ b/tests/test_rmdir.py @@ -1,17 +1,31 @@ '''test sftpretty.rmdir''' +import pytest -def test_rmdir(lsftp): +from common import SKIP_IF_ROOT, SKIP_IF_WIN +from pathlib import Path + + +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 +@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') + 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..07cf0bd9 100644 --- a/tests/test_truncate.py +++ b/tests/test_truncate.py @@ -1,60 +1,26 @@ '''test sftpretty.listdir''' +import pytest + from common import STARS8192 from io import BytesIO +from pathlib import Path -def test_truncate_smaller(lsftp): - '''test truncate, make file smaller''' - 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, 4096) - assert new_size == 4096 - lsftp.remove(rname) - - -def test_truncate_larger(lsftp): - '''test truncate, make file larger''' +@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, 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)