Repository: ReproNim/reproman

Issue:
ENH: recursive `get` for sessions (shell, ssh, etc) should support recursive mode of operation
It is necessary for any place where `get` is supposed to obtain a full hierarchy of directories like e.g. getting outputs of the job back to the local system.  

I will send a follow up initial PR for the plain orchestrator to fetch everything not just failed subjobs logs

Discussion:

Patch:
diff --git a/reproman/resource/session.py b/reproman/resource/session.py
index 776701c..ba80322 100644
--- a/reproman/resource/session.py
+++ b/reproman/resource/session.py
@@ -508,6 +508,50 @@ class Session(object):
         raise NotImplementedError
 
 
+    def transfer_recursive(self, src_path, dest_path, isdir_fct, listdir_fct, mkdir_fct, cp_fct):
+        """Recursively transfer files and directories.
+
+        Traversing the source tree contains a lot of the same logic whether 
+        we're putting or getting, so we abstract that out here.  The cost is 
+        that we need to pass the functions needed for the operation:
+
+            isdir_fct(src_path)
+
+            listdir_fct(src_path)
+
+            mkdir_fct(dest_path)
+
+            cp_fct(src_path, dest_path)
+
+        We unwrap the recursion so we can calculate the relative path names 
+        from the base source path.
+        """
+
+        # a simple copy now if the source is a file...
+        if not isdir_fct(src_path):
+            cp_fct(src_path, dest_path)
+            return
+        # ...otherwise we src_path for relative paths
+
+        stack = [src_path]
+        while stack:
+            path = stack.pop()
+            src_relpath = os.path.relpath(path, src_path)
+            dest_fullpath = os.path.join(dest_path, src_relpath)
+            if isdir_fct(path):
+                # For src_path, relpath() gives '.', so dest_fullpath ends 
+                # with "/.".  mkdir doesn't like this, so we handle src_path 
+                # as a special case.
+                if path == src_path:
+                    mkdir_fct(dest_path)
+                else:
+                    mkdir_fct(dest_fullpath)
+                stack.extend(os.path.join(path, p) for p in listdir_fct(path))
+            else:
+                cp_fct(path, dest_fullpath)
+        return
+
+
 @attr.s
 class POSIXSession(Session):
     """A Session which relies on commands present in any POSIX-compliant env
diff --git a/reproman/resource/shell.py b/reproman/resource/shell.py
index 316ed8c..4e9558e 100644
--- a/reproman/resource/shell.py
+++ b/reproman/resource/shell.py
@@ -84,7 +84,10 @@ class ShellSession(POSIXSession):
     @borrowdoc(Session)
     def get(self, src_path, dest_path=None, uid=-1, gid=-1):
         dest_path = self._prepare_dest_path(src_path, dest_path)
-        shutil.copy(src_path, dest_path)
+        if os.path.isdir(src_path):
+            shutil.copytree(src_path, dest_path)
+        else:
+            shutil.copy(src_path, dest_path)
         if uid > -1 or gid > -1:
             self.chown(dest_path, uid, gid, recursive=True)
 
@@ -157,4 +160,4 @@ class Shell(Resource):
             raise NotImplementedError
         if shared:
             raise NotImplementedError
-        return ShellSession()
\ No newline at end of file
+        return ShellSession()
diff --git a/reproman/resource/singularity.py b/reproman/resource/singularity.py
index bd87c42..537afec 100644
--- a/reproman/resource/singularity.py
+++ b/reproman/resource/singularity.py
@@ -186,8 +186,7 @@ class SingularitySession(POSIXSession):
 
         return (stdout, stderr)
 
-    @borrowdoc(Session)
-    def put(self, src_path, dest_path, uid=-1, gid=-1):
+    def _put_file(self, src_path, dest_path):
         dest_path = self._prepare_dest_path(src_path, dest_path,
                                             local=False, absolute_only=True)
         cmd = 'cat {} | singularity exec instance://{} tee {} > /dev/null'
@@ -195,19 +194,50 @@ class SingularitySession(POSIXSession):
                                     self.name,
                                     shlex_quote(dest_path)))
 
+    @borrowdoc(Session)
+    def put(self, src_path, dest_path, uid=-1, gid=-1):
+        self.transfer_recursive(
+            src_path, 
+            dest_path, 
+            os.path.isdir, 
+            os.listdir, 
+            self.mkdir, 
+            self._put_file
+        )
+
         if uid > -1 or gid > -1:
-            self.chown(dest_path, uid, gid)
+            self.chown(dest_path, uid, gid, recursive=True)
 
-    @borrowdoc(Session)
-    def get(self, src_path, dest_path=None, uid=-1, gid=-1):
+    def _get_file(self, src_path, dest_path):
         dest_path = self._prepare_dest_path(src_path, dest_path)
         cmd = 'singularity exec instance://{} cat {} > {}'
         self._runner.run(cmd.format(self.name,
                                     shlex_quote(src_path),
                                     shlex_quote(dest_path)))
 
+    @borrowdoc(Session)
+    def get(self, src_path, dest_path=None, uid=-1, gid=-1):
+        self.transfer_recursive(
+            src_path, 
+            dest_path, 
+            self.isdir, 
+            self.listdir, 
+            os.mkdir, 
+            self._get_file
+        )
+
         if uid > -1 or gid > -1:
-            self.chown(dest_path, uid, gid, remote=False)
+            self.chown(dest_path, uid, gid, remote=False, recursive=True)
+
+    def listdir(self, path):
+        cmd = ['singularity', 
+                'exec', 
+               'instance://{}'.format(self.name), 
+               'ls', 
+               '-1', 
+               path]
+        (stdout, stderr) = self._runner.run(cmd)
+        return [ f for f in stdout.split('\n') if f not in ('', '.', '..') ]
 
 
 @attr.s
diff --git a/reproman/resource/ssh.py b/reproman/resource/ssh.py
index 4153d8d..ebd1668 100644
--- a/reproman/resource/ssh.py
+++ b/reproman/resource/ssh.py
@@ -8,6 +8,8 @@
 """Resource sub-class to provide management of a SSH connection."""
 
 import attr
+import os
+import stat
 import getpass
 import uuid
 from ..log import LoggerHelper
@@ -218,18 +220,34 @@ class SSHSession(POSIXSession):
     @borrowdoc(Session)
     def put(self, src_path, dest_path, uid=-1, gid=-1):
         dest_path = self._prepare_dest_path(src_path, dest_path, local=False)
-        self.connection.put(src_path, dest_path)
+        sftp = self.connection.sftp()
+        self.transfer_recursive(
+            src_path, 
+            dest_path, 
+            os.path.isdir, 
+            os.listdir, 
+            sftp.mkdir, 
+            self.connection.put
+        )
 
         if uid > -1 or gid > -1:
-            self.chown(dest_path, uid, gid)
+            self.chown(dest_path, uid, gid, recursive=True)
 
     @borrowdoc(Session)
     def get(self, src_path, dest_path=None, uid=-1, gid=-1):
         dest_path = self._prepare_dest_path(src_path, dest_path)
-        self.connection.get(src_path, dest_path)
+        sftp = self.connection.sftp()
+        self.transfer_recursive(
+            src_path, 
+            dest_path, 
+            lambda f: stat.S_ISDIR(sftp.stat(f).st_mode), 
+            sftp.listdir, 
+            os.mkdir, 
+            self.connection.get
+        )
 
         if uid > -1 or gid > -1:
-            self.chown(dest_path, uid, gid, remote=False)
+            self.chown(dest_path, uid, gid, remote=False, recursive=True)
 
 
 @attr.s


Test patch:
diff --git a/reproman/interface/tests/test_run.py b/reproman/interface/tests/test_run.py
index c12531a..efacca7 100644
--- a/reproman/interface/tests/test_run.py
+++ b/reproman/interface/tests/test_run.py
@@ -493,3 +493,28 @@ def test_jobs_orc_error(context):
                 jobs(queries=[], status=True)
             assert "myshell" not in output.out
             assert "resurrection failed" in log.out
+
+
+def test_recursive_transfer(context):
+    run = context["run_fn"]
+    path = context["directory"]
+    jobs = context["jobs_fn"]
+
+    # Our script takes an inventory of the execute directory so we can be 
+    # sure that all of the files have transferred.  It then creates a tree 
+    # that we verify is returned.
+
+    create_tree(path, {"script": "find . > out_file ; mkdir out_dir ; touch out_dir/subfile", 
+                       "in_dir": {"subfile": ""}})
+
+    with swallow_outputs() as output:
+        run(command=["sh", "-e", "script"], 
+            inputs=["script", "in_dir"], 
+            outputs=["out_file", "out_dir"], 
+            resref="myshell")
+        try_fetch(lambda: jobs(queries=[], action="fetch", all_=True))
+        assert op.exists(op.join(path, "out_file"))
+        with open(op.join(path, "out_file")) as fo:
+            lines = [ line.strip() for line in fo ]
+        assert "./in_dir/subfile" in lines
+        assert op.exists(op.join(path, "out_dir", "subfile"))
diff --git a/reproman/resource/tests/test_session.py b/reproman/resource/tests/test_session.py
index 8dd551e..9ae51df 100644
--- a/reproman/resource/tests/test_session.py
+++ b/reproman/resource/tests/test_session.py
@@ -18,6 +18,7 @@ import uuid
 from ..session import get_updated_env, Session
 from ...support.exceptions import CommandError
 from ...utils import chpwd, swallow_logs
+from ...tests.utils import create_tree
 from ...tests.skip import mark
 from ...tests.fixtures import get_docker_fixture
 from ...tests.fixtures import get_singularity_fixture
@@ -260,11 +261,13 @@ def check_methods(resource_test_dir):
         assert not result
 
         # Create a temporary test file
-        temp_file = tempfile.NamedTemporaryFile(dir=resource_test_dir)
-        with temp_file as f:
-            f.write('ReproMan test content\nline 2\nline 3'.encode('utf8'))
-            f.flush()
-            local_path = temp_file.name
+        with tempfile.TemporaryDirectory(dir=resource_test_dir) as tdir:
+            create_tree(tdir,
+                        {'f0': 'ReproMan test content\nline 2\nline 3',
+                         'f1': 'f1',
+                         'd0': {'f2': 'f2',
+                                'd2': {'f3': 'f3'}}})
+            local_path = os.path.join(tdir, 'f0')
             remote_path = '{}/reproman upload/{}'.format(resource_test_dir,
                 uuid.uuid4().hex)
 
@@ -276,6 +279,13 @@ def check_methods(resource_test_dir):
             assert result
             # TODO: Check uid and gid of remote file
 
+            # Check recursive put().
+            remote_path_rec = '{}/recursive-put/{}'.format(
+                resource_test_dir, uuid.uuid4().hex)
+            session.put(tdir, remote_path_rec)
+            assert session.exists(remote_path_rec + "/d0/f2")
+            assert session.exists(remote_path_rec + "/d0/d2/f3")
+
             # We can use a relative name for the target
             basename_put_dir = os.path.join(resource_test_dir, "basename-put")
             if not os.path.exists(basename_put_dir):
@@ -330,6 +340,14 @@ def check_methods(resource_test_dir):
         result = session.isdir(test_dir)
         assert result
 
+        # Check listdir() method
+        if hasattr(session, 'listdir'):
+            subdir = uuid.uuid4().hex
+            subfile = uuid.uuid4().hex
+            session.mkdir(os.path.join(test_dir, subdir))
+            session.put('/etc/hosts', os.path.join(test_dir, subfile))
+            assert set(session.listdir(test_dir)) == set((subdir, subfile))
+
         # Check making parent dirs without setting flag
         test_dir = '{}/tmp/i fail/{}'.format(resource_test_dir, uuid.uuid4().hex)
         with pytest.raises(CommandError):