diff --git a/datalad_ria/tests/test_ssh_remote_io.py b/datalad_ria/tests/test_ssh_remote_io.py index 5964b571..7068091a 100644 --- a/datalad_ria/tests/test_ssh_remote_io.py +++ b/datalad_ria/tests/test_ssh_remote_io.py @@ -6,12 +6,32 @@ @pytest.fixture(autouse=False, scope="function") def ssh_remoteio(ria_sshserver_setup, ria_sshserver): + """Yield a ``SSHRemoteIO`` instance matching the RIA server setup""" ssh = SSHRemoteIO( 'ssh://{SSH_LOGIN}@{HOST}:{SSH_PORT}'.format(**ria_sshserver_setup) ) yield ssh +@pytest.fixture(autouse=False, scope="function") +def ssh_remote_wdir(ssh_remoteio, ria_sshserver_setup): + """Like ``ssh_remoteio``, but also yields a working directory + + It used ``mktemp -d`` on the remote end, and also remove the working + directory at the end. + """ + sshpath = PurePosixPath(ria_sshserver_setup['SSH_PATH']) + out, err = ssh_remoteio.ssh(f'mktemp -d {sshpath}/sshremotewdir.XXXXXXXX') + # hopefully catch more stupid errors of unexpectedness + assert '/sshremotewdir' in out + wdir = PurePosixPath(out.rstrip('\n')) + yield ssh_remoteio, wdir + # clean up + # only run dangerous 'rm -rf' if needed + if ssh_remoteio.exists(wdir): + ssh_remoteio.ssh(f'echo rm -rf "{wdir}"') + + def test_SSHRemoteIO_read_file(ssh_remoteio): # basic smoke test, just login and use the abstraction to read a file etcpasswd = ssh_remoteio.read_file('/etc/passwd') @@ -19,6 +39,8 @@ def test_SSHRemoteIO_read_file(ssh_remoteio): assert etcpasswd +# this is not using `ssh_remote_wdir`, because we want to go manual +# on all steps and have things break in a test, and not in a fixture def test_SSHRemoteIO_handledir(ssh_remoteio, ria_sshserver_setup): sshpath = PurePosixPath(ria_sshserver_setup['SSH_PATH']) targetdir = sshpath / 'testdir' @@ -43,3 +65,16 @@ def test_SSHRemoteIO_handledir(ssh_remoteio, ria_sshserver_setup): # ssh_remoteio.remove(targetdir) ssh_remoteio.remove_dir(targetdir) assert not ssh_remoteio.exists(targetdir) + + +def test_SSHRemoteIO_symlink(ssh_remote_wdir): + ssh_remoteio, targetdir = ssh_remote_wdir + targetfpath = targetdir / 'testfile' + assert not ssh_remoteio.exists(targetfpath) + ssh_remoteio.symlink('/etc/passwd', targetfpath) + assert ssh_remoteio.exists(targetfpath) + assert ssh_remoteio.read_file('/etc/passwd') \ + == ssh_remoteio.read_file(targetfpath) + # verify that we can remove a symlink + ssh_remoteio.remove(targetfpath) + assert not ssh_remoteio.exists(targetfpath)