Repository: vatlab/sos (language: python)

Issue:
Requiring shared directory for remote execution of tasks
Currently the sos task execution system is complicated because it needs to copy files around if files reside on directories that are not shared by the job submission machine and the head node. A lot of work, such as signature checking, directory mapping, and file synchronization has been done to enable a powerful, yet complicated and confusing system that can easily break.

I propose that we remove all the file mapping feature and require file systems to be shared. In this way input and output files have to exist on both local and remote file systems, and there is no need to map directories and copy files around. The system will be much easier to understand, configure, and a lot more robust.
subclass pathlib.Path for path
Breaks for python 3.10, and simple scripts like

```
file_target('a_1.txt').with_suffix('.bak')._md5
```

fails, because the functon we override to add `_md5`, named `_init` is removed from `pathlib` in python 3.10.

Supposedly, python 3.12 allows more graceful derivation of `path` from `Path` (https://discuss.python.org/t/make-pathlib-extensible/3428/46), but we will have to fix 3.10 for now.


Discussion / hints:
Agreed, considering most of our user cases so far are the simpler case of completely local or remote. 


Fix patch:
diff --git a/requirements_dev.txt b/requirements_dev.txt
index 7a55842c5..9ef6964d4 100644
--- a/requirements_dev.txt
+++ b/requirements_dev.txt
@@ -16,4 +16,5 @@ pytest
 pytest-cov
 pyzmq
 sos-pbs
+tabulate>=0.9.0
 tqdm
\ No newline at end of file
diff --git a/setup.py b/setup.py
index a5b648d9b..2437b29e2 100644
--- a/setup.py
+++ b/setup.py
@@ -112,7 +112,6 @@ def run(self):
 file_target = sos.targets:file_target
 sos_tempfile = sos.targets:sos_tempfile
 dynamic = sos.targets:dynamic
-remote = sos.targets:remote
 executable = sos.targets:executable
 sos_variable = sos.targets:sos_variable
 sos_step = sos.targets:sos_step
@@ -126,14 +125,8 @@ def run(self):
 [sos_actions]
 script = sos.actions:script
 sos_run = sos.actions:sos_run
-fail_if = sos.actions:fail_if
-warn_if = sos.actions:warn_if
-stop_if = sos.actions:stop_if
-done_if = sos.actions:done_if
-skip_if = sos.actions:skip_if
 download = sos.actions:download
 run = sos.actions:run
-
 bash = sos.actions_bash:bash
 csh = sos.actions_bash:csh
 tcsh = sos.actions_bash:tcsh
diff --git a/src/sos/__main__.py b/src/sos/__main__.py
index efefa41a5..4f8b45928 100755
--- a/src/sos/__main__.py
+++ b/src/sos/__main__.py
@@ -1057,22 +1057,6 @@ def cmd_remote(args, workflow_args):
             from .remote import run_command_on_hosts
 
             run_command_on_hosts(cfg, args.hosts, args.cmd, args.verbosity)
-        elif args.action == "push":
-            if not args.files:
-                raise ValueError(
-                    "Please specify files to push to remote host with option --files"
-                )
-            from .remote import push_to_hosts
-
-            push_to_hosts(cfg, args.hosts, args.files, args.verbosity)
-        elif args.action == "pull":
-            if not args.files:
-                raise ValueError(
-                    "Please specify files to pull from remote host with option --files"
-                )
-            from .remote import pull_from_host
-
-            pull_from_host(cfg, args.hosts, args.files, args.verbosity)
         else:
             raise ValueError(
                 "Unacceptable remote action. Use command 'sos remote -h' to check allowable actions."
diff --git a/src/sos/actions.py b/src/sos/actions.py
index f3c0c0de4..21507e50c 100644
--- a/src/sos/actions.py
+++ b/src/sos/actions.py
@@ -2,7 +2,6 @@
 #
 # Copyright (c) Bo Peng and the University of Texas MD Anderson Cancer Center
 # Distributed under the terms of the 3-clause BSD License.
-
 import copy
 import gzip
 import os
@@ -33,18 +32,13 @@
 from .parser import SoS_Script
 from .syntax import SOS_ACTION_OPTIONS
 from .targets import executable, file_target, path, paths, sos_targets
-from .utils import (StopInputGroup, TerminateExecution,
-                    TimeoutInterProcessLock, env, fileMD5, get_traceback,
+from .utils import (TimeoutInterProcessLock, env, fileMD5, get_traceback,
                     load_config_files, short_repr, textMD5, transcribe)
 
 __all__ = [
     "SoS_Action",
     "script",
     "sos_run",
-    "fail_if",
-    "warn_if",
-    "stop_if",
-    "download",
     "run",
     "perl",
     "report",
@@ -129,8 +123,7 @@ def action_wrapper(*args, **kwargs):
                             "shub",
                             "oras",
                     ):
-                        env.logger.warning(
-                            f"Container type {cty} might not be supported.")
+                        env.logger.warning(f"Container type {cty} might not be supported.")
                 elif engine is not None and engine != "local":
                     raise ValueError(f"Only docker and singularity container engines are supported: {engine} specified")
                 else:
@@ -412,8 +405,9 @@ def run(self, **kwargs):
             else:
                 raise RuntimeError(f"Unacceptable interpreter {self.interpreter}")
 
-            debug_script_path = os.path.dirname(os.path.abspath(kwargs["stderr"])) if ("stderr" in kwargs and kwargs["stderr"] is not False and
-                                os.path.isdir(os.path.dirname(os.path.abspath(kwargs["stderr"])))) else env.exec_dir
+            debug_script_path = os.path.dirname(os.path.abspath(kwargs["stderr"])) if (
+                "stderr" in kwargs and kwargs["stderr"] is not False and
+                os.path.isdir(os.path.dirname(os.path.abspath(kwargs["stderr"])))) else env.exec_dir
             debug_script_file = os.path.join(
                 debug_script_path,
                 f'{env.sos_dict["step_name"]}_{env.sos_dict["_index"]}_{str(uuid.uuid4())[:8]}{self.suffix}',
@@ -769,49 +763,6 @@ def script(script, interpreter="", suffix="", args="", entrypoint="", **kwargs):
     return SoS_ExecuteScript(script, interpreter, suffix, args, entrypoint).run(**kwargs)
 
 
-@SoS_Action(acceptable_args=["expr", "msg"])
-def fail_if(expr, msg=""):
-    """Raise an exception with `msg` if condition `expr` is False"""
-    if expr:
-        raise TerminateExecution(msg if msg else "error triggered by action fail_if")
-    return 0
-
-
-@SoS_Action(acceptable_args=["expr", "msg"])
-def warn_if(expr, msg=""):
-    """Yield an warning message `msg` if `expr` is False """
-    if expr:
-        env.logger.warning(msg)
-    return 0
-
-
-@SoS_Action(acceptable_args=["expr", "msg", "no_output"])
-def stop_if(expr, msg="", no_output=False):
-    """Abort the execution of the current step or loop and yield
-    an warning message `msg` if `expr` is False"""
-    if expr:
-        raise StopInputGroup(msg=msg, keep_output=not no_output)
-    return 0
-
-
-@SoS_Action(acceptable_args=["expr", "msg"])
-def done_if(expr, msg=""):
-    """Assuming that output has already been generated and stop
-    executing the rest of the substep"""
-    if expr:
-        raise StopInputGroup(msg=msg, keep_output=True)
-    return 0
-
-
-@SoS_Action(acceptable_args=["expr", "msg", "no_output"])
-def skip_if(expr, msg=""):
-    """Skip the current substep and set _output to empty. Output
-    will be removed if already generated."""
-    if expr:
-        raise StopInputGroup(msg=msg, keep_output=False)
-    return 0
-
-
 #
 # download file with progress bar
 #
diff --git a/src/sos/functions.py b/src/sos/functions.py
new file mode 100644
index 000000000..6eba89bd9
--- /dev/null
+++ b/src/sos/functions.py
@@ -0,0 +1,65 @@
+from .utils import (StopInputGroup, TerminateExecution, env)
+
+# g_action_map = {}
+
+# def _load_actions():
+#     global g_action_map  # pylint: disable=global-variable-not-assigned
+#     for _entrypoint in pkg_resources.iter_entry_points(group="sos_actions"):
+#         # import actions from entry_points
+#         # Grab the function that is the actual plugin.
+#         _name = _entrypoint.name
+#         try:
+#             _plugin = _entrypoint.load()
+#             g_action_map[_name] = _plugin
+#         except Exception as e:
+#             from .utils import get_logger
+
+#             # look for sos version requirement
+#             get_logger().warning(f"Failed to load script running action {_entrypoint.name}: {e}")
+
+# def sos_run_script(action, script, *args, **kwargs):
+#     '''Call script-execution actions.'''
+#     if not g_action_map:
+#         _load_actions()
+#     try:
+#         g_action_map[action](script, *args, **kwargs)
+#     except KeyError as e:
+#         raise RuntimeError(f'Undefined script running action {action}') from e
+
+
+def stop_if(expr, msg="", no_output=False):
+    """Abort the execution of the current step or loop and yield
+    an warning message `msg` if `expr` is False"""
+    if expr:
+        raise StopInputGroup(msg=msg, keep_output=not no_output)
+    return 0
+
+
+def done_if(expr, msg=""):
+    """Assuming that output has already been generated and stop
+    executing the rest of the substep"""
+    if expr:
+        raise StopInputGroup(msg=msg, keep_output=True)
+    return 0
+
+
+def skip_if(expr, msg=""):
+    """Skip the current substep and set _output to empty. Output
+    will be removed if already generated."""
+    if expr:
+        raise StopInputGroup(msg=msg, keep_output=False)
+    return 0
+
+
+def fail_if(expr, msg=""):
+    """Raise an exception with `msg` if condition `expr` is False"""
+    if expr:
+        raise TerminateExecution(msg if msg else "error triggered by action fail_if")
+    return 0
+
+
+def warn_if(expr, msg=""):
+    """Yield an warning message `msg` if `expr` is False """
+    if expr:
+        env.logger.warning(msg)
+    return 0
diff --git a/src/sos/hosts.py b/src/sos/hosts.py
index ccf89f7dc..22ce510b3 100755
--- a/src/sos/hosts.py
+++ b/src/sos/hosts.py
@@ -40,7 +40,6 @@
 #   file systems.
 #
 #   Keys for host configuration include:
-#   * path_map: path map between local and remote hosts
 #   * shared: paths that are shared between local and remote hosts
 #   * send_cmd (optional): alternative command to send files
 #   * receive_cmd (optional): alternative command to receive files
@@ -153,15 +152,8 @@ def target_exists(self, targets):
     def target_signature(self, targets):
         return targets.target_signature()
 
-    def send_to_host(self, items):
-        return {x: x for x in items}
-
-    def receive_from_host(self, items):
-        return {x: x for x in items}
-
     def prepare_task(self, task_id):
-        task_file = os.path.join(
-            os.path.expanduser("~"), ".sos", "tasks", task_id + ".task")
+        task_file = os.path.join(os.path.expanduser("~"), ".sos", "tasks", task_id + ".task")
         # add server restriction on task file
         if not os.path.isfile(task_file):
             raise ValueError(f"Missing task definition {task_file}")
@@ -183,19 +175,15 @@ def prepare_task(self, task_id):
             }
         }
         runtime["_runtime"]["workdir"] = (
-            task_vars["_runtime"]["workdir"]
-            if "workdir" in task_vars["_runtime"] else os.getcwd())
+            task_vars["_runtime"]["workdir"] if "workdir" in task_vars["_runtime"] else os.getcwd())
 
-        if ("max_mem" in self.config or "max_cores" in self.config or
-                "max_walltime" in self.config):
+        if ("max_mem" in self.config or "max_cores" in self.config or "max_walltime" in self.config):
             for key in ("max_mem", "max_cores", "max_walltime"):
                 if key in self.config:
                     runtime["_runtime"][key] = (
-                        format_HHMMSS(self.config[key])
-                        if key == "max_walltime" else self.config[key])
+                        format_HHMMSS(self.config[key]) if key == "max_walltime" else self.config[key])
 
-            if (self.config.get("max_mem", None) is not None and
-                    task_vars["_runtime"].get("mem", None) is not None and
+            if (self.config.get("max_mem", None) is not None and task_vars["_runtime"].get("mem", None) is not None and
                     self.config["max_mem"] < task_vars["_runtime"]["mem"]):
                 env.logger.error(
                     f'Task {task_id} requested more mem ({task_vars["_runtime"]["mem"]}) than allowed max_mem ({self.config["max_mem"]})'
@@ -210,8 +198,7 @@ def prepare_task(self, task_id):
                 return False
             if (self.config.get("max_walltime", None) is not None and
                     task_vars["_runtime"].get("walltime", None) is not None and
-                    expand_time(self.config["max_walltime"]) < expand_time(
-                        task_vars["_runtime"]["walltime"])):
+                    expand_time(self.config["max_walltime"]) < expand_time(task_vars["_runtime"]["walltime"])):
                 env.logger.error(
                     f'Task {task_id} requested more walltime ({task_vars["_runtime"]["walltime"]}) than allowed max_walltime ({self.config["max_walltime"]})'
                 )
@@ -237,8 +224,7 @@ def check_output(self, cmd, under_workdir=False, **kwargs):
             cmd = subprocess.list2cmdline(cmd)
         try:
             cmd = cfg_interpolate(cmd)
-            return subprocess.check_output(
-                cmd, shell=isinstance(cmd, str), **kwargs).decode()
+            return subprocess.check_output(cmd, shell=isinstance(cmd, str), **kwargs).decode()
         except Exception as e:
             env.logger.warning(f"Check output of {cmd} failed: {e}")
             raise
@@ -246,8 +232,7 @@ def check_output(self, cmd, under_workdir=False, **kwargs):
     def check_call(self, cmd, under_workdir=False, **kwargs):
         # get the output of command
         try:
-            return subprocess.check_call(
-                cmd, shell=isinstance(cmd, str), **kwargs)
+            return subprocess.check_call(cmd, shell=isinstance(cmd, str), **kwargs)
         except Exception as e:
             env.logger.warning(f"Check output of {cmd} failed: {e}")
             raise
@@ -280,12 +265,9 @@ def receive_result(self, task_id: str) -> Dict[str, Any]:
                     "exception": RuntimeError(tf.stderr),
                 }
             return {
-                "ret_code":
-                    1,
-                "task":
-                    task_id,
-                "exception":
-                    ValueError(f"No result is received for task {task_id}"),
+                "ret_code": 1,
+                "task": task_id,
+                "exception": ValueError(f"No result is received for task {task_id}"),
             }
 
         try:
@@ -314,13 +296,11 @@ def __init__(
         self.port = self.config.get("port", 22)
         self.pem_file = self.config.get("pem_file", None)
         self.shared_dirs = self._get_shared_dirs()
-        self.path_map = self._get_path_map()
         # we already test connect of remote hosts
         if test_connection:
             test_res = self.test_connection()
             if test_res != "OK":
-                raise RuntimeError(
-                    f"Failed to connect to {self.alias}: {test_res}")
+                raise RuntimeError(f"Failed to connect to {self.alias}: {test_res}")
 
     def target_exists(self, targets):
         try:
@@ -364,77 +344,37 @@ def _get_shared_dirs(self) -> List[Any]:
             return [value]
         if isinstance(value, Sequence):
             return value
-        raise ValueError(
-            "Option shared can only be a string or a list of strings")
-
-    def _get_path_map(self) -> Dict[str, str]:
-        res: Dict = {}
-        # if user-specified path_map, it overrides CONFIG
-        path_map = self.config.get("path_map", [])
-        #
-        if not path_map:
-            return res
-        if isinstance(path_map, str):
-            path_map = [path_map]
-        if isinstance(path_map, Sequence):
-            for v in path_map:
-                if " -> " not in v:
-                    raise ValueError(
-                        f"Path map should be separated as from -> to, {v} specified"
-                    )
-                if v.count(" -> ") > 1:
-                    raise ValueError(
-                        f"Path map should be separated as from -> to, {v} specified"
-                    )
-                res[v.split(" -> ")[0]] = v.split(" -> ")[1]
-        elif isinstance(path_map, dict):
-            for k, v in path_map.items():
-                res[k] = v
-        else:
-            raise ValueError(
-                f"Unacceptable path_mapue for configuration path_map: {path_map}"
-            )
-        return res
+        raise ValueError("Option shared can only be a string or a list of strings")
 
     def _get_control_master_options(self):
-        master_dir = os.path.join(
-            os.path.expanduser("~"), ".ssh", "controlmasters")
+        master_dir = os.path.join(os.path.expanduser("~"), ".ssh", "controlmasters")
         if not os.path.isdir(master_dir):
             try:
                 os.makedirs(master_dir, exist_ok=True)
             except Exception as e:
-                env.logger.debug(
-                    f"Failed to create ssh control master directory {master_dir}: {e}"
-                )
+                env.logger.debug(f"Failed to create ssh control master directory {master_dir}: {e}")
                 return ""
         return f"-o 'ControlMaster=auto' -o 'ControlPath={master_dir}/%r@%h:%p' -o 'ControlPersist=10m'"
 
     def _get_identify_file_options(self):
-        if ("pem_file" in self.config and
-                isinstance(self.config["pem_file"], str) and
+        if ("pem_file" in self.config and isinstance(self.config["pem_file"], str) and
                 self.config["pem_file"].strip() != ""):
             return f""" -i '{self.config["pem_file"]}' """
         return ""
 
     def _get_send_cmd(self, rename=False):
         if rename:
-            return (
-                "ssh " + self.cm_opts + self.pem_opts +
-                """ -q {host} -p {port} "mkdir -p {dest:dpq}" && """ +
-                """rsync -a --no-g -e 'ssh """ + self.cm_opts + self.pem_opts +
-                """ -p {port}' {source:aep} "{host}:{dest:dep}" && """ +
-                """ssh """ + self.cm_opts + self.pem_opts +
-                """ -q {host} -p {port} "mv {dest:dep}/{source:b} {dest:ep}" """
-            )
-        return ("ssh " + self.cm_opts + self.pem_opts +
-                """ -q {host} -p {port} "mkdir -p {dest:dpq}" """ +
-                """ && rsync -a --no-g -e "ssh -p {port} """ + self.cm_opts +
-                self.pem_opts + ''' " {source:aep} "{host}:{dest:dep}"''')
+            return ("ssh " + self.cm_opts + self.pem_opts + """ -q {host} -p {port} "mkdir -p {dest:dpq}" && """ +
+                    """rsync -a --no-g -e 'ssh """ + self.cm_opts + self.pem_opts +
+                    """ -p {port}' {source:aep} "{host}:{dest:dep}" && """ + """ssh """ + self.cm_opts + self.pem_opts +
+                    """ -q {host} -p {port} "mv {dest:dep}/{source:b} {dest:ep}" """)
+        return ("ssh " + self.cm_opts + self.pem_opts + """ -q {host} -p {port} "mkdir -p {dest:dpq}" """ +
+                """ && rsync -a --no-g -e "ssh -p {port} """ + self.cm_opts + self.pem_opts +
+                ''' " {source:aep} "{host}:{dest:dep}"''')
 
     def _get_receive_cmd(self, rename=False):
         if rename:
-            return ("""rsync -a --no-g -e 'ssh """ + self.cm_opts +
-                    self.pem_opts +
+            return ("""rsync -a --no-g -e 'ssh """ + self.cm_opts + self.pem_opts +
                     """ -p {port}' {host}:{source:e} "{dest:adep}" && """ +
                     '''mv "{dest:adep}/{source:b}" "{dest:aep}"''')
         return ("""rsync -a --no-g -e 'ssh """ + self.cm_opts + self.pem_opts +
@@ -445,15 +385,13 @@ def _get_execute_cmd(self, under_workdir=True, use_heredoc=True) -> str:
         if "execute_cmd" in self.config:
             return self.config["execute_cmd"]
         if use_heredoc:
-            return (
-                "ssh " + self.cm_opts + self.pem_opts +
-                """ -q {host} -p {port} <<'HEREDOC!!'\nbash --login -c '""" +
-                (" [ -d {workdir} ] || mkdir -p {workdir}; cd {workdir} && "
-                 if under_workdir else " ") + """ {cmd} '\nHEREDOC!!\n""")
-        return ("ssh " + self.cm_opts + self.pem_opts +
-                """ -q {host} -p {port} "bash --login -c '""" +
-                (" [ -d {workdir} ] || mkdir -p {workdir}; cd {workdir} && "
-                 if under_workdir else " ") + """ {cmd}'" """)
+            return ("ssh " + self.cm_opts + self.pem_opts +
+                    """ -q {host} -p {port} <<'HEREDOC!!'\nbash --login -c '""" +
+                    (" [ -d {workdir} ] || mkdir -p {workdir}; cd {workdir} && " if under_workdir else " ") +
+                    """ {cmd} '\nHEREDOC!!\n""")
+        return ("ssh " + self.cm_opts + self.pem_opts + """ -q {host} -p {port} "bash --login -c '""" +
+                (" [ -d {workdir} ] || mkdir -p {workdir}; cd {workdir} && " if under_workdir else " ") +
+                """ {cmd}'" """)
 
     def _get_query_cmd(self):
         return self.config.get(
@@ -464,91 +402,7 @@ def _get_query_cmd(self):
 
     def is_shared(self, path):
         fullpath = os.path.abspath(os.path.expanduser(path))
-        for sdir in self.shared_dirs:
-            if fullpath.startswith(sdir):
-                # issue 710, if a directory is both under path_map and shared, then it is not considered to be shared.
-                if not any(
-                        fullpath.startswith(mdir)
-                        for mdir in self.path_map.keys()):
-                    return True
-        return False
-
-    def _map_path(self, source):
-        result = {}
-        cwd = os.getcwd()
-        if isinstance(source, str) and source.startswith("#"):
-            result[path(source, host="localhost")] = path(
-                source, host=self.alias)
-        elif isinstance(source, (str, path)):
-            dest = os.path.abspath(os.path.expanduser(source))
-            # we use samefile to avoid problems with case-insensitive file system #522
-            # we also use the "cwd" name to avoid wrong case for cwd. For example,
-            # if the cwd = '/Users/user/Project'
-            # then, dest = '/USERS/USER/PROJECT/a.txt'
-            # would be converted to '/Users/user/Project/a.txt' before path mapping
-            if os.path.exists(dest[:len(cwd)]) and os.path.samefile(
-                    dest[:len(cwd)], cwd):
-                dest = cwd + dest[len(cwd):]
-            matched = [
-                k for k in self.path_map.keys()
-                if os.path.exists(dest[:len(k)]) and
-                os.path.samefile(dest[:len(k)], k)
-            ]
-            if matched:
-                # pick the longest key that matches
-                k = max(matched, key=len)
-                dest = self.path_map[k] + dest[len(k):]
-            else:
-                env.logger.debug(
-                    f"Path {source} is not under any specified paths of localhost and is mapped to {dest} on remote host."
-                )
-            result[source] = dest.replace("\\", "/")
-        elif isinstance(source, (Sequence, sos_targets)):
-            for src in source:
-                result.update(self._map_path(src))
-        else:
-            env.logger.debug(f"Ignore unmappable source {source}")
-            return {source: source}
-        return result
-
-    #
-    # Interface functions
-    #
-    def _map_var(self, source):
-        cwd = os.getcwd()
-        if isinstance(source, path):
-            source = str(source)
-        if isinstance(source, str):
-            if source.startswith("#"):
-                return path(source, host=self.alias)
-            dest = os.path.abspath(os.path.expanduser(source))
-            # we use samefile to avoid problems with case-insensitive file system #522
-            # we also use the "cwd" name to avoid wrong case for cwd. For example,
-            # if the cwd = '/Users/user/Project'
-            # then, dest = '/USERS/USER/PROJECT/a.txt'
-            # would be converted to '/Users/user/Project/a.txt' before path mapping
-            if os.path.exists(dest[:len(cwd)]) and os.path.samefile(
-                    dest[:len(cwd)], cwd):
-                dest = cwd + dest[len(cwd):]
-            matched = [
-                k for k in self.path_map.keys()
-                if os.path.exists(dest[:len(k)]) and
-                os.path.samefile(dest[:len(k)], k)
-            ]
-            if matched:
-                # pick the longest key that matches
-                k = max(matched, key=len)
-                dest = self.path_map[k] + dest[len(k):]
-            else:
-                env.logger.debug(
-                    f"Path {source} is not under any specified paths of localhost and is mapped to {dest} on remote host."
-                )
-            return dest.replace("\\", "/")
-        if isinstance(source, (Sequence, set, sos_targets)):
-            ret = [self._map_var(x) for x in source]
-            return [x for x in ret if x is not None]
-        env.logger.debug(f"Ignore unmappable source {source}")
-        return source
+        return any(fullpath.startswith(sdir) for sdir in self.shared_dirs)
 
     def test_connection(self):
         try:
@@ -592,179 +446,6 @@ def test_connection(self):
             return f"Failed to check remote connection {self.address}:{self.port}: {e}"
         return "OK"
 
-    def _reverse_map_var(self, dest):
-        if isinstance(dest, path):
-            dest = str(dest)
-        if isinstance(dest, str):
-            matched = [
-                l for l, r in self.path_map.items()
-                if dest.startswith(r) and (len(r) == len(dest) or r.endswith(
-                    "/") or r.endswith("\\") or dest[len(r)] in ("/", "\\"))
-            ]
-            if matched:
-                # pick the longest key that matches
-                k = max(matched, key=len)
-                dest = k + dest[len(self.path_map[k]):]
-            else:
-                env.logger.debug(
-                    f"Path {dest} is not under any specified paths of localhost and is mapped to {dest} on local host."
-                )
-            return dest.replace("\\", "/")
-        if isinstance(dest, (Sequence, set, sos_targets)):
-            ret = [self._reverse_map_var(x) for x in dest]
-            return [x for x in ret if x is not None]
-        env.logger.debug(f"Ignore unmappable source {dest}")
-        return dest
-
-    def _remote_abs(self, path):
-        # return an absolute path relative to remote host
-        path = str(path)
-        if os.path.isabs(path):
-            return path
-        return os.path.join(self._map_var(os.getcwd()), path)
-
-    def send_to_host(self, items):
-        # we only copy files and directories, not other types of targets
-        if isinstance(items, str):
-            p = [path(items)]
-        elif isinstance(items, path):
-            p = [items]
-        elif isinstance(items, Sequence):
-            ignored = [x for x in items if not isinstance(x, (str, path))]
-            if ignored:
-                env.logger.info(f"``Ignore`` {ignored}")
-            p = [path(x) for x in items if isinstance(x, (str, path))]
-        else:
-            env.logger.warning(
-                f"Unrecognized items to be sent to host: {items}")
-            return {}
-
-        items = sum(
-            [list(x.parent.glob(x.name)) for x in p],
-            [],
-        )
-
-        from .utils import find_symbolic_links
-
-        new_items = []
-        for item in items:
-            links = find_symbolic_links(item)
-            for link, realpath in links.items():
-                env.logger.info(f"Adding {realpath} for symbolic link {link}")
-            new_items.extend(links.values())
-        items.extend(new_items)
-
-        sending = self._map_path(items)
-
-        sent = {}
-        for source in sorted(sending.keys()):
-            dest = self._remote_abs(sending[source])
-            if self.is_shared(source):
-                if ("TASK" in env.config["SOS_DEBUG"] or
-                        "ALL" in env.config["SOS_DEBUG"]):
-                    env.log_to_file(
-                        "TASK", f"Skip sending {source} on shared file system")
-            else:
-                if ("TASK" in env.config["SOS_DEBUG"] or
-                        "ALL" in env.config["SOS_DEBUG"]):
-                    env.log_to_file(
-                        "TASK", f"Sending ``{source}`` to {self.alias}:{dest}")
-                cmd = cfg_interpolate(
-                    self._get_send_cmd(
-                        rename=os.path.basename(source) != os.path.basename(
-                            dest)),
-                    {
-                        "source": sos_targets(str(source).rstrip("/")),
-                        "dest": sos_targets(dest),
-                        "host": self.address,
-                        "port": self.port,
-                    },
-                )
-                if ("TASK" in env.config["SOS_DEBUG"] or
-                        "ALL" in env.config["SOS_DEBUG"]):
-                    env.log_to_file("TASK", cmd)
-                ret = subprocess.call(
-                    cmd,
-                    shell=True,
-                    stderr=subprocess.DEVNULL,
-                    stdout=subprocess.DEVNULL,
-                )
-                if ret != 0:
-                    raise RuntimeError(
-                        f'Failed to copy {source} to {self.alias} using command "{cmd}". The remote host might be unavailable.'
-                    )
-            sent[source] = dest
-        return sent
-
-    def receive_from_host(self, items):
-        if isinstance(items, str):
-            items = [items]
-        elif isinstance(items, path):
-            items = [str(items)]
-        elif isinstance(items, Sequence):
-            ignored = [x for x in items if not isinstance(x, (str, path))]
-            if ignored:
-                env.logger.info(f"``Ignore`` {ignored}")
-            items = [x for x in items if isinstance(x, (str, path))]
-        else:
-            env.logger.warning(
-                f"Unrecognized items to be retrieved from host: {items}")
-            return {}
-
-        # y could be path
-        receiving = {
-            self._remote_abs(y): str(x)
-            for x, y in self._map_path(items).items()
-        }
-        #
-        received = {}
-        for source in sorted(receiving.keys()):
-            dest = receiving[source]
-            dest_dir = os.path.dirname(dest)
-            if dest_dir and not os.path.isdir(dest_dir):
-                try:
-                    os.makedirs(dest_dir)
-                except Exception as e:
-                    env.logger.error(
-                        f"Failed to create destination directory {dest_dir}: {e}"
-                    )
-            if self.is_shared(dest) and os.path.basename(
-                    source) == os.path.basename(dest):
-                env.logger.debug(
-                    f"Skip retrieving ``{dest}`` from shared file system")
-                received[dest] = source
-            else:
-                cmd = cfg_interpolate(
-                    self._get_receive_cmd(
-                        rename=os.path.basename(source) != os.path.basename(
-                            dest)),
-                    {
-                        "source": sos_targets(str(source).rstrip("/")),
-                        "dest": sos_targets(dest),
-                        "host": self.address,
-                        "port": self.port,
-                    },
-                )
-                if ("TASK" in env.config["SOS_DEBUG"] or
-                        "ALL" in env.config["SOS_DEBUG"]):
-                    env.log_to_file("TASK", cmd)
-                try:
-                    ret = subprocess.call(
-                        cmd,
-                        shell=True,
-                        stderr=subprocess.DEVNULL,
-                        stdout=subprocess.DEVNULL,
-                    )
-                    if ret != 0:
-                        raise RuntimeError(f"command return {ret}")
-                    received[dest] = source
-                except Exception as e:
-                    raise RuntimeError(
-                        f'Failed to copy {source} from {self.alias} using command "{cmd}": {e}'
-                    ) from e
-        return received
-
-    #
     # Interface
     #
     def prepare_task(self, task_id):
@@ -776,8 +457,7 @@ def prepare_task(self, task_id):
             return False
 
     def _prepare_task(self, task_id):
-        task_file = os.path.join(
-            os.path.expanduser("~"), ".sos", "tasks", task_id + ".task")
+        task_file = os.path.join(os.path.expanduser("~"), ".sos", "tasks", task_id + ".task")
         if not os.path.isfile(task_file):
             raise ValueError(f"Missing task definition {task_file}")
         tf = TaskFile(task_id)
@@ -797,61 +477,26 @@ def _prepare_task(self, task_id):
             task_id: {},
         }
 
-        if (self.config.get("max_mem", None) is not None and
-                task_vars["_runtime"].get("mem", None) is not None and
+        if (self.config.get("max_mem", None) is not None and task_vars["_runtime"].get("mem", None) is not None and
                 self.config["max_mem"] < task_vars["_runtime"]["mem"]):
             raise ValueError(
                 f'Task {task_id} requested more mem ({task_vars["_runtime"]["mem"]}) than allowed max_mem ({self.config["max_mem"]})'
             )
-        if (self.config.get("max_cores", None) is not None and
-                task_vars["_runtime"].get("cores", None) is not None and
+        if (self.config.get("max_cores", None) is not None and task_vars["_runtime"].get("cores", None) is not None and
                 self.config["max_cores"] < task_vars["_runtime"]["cores"]):
             raise ValueError(
                 f"Task {task_id} requested more cores ({task_vars['_runtime']['cores']}) than allowed max_cores ({self.config['max_cores']})"
             )
         if (self.config.get("max_walltime", None) is not None and
                 task_vars["_runtime"].get("walltime", None) is not None and
-                expand_time(self.config["max_walltime"]) < expand_time(
-                    task_vars["_runtime"]["walltime"])):
+                expand_time(self.config["max_walltime"]) < expand_time(task_vars["_runtime"]["walltime"])):
             raise ValueError(
                 f'Task {task_id} requested more walltime ({task_vars["_runtime"]["walltime"]}) than allowed max_walltime ({self.config["max_walltime"]})'
             )
 
-        if task_vars["_input"] and not isinstance(task_vars["_input"],
-                                                  Undetermined):
-            sent = self.send_to_host(task_vars["_input"])
-            if sent:
-                env.logger.info(
-                    f"{task_id} ``sent`` {short_repr(sent.keys())} to {self.alias}"
-                )
-        if task_vars["_depends"] and not isinstance(task_vars["_depends"],
-                                                    Undetermined):
-            sent = self.send_to_host(task_vars["_depends"])
-            if sent:
-                env.logger.info(
-                    f"{task_id} ``sent`` {short_repr(sent.keys())} to {self.alias}"
-                )
-        if "to_host" in task_vars["_runtime"]:
-            if not isinstance(
-                    task_vars["_runtime"]["to_host"],
-                (str, Sequence)) or (isinstance(
-                    task_vars["_runtime"]["to_host"], Sequence) and not all(
-                        isinstance(x, str)
-                        for x in task_vars["_runtime"]["to_host"])):
-                raise ValueError(
-                    f'Parameter to_host accepts a list of paths (strings). {task_vars["_runtime"]["to_host"]} provided'
-                )
-
-            sent = self.send_to_host(task_vars["_runtime"]["to_host"])
-            if sent:
-                env.logger.info(
-                    f"{task_id} ``sent`` {short_repr(sent.keys())} to {self.alias}"
-                )
-
         # map variables
         runtime["_runtime"]["workdir"] = (
-            task_vars["_runtime"]["workdir"] if "workdir"
-            in task_vars["_runtime"] else path.cwd().to_named_path())
+            task_vars["_runtime"]["workdir"] if "workdir" in task_vars["_runtime"] else str(path.cwd()))
 
         if runtime["_runtime"]["workdir"].startswith("#"):
             try:
@@ -861,21 +506,16 @@ def _prepare_task(self, task_id):
                     f'Working directory {runtime["_runtime"]["workdir"]} does not exist on remote host {self.alias}: {e}'
                 ) from e
         elif path(runtime["_runtime"]["workdir"]).is_absolute():
-            env.logger.debug(
-                f'Absolute path {path(runtime["_runtime"]["workdir"])} used as workdir.'
-            )
+            env.logger.debug(f'Absolute path {path(runtime["_runtime"]["workdir"])} used as workdir.')
 
-        env.log_to_file("TASK",
-                        f'Set workdir to {runtime["_runtime"]["workdir"]}')
+        env.log_to_file("TASK", f'Set workdir to {runtime["_runtime"]["workdir"]}')
 
         # server restrictions #488
         for key in ("max_mem", "max_cores", "max_walltime"):
             if key in self.config:
                 runtime["_runtime"][key] = (
-                    format_HHMMSS(self.config[key])
-                    if key == "max_walltime" else self.config[key])
-        runtime["_runtime"]["localhost"] = get_config(
-            ["hosts", self.alias], allowed_keys=["shared", "paths"])
+                    format_HHMMSS(self.config[key]) if key == "max_walltime" else self.config[key])
+        runtime["_runtime"]["localhost"] = get_config(["hosts", self.alias], allowed_keys=["shared", "paths"])
         # only update task file if there are runtime information
         if len(runtime) > 1 or runtime["_runtime"] or runtime != old_runtime:
             tf.runtime = runtime
@@ -900,31 +540,23 @@ def send_job_file(self, job_file, dir="tasks"):
         try:
             subprocess.check_call(send_cmd, shell=True)
         except subprocess.CalledProcessError as e:
-            raise RuntimeError(
-                f"Failed to copy job {job_file} to {self.alias} using command {send_cmd}: {e}"
-            ) from e
-
-    def check_output(self,
-                     cmd: object,
-                     under_workdir=False,
-                     **kwargs) -> object:
+            raise RuntimeError(f"Failed to copy job {job_file} to {self.alias} using command {send_cmd}: {e}") from e
+
+    def check_output(self, cmd: object, under_workdir=False, **kwargs) -> object:
         if isinstance(cmd, list):
             cmd = subprocess.list2cmdline(cmd)
         try:
             cmd = cfg_interpolate(
-                self._get_execute_cmd(
-                    under_workdir=under_workdir, use_heredoc="." in cmd),
+                self._get_execute_cmd(under_workdir=under_workdir, use_heredoc="." in cmd),
                 {
                     "host": self.address,
                     "port": self.port,
                     "cmd": cmd.replace("'", r"'\''"),
-                    "workdir": self._map_var(os.getcwd()),
+                    "workdir": os.getcwd(),
                 },
             )
         except Exception as e:
-            raise ValueError(
-                f'Failed to run command {cmd}: {e} ({env.sos_dict["CONFIG"]})'
-            ) from e
+            raise ValueError(f'Failed to run command {cmd}: {e} ({env.sos_dict["CONFIG"]})') from e
         if "TASK" in env.config["SOS_DEBUG"] or "ALL" in env.config["SOS_DEBUG"]:
             env.log_to_file("TASK", f"Executing command ``{cmd}``")
         try:
@@ -938,13 +570,12 @@ def check_call(self, cmd, under_workdir=False, **kwargs):
             cmd = subprocess.list2cmdline(cmd)
         try:
             cmd = cfg_interpolate(
-                self._get_execute_cmd(
-                    under_workdir=under_workdir, use_heredoc="." in cmd),
+                self._get_execute_cmd(under_workdir=under_workdir, use_heredoc="." in cmd),
                 {
                     "host": self.address,
                     "port": self.port,
                     "cmd": cmd.replace("'", r"'\''"),
-                    "workdir": self._map_var(os.getcwd()),
+                    "workdir": os.getcwd(),
                 },
             )
         except Exception as e:
@@ -962,13 +593,12 @@ def run_command(self, cmd, wait_for_task, realtime=False, **kwargs):
             cmd = subprocess.list2cmdline(cmd)
         try:
             cmd = cfg_interpolate(
-                self._get_execute_cmd(
-                    under_workdir=False, use_heredoc="." in cmd),
+                self._get_execute_cmd(under_workdir=False, use_heredoc="." in cmd),
                 {
                     "host": self.address,
                     "port": self.port,
                     "cmd": cmd.replace("'", r"'\''"),
-                    "workdir": self._map_var(os.getcwd()),
+                    "workdir": os.getcwd(),
                 },
             )
         except Exception as e:
@@ -1005,9 +635,7 @@ def receive_result(self, task_id: str) -> Dict[str, int]:
         # it is possible that local files are readonly (e.g. a pluse file) so we first need to
         # make sure the files are readable and remove them. Also, we do not want any file that is
         # obsolete to appear as new after copying
-        for lfile in glob.glob(
-                os.path.join(
-                    os.path.expanduser("~"), ".sos", "tasks", task_id + ".*")):
+        for lfile in glob.glob(os.path.join(os.path.expanduser("~"), ".sos", "tasks", task_id + ".*")):
             if not os.access(lfile, os.W_OK):
                 os.chmod(lfile, stat.S_IREAD | stat.S_IWRITE)
             # os.remove(lfile)
@@ -1024,16 +652,14 @@ def receive_result(self, task_id: str) -> Dict[str, int]:
             ret = subprocess.call(receive_cmd, shell=True)
             if ret != 0:
                 raise RuntimeError(
-                    f"Failed to retrieve result of job {task_id} from {self.alias} with cmd\n{receive_cmd}"
-                )
+                    f"Failed to retrieve result of job {task_id} from {self.alias} with cmd\n{receive_cmd}")
 
         tf = TaskFile(task_id)
         params = tf.params
         res = tf.result
 
         if not res:
-            env.logger.debug(
-                f"Result for {task_id} is not received (no result)")
+            env.logger.debug(f"Result for {task_id} is not received (no result)")
             return {
                 "ret_code": 1,
                 "task": task_id,
@@ -1041,8 +667,7 @@ def receive_result(self, task_id: str) -> Dict[str, int]:
                 "output": sos_targets(),
             }
 
-        if ("ret_code" in res and res["ret_code"] != 0) or ("succ" in res and
-                                                            res["succ"] != 0):
+        if ("ret_code" in res and res["ret_code"] != 0) or ("succ" in res and res["succ"] != 0):
             _show_err_and_out(task_id, res)
             env.logger.info(f"Ignore remote results for failed job {task_id}")
             return res
@@ -1052,28 +677,11 @@ def receive_result(self, task_id: str) -> Dict[str, int]:
         # do we need to copy files? We need to consult original task file
         # not the converted one
         job_dict = params.sos_dict
-        if ("_output" in job_dict and job_dict["_output"] and
-                not isinstance(job_dict["_output"], Undetermined) and
+        if ("_output" in job_dict and job_dict["_output"] and not isinstance(job_dict["_output"], Undetermined) and
                 env.config["run_mode"] != "dryrun"):
-            received = self.receive_from_host(
-                [x for x in job_dict["_output"] if isinstance(x, (str, path))])
+            received = {x:x for x in job_dict["_output"] if isinstance(x, (str, path))}
             if received:
-                env.logger.info(
-                    f"{task_id} ``received`` {short_repr(received.keys())} from {self.alias}"
-                )
-        if "from_host" in job_dict[
-                "_runtime"] and env.config["run_mode"] != "dryrun":
-            if isinstance(job_dict["_runtime"]["from_host"], (Sequence, str)):
-                received = self.receive_from_host(
-                    job_dict["_runtime"]["from_host"])
-                if received:
-                    env.logger.info(
-                        f"{task_id} ``received`` {short_repr(received.keys())} from {self.alias}"
-                    )
-            else:
-                env.logger.warning(
-                    f"Expecting a string or list of string from from_host: {job_dict['_runtime']['from_host']} received"
-                )
+                env.logger.info(f"{task_id} ``received`` {short_repr(received.keys())} from {self.alias}")
         # we need to translate result from remote path to local
         if "output" in res:
             if "_output" not in job_dict:
@@ -1081,8 +689,7 @@ def receive_result(self, task_id: str) -> Dict[str, int]:
                 env.logger.warning("Missing _output in task dict")
                 res["output"] = sos_targets()
             elif job_dict["_output"].undetermined():
-                res["output"] = sos_targets(
-                    self._reverse_map_var(res["output"]))
+                res["output"] = sos_targets(res["output"])
             else:
                 res["output"] = job_dict["_output"]
         if "subtasks" in res:
@@ -1093,12 +700,9 @@ def receive_result(self, task_id: str) -> Dict[str, int]:
                         env.logger.warning("Missing _output in subparams")
                         res["subtasks"][tid]["output"] = sos_targets()
                     elif subparams.sos_dict["_output"].undetermined():
-                        res["subtasks"][tid]["output"] = sos_targets(
-                            self._reverse_map_var(
-                                res["subtasks"][tid]["output"]))
+                        res["subtasks"][tid]["output"] = sos_targets(res["subtasks"][tid]["output"])
                     else:
-                        res["subtasks"][tid]["output"] = subparams.sos_dict[
-                            "_output"]
+                        res["subtasks"][tid]["output"] = subparams.sos_dict["_output"]
         return res
 
 
@@ -1139,13 +743,11 @@ def _detect_host(self) -> str:
         #
         for host, host_info in env.sos_dict["CONFIG"]["hosts"].items():
             # find by key hostname
-            if "hostname" in host_info and host_info["hostname"].lower(
-            ) == hostname:
+            if "hostname" in host_info and host_info["hostname"].lower() == hostname:
                 return host
             # find by key hostname
             if "hostname" in host_info:
-                hn = get_config(
-                    "hosts", host, "hostname", expected_type=str).lower()
+                hn = get_config("hosts", host, "hostname", expected_type=str).lower()
                 if hn.split(".")[0] == hostname or hn == hostname.split(".")[0]:
                     return host
             # find by alias
@@ -1154,8 +756,8 @@ def _detect_host(self) -> str:
             # find by address
             if "address" in host_info:
                 addr = get_config("hosts", host, "address", expected_type=str)
-                if (addr.split("@")[-1].lower() == hostname or addr.split(
-                        ".", 1)[0].split("@")[-1].lower() == hostname):
+                if (addr.split("@")[-1].lower() == hostname or
+                        addr.split(".", 1)[0].split("@")[-1].lower() == hostname):
                     return host
                 if any(ip == addr.split("@")[-1] for ip in ips):
                     return host
@@ -1167,21 +769,13 @@ def _get_local_host(self) -> str:
             load_config_files()
         # look for an entry with gethost
         if "hosts" not in env.sos_dict["CONFIG"]:
-            env.sos_dict["CONFIG"]["hosts"] = {
-                "localhost": {
-                    "address": "localhost",
-                    "alias": "localhost"
-                }
-            }
+            env.sos_dict["CONFIG"]["hosts"] = {"localhost": {"address": "localhost", "alias": "localhost"}}
             return "localhost"
         #
         # check if a key localhost is defined
         if "localhost" in env.sos_dict["CONFIG"]:
-            if (env.sos_dict["CONFIG"]["localhost"]
-                    not in env.sos_dict["CONFIG"]["hosts"]):
-                raise ValueError(
-                    f"Undefined localhost {env.sos_dict['CONFIG']['localhost']}"
-                )
+            if env.sos_dict["CONFIG"]["localhost"] not in env.sos_dict["CONFIG"]["hosts"]:
+                raise ValueError(f"Undefined localhost {env.sos_dict['CONFIG']['localhost']}")
             return env.sos_dict["CONFIG"]["localhost"]
         env.sos_dict["CONFIG"]["localhost"] = "localhost"
         return "localhost"
@@ -1201,9 +795,7 @@ def _get_remote_host(self, alias: Optional[str]) -> str:
         if alias in env.sos_dict["CONFIG"]["hosts"]:
             return alias
         # assuming the host name is a name or IP address
-        env.logger.debug(
-            f"Assuming {alias} to be a hostname or IP address not defined in hosts file"
-        )
+        env.logger.debug(f"Assuming {alias} to be a hostname or IP address not defined in hosts file")
         env.sos_dict["CONFIG"]["hosts"][alias] = {
             "address": alias,
             "alias": alias,
@@ -1242,89 +834,46 @@ def _get_config(self, alias: Optional[str]) -> None:
             )
 
             same_host = LOCAL == REMOTE
-            if same_host and LOCAL != "localhost" and LOCAL != DETECTED and DETECTED not in env.sos_dict[
-                    'CONFIG']['hosts']:
+            if same_host and LOCAL != "localhost" and LOCAL != DETECTED and DETECTED not in env.sos_dict['CONFIG'][
+                    'hosts']:
                 # if "localhost" is defined, but does not match by ip address etc,
                 # we assume that the matched_host is a separate host with the same
                 # configuration (see #1407 for details)
-                env.logger.debug(
-                    f'Specified host {LOCAL} does not match detected host {DETECTED}.'
-                )
-                local_cfg = copy.deepcopy(
-                    env.sos_dict["CONFIG"]["hosts"][LOCAL])
+                env.logger.debug(f'Specified host {LOCAL} does not match detected host {DETECTED}.')
+                local_cfg = copy.deepcopy(env.sos_dict["CONFIG"]["hosts"][LOCAL])
                 env.sos_dict["CONFIG"]["hosts"][DETECTED] = local_cfg
                 LOCAL = DETECTED
 
             cfg = env.sos_dict["CONFIG"]["hosts"]
             # if local and remote hosts are the same
-            if (same_host or
-                    "address" not in env.sos_dict["CONFIG"]["hosts"][REMOTE] or
+            if (same_host or "address" not in env.sos_dict["CONFIG"]["hosts"][REMOTE] or
                 ("address" in env.sos_dict["CONFIG"]["hosts"][REMOTE] and
-                 env.sos_dict["CONFIG"]["hosts"][REMOTE]["address"]
-                 == "localhost")):
+                 env.sos_dict["CONFIG"]["hosts"][REMOTE]["address"] == "localhost")):
                 # there would be no path map
-                self.config["path_map"] = []
                 self.config["shared"] = ["/"]
                 # do not override address setting to use localhost
                 # because the address should be used in #1407
                 # self.config["address"] = "localhost"
             else:
-                self.config["path_map"] = []
 
                 def normalize_value(x):
                     x = cfg_interpolate(x)
                     return x if x.endswith(os.sep) else (x + os.sep)
 
                 if "shared" in cfg[LOCAL] and "shared" in cfg[REMOTE]:
-                    common = set(cfg[LOCAL]["shared"].keys()) & set(
-                        cfg[REMOTE]["shared"].keys())
+                    common = set(cfg[LOCAL]["shared"].keys()) & set(cfg[REMOTE]["shared"].keys())
                     if common:
-                        lcl_shrd = get_config(
-                            "hosts", LOCAL, "shared", expected_type=dict)
-                        rmt_shrd = get_config(
-                            "hosts", REMOTE, "shared", expected_type=dict)
-                        self.config["shared"] = [
-                            normalize_value(lcl_shrd[x]) for x in common
-                        ]
-                        self.config["path_map"] = [
-                            f"{normalize_value(lcl_shrd[x])} -> {normalize_value(rmt_shrd[x])}"
-                            for x in common
-                        ]
-                # if paths are defined for both local and remote host, define path_map
-                if ("paths" in cfg[LOCAL] and
-                        cfg[LOCAL]["paths"]) and ("paths" in cfg[REMOTE] and
-                                                  cfg[REMOTE]["paths"]):
-                    if any(k not in cfg[REMOTE]["paths"]
-                           for k in cfg[LOCAL]["paths"].keys()):
-                        env.logger.debug(
-                            f'One or more local paths {", ".join(cfg[LOCAL]["paths"].keys())} cannot be mapped to remote host {REMOTE} with paths {",".join(cfg[REMOTE]["paths"].keys())}'
-                        )
-                    #
-                    lpaths = get_config(
-                        "hosts", LOCAL, "paths", expected_type=dict)
-                    rpaths = get_config(
-                        "hosts", REMOTE, "paths", expected_type=dict)
-                    self.config["path_map"].extend([
-                        f"{normalize_value(lpaths[x])} -> {normalize_value(rpaths[x])}"
-                        for x in lpaths.keys()
-                        if x in rpaths
-                    ])
+                        lcl_shrd = get_config("hosts", LOCAL, "shared", expected_type=dict)
+                        self.config["shared"] = [normalize_value(lcl_shrd[x]) for x in common]
                 if "pem_file" in cfg[LOCAL]:
                     if isinstance(cfg[LOCAL]["pem_file"], dict):
                         if REMOTE in cfg[LOCAL]["pem_file"]:
-                            self.config["pem_file"] = get_config(
-                                "hosts",
-                                LOCAL,
-                                "pem_file",
-                                REMOTE,
-                                expected_type=str)
+                            self.config["pem_file"] = get_config("hosts", LOCAL, "pem_file", REMOTE, expected_type=str)
                     elif isinstance(cfg[LOCAL]["pem_file"], str):
-                        self.config["pem_file"] = get_config(
-                            "hosts", LOCAL, "pem_file", expected_type=str)
+                        self.config["pem_file"] = get_config("hosts", LOCAL, "pem_file", expected_type=str)
                     else:
                         raise ValueError(
-                            f"Option pem_file should be a string or dictionary, {cfg[LOCAL]['pem_file']} provided."
-                        )
+                            f"Option pem_file should be a string or dictionary, {cfg[LOCAL]['pem_file']} provided.")
         elif LOCAL == REMOTE:
             # now we have checked local and remote are not defined, but they are the same, so
             # it is safe to assume that they are both local hosts
@@ -1333,16 +882,14 @@ def normalize_value(x):
                 "alias": LOCAL,
             }
         else:
-            raise ValueError(
-                f"Undefined local and remote hosts {LOCAL} and {REMOTE}.")
+            raise ValueError(f"Undefined local and remote hosts {LOCAL} and {REMOTE}.")
         #
         self.config["alias"] = self.alias
         self.description = self.config.get("description", "")
 
         # standardize parameters max_walltime, max_cores, and max_mem for the host
         if "max_walltime" in self.config:
-            self.config["max_walltime"] = format_HHMMSS(
-                self.config["max_walltime"])
+            self.config["max_walltime"] = format_HHMMSS(self.config["max_walltime"])
         if "max_cores" in self.config:
             if not isinstance(self.config["max_cores"], int):
                 raise ValueError("An integer is expected for max_cores")
@@ -1351,36 +898,27 @@ def normalize_value(x):
 
     def _get_task_and_workflow_engine(self):
         if self._engine_type == "process":
-            task_engine = BackgroundProcess_TaskEngine(
-                self.host_instances[self.alias])
-            workflow_engine = BackgroundProcess_WorkflowEngine(
-                self.host_instances[self.alias])
+            task_engine = BackgroundProcess_TaskEngine(self.host_instances[self.alias])
+            workflow_engine = BackgroundProcess_WorkflowEngine(self.host_instances[self.alias])
         else:
             task_engine = None
             workflow_engine = None
 
-            for entrypoint in pkg_resources.iter_entry_points(
-                    group="sos_taskengines"):
+            for entrypoint in pkg_resources.iter_entry_points(group="sos_taskengines"):
                 try:
                     if entrypoint.name == self._engine_type:
-                        task_engine = entrypoint.load()(
-                            self.host_instances[self.alias])
+                        task_engine = entrypoint.load()(self.host_instances[self.alias])
                         break
                 except Exception as e:
-                    env.logger.debug(
-                        f"Failed to load task engine {self._engine_type}: {e}")
+                    env.logger.debug(f"Failed to load task engine {self._engine_type}: {e}")
 
-            for entrypoint in pkg_resources.iter_entry_points(
-                    group="sos_workflowengines"):
+            for entrypoint in pkg_resources.iter_entry_points(group="sos_workflowengines"):
                 try:
                     if entrypoint.name == self._engine_type:
-                        workflow_engine = entrypoint.load()(
-                            self.host_instances[self.alias])
+                        workflow_engine = entrypoint.load()(self.host_instances[self.alias])
                         break
                 except Exception as e:
-                    env.logger.debug(
-                        f"Failed to load workflow engine {self._engine_type}: {e}"
-                    )
+                    env.logger.debug(f"Failed to load workflow engine {self._engine_type}: {e}")
 
             if task_engine is None and workflow_engine is None:
                 raise ValueError(
@@ -1388,22 +926,18 @@ def _get_task_and_workflow_engine(self):
                 )
         return task_engine, workflow_engine
 
-    def _get_host_agent(self, start_engine: bool,
-                        test_connection: bool) -> None:
+    def _get_host_agent(self, start_engine: bool, test_connection: bool) -> None:
         if "queue_type" not in self.config:
             self._engine_type = "process"
         else:
             self._engine_type = self.config["queue_type"].strip()
         # if there is no engine, or if the engine was stopped
-        if self.alias not in self.host_instances or (
-                hasattr(self.host_instances[self.alias], "_task_engine") and
-                self.host_instances[self.alias]._task_engine._is_stopped):
+        if self.alias not in self.host_instances or (hasattr(self.host_instances[self.alias], "_task_engine") and
+                                                     self.host_instances[self.alias]._task_engine._is_stopped):
             if self.config["address"] == "localhost":
-                self.host_instances[self.alias] = LocalHost(
-                    self.config, test_connection=test_connection)
+                self.host_instances[self.alias] = LocalHost(self.config, test_connection=test_connection)
             else:
-                self.host_instances[self.alias] = RemoteHost(
-                    self.config, test_connection=test_connection)
+                self.host_instances[self.alias] = RemoteHost(self.config, test_connection=test_connection)
 
             task_engine, workflow_engine = self._get_task_and_workflow_engine()
             self.host_instances[self.alias]._task_engine = task_engine
@@ -1421,23 +955,11 @@ def _get_host_agent(self, start_engine: bool,
             self._workflow_engine = self._host_agent._workflow_engine
         # it is possible that Host() is initialized before with start_engine=False
         # and called again to start engine
-        if (start_engine and self._task_engine is not None and
-                not self._task_engine.is_alive()):
+        if (start_engine and self._task_engine is not None and not self._task_engine.is_alive()):
             self._task_engine.start()
 
     # public interface
     #
-    # based on Host definition
-    #
-    def send_to_host(self, items):
-        return self._host_agent.send_to_host(items)
-
-    def receive_from_host(self, items):
-        return self._host_agent.receive_from_host(items)
-
-    def map_var(self, rvars):
-        return self._host_agent._map_var(rvars)
-
     def target_exists(self, targets):
         return self._host_agent.target_exists(targets)
 
@@ -1446,9 +968,7 @@ def target_signature(self, targets):
 
     def submit_task(self, task_id: str) -> str:
         if not self._task_engine:
-            raise RuntimeError(
-                f"No task engine or invalid engine definition defined for host {self.alias}"
-            )
+            raise RuntimeError(f"No task engine or invalid engine definition defined for host {self.alias}")
         return self._task_engine.submit_task(task_id)
 
     def check_status(self, tasks: List[str]) -> List[str]:
@@ -1460,8 +980,5 @@ def retrieve_results(self, tasks: List[str]):
 
     def execute_workflow(self, script, cmd, **template_args):
         if not self._workflow_engine:
-            raise RuntimeError(
-                f"No workflow engine or invalid engine definition defined for host {self.alias}"
-            )
-        return self._workflow_engine.execute_workflow(script, cmd,
-                                                      **template_args)
+            raise RuntimeError(f"No workflow engine or invalid engine definition defined for host {self.alias}")
+        return self._workflow_engine.execute_workflow(script, cmd, **template_args)
diff --git a/src/sos/parser.py b/src/sos/parser.py
index f4706a271..87b8f4dd3 100755
--- a/src/sos/parser.py
+++ b/src/sos/parser.py
@@ -490,6 +490,7 @@ def wrap_script(self) -> None:
             self._script = repr(self._script)
         self.statements[-1] = [
             "!",
+            #f'sos_run_script("{self._action}", {self._script}{(", " + opt) if opt else ""})\n',
             f'{self._action}({self._script}{(", " + opt) if opt else ""})\n',
         ]
         self.values = []
diff --git a/src/sos/remote.py b/src/sos/remote.py
index c09a0c84b..67ab466f6 100755
--- a/src/sos/remote.py
+++ b/src/sos/remote.py
@@ -186,97 +186,18 @@ def test_cmd(host, cmd):
         return str(e)
 
 
-def test_paths(host):
-    if host.address == "localhost":
-        return "OK"
-    # shared means, if localhost creates a file, it should be
-    # instantly available on the remote host
-    if not host.path_map:
-        return "No path_map between local and remote host."
-    import random
-
-    tID = random.randint(1, 100000)
-    for local, remote in host.path_map.items():
-        if local in host.shared_dirs:
-            # will be tested by 'shared'
-            continue
-        # now, let us see if two directory has the same files?
-        if not os.path.isdir(local):
-            return f"Mapped directory {local} does not exist."
-        # remote?
-        try:
-            host.check_output(f"ls -a {path(remote):q}")
-        except Exception:
-            return f"Failed to access shared directory {remote} on remote host."
-
-        # test if local directory is writable
-        try:
-            with open(os.path.join(local, f".sos_test_{tID}.txt"),
-                      "w") as tFile:
-                tFile.write(f"{tID}")
-        except Exception:
-            return f"Failed to write to mapped directory {local}"
-        # test if file can be sent
-        try:
-            host.send_to_host(os.path.join(local, f".sos_test_{tID}.txt"))
-        except Exception as e:
-            return (
-                f"Failed to send files under {local} to remote host under {remote}: {e}"
-            )
-
-        # the file should be available on remote host
-        try:
-            remote_content = host.check_output(
-                f"cat {remote}/.sos_test_{tID}.txt")
-        except Exception as e:
-            return (
-                f"Failed to send files under {local} to remote host under {remote}: {e}"
-            )
-        if remote_content != str(tID):
-            return f"Content of file sent does not match: {tID} sent, {remote_content} received"
-        # test retrieving files
-        # remove local file
-        os.remove(os.path.join(local, f".sos_test_{tID}.txt"))
-        # copy file back
-        try:
-            host.receive_from_host(os.path.join(local, f".sos_test_{tID}.txt"))
-        except Exception as e:
-            return f"Failed to receive file from remote host {remote}: {e}"
-        #
-        if not os.path.isfile(os.path.join(local, f".sos_test_{tID}.txt")):
-            return (
-                f"Failed to receive file from remote host {remote}: file does not exist"
-            )
-        # check file content?
-        with open(os.path.join(local, f".sos_test_{tID}.txt")) as tFile:
-            remote_content = tFile.read()
-        if remote_content != str(tID):
-            return f"Content of received file does not match: {tID} expected, {remote_content} received."
-        # if everything ok, remove local and remote test files
-        os.remove(os.path.join(local, f".sos_test_{tID}.txt"))
-        #
-        try:
-            remote_content = host.check_output(
-                f"rm {remote}/.sos_test_{tID}.txt")
-        except Exception as e:
-            return f"Failed to remove test file on remote host: {e}"
-    return "OK"
-
-
 def test_shared(host):
     if host.address == "localhost":
         return "OK (localhost)"
     # shared means, if localhost creates a file, it should be
     # instantly available on the remote host
     for local in host.shared_dirs:
-        if local not in host.path_map:
-            return f"shared directory {local} not in path_map"
         # now, let us see if two directory has the same files?
         if not os.path.isdir(local):
             return f"shared directory {local} does not exist."
         local_files = os.listdir(local)
         # remote?
-        remote = host.path_map[local]
+        remote = local
         try:
             remote_files = host.check_output(f"ls -a {path(remote):q}")
         except Exception:
@@ -304,7 +225,7 @@ def test_queue(host, cmd=None, verbosity=1):
     except Exception as e:
         if verbosity > 2:
             env.logger.warning(e)
-        return [host, "?", "?", "-", "-", "-", "-", "-"]
+        return [host, "?", "?", "-", "-", "-", "-"]
     ssh_res = test_ssh(h._host_agent)
     return [
         h.alias,
@@ -314,7 +235,6 @@ def test_queue(host, cmd=None, verbosity=1):
         test_scp(h._host_agent) if ssh_res.startswith("OK") else "-",
         test_cmd(h._host_agent, [h.config.get("sos", "sos"), "-h"])
         if ssh_res.startswith("OK") else "-",
-        test_paths(h._host_agent) if ssh_res.startswith("OK") else "-",
         test_shared(h._host_agent) if ssh_res.startswith("OK") else "-",
     ] + ([] if cmd is None else [test_cmd(h._host_agent, cmd)])
 
@@ -584,69 +504,3 @@ def run_command_on_hosts(cfg, hosts, cmd, verbosity):
             if verbosity and verbosity > 2:
                 sys.stderr.write(get_traceback())
             env.logger.error(str(e))
-
-
-def push_to_hosts(cfg, hosts, items, verbosity):
-    env.verbosity = verbosity
-    if not hosts:
-        hosts = cfg.get("hosts", [])
-    if not hosts:
-        env.logger.warning(
-            "No remote host or task queue is defined in ~/.sos/hosts.yml.")
-        return
-    for host in hosts:
-        try:
-            env.logger.info(f'Pushing ``{" ".join(items)}`` to ``{host}``')
-
-            h = Host(host, start_engine=False)
-            #
-            sent = h.send_to_host(items)
-            #
-            env.logger.info("{} item{} sent:\n{}".format(
-                len(sent),
-                " is" if len(sent) <= 1 else "s are",
-                "\n".join([
-                    f"{x} => {sent[x]}" for x in sorted(sent.keys())
-                ]),
-            ))
-        except Exception as e:
-            from .utils import get_traceback
-
-            if verbosity and verbosity > 2:
-                sys.stderr.write(get_traceback())
-            env.logger.error(str(e))
-            sys.exit(1)
-
-
-def pull_from_host(cfg, hosts, items, verbosity):
-    env.verbosity = verbosity
-    if not hosts:
-        hosts = cfg.get("hosts", [])
-    if not hosts:
-        env.logger.warning(
-            "No remote host or task queue is defined in ~/.sos/hosts.yml.")
-        return
-    if len(hosts) > 1:
-        raise ValueError("Can only pull from a single remote host.")
-    try:
-        env.logger.info(f'Pulling ``{" ".join(items)}`` from ``{hosts[0]}``')
-
-        host = Host(hosts[0], start_engine=False)
-        #
-        received = host.receive_from_host(items)
-        #
-        print("{} item{} received:\n{}".format(
-            len(received),
-            " is" if len(received) <= 1 else "s are",
-            "\n".join([
-                f"{x} <= {received[x]}"
-                for x in sorted(received.keys())
-            ]),
-        ))
-    except Exception as e:
-        from .utils import get_traceback
-
-        if verbosity and verbosity > 2:
-            sys.stderr.write(get_traceback())
-        env.logger.error(str(e))
-        sys.exit(1)
diff --git a/src/sos/runtime.py b/src/sos/runtime.py
index 9646a954c..f35bc9007 100644
--- a/src/sos/runtime.py
+++ b/src/sos/runtime.py
@@ -4,13 +4,14 @@
 # Distributed under the terms of the 3-clause BSD License.
 import pkg_resources
 
+from .functions import (done_if, fail_if, skip_if, stop_if, warn_if)
 # backward compatibility #1337
 from .pattern import expand_pattern
 from .targets import path, paths
 from .utils import get_output, sos_get_param
 
 # silent pyflakes
-sos_get_param, get_output, path, paths, expand_pattern
+sos_get_param, get_output, path, paths, expand_pattern, done_if, fail_if, skip_if, stop_if, warn_if
 
 
 def _load_group(group: str) -> None:
@@ -41,13 +42,12 @@ def _load_group(group: str) -> None:
                         continue
             if _name == "run":
                 # this is critical so we print the warning
-                get_logger().warning(
-                    f"Failed to load target {_entrypoint.name}: {e}")
+                get_logger().warning(f"Failed to load target {_entrypoint.name}: {e}")
             else:
-                get_logger().debug(
-                    f"Failed to load target {_entrypoint.name}: {e}")
+                get_logger().debug(f"Failed to load target {_entrypoint.name}: {e}")
 
 
 _load_group("sos_targets")
+# for backward compatibility
 _load_group("sos_actions")
 _load_group("sos_functions")
diff --git a/src/sos/section_analyzer.py b/src/sos/section_analyzer.py
index 721a387de..cdd4c4d4c 100644
--- a/src/sos/section_analyzer.py
+++ b/src/sos/section_analyzer.py
@@ -11,8 +11,7 @@
 from .executor_utils import __null_func__, prepare_env, strip_param_defs
 from .parser import SoS_Step
 from .syntax import SOS_TARGETS_OPTIONS
-from .targets import (dynamic, file_target, named_output, remote, sos_step,
-                      sos_targets)
+from .targets import dynamic, file_target, named_output, sos_step, sos_targets
 from .utils import env
 
 # imported for eval, assert to reduce warning
@@ -342,7 +341,7 @@ def get_step_depends(section):
             })
             args, kwargs = SoS_eval(
                 f"__null_func__({value})", extra_dict=env.sos_dict.dict())
-            if any(isinstance(x, (dynamic, remote)) for x in args):
+            if any(isinstance(x, dynamic) for x in args):
                 dynamic_depends = True
             else:
                 step_depends.extend(sos_targets(*args, **kwargs))
@@ -398,7 +397,7 @@ def get_step_input(section, default_input):
                 step_input = sos_targets()
             else:
                 step_input = default_input
-        elif not any(isinstance(x, (dynamic, remote)) for x in args):
+        elif not any(isinstance(x, dynamic) for x in args):
             step_input = sos_targets(*args)
     except SyntaxError:
         raise
@@ -495,7 +494,7 @@ def get_step_output(section, default_output, analysis_type):
             })
             args, kwargs = SoS_eval(
                 f"__null_func__({value})", extra_dict=env.sos_dict.dict())
-            if any(isinstance(x, (dynamic, remote)) for x in args):
+            if any(isinstance(x, dynamic) for x in args):
                 raise ValueError(
                     f"Auxiliary step does not allow dynamic or remote input or output: {value} provided"
                 )
diff --git a/src/sos/syntax.py b/src/sos/syntax.py
index 05c263689..fb932cdfe 100644
--- a/src/sos/syntax.py
+++ b/src/sos/syntax.py
@@ -28,8 +28,6 @@
     "env",
     "prepend_path",
     "queue",
-    "to_host",
-    "from_host",
     "trunk_size",
     "trunk_workers",
     "tags",
diff --git a/src/sos/targets.py b/src/sos/targets.py
index df43b9073..b587b6993 100644
--- a/src/sos/targets.py
+++ b/src/sos/targets.py
@@ -21,12 +21,10 @@
 import fasteners
 import pkg_resources
 
-from .controller import (request_answer_from_controller,
-                         send_message_to_controller)
+from .controller import (request_answer_from_controller, send_message_to_controller)
 from .eval import get_config, interpolate
 from .pattern import extract_pattern
-from .utils import (Error, env, fileMD5, objectMD5, pickleable, short_repr,
-                    stable_repr, textMD5)
+from .utils import (Error, env, fileMD5, objectMD5, pickleable, short_repr, stable_repr, textMD5)
 
 __all__ = ["dynamic", "executable", "env_variable", "sos_variable"]
 
@@ -39,8 +37,7 @@ def is_basic_type(obj):
     if isinstance(obj, (tuple, list, set)):
         return all(is_basic_type(x) for x in obj)
     if isinstance(obj, dict):
-        return all(is_basic_type(x) for x in obj.keys()) and all(
-            is_basic_type(x) for x in obj.values())
+        return all(is_basic_type(x) for x in obj.keys()) and all(is_basic_type(x) for x in obj.values())
     if isinstance(obj, (file_target, path, paths)):
         return True
     # we support types defined in numpy and pandas, but not others
@@ -103,30 +100,23 @@ def set_traced(self):
     def set(self, *args, **kwargs):
         if args:
             if len(args) != 2:
-                raise ValueError(
-                    "set(name, value) or set(name=value) is expected.")
+                raise ValueError("set(name, value) or set(name=value) is expected.")
             if not is_basic_type(args[1]):
-                env.logger.warning(
-                    f"Failed to set attribute: {args[1]} is or contains unsupported data type."
-                )
+                env.logger.warning(f"Failed to set attribute: {args[1]} is or contains unsupported data type.")
                 return self
             if hasattr(self, args[0]):
                 raise ValueError(
-                    f"Attribute {args[0]} conflicts with another attribute of type {self.__class__.__name__}."
-                )
+                    f"Attribute {args[0]} conflicts with another attribute of type {self.__class__.__name__}.")
             self._dict[args[0]] = args[1]
         #
         if kwargs:
             for name, value in kwargs.items():
                 if not is_basic_type(value):
-                    env.logger.warning(
-                        f"Failed to set attribute: {value} is or contains unsupported data type."
-                    )
+                    env.logger.warning(f"Failed to set attribute: {value} is or contains unsupported data type.")
                     return self
                 if hasattr(self, name):
                     raise ValueError(
-                        f"Attribute {name} conflicts with another attribute of {value.__class__.__name__}."
-                    )
+                        f"Attribute {name} conflicts with another attribute of {value.__class__.__name__}.")
             self._dict.update(kwargs)
         return self
 
@@ -139,8 +129,7 @@ def __getattr__(self, name):
         except Exception as e:
             # if name in self._dict:
             # return self._dict.get(name)
-            raise AttributeError(
-                f"{self.__class__.__name__} object has no attribute {name}") from e
+            raise AttributeError(f"{self.__class__.__name__} object has no attribute {name}") from e
 
     def target_exists(self, mode="any"):
         # mode should be 'any', 'target', or 'signature'
@@ -169,8 +158,7 @@ def validate(self, sig):
         return self.target_signature() == sig
 
     def __eq__(self, obj):
-        return (isinstance(obj, self.__class__) and
-                self.target_signature() == obj.target_signature())
+        return (isinstance(obj, self.__class__) and self.target_signature() == obj.target_signature())
 
 
 class sos_variable(BaseTarget):
@@ -197,8 +185,6 @@ def __eq__(self, other):
 
     def __format__(self, format_spec):
         # handling special !q conversion flag
-        if format_spec and format_spec[0] == "R":
-            return self._var.__format__(format_spec[1:])
         return str(self).__format__(format_spec)
 
 
@@ -226,8 +212,6 @@ def __hash__(self):
 
     def __format__(self, format_spec):
         # handling special !q conversion flag
-        if format_spec and format_spec[0] == "R":
-            return self._var.__format__(format_spec[1:])
         return str(self).__format__(format_spec)
 
 
@@ -310,16 +294,13 @@ def target_signature(self, mode="any"):
         return textMD5(f"sos_step({self._step_name})")
 
     def __eq__(self, other):
-        return isinstance(other,
-                          sos_step) and self._step_name == other._step_name
+        return isinstance(other, sos_step) and self._step_name == other._step_name
 
     def __hash__(self):
         return hash(repr(self))
 
     def __format__(self, format_spec):
         # handling special !q conversion flag
-        if format_spec and format_spec[0] == "R":
-            return self._step_name.__format__(format_spec[1:])
         return str(self).__format__(format_spec)
 
 
@@ -343,8 +324,7 @@ def target_signature(self, mode="any"):
         return textMD5(f"named_output({self._output_name})")
 
     def __eq__(self, other):
-        return (isinstance(other, named_output) and
-                self._output_name == other._output_name)
+        return (isinstance(other, named_output) and self._output_name == other._output_name)
 
     def __hash__(self):
         return hash(repr(self))
@@ -369,88 +349,16 @@ def target_signature(self, mode="any"):
 
     def resolve(self):
         if isinstance(self._target, str):
-            return sorted(
-                [x for x in glob.glob(self._target) if os.path.isfile(x)])
-        if isinstance(self._target, Sequence) and all(
-                isinstance(x, str) for x in self._target):
-            return sorted(
-                sum(
-                    [[x
-                      for x in glob.glob(t)
-                      if os.path.isfile(x)]
-                     for t in self._target],
-                    [],
-                ))
+            return sorted([x for x in glob.glob(self._target) if os.path.isfile(x)])
+        if isinstance(self._target, Sequence) and all(isinstance(x, str) for x in self._target):
+            return sorted(sum(
+                [[x for x in glob.glob(t) if os.path.isfile(x)] for t in self._target],
+                [],
+            ))
         return self._target
 
     def __format__(self, format_spec):
         # handling special !q conversion flag
-        if format_spec and format_spec[0] == "R":
-            return sos_targets(self._target).__format__(format_spec[1:])
-        return str(self).__format__(format_spec)
-
-
-class remote(BaseTarget):
-    """A remote target is not tracked and not translated during task execution"""
-
-    def __init__(self, *targets, host=None):
-        super().__init__()
-        self.__unresolvable_object__ = True
-        self._host = host
-        if len(targets) == 1:
-            self._target = targets[0]
-        else:
-            # multi-item targets
-            self._target = targets
-        if isinstance(self._target,
-                      Sequence) and not isinstance(self._target, str):
-            self.__flattenable__ = True
-
-    def target_name(self):
-        if isinstance(self._target, str):
-            return file_target(self._target).target_name()
-        if isinstance(self._target, BaseTarget):
-            return self._target.target_name()
-        return repr(self._target)
-
-    def target_exists(self, mode="any"):
-        if not self._host and not env.config["default_queue"]:
-            return True
-        try:
-            from .hosts import Host
-
-            h = Host(self._host if self._host else env.config["default_queue"])
-            return h.target_exists(sos_targets(self._target))
-        except Exception as e:
-            env.logger.debug(
-                f'Failed to check existence of {self._target} on {self._host if self._host else env.config["default_queue"]}: {e}'
-            )
-            return True
-
-    def target_signature(self, mode="any"):
-        if not self._host and not env.config["default_queue"]:
-            return textMD5(self.target_name())
-        try:
-            from .hosts import Host
-
-            h = Host(self._host if self._host else env.config["default_queue"])
-            return h.target_signature(sos_targets(self._target))
-        except Exception as e:
-            env.logger.debug(
-                f'Failed to check existence of {self._target} on {self._host if self._host else env.config["default_queue"]}: {e}'
-            )
-            return textMD5(self.target_name())
-
-    def resolve(self):
-        return self._target
-
-    def flatten(self):
-        return [remote(x) for x in self._target]
-
-    def __format__(self, format_spec):
-        # handling special !q conversion flag
-        if format_spec and format_spec[0] == "R":
-            return sos_targets(self._target).__format__(format_spec[1:])
         return str(self).__format__(format_spec)
 
 
@@ -469,19 +377,14 @@ def __init__(self, cmd, version=None):
             self._version = tuple(version)
 
     def __eq__(self, other):
-        return (isinstance(other, executable) and self._cmd == other._cmd and
-                self._version == other._version)
+        return (isinstance(other, executable) and self._cmd == other._cmd and self._version == other._version)
 
     def target_exists(self, mode="any"):
-        if mode in ("any", "target") and shutil.which(
-                shlex.split(self._cmd)[0]):
+        if mode in ("any", "target") and shutil.which(shlex.split(self._cmd)[0]):
             if self._version:
                 try:
                     output = subprocess.check_output(
-                        self._cmd,
-                        stderr=subprocess.STDOUT,
-                        shell=True,
-                        timeout=5).decode()
+                        self._cmd, stderr=subprocess.STDOUT, shell=True, timeout=5).decode()
                 except subprocess.TimeoutExpired as e:
                     env.logger.warning(str(e))
                     return False
@@ -510,8 +413,6 @@ def __hash__(self):
 
     def __format__(self, format_spec):
         # handling special !q conversion flag
-        if format_spec and format_spec[0] == "R":
-            return self._cmd.__format__(format_spec[1:])
         return str(self).__format__(format_spec)
 
 
@@ -546,11 +447,10 @@ class path(type(Path())):
             lambda x: os.path.splitext(x)[0],
         "x":
             lambda x: os.path.splitext(x)[1],
-        "q": (lambda x: subprocess.list2cmdline([x]))
-             if sys.platform == "win32" else quote,
+        "q": (lambda x: subprocess.list2cmdline([x])) if sys.platform == "win32" else quote,
         "p":
-            lambda x: ("/" if len(x) > 1 and x[1] == ":" else "") + x.replace(
-                "\\", "/").replace(":/", "/").replace(":", "/"),
+            lambda x:
+            ("/" if len(x) > 1 and x[1] == ":" else "") + x.replace("\\", "/").replace(":/", "/").replace(":", "/"),
         "r":
             repr,
         "s":
@@ -560,6 +460,7 @@ class path(type(Path())):
             lambda x: x,
         "!":
             lambda x: x,
+        # this is deprecated $1535
         "R":
             lambda x: x,
     }
@@ -575,143 +476,25 @@ def is_relative_to(self, *other):
         except ValueError:
             return False
 
-    @staticmethod
-    def names(host=None):
-        if host is None:
-            if "__host__" not in env.sos_dict:
-                env.logger.debug(
-                   'Incomplete sos environment: missing __host__ definition, assuming "localhost".'
-                )
-            host = env.sos_dict.get("__host__", "localhost")
-        if "CONFIG" not in env.sos_dict or "hosts" not in env.sos_dict["CONFIG"]:
-            raise RuntimeError(
-                "Incomplete sos environment: missing hosts definition.")
-        if host not in env.sos_dict["CONFIG"]["hosts"]:
-            raise RuntimeError(
-                "Incomplete sos environment: undefined host {host}")
-        if "paths" not in env.sos_dict["CONFIG"]["hosts"][host]:
-            return []
-        return list(get_config(["hosts", host, "paths"]).keys())
-
     # the PathLike interface defines __fspath__ as str()
     def __str__(self):
-        return super(path, self.expandname().expanduser()).__str__()
+        return super(path, self.expanduser()).__str__()
 
     def __repr__(self):
         raw_str = super().__str__().replace(self._flavour.sep, '/')
         return f"{self.__class__.__name__}({raw_str})"
 
-    def to_named_path(self, host=None):
-        try:
-            if self._parts[0].startswith("#"):
-                return self
-            #
-            p = self if self.is_absolute() else self.resolve()
-            host = env.sos_dict.get("__host__",
-                                    "localhost") if host is None else host
-            cfg = get_config(
-                "hosts",
-                host,
-                expected_type=dict,
-                allowed_keys=["shared", "paths"])
-            if "paths" not in cfg:
-                raise ValueError("No path is defined in host defintion.")
-            relative_paths = [(name, path)
-                              for name, path in cfg["paths"].items()
-                              if p.is_relative_to(path)]
-            if not relative_paths:
-                raise ValueError(
-                    f"{self} is not relative to any of the pre-defined paths for host {host}."
-                )
-            if len(relative_paths) > 1:
-                env.logger.debug(
-                    f"{self} is relative to more than one pre-defined paths for host {host}"
-                )
-                max_length = max(len(str(x[1])) for x in relative_paths)
-                relative_paths = [
-                    x for x in relative_paths if len(str(x[1])) == max_length
-                ]
-            # return the anchored
-            related = str(p)[len(relative_paths[0][1]):]
-            if related and not related.startswith("/"):
-                related = "/" + related
-            return "#" + relative_paths[0][0] + related
-        except Exception as e:
-            raise ValueError(
-                f"Failed to relate {self} with any of the named paths: {e}") from e
-
-    def expandname(self, host=None):
-        if not self._parts or self._parts[0][:1] != "#":
-            return self
-        try:
-            # this is the case for task execution where paths is directly specified in
-            # _runtime.
-            if "_runtime" in env.sos_dict and "localhost" in env.sos_dict[
-                    "_runtime"]:
-                cfg = env.sos_dict["_runtime"]["localhost"]
-            # this is the case for the main program, or when the task is executed
-            else:
-                cfg = get_config(
-                    "hosts",
-                    env.sos_dict.get("__host__", "localhost")
-                    if host is None else host,
-                    expected_type=dict,
-                )
-            try:
-                return self._from_parts([cfg["paths"][self._parts[0][1:]]] +
-                                        self._parts[1:])
-            except KeyError:
-                return self._from_parts([cfg["shared"][self._parts[0][1:]]] +
-                                        self._parts[1:])
-        except Exception as e:
-            # if self._parts[0] == '#cwd':
-            #     return self._from_parts(
-            #         [self.cwd()] + self._parts[1:]
-            #     )
-            if host is None and "__host__" not in env.sos_dict:
-                return self
-                # raise RuntimeError(
-                #     "Incomplete sos environment: missing __host__ definition."
-                # )
-            if "CONFIG" not in env.sos_dict or "hosts" not in env.sos_dict[
-                    "CONFIG"]:
-                raise RuntimeError(
-                    "Incomplete sos environment: missing hosts definition.") from e
-            if host is not None and host not in env.sos_dict["CONFIG"]["hosts"]:
-                raise RuntimeError(
-                    f"Incomplete sos environment: undefined host {host}") from e
-            if (env.sos_dict.get("__host__", "localhost")
-                  not in env.sos_dict["CONFIG"]["hosts"]):
-                raise RuntimeError(
-                    f'Incomplete sos environment: undefined host {env.sos_dict.get("__host__", "locahost")}'
-                ) from e
-            if ("paths" not in env.sos_dict["CONFIG"]["hosts"][env.sos_dict.get(
-                    "__host__", "localhost")]):
-                raise RuntimeError(
-                    f'Incomplete sos environment: paths not defined for host {env.sos_dict.get("__host__", "localhost")}'
-                ) from e
-            name = self._parts[0][1:]
-            if (name not in env.sos_dict["CONFIG"]["hosts"]
-                [env.sos_dict.get("__host__", "localhost"
-                                 ) if host is None else host]["paths"]):
-                raise ValueError(
-                    f'Named path "{name}" not defined for host {env.sos_dict.get("__host__", "localhost") if host is None else host}'
-                ) from e
-
     def fullname(self):
         return os.path.abspath(str(self))
 
     def __eq__(self, other):
         return os.path.abspath(self.fullname()) == os.path.abspath(
-            (other
-             if isinstance(other, file_target) else path(other)).fullname())
+            (other if isinstance(other, file_target) else path(other)).fullname())
 
     def __add__(self, part):
         if isinstance(part, (str, path)):
             return self.__class__(str(self) + str(part))
-        raise ValueError(
-            f"Cannot concatenate path to {part} of type {type(part).__name__}: expect a string or path"
-        )
+        raise ValueError(f"Cannot concatenate path to {part} of type {type(part).__name__}: expect a string or path")
 
     def __format__(self, format_spec):
         # handling special !q conversion flag
@@ -737,9 +520,7 @@ def zap(self):
         if not self.exists() or not self.is_file():
             raise FileNotFoundError(str(self))
         with open(zap_file, "w") as md5:
-            md5.write(
-                f"{self.resolve()}\t{os.path.getmtime(self)}\t{os.path.getsize(self)}\t{fileMD5(self)}\n"
-            )
+            md5.write(f"{self.resolve()}\t{os.path.getmtime(self)}\t{os.path.getsize(self)}\t{fileMD5(self)}\n")
         self.unlink()
 
 
@@ -750,23 +531,22 @@ def __init__(self, *args, **kwargs):
         # this is path segments
         super().__init__(*args, **kwargs)
         if len(args) == 1 and isinstance(args[0], file_target):
-            self._md5 = args[0]._md5
+            self._md5 = args[0]._md5 if hasattr(args[0], '_md5') else None
         else:
             self._md5 = None
 
+    # this only works for Python 3.9. Starting from 3.10, newly constructed file_target
+    # from calls such as file_target().with_suffix() will not have _md5
     def _init(self, template=None):
         super()._init(template)
         self._md5 = None
 
     def create_placeholder(self):
         # create an empty placeholder file
-        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config[
-                "SOS_DEBUG"]:
+        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config["SOS_DEBUG"]:
             env.log_to_file("TARGET", f"Create placeholder target {self}")
         self.touch()
-        send_message_to_controller(
-            ["workflow_sig", "placeholder", "file_target",
-             str(self)])
+        send_message_to_controller(["workflow_sig", "placeholder", "file_target", str(self)])
 
     def target_exists(self, mode="any"):
         try:
@@ -794,12 +574,13 @@ def target_name(self):
 
     def target_signature(self):
         """Return file signature"""
+        if not hasattr(self, '_md5'):
+            self._md5 = None
         full_md5 = env.config['sig_type'] == 'md5'
         if self.exists():
             if full_md5:
                 md5_file = self + '.md5'
-                if not md5_file.exists(
-                ) or os.path.getmtime(self) > os.path.getmtime(md5_file):
+                if not md5_file.exists() or os.path.getmtime(self) > os.path.getmtime(md5_file):
                     self._md5, md5 = fileMD5(self, sig_type='both')
                     # write md5 file
                     with open(md5_file, 'w') as mfile:
@@ -819,11 +600,12 @@ def target_signature(self):
     def sig_file(self):
         # self.resolve() does not resolve non-existing file and cannot be used here
         return os.path.join(
-            os.path.expanduser('~'), '.sos', 'signatures',
-            f"{textMD5(os.path.abspath(self))}.file_info")
+            os.path.expanduser('~'), '.sos', 'signatures', f"{textMD5(os.path.abspath(self))}.file_info")
 
     def validate(self, sig=None):
         """Check if file matches its signature"""
+        if not hasattr(self, '_md5'):
+            self._md5 = None
         if env.config['sig_type'] == 'md5':
             md5_file = self + '.md5'
             if md5_file.exists():
@@ -854,12 +636,12 @@ def validate(self, sig=None):
 
     def write_sig(self):
         """Write signature to sig store"""
+        if not hasattr(self, '_md5'):
+            self._md5 = None
         if not self._md5:
             self._md5 = fileMD5(self)
         with open(self.sig_file(), "w") as sig:
-            sig.write(
-                f"{os.path.getmtime(self)}\t{os.path.getsize(self)}\t{self._md5}"
-            )
+            sig.write(f"{os.path.getmtime(self)}\t{os.path.getsize(self)}\t{self._md5}")
 
     def __hash__(self):
         return hash(repr(self))
@@ -869,8 +651,7 @@ def __fspath__(self):
         return super().__fspath__()
 
     def __eq__(self, obj):
-        return isinstance(
-            obj, file_target) and os.path.abspath(self) == os.path.abspath(obj)
+        return isinstance(obj, file_target) and os.path.abspath(self) == os.path.abspath(obj)
 
     def __deepcopy__(self, memo):
         ft = file_target(self)
@@ -882,7 +663,7 @@ def __reduce__(self):
             self.__class__,
             super().__reduce__()[1],
             {
-                "_md5": self._md5,
+                "_md5": self._md5 if hasattr(self, '_md5') else None,
                 "_dict": self._dict
             },
         ])
@@ -893,8 +674,7 @@ class sos_tempfile(file_target):
     def __new__(cls, path=None, name=None, suffix=None, prefix=None, dir=None):
         if cls is Path:
             cls = WindowsPath if os.name == "nt" else PosixPath
-        filename = request_answer_from_controller(
-            ["sos_tempfile", path, name, suffix, prefix, dir])
+        filename = request_answer_from_controller(["sos_tempfile", path, name, suffix, prefix, dir])
         return cls._from_parts([filename])
 
 
@@ -911,8 +691,7 @@ def __init__(self, *args):
 
         for t in self._paths:
             if isinstance(t, paths):
-                raise RuntimeError(
-                    f"Nested paths {t} were introduced by {args}")
+                raise RuntimeError(f"Nested paths {t} were introduced by {args}")
             if not isinstance(t, path):
                 raise RuntimeError(f"Unrecognized path {t}")
 
@@ -933,9 +712,7 @@ def __append__(self, arg):
                 self._paths.append(path(arg))
         elif isinstance(arg, sos_targets):
             if not all(isinstance(x, file_target) for x in arg._targets):
-                raise ValueError(
-                    f"Cannot convert a sos_targets object {arg} with non-file target to paths"
-                )
+                raise ValueError(f"Cannot convert a sos_targets object {arg} with non-file target to paths")
             self._paths.extend([path(str(x)) for x in arg._targets])
         elif isinstance(arg, file_target):
             self._paths.append(path(str(arg)))
@@ -966,9 +743,7 @@ def __fspath__(self):
             return self._paths[0].__fspath__()
         if not self._paths:
             raise ValueError("Cannot treat an empty paths as single path")
-        raise ValueError(
-            f"Cannot treat an paths object {self} with more than one paths as a single path"
-        )
+        raise ValueError(f"Cannot treat an paths object {self} with more than one paths as a single path")
 
     def __format__(self, format_spec):
         if "," in format_spec:
@@ -983,18 +758,14 @@ def __getattr__(self, key):
         if len(self._paths) == 1:
             return getattr(self._paths[0], key)
         if len(self._paths) == 0:
-            raise AttributeError(
-                f"Cannot get attribute {key} from empty target list")
-        raise AttributeError(
-            f"Cannot get attribute {key} from group of {len(self)} targets {self!r}"
-        )
+            raise AttributeError(f"Cannot get attribute {key} from empty target list")
+        raise AttributeError(f"Cannot get attribute {key} from group of {len(self)} targets {self!r}")
 
     def __hash__(self):
         return hash(repr(self))
 
     def __eq__(self, other):
-        return self._paths == other._paths if isinstance(other,
-                                                         paths) else other
+        return self._paths == other._paths if isinstance(other, paths) else other
 
     def __repr__(self):
         return "[" + ", ".join(repr(x) for x in self._paths) + "]"
@@ -1024,10 +795,8 @@ def __init__(self, indexes, labels=None, parent=None):
     def add_last(self, n, parent):
         # add the last n elements of parent to group
         # this has to be called after the elements have been appended
-        self._indexes.extend(
-            range(len(parent._targets) - n, len(parent._targets)))
-        self._labels.extend(parent._labels[len(parent._targets) -
-                                           n:len(parent._targets)])
+        self._indexes.extend(range(len(parent._targets) - n, len(parent._targets)))
+        self._labels.extend(parent._labels[len(parent._targets) - n:len(parent._targets)])
         return self
 
     def extend(self, grp, start, parent):
@@ -1047,11 +816,7 @@ def idx_to_targets(self, parent):
         return ret
 
     def __getstate__(self):
-        return {
-            'indexes': self._indexes,
-            'labels': self._labels,
-            'properties': self._dict
-        }
+        return {'indexes': self._indexes, 'labels': self._labels, 'properties': self._dict}
 
     def __setstate__(self, sdict):
         self._indexes = sdict["indexes"]
@@ -1090,15 +855,12 @@ def __init__(
         else:
             self._undetermined = not bool(args)
         for arg in args:
-            self.__append__(
-                arg, default_source=_source, verify_existence=_verify_existence)
+            self.__append__(arg, default_source=_source, verify_existence=_verify_existence)
         for src, value in kwargs.items():
-            self.__append__(
-                value, source=src, verify_existence=_verify_existence)
+            self.__append__(value, source=src, verify_existence=_verify_existence)
         for t in self._targets:
             if isinstance(t, sos_targets):
-                raise RuntimeError(
-                    f"Nested sos_targets {t} were introduced by {args}")
+                raise RuntimeError(f"Nested sos_targets {t} were introduced by {args}")
             if not isinstance(t, BaseTarget):
                 raise RuntimeError(f"Unrecognized target {t}")
         if _verify_existence:
@@ -1139,11 +901,7 @@ def touch(self):
     def valid(self):
         return self._targets or self._undetermined is False
 
-    def __append__(self,
-                   arg,
-                   source="",
-                   default_source="",
-                   verify_existence=False):
+    def __append__(self, arg, source="", default_source="", verify_existence=False):
         src = source if source else default_source
         if isinstance(arg, paths):
             self._targets.extend([file_target(x) for x in arg._paths])
@@ -1197,8 +955,7 @@ def __append__(self,
             for t in list(arg):
                 self.__append__(t, source=src)
         elif arg is not None:
-            raise RuntimeError(
-                f"Unrecognized targets {arg} of type {arg.__class__.__name__}")
+            raise RuntimeError(f"Unrecognized targets {arg} of type {arg.__class__.__name__}")
 
     def set_labels(self, source):
         if isinstance(source, str):
@@ -1206,16 +963,13 @@ def set_labels(self, source):
         elif len(source) == len(self._targets):
             self._labels = source
         else:
-            raise ValueError(
-                f"Invalid source {source} for sos_target with {len(self)} targets."
-            )
+            raise ValueError(f"Invalid source {source} for sos_target with {len(self)} targets.")
 
     labels = property(lambda self: self._labels, set_labels)
 
     targets = property(lambda self: self._targets)
 
-    groups = property(
-        lambda self: [x.idx_to_targets(self) for x in self._groups])
+    groups = property(lambda self: [x.idx_to_targets(self) for x in self._groups])
 
     def _get_group(self, index):
         return self._groups[index].idx_to_targets(self)
@@ -1234,8 +988,7 @@ def later_than(self, other):
         if not file_lhs:
             return False
         # now we have both.
-        return min(x.stat().st_mtime for x in file_lhs) > max(
-            x.stat().st_mtime for x in file_rhs)
+        return min(x.stat().st_mtime for x in file_lhs) > max(x.stat().st_mtime for x in file_rhs)
 
     def extend(self, another, source="", keep_groups=False):
         if isinstance(another, sos_targets):
@@ -1265,15 +1018,11 @@ def extend(self, another, source="", keep_groups=False):
         if arg._groups:
             # if source is specified, it will override labels of all groups
             if not self._groups:
-                self._groups = [
-                    _sos_group(range(n_old), parent=self) for g in arg._groups
-                ]
+                self._groups = [_sos_group(range(n_old), parent=self) for g in arg._groups]
             if len(self._groups) == 1 and len(arg._groups) > 1:
                 # 1 vs more, we duplicate itself
                 self._groups = [
-                    _sos_group(
-                        self._groups[0]._indexes,
-                        self._groups[0]._labels).set(**self._groups[0]._dict)
+                    _sos_group(self._groups[0]._indexes, self._groups[0]._labels).set(**self._groups[0]._dict)
                     for ag in arg._groups
                 ]
                 for g, ag in zip(self._groups, arg._groups):
@@ -1354,9 +1103,7 @@ def select(self, i):
         if isinstance(i, str):
             ret = sos_targets()
             ret._undetermined = self._undetermined
-            ret._targets = [
-                x for x, y in zip(self._targets, self._labels) if y == i
-            ]
+            ret._targets = [x for x, y in zip(self._targets, self._labels) if y == i]
             index_map = {
                 o_idx: n_idx for n_idx, o_idx in zip(
                     range(len(ret._targets)),
@@ -1368,11 +1115,7 @@ def select(self, i):
             for grp in self._groups:
                 ret._groups.append(
                     _sos_group(
-                        [
-                            index_map[x]
-                            for x, y in zip(grp._indexes, grp._labels)
-                            if y == i
-                        ],
+                        [index_map[x] for x, y in zip(grp._indexes, grp._labels) if y == i],
                         labels=i,
                     ).set(**grp._dict))
             return ret
@@ -1394,27 +1137,19 @@ def select(self, i):
             ret._groups = []
             if not self._groups:
                 return ret
-            index_map = {
-                o_idx: n_idx
-                for n_idx, o_idx in zip(range(len(ret._targets)), kept)
-            }
+            index_map = {o_idx: n_idx for n_idx, o_idx in zip(range(len(ret._targets)), kept)}
             kept = set(kept)
             for grp in self._groups:
                 ret._groups.append(
                     _sos_group(
                         [index_map[x] for x in grp._indexes if x in kept],
-                        [
-                            y for x, y in zip(grp._indexes, grp._labels)
-                            if x in kept
-                        ],
+                        [y for x, y in zip(grp._indexes, grp._labels) if x in kept],
                     ).set(**grp._dict))
             return ret
         ret = sos_targets()
         ret._undetermined = self._undetermined
-        ret._targets = ([self._targets[i]]
-                        if isinstance(i, int) else self._targets[i])
-        ret._labels = [self._labels[i]] if isinstance(
-            i, int) else self._labels[i]
+        ret._targets = ([self._targets[i]] if isinstance(i, int) else self._targets[i])
+        ret._labels = [self._labels[i]] if isinstance(i, int) else self._labels[i]
         ret._groups = []
         return ret
 
@@ -1422,9 +1157,7 @@ def __getitem__(self, i):
         if isinstance(i, str):
             ret = sos_targets()
             ret._undetermined = self._undetermined
-            ret._targets = [
-                x for x, y in zip(self._targets, self._labels) if y == i
-            ]
+            ret._targets = [x for x, y in zip(self._targets, self._labels) if y == i]
             index_map = {
                 o_idx: n_idx for n_idx, o_idx in zip(
                     range(len(ret._targets)),
@@ -1436,11 +1169,7 @@ def __getitem__(self, i):
             for grp in self._groups:
                 ret._groups.append(
                     _sos_group(
-                        [
-                            index_map[x]
-                            for x, y in zip(grp._indexes, grp._labels)
-                            if y == i
-                        ],
+                        [index_map[x] for x, y in zip(grp._indexes, grp._labels) if y == i],
                         labels=i,
                     ).set(**grp._dict))
             if not ret._targets:
@@ -1449,14 +1178,11 @@ def __getitem__(self, i):
         return self._targets[i]
 
     def target_signature(self):
-        return tuple((x.target_signature(), y)
-                     for x, y in zip(self._targets, self._labels))
+        return tuple((x.target_signature(), y) for x, y in zip(self._targets, self._labels))
 
     def validate(self, sig):
         return (isinstance(sig, tuple) and len(sig) == len(self._targets) and
-                all(
-                    x.validate(sig[0]) and src == sig[1]
-                    for x, src, sig in zip(self._targets, self._labels, sig)))
+                all(x.validate(sig[0]) and src == sig[1] for x, src, sig in zip(self._targets, self._labels, sig)))
 
     def target_exists(self, mode="any"):
         return all(x.target_exists(mode) for x in self._targets)
@@ -1470,12 +1196,10 @@ def __getattr__(self, name):
                     return getattr(self._targets[0], name)
                 except Exception as e:
                     raise AttributeError(
-                        f"{self.__class__.__name__} object or its first child has no attribute {name}"
-                    ) from e
+                        f"{self.__class__.__name__} object or its first child has no attribute {name}") from e
             else:
                 raise AttributeError(
-                    f"{self.__class__.__name__} object of length {len(self)} has no attribute {name}"
-                ) from e
+                    f"{self.__class__.__name__} object of length {len(self)} has no attribute {name}") from e
 
     def target_name(self):
         return f"sos_targets([{','.join(x.target_name() for x in self._targets)}],_labels=[{','.join(self._labels)}])"
@@ -1491,11 +1215,8 @@ def _dedup(self):
 
     def paired_with(self, name, properties):
         # can pair with sos_targets
-        if not isinstance(properties,
-                          sos_targets) and not is_basic_type(properties):
-            env.logger.warning(
-                f'Failed to paired_with with value "{properties}" as it contains unsupported data type'
-            )
+        if not isinstance(properties, sos_targets) and not is_basic_type(properties):
+            env.logger.warning(f'Failed to paired_with with value "{properties}" as it contains unsupported data type')
             return self
         if isinstance(properties, (bool, int, float, str, bytes)):
             for target in self._targets:
@@ -1508,28 +1229,20 @@ def paired_with(self, name, properties):
             for target, property in zip(self._targets, properties):
                 target.set(name, property)
         else:
-            raise ValueError(
-                f"Unacceptable properties {properties} for function paired_with"
-            )
+            raise ValueError(f"Unacceptable properties {properties} for function paired_with")
         return self
 
     def remove_targets(self, type, kept=None):
         """Remove targets of certain type"""
         if kept is None:
-            kept = [
-                i for i, x in enumerate(self._targets)
-                if not isinstance(x, type)
-            ]
+            kept = [i for i, x in enumerate(self._targets) if not isinstance(x, type)]
         if len(kept) == len(self._targets):
             return self
         self._targets = [self._targets[x] for x in kept]
         self._labels = [self._labels[x] for x in kept]
         if not self._groups:
             return self
-        index_map = {
-            o_idx: n_idx
-            for n_idx, o_idx in zip(range(len(self._targets)), kept)
-        }
+        index_map = {o_idx: n_idx for n_idx, o_idx in zip(range(len(self._targets)), kept)}
         kept = set(kept)
         for idx, grp in enumerate(self._groups):
             self._groups[idx] = _sos_group(
@@ -1538,23 +1251,11 @@ def remove_targets(self, type, kept=None):
             ).set(**grp._dict)
         return self
 
-    def resolve_remote(self):
-        """If target is of remote type, resolve it"""
-        for idx, target in enumerate(self._targets):
-            if isinstance(target, remote):
-                resolved = target.resolve()
-                if isinstance(resolved, str):
-                    resolved = interpolate(resolved, env.sos_dict.dict())
-                self._targets[idx] = file_target(resolved).set(**target._dict)
-        return self
-
     def group_with(self, name, properties):
         if not self._groups:
             self._group(by="all")
         if not is_basic_type(properties):
-            env.logger.warning(
-                f"Failed to set {properties} as it is or contains unsupported data type"
-            )
+            env.logger.warning(f"Failed to set {properties} as it is or contains unsupported data type")
             return self
         if isinstance(properties, (bool, int, float, str, bytes)):
             for group in self._groups:
@@ -1568,8 +1269,7 @@ def group_with(self, name, properties):
                 group.set(name, property)
         else:
             raise ValueError(
-                f"Unacceptable properties {properties} of type {properties.__class__.__name__} for function group_with"
-            )
+                f"Unacceptable properties {properties} of type {properties.__class__.__name__} for function group_with")
         return self
 
     def get(self, name, default=None):
@@ -1585,16 +1285,12 @@ def _add_groups(self, grps):
         for grp in grps:
             if not isinstance(grp, sos_targets):
                 raise RuntimeError(
-                    f"_output should be of type sos_targets: {grp} of type {grp.__class__.__name__} returned."
-                )
+                    f"_output should be of type sos_targets: {grp} of type {grp.__class__.__name__} returned.")
             start_idx = len(self._targets)
             grp_size = len(grp)
             self._targets.extend(grp._targets)
             self._labels.extend(grp._labels)
-            self._groups.append(
-                _sos_group(
-                    range(start_idx, start_idx + grp_size),
-                    labels=grp._labels).set(**grp._dict))
+            self._groups.append(_sos_group(range(start_idx, start_idx + grp_size), labels=grp._labels).set(**grp._dict))
         # in theory the groups should not overlap but in rare cases when
         # output is for example dynamic, they could overlap.
         return self._dedup()
@@ -1607,8 +1303,7 @@ def _duplicate_groups(self, n):
         n_grps = len(self._groups)
         for _ in range(n - 1):
             for grp in self._groups[:n_grps]:
-                self._groups.append(
-                    _sos_group(grp._indexes, grp._labels).set(**grp._dict))
+                self._groups.append(_sos_group(grp._indexes, grp._labels).set(**grp._dict))
         return self
 
     def _num_groups(self):
@@ -1622,14 +1317,11 @@ def _group(self, by):
             self._groups = []
 
         if by == "single":
-            self._groups = [
-                _sos_group([x], parent=self) for x in range(len(self))
-            ]
+            self._groups = [_sos_group([x], parent=self) for x in range(len(self))]
         elif by == "all":
             # default option
             self._groups = [_sos_group(range(len(self)), self._labels)]
-        elif isinstance(by, str) and (by.startswith("pairsource") or
-                                      by.startswith("pairlabel")):
+        elif isinstance(by, str) and (by.startswith("pairsource") or by.startswith("pairlabel")):
             labels = list(dict.fromkeys(self.labels))
             if len(labels) == 1:
                 raise ValueError("Cannot pairlabel input with a single label.")
@@ -1645,15 +1337,11 @@ def _group(self, by):
                     raise ValueError(f"Invalid pairsource option {by}") from e
             src_sizes = {s: self.labels.count(s) for s in labels}
             if max(src_sizes.values()) % grp_size != 0:
-                raise ValueError(
-                    f"Cannot use group size {grp_size} (option {by}) for source of size {src_sizes}"
-                )
+                raise ValueError(f"Cannot use group size {grp_size} (option {by}) for source of size {src_sizes}")
             n_groups = max(src_sizes.values()) // grp_size
             indexes = [[] for x in range(n_groups)]
             for s in labels:
-                lookup = [
-                    idx for idx, src in enumerate(self.labels) if src == s
-                ]
+                lookup = [idx for idx, src in enumerate(self.labels) if src == s]
                 if src_sizes[s] > n_groups and src_sizes[s] % n_groups == 0:
                     gs = src_sizes[s] // n_groups
                     for i in range(n_groups):
@@ -1662,20 +1350,13 @@ def _group(self, by):
                 elif n_groups >= src_sizes[s] and n_groups % src_sizes[s] == 0:
                     for i in range(n_groups):
                         # (0 ), (0, ), (1, ), (1, ) ...
-                        indexes[i].append(lookup[i //
-                                                 (n_groups // src_sizes[s])])
+                        indexes[i].append(lookup[i // (n_groups // src_sizes[s])])
                 else:
-                    raise ValueError(
-                        f'Cannot use group size {grp_size} (by="{by}") for source of size {src_sizes}'
-                    )
-            self._groups = [
-                _sos_group(indexes[x], parent=self) for x in range(n_groups)
-            ]
+                    raise ValueError(f'Cannot use group size {grp_size} (by="{by}") for source of size {src_sizes}')
+            self._groups = [_sos_group(indexes[x], parent=self) for x in range(n_groups)]
         elif isinstance(by, str) and by.startswith("pairs"):
             if len(self) % 2 != 0:
-                raise ValueError(
-                    f"Paired by has to have even number of input files: {len(self)} provided"
-                )
+                raise ValueError(f"Paired by has to have even number of input files: {len(self)} provided")
             if by == "pairs":
                 grp_size = 1
             else:
@@ -1685,19 +1366,16 @@ def _group(self, by):
                     raise ValueError(f"Invalid pairs option {by}") from e
             if grp_size == 1:
                 self._groups = [
-                    _sos_group(x, parent=self) for x in zip(
-                        range(0,
-                              len(self) // 2), range(len(self) // 2, len(self)))
+                    _sos_group(x, parent=self) for x in zip(range(0,
+                                                                  len(self) // 2), range(len(self) // 2, len(self)))
                 ]
             else:
                 if len(self) % grp_size != 0:
                     raise ValueError(
-                        f"Paired by with group size {grp_size} is not possible with input of size {len(self)}"
-                    )
+                        f"Paired by with group size {grp_size} is not possible with input of size {len(self)}")
                 self._groups = [
                     _sos_group(
-                        list(range(x[0], x[0] + grp_size)) +
-                        list(range(x[1], x[1] + grp_size)),
+                        list(range(x[0], x[0] + grp_size)) + list(range(x[1], x[1] + grp_size)),
                         parent=self,
                     ) for x in zip(
                         range(0,
@@ -1720,14 +1398,13 @@ def _group(self, by):
             else:
                 if len(self) % grp_size != 0:
                     raise ValueError(
-                        f"Paired by with group size {grp_size} is not possible with input of size {len(self)}"
-                    )
+                        f"Paired by with group size {grp_size} is not possible with input of size {len(self)}")
                 f1, f2 = tee(range(len(self) // grp_size))
                 next(f2, None)
                 self._groups = [
                     _sos_group(
-                        list(range(x[0] * grp_size, (x[0] + 1) * grp_size)) +
-                        list(range(x[1] * grp_size, (x[1] + 1) * grp_size)),
+                        list(range(x[0] * grp_size,
+                                   (x[0] + 1) * grp_size)) + list(range(x[1] * grp_size, (x[1] + 1) * grp_size)),
                         parent=self,
                     ) for x in zip(f1, f2)
                 ]
@@ -1739,15 +1416,11 @@ def _group(self, by):
                     grp_size = int(by[12:])
                 except Exception as e:
                     raise ValueError(f"Invalid pairs option {by}") from e
-            self._groups = [
-                _sos_group(x, parent=self)
-                for x in combinations(range(len(self)), grp_size)
-            ]
+            self._groups = [_sos_group(x, parent=self) for x in combinations(range(len(self)), grp_size)]
         elif by == "source" or by == "label":
             labels = list(dict.fromkeys(self.labels))
             self._groups = [
-                _sos_group([i for i, x in enumerate(self._labels) if x == src],
-                           parent=self) for src in labels
+                _sos_group([i for i, x in enumerate(self._labels) if x == src], parent=self) for src in labels
             ]
         elif isinstance(by, int) or (isinstance(by, str) and by.isdigit()):
             by = int(by)
@@ -1756,12 +1429,8 @@ def _group(self, by):
                     f"Number of samples ({len(self)}) is not a multiple of by ({by}). The last group would have less files than the other groups."
                 )
             if by < 1:
-                raise ValueError(
-                    "Value of paramter by should be a positive number.")
-            self._groups = [
-                _sos_group(range(i, min(i + by, len(self))), parent=self)
-                for i in range(0, len(self), by)
-            ]
+                raise ValueError("Value of paramter by should be a positive number.")
+            self._groups = [_sos_group(range(i, min(i + by, len(self))), parent=self) for i in range(0, len(self), by)]
         elif callable(by):
             try:
                 self._groups = []
@@ -1774,28 +1443,22 @@ def _group(self, by):
                     ) from e
                 _target_index = None
                 for grp in by(self):
-                    if isinstance(grp, Sequence) and all(
-                            isinstance(x, int) for x in grp):
+                    if isinstance(grp, Sequence) and all(isinstance(x, int) for x in grp):
                         if any(x < 0 or x >= len(self._targets) for x in grp):
-                            raise ValueError(
-                                f"Index out of range (< {len(self._targets)}): {grp}"
-                            )
+                            raise ValueError(f"Index out of range (< {len(self._targets)}): {grp}")
                         self._groups.append(_sos_group(grp, parent=self))
                     else:
                         index = []
                         for x in sos_targets(grp):
                             try:
                                 if _target_index is None:
-                                    _target_index = {v:k for k,v in enumerate(self._targets)}
+                                    _target_index = {v: k for k, v in enumerate(self._targets)}
                                 index.append(_target_index[x])
                             except Exception as e:
-                                raise ValueError(
-                                    f"Returned target is not one of the targets. {x}"
-                                ) from e
+                                raise ValueError(f"Returned target is not one of the targets. {x}") from e
                         self._groups.append(_sos_group(index, parent=self))
             except Exception as e:
-                raise ValueError(
-                    f"Failed to apply customized grouping method: {e}") from e
+                raise ValueError(f"Failed to apply customized grouping method: {e}") from e
         else:
             raise ValueError(f"Unsupported by option ``{by}``!")
         return self
@@ -1820,16 +1483,14 @@ def _handle_paired_with(self, paired_with):
             try:
                 var_name = ["_" + x for x in paired_with]
             except Exception as e:
-                raise ValueError(
-                    f"Invalid value for option paired_with {paired_with}") from e
+                raise ValueError(f"Invalid value for option paired_with {paired_with}") from e
             var_value = []
             for vn in var_name:
                 if vn[1:] not in env.sos_dict:
                     raise ValueError(f"Variable {vn[1:]} does not exist.")
                 var_value.append(env.sos_dict[vn[1:]])
         else:
-            raise ValueError(
-                f"Unacceptable value for parameter paired_with: {paired_with}")
+            raise ValueError(f"Unacceptable value for parameter paired_with: {paired_with}")
         #
         for vn, vv in zip(var_name, var_value):
             # set paired with values to step_input
@@ -1855,16 +1516,14 @@ def _handle_group_with(self, group_with):
             try:
                 var_name = ["_" + x for x in group_with]
             except Exception as e:
-                raise ValueError(
-                    f"Invalud value for option group_with {group_with}") from e
+                raise ValueError(f"Invalud value for option group_with {group_with}") from e
             var_value = []
             for vn in var_name:
                 if vn[1:] not in env.sos_dict:
                     raise ValueError(f"Variable {vn[1:]} does not exist.")
                 var_value.append(env.sos_dict[vn[1:]])
         else:
-            raise ValueError(
-                f"Unacceptable value for parameter group_with: {group_with}")
+            raise ValueError(f"Unacceptable value for parameter group_with: {group_with}")
         #
         for vn, vv in zip(var_name, var_value):
             self.group_with(vn, vv)
@@ -1878,8 +1537,7 @@ def _handle_extract_pattern(self, pattern):
         elif isinstance(pattern, Iterable):
             patterns = pattern
         else:
-            raise ValueError(
-                f"Unacceptable value for parameter pattern: {pattern}")
+            raise ValueError(f"Unacceptable value for parameter pattern: {pattern}")
         #
         for pattern in patterns:
             res = extract_pattern(pattern, self._targets)
@@ -1897,20 +1555,14 @@ def _handle_for_each(self, for_each):
             if all(isinstance(x, dict) for x in for_each):
                 keys = [tuple(sorted(x.keys())) for x in for_each]
                 # the keys should be all the same, or all different.
-                if len(set(keys)) == 1 and all(
-                        isinstance(x, str) for x in keys[0]):
+                if len(set(keys)) == 1 and all(isinstance(x, str) for x in keys[0]):
                     # this is a special case for specified contexts. #1403
-                    for_each = [{
-                        ",".join(keys[0]):
-                            [[x[key] for key in keys[0]] for x in for_each]
-                    }]
+                    for_each = [{",".join(keys[0]): [[x[key] for key in keys[0]] for x in for_each]}]
                 elif len(set(keys)) != len(keys):
                     raise ValueError(
-                        "List of dictionaries for parameter for_each should have all different, or all the same keys."
-                    )
+                        "List of dictionaries for parameter for_each should have all different, or all the same keys.")
         else:
-            raise ValueError(
-                f"Unacceptable value for parameter for_each: {for_each}")
+            raise ValueError(f"Unacceptable value for parameter for_each: {for_each}")
         #
         for fe_all in for_each:
             if isinstance(fe_all, dict):
@@ -1924,8 +1576,7 @@ def _handle_for_each(self, for_each):
                             v = list(v)
                         if any(len(_v) != len(names) for _v in v):
                             raise ValueError(
-                                f"Unable to unpack object {short_repr(v)} for variables {k} (of length {len(names)})"
-                            )
+                                f"Unable to unpack object {short_repr(v)} for variables {k} (of length {len(names)})")
                         fe_iter_names.extend(names)
                         fe_values.extend(list(zip(*v)))
                     else:
@@ -1948,9 +1599,7 @@ def _handle_for_each(self, for_each):
                     if name.split(".")[0] not in env.sos_dict:
                         raise ValueError(f"Variable {name} does not exist.")
                     if "." in name:
-                        fe_values.append(
-                            getattr(env.sos_dict[name.split(".")[0]],
-                                    name.split(".", 1)[-1]))
+                        fe_values.append(getattr(env.sos_dict[name.split(".")[0]], name.split(".", 1)[-1]))
                     else:
                         fe_values.append(env.sos_dict[name])
 
@@ -1961,14 +1610,10 @@ def _handle_for_each(self, for_each):
                     try:
                         import pandas as pd
 
-                        if not isinstance(values,
-                                          (pd.DataFrame, pd.Series, pd.Index)):
-                            raise ValueError(
-                                f"Unacceptable for_each data type {values.__class__.__name__}"
-                            )
+                        if not isinstance(values, (pd.DataFrame, pd.Series, pd.Index)):
+                            raise ValueError(f"Unacceptable for_each data type {values.__class__.__name__}")
                     except Exception as e:
-                        raise ValueError(
-                            f"Cannot iterate through variable {name}: {e}") from e
+                        raise ValueError(f"Cannot iterate through variable {name}: {e}") from e
                 if loop_size is None:
                     loop_size = len(values)
                 elif loop_size != len(values):
@@ -1986,21 +1631,15 @@ def _handle_for_each(self, for_each):
                 for idx in range(n_grps):
                     for var_name, values in zip(fe_iter_names, fe_values):
                         if isinstance(values, Sequence):
-                            self._groups[n_grps * vidx + idx].set(
-                                var_name, values[vidx])
+                            self._groups[n_grps * vidx + idx].set(var_name, values[vidx])
                         elif isinstance(values, pd.DataFrame):
-                            self._groups[n_grps * vidx + idx].set(
-                                var_name, values.iloc[vidx])
+                            self._groups[n_grps * vidx + idx].set(var_name, values.iloc[vidx])
                         elif isinstance(values, pd.Series):
-                            self._groups[n_grps * vidx + idx].set(
-                                var_name, values.iloc[vidx])
+                            self._groups[n_grps * vidx + idx].set(var_name, values.iloc[vidx])
                         elif isinstance(values, pd.Index):
-                            self._groups[n_grps * vidx + idx].set(
-                                var_name, values[vidx])
+                            self._groups[n_grps * vidx + idx].set(var_name, values[vidx])
                         else:
-                            raise ValueError(
-                                f"Failed to iterate through for_each variable {short_repr(values)}"
-                            )
+                            raise ValueError(f"Failed to iterate through for_each variable {short_repr(values)}")
 
     def __hash__(self):
         return hash(repr(self))
@@ -2008,9 +1647,7 @@ def __hash__(self):
     def __eq__(self, other):
         try:
             # allow compare to any object as long as it can be converted to sos_targets
-            return self._targets == (
-                other._targets if isinstance(other, sos_targets) else
-                sos_targets(other)._targets)
+            return self._targets == (other._targets if isinstance(other, sos_targets) else sos_targets(other)._targets)
         except Exception:
             return False
 
@@ -2019,31 +1656,24 @@ def __add__(self, part):
             return self._targets[0].__add__(part)
         if len(self._targets) == 0:
             raise ValueError(f"Cannot add {part} to empty target list")
-        raise ValueError(
-            f"Cannot add {part} to group of {len(self)} targets {self!r}")
+        raise ValueError(f"Cannot add {part} to group of {len(self)} targets {self!r}")
 
     def __fspath__(self):
         if len(self._targets) == 1:
             return self._targets[0].__fspath__()
         if len(self._targets) == 0:
-            raise ValueError(
-                "Cannot treat an empty sos_targets as single target")
-        raise ValueError(
-            f"Cannot treat an sos_targets object {self} with more than one targets as a single target"
-        )
+            raise ValueError("Cannot treat an empty sos_targets as single target")
+        raise ValueError(f"Cannot treat an sos_targets object {self} with more than one targets as a single target")
 
     def __repr__(self):
-        return (("[" + ", ".join(repr(x) for x in self._targets) +
-                 "]") if self.valid() else
+        return (("[" + ", ".join(repr(x) for x in self._targets) + "]") if self.valid() else
                 ("Unspecified" if self.unspecified() else self._undetermined))
 
     def __short_repr__(self):
-        grp_info = "" if self._num_groups(
-        ) <= 1 else f" in {self._num_groups()} groups"
+        grp_info = "" if self._num_groups() <= 1 else f" in {self._num_groups()} groups"
         if self.valid():
             if len(self._targets) <= 2:
-                return " ".join([x.target_name() for x in self._targets
-                                ]) + grp_info
+                return " ".join([x.target_name() for x in self._targets]) + grp_info
             return (" ".join([x.target_name() for x in self._targets[:2]]) +
                     f"... ({len(self._targets)} items{grp_info})")
         return "Unspecified" if self.unspecified() else self._undetermined
@@ -2052,8 +1682,7 @@ def __stable_repr__(self):
         return repr(self)
 
     def __str__(self):
-        return (self.__format__("") if self.valid() else
-                ("Unspecified" if self.unspecified() else self._undetermined))
+        return (self.__format__("") if self.valid() else ("Unspecified" if self.unspecified() else self._undetermined))
 
     def __format__(self, format_spec):
         if not self.valid():
@@ -2094,11 +1723,9 @@ def __init__(
         if not sdict:
             sdict = env.sos_dict
         if not input_files.valid():
-            raise RuntimeError(
-                "Input files of step signature cannot be undetermined.")
+            raise RuntimeError("Input files of step signature cannot be undetermined.")
         if not dependent_files.valid():
-            raise RuntimeError(
-                "Dependent files of step signature cannot be undetermined.")
+            raise RuntimeError("Dependent files of step signature cannot be undetermined.")
 
         self.input_files = input_files.remove_targets(type=sos_step)
         self.dependent_files = dependent_files.remove_targets(type=sos_step)
@@ -2108,8 +1735,9 @@ def __init__(
 
         # signatures that exist before execution and might change during execution
         self.init_signature = {
-            x: deepcopy(sdict[x]) for x in sorted(signature_vars) if
-            x in sdict and not callable(sdict[x]) and pickleable(sdict[x], x)
+            x: deepcopy(sdict[x])
+            for x in sorted(signature_vars)
+            if x in sdict and not callable(sdict[x]) and pickleable(sdict[x], x)
         }
 
     def identify_local_args(self):
@@ -2141,35 +1769,27 @@ def write(self):
             return self.content
         if self.output_files.undetermined():
             self.output_files = env.sos_dict["_output"]
-            env.log_to_file(
-                "TARGET",
-                f'Set undetermined output files to {env.sos_dict["_output"]}')
+            env.log_to_file("TARGET", f'Set undetermined output files to {env.sos_dict["_output"]}')
         input_sig = {}
         for f in self.input_files:
             try:
                 input_sig[str(f)] = f.target_signature()
             except Exception:
-                env.logger.debug(
-                    f"Failed to create signature: input target {f} does not exist"
-                )
+                env.logger.debug(f"Failed to create signature: input target {f} does not exist")
                 return False
         output_sig = {}
         for f in self.output_files:
             try:
                 output_sig[str(f)] = f.target_signature()
             except Exception:
-                env.logger.debug(
-                    f"Failed to create signature: output target {f} does not exist"
-                )
+                env.logger.debug(f"Failed to create signature: output target {f} does not exist")
                 return False
         dependent_sig = {}
         for f in self.dependent_files:
             try:
                 dependent_sig[str(f)] = f.target_signature()
             except Exception:
-                env.logger.debug(
-                    f"Failed to create signature: dependent target {f} does not exist"
-                )
+                env.logger.debug(f"Failed to create signature: dependent target {f} does not exist")
                 return False
         init_context_sig = {
             var: objectMD5(self.init_signature[var])
@@ -2202,9 +1822,7 @@ def validate(self, signature):
         """Check if ofiles and ifiles match signatures recorded in md5file"""
         if not signature:
             return "Empty signature"
-        sig_files = (
-            self.input_files._targets + self.output_files._targets +
-            self.dependent_files._targets)
+        sig_files = self.input_files._targets + self.output_files._targets + self.dependent_files._targets
         for x in sig_files:
             if not x.target_exists("any"):
                 return f"Missing target {x}"
@@ -2222,9 +1840,7 @@ def validate(self, signature):
                     if env.sos_dict[key] != value:
                         return f"Context variable {key} value mismatch: {short_repr(value)} saved, {short_repr(env.sos_dict[key])} current"
                 except Exception as e:
-                    env.logger.debug(
-                        f"Variable {key} of type {type(value).__name__} cannot be compared: {e}"
-                    )
+                    env.logger.debug(f"Variable {key} of type {type(value).__name__} cannot be compared: {e}")
         elif "init_context_sig" in signature:
             for key, value in signature["init_context_sig"].items():
                 if key not in env.sos_dict:
@@ -2252,23 +1868,19 @@ def validate(self, signature):
                             target_class = eval(target_type)
                         else:
                             # check registry
-                            for entrypoint in pkg_resources.iter_entry_points(
-                                    group="sos_targets"):
+                            for entrypoint in pkg_resources.iter_entry_points(group="sos_targets"):
                                 if entrypoint.name.strip() == target_type:
                                     target_class = entrypoint.load()
                                     break
                         if target_class is None:
-                            raise ValueError(
-                                f"Failed to identify target class {target_type}"
-                            )
+                            raise ValueError(f"Failed to identify target class {target_type}")
                         # parameter of class?
                         freal = eval(f, {target_type: target_class})
                     else:
                         freal = file_target(f)
                     if not freal.validate(m):
                         return f"Target {f} does not exist or does not match saved signature {m}"
-                    res[cur_type].append(freal.target_name(
-                    ) if isinstance(freal, file_target) else freal)
+                    res[cur_type].append(freal.target_name() if isinstance(freal, file_target) else freal)
                     files_checked[freal.target_name()] = True
                 except Exception as e:
                     env.logger.debug(f"Wrong md5 in signature: {e}")
@@ -2349,8 +1961,7 @@ def lock(self):
             return
         # we will need to lock on a file that we do not really write to
         # otherwise the lock will be broken when we write to it.
-        self._lock = fasteners.InterProcessLock(
-            os.path.join(env.temp_dir, self.sig_id + ".lock"))
+        self._lock = fasteners.InterProcessLock(os.path.join(env.temp_dir, self.sig_id + ".lock"))
         if not self._lock.acquire(blocking=False):
             self._lock = None
             raise UnavailableLock((
@@ -2367,9 +1978,7 @@ def release(self, quiet=False):
         if not self.sig_id:
             return
         if not hasattr(self, "_lock") or self._lock is None:
-            env.logger.warning(
-                f"Releasing an non-existent or released lock for {self.sig_id}."
-            )
+            env.logger.warning(f"Releasing an non-existent or released lock for {self.sig_id}.")
             return
         if self._lock:
             try:
@@ -2380,9 +1989,7 @@ def release(self, quiet=False):
                 )
             except Exception as e:
                 if not quiet:
-                    env.logger.warning(
-                        f"Unable to release lock for output files {self.output_files}: {e}"
-                    )
+                    env.logger.warning(f"Unable to release lock for output files {self.output_files}: {e}")
             finally:
                 self._lock = None
 
@@ -2390,8 +1997,7 @@ def set_output(self, files: sos_targets):
         if not self.sig_id:
             return
         # add signature file if input and output files are dynamic
-        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config[
-                "SOS_DEBUG"]:
+        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config["SOS_DEBUG"]:
             env.log_to_file("TARGET", f"Set output of signature to {files}")
         self.output_files = files
 
@@ -2403,11 +2009,8 @@ def write(self):
         if not self.sig_id:
             return False
         if not self.output_files.valid():
-            raise ValueError(
-                f"Cannot write signature with undetermined output {self.output_files}"
-            )
-        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config[
-                "SOS_DEBUG"]:
+            raise ValueError(f"Cannot write signature with undetermined output {self.output_files}")
+        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config["SOS_DEBUG"]:
             env.log_to_file(
                 "TARGET",
                 f"write signature {self.sig_id} with output {self.output_files}",
@@ -2422,21 +2025,9 @@ def write(self):
             "tracked_files",
             self.sig_id,
             repr({
-                "input_files": [
-                    str(f.resolve())
-                    for f in self.input_files
-                    if isinstance(f, file_target)
-                ],
-                "dependent_files": [
-                    str(f.resolve())
-                    for f in self.dependent_files
-                    if isinstance(f, file_target)
-                ],
-                "output_files": [
-                    str(f.resolve())
-                    for f in self.output_files
-                    if isinstance(f, file_target)
-                ],
+                "input_files": [str(f.resolve()) for f in self.input_files if isinstance(f, file_target)],
+                "dependent_files": [str(f.resolve()) for f in self.dependent_files if isinstance(f, file_target)],
+                "output_files": [str(f.resolve()) for f in self.output_files if isinstance(f, file_target)],
             }),
         ])
         return True
@@ -2445,14 +2036,11 @@ def validate(self):
         """Check if ofiles and ifiles match signatures recorded in md5file"""
         if not self.sig_id:
             return "no signature for steps with nested workflow"
-        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config[
-                "SOS_DEBUG"]:
+        if "TARGET" in env.config["SOS_DEBUG"] or "ALL" in env.config["SOS_DEBUG"]:
             env.log_to_file("TARGET", f"Validating {self.sig_id}")
         #
         # file not exist?
-        sig_files = (
-            self.input_files._targets + self.output_files._targets +
-            self.dependent_files._targets)
+        sig_files = self.input_files._targets + self.output_files._targets + self.dependent_files._targets
         for x in sig_files:
             if not x.target_exists("any"):
                 return f"Missing target {x}"
diff --git a/src/sos/task_executor.py b/src/sos/task_executor.py
index 9ad304293..ffcaf5205 100644
--- a/src/sos/task_executor.py
+++ b/src/sos/task_executor.py
@@ -156,8 +156,7 @@ def execute_single_task(self, task_id, params, runtime, sig_content, quiet=False
                 "_depends",
         ]:
             if key in sos_dict and isinstance(sos_dict[key], sos_targets):
-                # resolve remote() target
-                env.sos_dict.set(key, sos_dict[key].remove_targets(type=sos_step).resolve_remote())
+                env.sos_dict.set(key, sos_dict[key].remove_targets(type=sos_step))
 
         # when no output is specified, we just treat the task as having no output (determined)
         env.sos_dict["_output"]._undetermined = False
diff --git a/src/sos/templates/parts/hover_doc.tpl b/src/sos/templates/parts/hover_doc.tpl
index 091d97138..8e7ab68e7 100644
--- a/src/sos/templates/parts/hover_doc.tpl
+++ b/src/sos/templates/parts/hover_doc.tpl
@@ -26,8 +26,8 @@ let sos_keywords = new Set([
   'sos_run', 'fail_if', 'done_if', 'warn_if', 'skip_if',
   'get_output', 'expand_pattern',
   // task options
-  'walltime', 'cores', 'mem', 'queue', 'to_host', 'nodes',
-  'from_host', 'trunk_size', 'trunk_workers', 'workdir',
+  'walltime', 'cores', 'mem', 'queue', 'nodes',
+  'trunk_size', 'trunk_workers', 'workdir',
   'shared', 'env', 'prepend_path', 'tags',
   // targets
   'file_target', 'executable', 'sos_variable', 'env_variable',
diff --git a/src/sos/workflow_engines.py b/src/sos/workflow_engines.py
index 1f5e027b6..270e453e1 100644
--- a/src/sos/workflow_engines.py
+++ b/src/sos/workflow_engines.py
@@ -88,18 +88,8 @@ def execute_workflow(self, filename, command, **template_args):
             # copy the files over
             self.agent.send_job_file(f"~/.sos/config_{self.alias}.yml", dir=".")
 
-        self.local_filename = filename
-
         self.job_name = workflow.calc_md5(workflow_args)
-
-        if os.path.isfile(filename):
-            ret = self.agent.send_to_host([filename])
-        elif os.path.isfile(filename + ".sos"):
-            ret = self.agent.send_to_host([filename + ".sos"])
-        elif os.path.isfile(filename + ".ipynb"):
-            ret = self.agent.send_to_host([filename + ".ipynb"])
-
-        self.filename = list(ret.values())[0]
+        self.filename = filename
         self.command = self.remove_arg(command, "-r")
         # -c only point to local config file.
         self.command = self.remove_arg(self.command, "-c")


Test patch:
diff --git a/test/build_test_docker.sh b/test/build_test_docker.sh
index f4452c6e6..9282a140f 100644
--- a/test/build_test_docker.sh
+++ b/test/build_test_docker.sh
@@ -42,8 +42,7 @@ ADD authorized_keys /root/.ssh/authorized_keys
 
 ARG  SHA=LATEST
 RUN  SHA=$SHA git clone http://github.com/vatlab/sos sos
-RUN  cd sos && pip install . -U
-RUN  pip install sos-pbs
+RUN  pip install sos sos-pbs
 
 RUN  echo "export TS_SLOTS=10" >> /root/.bash_profile
 
@@ -62,7 +61,7 @@ docker build --build-arg SHA=$SHA -t eg_sshd .
 
 #
 # start docker image
-docker run -d -P --env TS_SLOTS=10 --name test_sos eg_sshd
+docker run -d -v `pwd`:`pwd` -P --env TS_SLOTS=10 --name test_sos eg_sshd
 
 # get the port
 PORT22=$(docker port test_sos 22 | cut -f2 -d:)
diff --git a/test/conftest.py b/test/conftest.py
index 7128eb66d..4868c8d4f 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -35,15 +35,15 @@ def get_config(text_or_dict):
 
 @pytest.fixture
 def script_factory():
-    filename = tempfile.NamedTemporaryFile(suffix='.sos', delete=False).name
+    default_filename = tempfile.NamedTemporaryFile(suffix='.sos', delete=False).name
 
-    def get_script(text):
-        with open(filename, 'w') as conf:
+    def get_script(text, filename=None):
+        with open(filename or default_filename, 'w') as conf:
             conf.write(textwrap.dedent(text))
-        return filename
+        return filename or default_filename
 
     yield get_script
-    os.remove(filename)
+    os.remove(default_filename)
 
 
 @pytest.fixture
diff --git a/test/run_tests.py b/test/run_tests.py
index 087407524..923b54eb5 100755
--- a/test/run_tests.py
+++ b/test/run_tests.py
@@ -66,6 +66,7 @@ def test_failed(test_names, return_code):
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser('run_tests')
+    parser.add_argument('tests', nargs='*', help='Tests that will be executed instead of gathering all tests.')
     parser.add_argument('-b', '--batch', default=5, type=int, help='Group tests')
     parser.add_argument(
         '-l',
@@ -78,33 +79,36 @@ def test_failed(test_names, return_code):
     parser.add_argument('-x', '--exitfirst', help='Stop when one test fails')
     args = parser.parse_args()
 
-    print('Collecting tests')
-    all_tests = get_testcases()
-    print(f'{len(all_tests)} tests are collected.')
+    if args.tests:
+        all_tests = args.tests
+    else:
+        print('Collecting tests')
+        all_tests = get_testcases()
+        print(f'{len(all_tests)} tests are collected.')
 
-    if args.lastfailed is not None:
-        if not os.path.isfile(LOGFILE):
-            sys.exit(f'Log file {LOGFILE} does not exists.')
-        test_results = {}
-        with open(LOGFILE) as fl:
-            for line in fl:
-                if not line.strip():
-                    continue
-                try:
-                    fields = line.split()
-                    if len(fields) >= 2:
-                        tst = fields[-2]
-                        res = fields[-1].strip()
-                    else:
-                        tst = fields[-1]
-                        res = 'FAILED'
-                except Exception:
-                    print(f'Invalid log line: {line}')
-                test_results[tst] = res.strip()
-        all_tests = [x for x, y in test_results.items() if y == 'FAILED' and x in all_tests]
-        # if args.lastfailed != 0:
-        #     all_tests = all_tests[-args.lastfailed:]
-        print(f'Running {len(all_tests)} failed tests.')
+        if args.lastfailed is not None:
+            if not os.path.isfile(LOGFILE):
+                sys.exit(f'Log file {LOGFILE} does not exists.')
+            test_results = {}
+            with open(LOGFILE) as fl:
+                for line in fl:
+                    if not line.strip():
+                        continue
+                    try:
+                        fields = line.split()
+                        if len(fields) >= 2:
+                            tst = fields[-2]
+                            res = fields[-1].strip()
+                        else:
+                            tst = fields[-1]
+                            res = 'FAILED'
+                    except Exception:
+                        print(f'Invalid log line: {line}')
+                    test_results[tst] = res.strip()
+            all_tests = [x for x, y in test_results.items() if (y == 'FAILED' and x in all_tests) or x not in test_results]
+            # if args.lastfailed != 0:
+            #     all_tests = all_tests[-args.lastfailed:]
+            print(f'Running {len(all_tests)} failed tests.')
 
     failed_tests = []
     nbatch = len(all_tests) // args.batch + 1
@@ -121,7 +125,7 @@ def test_failed(test_names, return_code):
         failed_tests = retried_failed_tests
 
     if failed_tests:
-        print(f'Failed tests (logged to {LOGFILE}):\n' + '\n'.join(failed_tests))
+        print(f'\n\n{len(failed_tests)} failed tests (logged to {LOGFILE}):\n' + '\n'.join(failed_tests))
     else:
         print(f'All {len(all_tests)} tests complete successfully.')
     sys.exit(0 if not failed_tests else 1)
diff --git a/test/test_convert.py b/test/test_convert.py
index 28a2dd076..24b275119 100644
--- a/test/test_convert.py
+++ b/test/test_convert.py
@@ -10,7 +10,7 @@
 from sos.converter import extract_workflow
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_script_to_html(temp_factory, clear_now_and_after):
     '''Test sos show script --html'''
     clear_now_and_after('temp1.sos.html', 'temp2.sos.html')
diff --git a/test/test_dag.py b/test/test_dag.py
index 90264640e..321cb7244 100644
--- a/test/test_dag.py
+++ b/test/test_dag.py
@@ -31,9 +31,9 @@ def sorted_dot(dot):
         return sorted([x.strip() for x in dot.split('\n') if x.strip() and not 'digraph' in x])
 
     if isinstance(content, str):
-        assert sorted_dot(dot) == sorted_dot(content)
+        assert sorted_dot(dot) == sorted_dot(content), f'expect {content}, got {dot}'
     else:
-        assert sorted_dot(dot) in [sorted_dot(x) for x in content]
+        assert sorted_dot(dot) in [sorted_dot(x) for x in content], f'expect {content}, got {dot}'
 
 
 def get_initial_dag(test):
@@ -44,7 +44,7 @@ def get_initial_dag(test):
     return dag
 
 
-def test_simple_dag(temp_factory):
+def test_simple_dag1(temp_factory):
     '''Test DAG with simple dependency'''
     temp_factory('a.txt', 'a1.txt')
     # basica case
@@ -69,6 +69,11 @@ def test_simple_dag(temp_factory):
             A_3 -> A_4;
             }
             '''))
+
+
+def test_simple_dag2(temp_factory):
+    '''Test DAG with simple dependency'''
+    temp_factory('a.txt', 'a1.txt')
     # basica case
     # 1 -> 2 -> 3 -> 4
     assertDAG(
@@ -90,6 +95,11 @@ def test_simple_dag(temp_factory):
             A_3 -> A_4;
             }
             '''))
+
+
+def test_simple_dag3(temp_factory):
+    '''Test DAG with simple dependency'''
+    temp_factory('a.txt')
     #
     # 1 -> 2 -> 3 -> 4
     #
@@ -121,7 +131,11 @@ def test_simple_dag(temp_factory):
             A_3 -> A_4;
             }
             '''))
-    #
+
+
+def test_simple_dag4(temp_factory):
+    '''Test DAG with simple dependency'''
+    temp_factory('a.txt', 'a1.txt')
     # 1 -> 2
     # 3 -> 4 (3 does not have any input)
     #
@@ -152,7 +166,11 @@ def test_simple_dag(temp_factory):
             B_3 -> B_4;
             }
             '''))
-    #
+
+
+def test_simple_dag5(temp_factory):
+    '''Test DAG with simple dependency'''
+    temp_factory('a.txt', 'a1.txt')
     # 1 -> 2
     # 3 -> 4 (3 depends on something else)
     #
@@ -183,7 +201,11 @@ def test_simple_dag(temp_factory):
             B_3 -> B_4;
             }
             '''))
-    #
+
+
+def test_simple_dag6(temp_factory):
+    '''Test DAG with simple dependency'''
+    temp_factory('a.txt', 'a1.txt')
     # (1) -> 2
     # (1) -> 3 -> 4
     #
@@ -806,7 +828,7 @@ def test_reverse_shared_variable(clear_now_and_after):
     assert env.sos_dict['b'] == 1
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_chained_depends(temp_factory):
     '''Test chain dependent'''
     temp_factory('a.bam', 'a.bam.bai', 'a.vcf')
@@ -829,7 +851,6 @@ def test_chained_depends(temp_factory):
         targets=['a.vcf'])
 
 
-@pytest.mark.skipif(True, reason='This test is failing')
 def test_output_of_dag(clear_now_and_after):
     '''Test output of dag'''
     #
diff --git a/test/test_docker_actions.py b/test/test_docker_actions.py
index a00c3d530..36d2dda0b 100644
--- a/test/test_docker_actions.py
+++ b/test/test_docker_actions.py
@@ -90,16 +90,6 @@ def test_bash_in_docker():
         ''')
 
 
-#    @pytest.mark.skipif(not has_docker or sys.platform != 'win32', reason='Skip test because docker is not installed.')
-#    def testBatchScriptInDocker():
-#        '''Test action powershell in docker environment'''
-#        script = execute_workflow(r'''
-#           [0]
-#           run:  container='docker://microsoft/windowsservercore'
-#           dir c:\
-#           ''')
-
-
 @pytest.mark.skipif(
     not has_docker or sys.platform == 'win32',
     reason='Skip test because docker is not installed.')
diff --git a/test/test_execute.py b/test/test_execute.py
index 1c7c04b54..783890ed9 100644
--- a/test/test_execute.py
+++ b/test/test_execute.py
@@ -1775,7 +1775,6 @@ def test_step_id_vars():
         """)
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_reexecution_of_dynamic_depends(clear_now_and_after):
     """Testing the rerun of steps to verify dependency"""
     clear_now_and_after("a.bam", "a.bam.bai")
@@ -1804,7 +1803,6 @@ def test_reexecution_of_dynamic_depends(clear_now_and_after):
     assert res["__completed__"]["__step_skipped__"] == 1
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_traced_function(clear_now_and_after):
     clear_now_and_after("a.bam", "a.bam.bai")
     script = """
@@ -2314,7 +2312,6 @@ def test_remove_empty_groups_empty_named(clear_now_and_after):
         """)
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_multi_depends(clear_now_and_after, temp_factory):
     """Test a step with multiple depdendend steps"""
 
@@ -2378,10 +2375,7 @@ def test_execute_ipynb(sample_workflow):
     Base_Executor(wf).run()
 
 
-@pytest.mark.skipif(
-    True,
-    reason="Skip test because travis fails on this test for unknown reason, also due to a bug in psutil under windows",
-)
+@pytest.mark.skip(reason="Skip test because travis fails on this test for unknown reason, also due to a bug in psutil under windows")
 def test_kill_worker(script_factory):
     """Test if the workflow can error out after a worker is killed"""
     import time
@@ -2402,17 +2396,7 @@ def test_kill_worker(script_factory):
     while True:
         children = proc.children(recursive=True)
         if children:
-            children[0].terminate()
-            break
-        time.sleep(0.1)
-    ret.wait()
-
-    ret = subprocess.Popen(["sos", "run", script_file])
-    proc = psutil.Process(ret.pid)
-    while True:
-        children = proc.children(recursive=True)
-        if children:
-            children[0].kill()
+            children[-1].terminate()
             break
         time.sleep(0.1)
     ret.wait()
@@ -2457,7 +2441,7 @@ def test_kill_substep_worker(script_factory):
     ret.wait()
 
 
-@pytest.mark.skipif(True, reason="This test needs to be improved to make it consistent")
+@pytest.mark.skip(reason="This test needs to be improved to make it consistent")
 def test_kill_task(script_factory):
     """Test if the workflow can error out after a worker is killed"""
     subprocess.call(["sos", "purge", "--all"])
@@ -2501,7 +2485,7 @@ def test_kill_task(script_factory):
     assert ret.returncode != 0
 
 
-@pytest.mark.skipif(True, reason="This test needs to be improved to make it consistent")
+@pytest.mark.skip(reason="This test needs to be improved to make it consistent")
 def test_restart_orphaned_tasks(script_factory):
     """Test restarting orphaned tasks which displays as running at first."""
     import time
@@ -2596,7 +2580,6 @@ def test_concurrent_running_tasks(script_factory):
     assert ret2.returncode == 0
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_reexecute_task_with_missing_output(clear_now_and_after):
     '''Issue #1493'''
     clear_now_and_after([f'a_{i}.txt' for i in range(10)])
diff --git a/test/test_execute_2.py b/test/test_execute_2.py
index 65748a073..651f1e8b0 100644
--- a/test/test_execute_2.py
+++ b/test/test_execute_2.py
@@ -58,7 +58,7 @@ def test_removal_of_output_from_failed_step(clear_now_and_after):
     assert not os.path.isfile("result.csv")
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_error_handling_of_substeps(clear_now_and_after):
     clear_now_and_after(
         [f"test_{i}.txt" for i in range(10)],
@@ -194,7 +194,7 @@ def test_for_each_as_target_property_nested_list(temp_factory):
     ]
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_rerun_with_zap(clear_now_and_after):
     clear_now_and_after([f"zapped_example_{i}.txt.zapped" for i in range(3)])
     clear_now_and_after([f"zapped_example_{i}.bak" for i in range(3)])
diff --git a/test/test_parser.py b/test/test_parser.py
index 8a1d56f44..9cc284829 100644
--- a/test/test_parser.py
+++ b/test/test_parser.py
@@ -1470,10 +1470,10 @@ def test_cell():
 """)
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_overwrite_keyword(clear_now_and_after):
     """Test overwrite sos keyword with user defined one."""
-    clear_now_and_after("a.txt")
+    clear_now_and_after("a.txt", "b.txt")
     #
     execute_workflow("""
 def run(script):
@@ -1483,15 +1483,16 @@ def run(script):
 run:
     touch a.txt
 """)
-    assert not os.path.isfile("a.txt")
+    assert os.path.isfile("a.txt")
     #
     execute_workflow("""
 parameter: run = 5
 
 [1]
 run:
-    touch a.txt
+    touch b.txt
 """)
+    assert os.path.isfile("b.txt")
 
 
 def test_help_message(sample_workflow):
diff --git a/test/test_python3_target.py b/test/test_python3_target.py
index ade3786fa..3ee23a47f 100644
--- a/test/test_python3_target.py
+++ b/test/test_python3_target.py
@@ -61,41 +61,41 @@ def test_py_module_with_version():
 ''')
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_upgrade_py_module():
     '''Test upgrade py module #1246'''
-    # first install tabulate == 0.7.5
+    # first install tabulate == 0.8.5
     execute_workflow(r'''
 [10]
-depends: Py_Module('tabulate==0.7.5', autoinstall=True)
+depends: Py_Module('tabulate==0.8.9', autoinstall=True)
 ''')
     # test should pass
     execute_workflow(r'''
 [10]
-depends: Py_Module('tabulate'), Py_Module('tabulate==0.7.5')
+depends: Py_Module('tabulate'), Py_Module('tabulate==0.8.9')
 ''')
     # test for newer version should fail
 
     with pytest.raises(Exception):
         execute_workflow(r'''
 [10]
-depends: Py_Module('tabulate==0.8.3')
+depends: Py_Module('tabulate==0.9.0')
 ''')
 
     # auto install should work
     execute_workflow(r'''
 [10]
-depends: Py_Module('tabulate==0.8.3', autoinstall=True)
+depends: Py_Module('tabulate==0.9.0', autoinstall=True)
 ''')
 
     # test for old version should fail
     execute_workflow(r'''
         [10]
-        depends: Py_Module('tabulate'), Py_Module('tabulate==0.8.3')
+        depends: Py_Module('tabulate'), Py_Module('tabulate==0.9.0')
         ''')
     # test for old version should fail
     with pytest.raises(Exception):
         execute_workflow(r'''
         [10]
-        depends: Py_Module('tabulate==0.7.5')
+        depends: Py_Module('tabulate==0.8.9')
         ''')
diff --git a/test/test_r_targets.py b/test/test_r_targets.py
index 990b245dc..bdb8ba781 100644
--- a/test/test_r_targets.py
+++ b/test/test_r_targets.py
@@ -13,9 +13,7 @@
 
 
 @pytest.mark.xfail(reason="Sometimes there is no permission to install to R")
-@pytest.mark.skipif(
-    not shutil.which("Rscript") or "TRAVIS" in os.environ,
-    reason="R not installed")
+@pytest.mark.skipif(not shutil.which("Rscript") or "TRAVIS" in os.environ, reason="R not installed")
 def test_r_library():
     """Test target R_Library"""
     execute_workflow("""
@@ -26,7 +24,7 @@ def test_r_library():
         """)
 
 
-@pytest.mark.skip(reason="temporary skip")
+@pytest.mark.xfail(reason="Sometimes there is no permission to install to R")
 @pytest.mark.skipif(not shutil.which("Rscript"), reason="R not installed")
 def test_depends_r_library():
     """Testing depending on R_library"""
diff --git a/test/test_remote.py b/test/test_remote.py
index fd4e95c69..b5adc33ec 100644
--- a/test/test_remote.py
+++ b/test/test_remote.py
@@ -23,126 +23,6 @@
         has_docker = False
 
 
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_to_host_option(clear_now_and_after):
-    """Test from_remote option"""
-    clear_now_and_after("to_host_testfile.txt", "to_host_linecount.txt")
-    with open("to_host_testfile.txt", "w") as tf:
-        for i in range(100):
-            tf.write(f"line {i+1}\n")
-    execute_workflow(
-        """
-        [10]
-        output: 'to_host_linecount.txt'
-        task: to_host='to_host_testfile.txt'
-        sh:
-            wc -l to_host_testfile.txt > to_host_linecount.txt
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-            "sig_mode": "force",
-        },
-    )
-    assert os.path.isfile("to_host_linecount.txt")
-    with open("to_host_linecount.txt") as lc:
-        assert lc.read().strip().startswith("100")
-
-
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_to_host_option_with_named_path(clear_now_and_after):
-    """Test from_remote option"""
-    clear_now_and_after(
-        os.path.expanduser("~/to_host_named_testfile.txt"),
-        "to_host_named_linecount.txt",
-    )
-    with open(os.path.expanduser("~/to_host_named_testfile.txt"), "w") as tf:
-        for i in range(200):
-            tf.write(f"line {i+1}\n")
-    execute_workflow(
-        """
-        [10]
-        output: 'to_host_named_linecount.txt'
-        task: to_host='#home/to_host_named_testfile.txt'
-        sh:
-            wc -l ~/to_host_named_testfile.txt > to_host_named_linecount.txt
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-            "sig_mode": "force",
-        },
-    )
-    assert os.path.isfile("to_host_named_linecount.txt")
-    with open("to_host_named_linecount.txt") as lc:
-        assert lc.read().strip().startswith("200")
-
-
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_from_host_option(clear_now_and_after):
-    """Test from_remote option"""
-    clear_now_and_after("llp")
-    execute_workflow(
-        """
-        [10]
-        task: from_host='llp'
-        with open('llp', 'w') as llp:
-            llp.write("LLP")
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            "wait_for_task": True,
-            "default_queue": "docker",
-            "sig_mode": "force",
-        },
-    )
-    assert os.path.isfile("llp")
-
-
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_local_from_host_option(clear_now_and_after):
-    """Test from_remote option"""
-    clear_now_and_after("llp")
-    execute_workflow(
-        """
-        [10]
-        task: from_host='llp'
-        sh:
-        echo "LLP" > llp
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            # do not wait for jobs
-            "wait_for_task": True,
-            "sig_mode": "force",
-            "default_queue": "localhost",
-        },
-    )
-    assert os.path.isfile("llp")
-
-
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_local_from_host_option_with_named_path(clear_now_and_after):
-    """Test from_remote option"""
-    clear_now_and_after(os.path.expanduser("~/llp"))
-    execute_workflow(
-        """
-        [10]
-        task: from_host='#home/llp'
-        sh:
-        echo "LLP" > ~/llp
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            # do not wait for jobs
-            "wait_for_task": True,
-            "sig_mode": "force",
-            "default_queue": "localhost",
-        },
-    )
-    assert os.path.isfile(os.path.expanduser("~/llp"))
-
-
 def test_worker_procs():
     # test -j option
     execute_workflow(
@@ -182,7 +62,7 @@ def test_worker_procs_with_task():
 
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
 def test_remote_execute(clear_now_and_after, script_factory):
-    clear_now_and_after("result_remote.txt", "result_remote1.txt", "local.txt")
+    clear_now_and_after("result_remote.txt", "result_remote1.txt", "local.txt", "remote_exec.sos")
 
     test_remote_sos = script_factory("""
         [10]
@@ -193,40 +73,22 @@ def test_remote_execute(clear_now_and_after, script_factory):
         run:
             cp local.txt result_remote.txt
             echo 'adf' >> 'result_remote.txt'
-        """)
+        """, filename='remote_exec.sos')
     with open("local.txt", "w") as w:
         w.write("something")
 
-    assert 0 == subprocess.call("sos remote push docker --files local.txt -c ~/docker.yml", shell=True)
-
     assert 0 == subprocess.call(
         f"sos run {test_remote_sos} -c ~/docker.yml -r docker -s force",
         shell=True,
     )
 
-    assert not file_target("result_remote.txt").target_exists()
-
-    # self.assertEqual(subprocess.call('sos preview result_remote.txt -c ~/docker.yml -r docker', shell=True), 0)
-    # self.assertNotEqual(subprocess.call('sos preview result_remote.txt', shell=True), 0)
-    assert 0 == subprocess.call("sos remote pull docker --files result_remote.txt -c ~/docker.yml", shell=True)
-
     assert file_target("result_remote.txt").target_exists()
-
     # self.assertEqual(subprocess.call('sos preview result_remote.txt', shell=True), 0)
     with open("result_remote.txt") as w:
         content = w.read()
         assert "something" in content
         assert "adf" in content
 
-    # test sos remote run
-    assert 0 == subprocess.call(
-        "sos remote run docker -c  ~/docker.yml --cmd cp result_remote.txt result_remote1.txt ",
-        shell=True,
-    )
-    assert 0 == subprocess.call("sos remote pull docker --files result_remote1.txt -c ~/docker.yml", shell=True)
-
-    assert file_target("result_remote1.txt").target_exists()
-
 
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
 def test_remote_workflow_remote_queue(script_factory):
@@ -242,65 +104,9 @@ def test_remote_workflow_remote_queue(script_factory):
     assert 0 == subprocess.call(f"sos run {test_r_q} -c ~/docker.yml -r ts -q ts", shell=True)
 
 
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_signature_of_remote_target(clear_now_and_after, monkeypatch):
-    """Test remote() target"""
-    monkeypatch.setenv("SOS_DEBUG", "TASK,-")
-    clear_now_and_after("remote_file.txt", "result.txt")
-
-    with open("remote_file.txt", "w") as rf:
-        rf.write("""line1
-        line2
-        line3
-        """)
-    assert 0 == subprocess.call("sos remote push docker --files remote_file.txt -c ~/docker.yml", shell=True)
-    os.remove("remote_file.txt")
-    #
-    wf = """
-        input: remote('remote_file.txt')
-        output: 'result.txt'
-
-        task:
-        sh: expand=True
-            wc -l {_input} > {_output}
-        """
-    execute_workflow(
-        wf,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-        },
-    )
-    assert file_target("result.txt").target_exists()
-    assert open("result.txt").read().strip().startswith("3")
-    #
-    # now change the remote file
-    with open("remote_file.txt", "w") as rf:
-        rf.write("""line1
-        line2
-        line3
-        line4
-        line5
-        """)
-    assert 0 == subprocess.call("sos remote push docker --files remote_file.txt -c ~/docker.yml", shell=True)
-    os.remove("remote_file.txt")
-    os.remove("result.txt")
-    #
-    execute_workflow(
-        wf,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-        },
-    )
-    assert file_target("result.txt").target_exists()
-    assert open("result.txt").read().strip().startswith("5")
-
-
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
 def test_remote_exec(clear_now_and_after):
     clear_now_and_after("result_exec.txt")
-    root_dir = "/root/build" if "TRAVIS" in os.environ else "/root"
     execute_workflow(
         """
         output: 'result_exec.txt'
@@ -320,64 +126,11 @@ def test_remote_exec(clear_now_and_after):
     with open(file_target("result_exec.txt")) as res:
         result = res.read()
         assert "Output: result_exec.txt" in result
-        assert f"PWD: {root_dir}" in result
-
-
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_remote_exec_named_path(clear_now_and_after):
-    clear_now_and_after("result_named_path.txt")
-    root_dir = "/root/build" if "TRAVIS" in os.environ else "/root"
-
-    execute_workflow(
-        """
-        output: '#home/result_named_path.txt'
-
-        task:
-        sh: expand=True
-            echo Output: {_output} > {_output}
-            echo PWD: `pwd`. >> {_output}
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-            "sig_mode": "force",
-        },
-    )
-    assert file_target("#home/result_named_path.txt").target_exists()
-    with open(file_target("#home/result_named_path.txt")) as res:
-        result = res.read()
-        print(result)
-        assert "Output: /root/result_named_path.txt" in result
-        assert f"PWD: {root_dir}" in result
-
-
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_remote_exec_workdir_named_path(clear_now_and_after):
-    clear_now_and_after(file_target("#home/wd/result_workdir_named_path.txt"))
-    execute_workflow(
-        """
-        output: '#home/wd/result_workdir_named_path.txt'
-
-        task: workdir='/root'
-        sh: expand=True
-            echo Output: {_output} > {_output}
-            echo PWD: `pwd`. >> {_output}
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-            "sig_mode": "force",
-        },
-    )
-    assert file_target("#home/wd/result_workdir_named_path.txt").target_exists()
-    with open(file_target("#home/wd/result_workdir_named_path.txt")) as res:
-        result = res.read()
-        assert "Output: /root/wd/result_workdir_named_path.txt" in result
-        assert "PWD: /root." in result
+        assert f"PWD: {os.getcwd()}" in result
 
 
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_remote_exec_workdir_wo_named_path(clear_now_and_after):
+def test_remote_exec_workdir(clear_now_and_after):
     clear_now_and_after(file_target("result_workdir_wo_named.txt"))
     with pytest.raises(Exception):
         execute_workflow(
diff --git a/test/test_signature.py b/test/test_signature.py
index d20f66c25..ef9bc2bf2 100644
--- a/test/test_signature.py
+++ b/test/test_signature.py
@@ -754,7 +754,7 @@ def test_signature_with_dependency_tracing_and_vars(clear_signatures,
     assert res['__completed__']['__substep_completed__'] == 2
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 def test_skip_mode(clear_signatures, temp_factory, clear_now_and_after):
     '''Test skipping mode of signature'''
     clear_now_and_after([f'a_{i}.bak' for i in range(4)])
diff --git a/test/test_singularity.py b/test/test_singularity.py
index 9fae476cb..5c8c6ed19 100644
--- a/test/test_singularity.py
+++ b/test/test_singularity.py
@@ -25,7 +25,7 @@ def test_bash_in_singularity():
         ''')
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 @pytest.mark.skipif(
     not shutil.which('singularity') or sys.platform == 'win32' or
     'TRAVIS' in os.environ or 'APPVEYOR' in os.environ,
@@ -47,7 +47,7 @@ def test_singularity_build_linux_image(self):
         ''')
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 @pytest.mark.skipif(
     not shutil.which('singularity') or sys.platform == 'win32' or
     'TRAVIS' in os.environ or 'APPVEYOR' in os.environ,
@@ -59,7 +59,7 @@ def test_singularity_build_from_shub(self):
         ''')
 
 
-@pytest.mark.skip(reason="temporary skip")
+
 @pytest.mark.skipif(
     not shutil.which('singularity') or sys.platform == 'win32',
     reason='Skip test because docker is not installed.')
diff --git a/test/test_target.py b/test/test_target.py
index 41face2af..30bc39b76 100644
--- a/test/test_target.py
+++ b/test/test_target.py
@@ -94,12 +94,7 @@ def test_target_set_get():
 
 def test_target_group_by():
     """Test new option group_by to sos_targets"""
-    res = sos_targets(
-        "e.txt",
-        "f.ext",
-        a=["a.txt", "b.txt"],
-        b=["c.txt", "d.txt"],
-        group_by=1)
+    res = sos_targets("e.txt", "f.ext", a=["a.txt", "b.txt"], b=["c.txt", "d.txt"], group_by=1)
     assert len(res.groups) == 6
     assert res.labels == ["", "", "a", "a", "b", "b"]
     #
@@ -110,10 +105,7 @@ def test_target_group_by():
 def test_target_paired_with():
     """Test paired_with targets with vars"""
     res = sos_targets(
-        "e.txt",
-        "f.ext",
-        a=["a.txt", "b.txt"],
-        b=["c.txt", "d.txt"],
+        "e.txt", "f.ext", a=["a.txt", "b.txt"], b=["c.txt", "d.txt"],
         group_by=1).paired_with("_name", ["e", "f", "a", "b", "c", "d"])
     for i, n in enumerate(["e", "f", "a", "b", "c", "d"]):
         assert res[i]._name == n
@@ -124,17 +116,13 @@ def test_target_paired_with():
     #
     # test assert for length difference
     with pytest.raises(Exception):
-        sos_targets("e.txt",
-                    "f.ext").paired_with("name", ["e", "f", "a", "b", "c", "d"])
+        sos_targets("e.txt", "f.ext").paired_with("name", ["e", "f", "a", "b", "c", "d"])
 
 
 def test_target_group_with():
     """Test group_with targets with vars"""
     res = sos_targets(
-        "e.txt",
-        "f.ext",
-        a=["a.txt", "b.txt"],
-        b=["c.txt", "d.txt"],
+        "e.txt", "f.ext", a=["a.txt", "b.txt"], b=["c.txt", "d.txt"],
         group_by=2).group_with("name", ["a1", "a2", "a3"])
     for i, n in enumerate(["a1", "a2", "a3"]):
         assert res.groups[i].name == n
@@ -168,8 +156,7 @@ def test_group_with_with_no_output():
 def test_merging_of_sos_targets():
     """Test merging of multiple sos targets"""
     # merge 0 to 0
-    res = sos_targets("a.txt", "b.txt",
-                      sos_targets("c.txt", "d.txt", group_by=1))
+    res = sos_targets("a.txt", "b.txt", sos_targets("c.txt", "d.txt", group_by=1))
     assert len(res) == 4
     assert len(res.groups) == 2
     assert res.groups[0] == ["a.txt", "b.txt", "c.txt"]
@@ -217,15 +204,12 @@ def test_target_format():
         (sos_targets("a b.txt"), "x", ".txt"),
     ]:
         if isinstance(res, str):
-            assert interpolate(
-                f"{{target:{fmt}}}", globals(),
-                locals()) == res, "Interpolation of {}:{} should be {}".format(
-                    target, fmt, res)
+            assert interpolate(f"{{target:{fmt}}}", globals(),
+                               locals()) == res, "Interpolation of {}:{} should be {}".format(target, fmt, res)
 
         else:
-            assert interpolate(f"{{target:{fmt}}}", globals(), locals(
-            )) in res, "Interpolation of {}:{} should be one of {}".format(
-                target, fmt, res)
+            assert interpolate(f"{{target:{fmt}}}", globals(),
+                               locals()) in res, "Interpolation of {}:{} should be one of {}".format(target, fmt, res)
 
 
 def test_iter_targets():
@@ -382,9 +366,7 @@ def test_depends_executable():
     file_target("a.txt").unlink()
 
 
-@pytest.mark.skipif(
-    sys.platform == "win32",
-    reason="Windows executable cannot be created with chmod.")
+@pytest.mark.skipif(sys.platform == "win32", reason="Windows executable cannot be created with chmod.")
 def test_output_executable(clear_now_and_after):
     """Testing target executable."""
     # change $PATH so that lls can be found at the current
@@ -446,9 +428,7 @@ def test_depends_env_variable():
     file_target("a.txt").unlink()
 
 
-@pytest.mark.skipif(
-    sys.platform == "win32",
-    reason="Windows executable cannot be created with chmod.")
+@pytest.mark.skipif(sys.platform == "win32", reason="Windows executable cannot be created with chmod.")
 def test_provides_executable():
     """Testing provides executable target."""
     # change $PATH so that lls can be found at the current
@@ -499,7 +479,8 @@ def test_shared_var_in_paired_with(temp_factory):
 
 def test_shared_var_in_for_each(temp_factory, clear_now_and_after):
     temp_factory("1.txt", "2.txt")
-    clear_now_and_after("1.out", "2.out", "1.out2", "2.out2", '2.out_2.out2', '1.out_1.out2', '1.out_2.out2', '2.out_1.out2')
+    clear_now_and_after("1.out", "2.out", "1.out2", "2.out2", '2.out_2.out2', '1.out_1.out2', '1.out_2.out2',
+                        '2.out_1.out2')
     script = SoS_Script("""
 [work_1: shared = {'data': 'step_output'}]
 input: "1.txt", "2.txt", group_by = 'single', pattern = '{name}.{ext}'
@@ -687,36 +668,3 @@ def test_temp_file():
         assert sos_tempfile(file_target('a.txt')) == sos_tempfile(file_target('a.txt'))
 
     """)
-
-
-@pytest.mark.skip(reason="temporary skip")
-def test_named_path():
-    """Test the use of option name of path"""
-    execute_workflow(
-        """
-        import os
-        # windows might not have HOME
-        if 'HOME' in os.environ:
-            assert path('#home') == os.environ['HOME']
-            assert 'home' in path.names()
-            assert 'home' in path.names('docker')
-        """,
-        options={
-            "config_file": os.path.join(os.path.expanduser("~"), "docker.yml")
-        },
-    )
-
-
-@pytest.mark.skip(reason="temporary skip")
-@pytest.mark.skipif(
-    sys.platform == 'win32', reason='Graphviz not available under windows')
-def test_to_named_path_path():
-    execute_workflow(
-        """
-        [10: shared="a"]
-        a = path('/root/xxx/whatever').to_named_path(host='docker')
-        """,
-        options={
-            "config_file": os.path.join(os.path.expanduser("~"), "docker.yml"),
-        })
-    assert env.sos_dict['a'] == '#home/xxx/whatever'
diff --git a/test/test_task.py b/test/test_task.py
index a438d4286..66b9034ce 100644
--- a/test/test_task.py
+++ b/test/test_task.py
@@ -2,7 +2,6 @@
 #
 # Copyright (c) Bo Peng and the University of Texas MD Anderson Cancer Center
 # Distributed under the terms of the 3-clause BSD License.
-
 import os
 import shutil
 import subprocess
@@ -21,13 +20,11 @@
 has_docker = sys.platform != "win32"
 try:
     if sys.platform != "win32":
-        subprocess.check_output(
-            "docker ps | grep test_sos", shell=True).decode()
+        subprocess.check_output("docker ps | grep test_sos", shell=True).decode()
 except subprocess.CalledProcessError:
     subprocess.call("sh build_test_docker.sh", shell=True)
     try:
-        subprocess.check_output(
-            "docker ps | grep test_sos", shell=True).decode()
+        subprocess.check_output("docker ps | grep test_sos", shell=True).decode()
     except subprocess.CalledProcessError:
         print("Failed to set up a docker machine with sos")
         has_docker = False
@@ -48,23 +45,23 @@ def cd_new(path):
 
 def get_tasks():
     from sos.signatures import WorkflowSignatures
-    env.exec_dir = os.path.join(
-        os.path.expanduser("~"), ".sos", textMD5(os.getcwd()))
+    env.exec_dir = os.path.join(os.path.expanduser("~"), ".sos", textMD5(os.getcwd()))
     db = WorkflowSignatures()
     conn = db.conn
     #conn = sqlite3.connect(os.path.join(env.exec_dir, "workflow_signatures.db"))
     cur = conn.cursor()
     cur.execute('SELECT DISTINCT id FROM workflows WHERE entry_type = "task"')
-    return [x[0] for x in cur.fetchall() if os.path.isfile(
-        os.path.join(os.path.expanduser('~'), '.sos', 'tasks', x[0] + '.task'
-    ))]
+    return [
+        x[0]
+        for x in cur.fetchall()
+        if os.path.isfile(os.path.join(os.path.expanduser('~'), '.sos', 'tasks', x[0] + '.task'))
+    ]
 
 
 def test_task_file():
     """Test task file handling"""
     for ext in (".pulse", ".out", ".err", ".task", ".sh"):
-        filename = os.path.join(
-            os.path.expanduser("~"), ".sos", "tasks", "ffffffffffffffff" + ext)
+        filename = os.path.join(os.path.expanduser("~"), ".sos", "tasks", "ffffffffffffffff" + ext)
         if os.path.isfile(filename):
             os.remove(filename)
     params = TaskParams(
@@ -79,9 +76,7 @@ def test_task_file():
     assert a.tags == "a b"
     for ext in (".pulse", ".out", ".err", ".sh"):
         with open(
-                os.path.join(
-                    os.path.expanduser("~"), ".sos", "tasks",
-                    "ffffffffffffffff" + ext),
+                os.path.join(os.path.expanduser("~"), ".sos", "tasks", "ffffffffffffffff" + ext),
                 "w",
         ) as fh:
             fh.write(ext)
@@ -321,13 +316,11 @@ def test_kill_and_purge():
         output = subprocess.check_output(["sos", "status", "-v", "1"]).decode()
         if "killed" in output or "aborted" in output or "completed" in output:
             break
-        assert i <= 10, "Task should be killed within 10 seconds, got {}".format(
-            output)
+        assert i <= 10, "Task should be killed within 10 seconds, got {}".format(output)
         time.sleep(1)
     # test purge by status
     subprocess.call(["sos", "purge", "--status", "aborted"])
-    assert "killed" not in subprocess.check_output(["sos", "status", "-v",
-                                                    "3"]).decode()
+    assert "killed" not in subprocess.check_output(["sos", "status", "-v", "3"]).decode()
     # purge by all is not tested because it is dangerous
 
 
@@ -433,8 +426,7 @@ def test_task_tags():
             "worker_procs": ["4"],
         },
     ).run()
-    ret = subprocess.check_output(
-        f"sos status -t {tag}", shell=True).decode()
+    ret = subprocess.check_output(f"sos status -t {tag}", shell=True).decode()
     assert len(ret.splitlines()) == 5, f"Obtained {ret}"
     # test multiple tags
     tag1 = f"tag{random.randint(1, 100000)}"
@@ -459,8 +451,7 @@ def test_task_tags():
             "workflow": "default",
         },
     ).run()
-    ret = subprocess.check_output(
-        f"sos status -t {tag2}", shell=True).decode()
+    ret = subprocess.check_output(f"sos status -t {tag2}", shell=True).decode()
     assert len(ret.splitlines()) == 2, f"Obtained {ret}"
 
 
@@ -482,7 +473,6 @@ def test_max_mem():
         )
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_local_runtime_max_walltime():
     """Test server max_walltime option"""
     # gives warning, but do not kill
@@ -519,8 +509,7 @@ def test_max_cores():
         )
 
 
-@pytest.mark.skip(reason="temporary skip")
-@pytest.mark.skipIf(not has_docker, reason="Docker container not usable")
+@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
 def test_override_max_cores():
     """Test use queue_args to override server restriction max_cores"""
     execute_workflow(
@@ -540,12 +529,10 @@ def test_override_max_cores():
     )
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_list_hosts():
     """test list hosts using sos status -q"""
     for v in ["0", "1", "3", "4"]:
-        output = subprocess.check_output(
-            ["sos", "remote", "list", "-c", "~/docker.yml", "-v", v]).decode()
+        output = subprocess.check_output(["sos", "remote", "list", "-c", "~/docker.yml", "-v", v]).decode()
         assert "local_limited" in output, f"local_limited not in \n{output}\n for verbosity {v}"
 
 
@@ -583,25 +570,16 @@ def test_purge_all_with_option():
         tasks = get_tasks()
         subprocess.call("sos purge -s failed", shell=True)
     # check tasks
-    taskstatus = [
-        x.split()[0] for x in subprocess.check_output(
-            "sos status -v1", shell=True).decode().splitlines()
-    ]
+    taskstatus = [x.split()[0] for x in subprocess.check_output("sos status -v1", shell=True).decode().splitlines()]
     assert all(x in taskstatus for x in tasks)
     # purge one of them
     subprocess.call(f"sos purge {tasks[0]}", shell=True)
-    taskstatus = [
-        x.split()[0] for x in subprocess.check_output(
-            "sos status -v1", shell=True).decode().splitlines()
-    ]
+    taskstatus = [x.split()[0] for x in subprocess.check_output("sos status -v1", shell=True).decode().splitlines()]
     assert tasks[0] not in taskstatus
     assert tasks[1] in taskstatus
     #
     subprocess.call("sos purge --all", shell=True)
-    taskstatus = [
-        x.split()[0] for x in subprocess.check_output(
-            "sos status -v1", shell=True).decode().splitlines()
-    ]
+    taskstatus = [x.split()[0] for x in subprocess.check_output("sos status -v1", shell=True).decode().splitlines()]
     assert tasks[1] not in taskstatus
 
 
@@ -652,7 +630,6 @@ def test_task_no_signature(purge_tasks):
         assert time.time() - st > 1
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_task_with_signature(purge_tasks, clear_now_and_after):
     """Test re-execution of tasks"""
     # now with a real signature
@@ -714,7 +691,6 @@ def test_output_in_task():
         options={"default_queue": "localhost"})
 
 
-@pytest.mark.skip(reason="temporary skip")
 def test_repeated_tasks():
     """Test statement before task #1142 """
     for i in range(5):
@@ -801,47 +777,21 @@ def test_output_from_master_task():
         options={"default_queue": "localhost"})
 
 
-@pytest.mark.skip(reason="temporary skip")
-@pytest.mark.skipIf(not has_docker, reason="Docker container not usable")
-def test_remote_input_target(clear_now_and_after):
-    """Test the use of remote target"""
-    clear_now_and_after("vars.sh", "vars1.sh")
-    execute_workflow(
-        """
-        [10]
-        input: remote('/lib/init/vars.sh')
-        output: f'vars1.sh'
-
-        task:
-
-        with open(_input, 'r') as inf, open(_output, 'w') as outf:
-            outf.write(inf.read())
-        """,
-        options={
-            "config_file": "~/docker.yml",
-            "default_queue": "docker",
-            "sig_mode": "force",
-        },
-    )
-    assert not os.path.isfile("vars.sh")
-    assert os.path.isfile("vars1.sh")
-
-
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_delayed_interpolation(clear_now_and_after):
+def test_remotely_generated_output(clear_now_and_after):
     """Test delayed interpolation with expression involving remote objects"""
     # purge all previous tasks
     clear_now_and_after('test.py', 'test.py.bak')
     execute_workflow(
         """
         [10]
-        output: remote('test.py')
+        output: 'test.py'
         task:
         run:
             touch test.py
 
         [20]
-        output: remote(f"{_input:R}.bak")
+        output: f"{_input:R}.bak"
         task:
         run: expand=True
             cp {_input} {_output}
@@ -855,24 +805,30 @@ def test_delayed_interpolation(clear_now_and_after):
         },
     )
     # this file is remote only
-    assert not os.path.isfile("test.py")
-    assert not os.path.isfile("test.py.bak")
+    assert os.path.isfile("test.py")
+    assert os.path.isfile("test.py.bak")
 
 
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_remote_output_target(clear_now_and_after):
+def test_remote_output(clear_now_and_after):
     """Test the use of remote target"""
-    clear_now_and_after("vars.sh", "vars1.sh")
+    clear_now_and_after("line_cnt.txt")
+    with open("data.txt", "w") as data:
+        for i in range(100):
+            data.write(f"row {i}\n")
+    with open("line_cnt.txt", "w") as res:
+        res.write("50")
+
     execute_workflow(
         """
         [10]
-        input: remote('/lib/init/vars.sh')
-        output: remote(f'vars1.sh')
+        input: 'data.txt'
+        output: "line_cnt.txt"
 
         task:
 
         with open(_input, 'r') as inf, open(_output, 'w') as outf:
-            outf.write(inf.read())
+            outf.write(f'{len(inf.readlines())}')
         """,
         options={
             "config_file": "~/docker.yml",
@@ -880,60 +836,32 @@ def test_remote_output_target(clear_now_and_after):
             "sig_mode": "force",
         },
     )
-    assert not os.path.isfile("vars.sh")
-    assert not os.path.isfile("vars1.sh")
+    with open("line_cnt.txt") as res:
+        assert res.read() == "100"
 
 
-@pytest.mark.skipif(not has_docker, reason="Docker container not usable")
-def test_remote_output_target_with_trunksize(clear_now_and_after):
-    clear_now_and_after("vars.sh", "vars1.sh")
-
+def test_runtime_max_walltime():
+    """Test server max_walltime option"""
+    # only warning messages will be yielded
     execute_workflow(
-        """\
-        [10]
-        import os
-        input: remote('/lib/init/vars.sh'), remote('/lib/init/init-d-script'), group_by=1
-        output: remote(os.path.basename(str(_input)))
-
-        task: trunk_size=2
-
-        with open(_input, 'r') as inf, open(_output, 'w') as outf:
-            outf.write(inf.read())""",
+        """
+    [10]
+    task:
+    import time
+    time.sleep(25)
+    """,
         options={
             "config_file": "~/docker.yml",
-            "default_queue": "docker",
+            "default_queue": "docker_limited",
             "sig_mode": "force",
         },
     )
-    assert not os.path.isfile("vars.sh")
-    assert not os.path.isfile("init-d-script")
-
-
-@pytest.mark.skip(reason="temporary skip")
-def test_runtime_max_walltime():
-    """Test server max_walltime option"""
-    with pytest.raises(Exception):
-        execute_workflow(
-            """
-        [10]
-        task:
-        import time
-        time.sleep(25)
-        """,
-            options={
-                "config_file": "~/docker.yml",
-                "default_queue": "docker_limited",
-                "sig_mode": "force",
-            },
-        )
 
 
-@pytest.mark.skip(reason="temporary skip")
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
 def test_sync_master_task(clear_now_and_after):
     """Test sync input and output with remote host with trunksize"""
-    clear_now_and_after([f"test_{i}.txt" for i in range(4)],
-                        [f"test_{i}.bak" for i in range(4)])
+    clear_now_and_after([f"test_{i}.txt" for i in range(4)], [f"test_{i}.bak" for i in range(4)])
     import random
 
     val = random.randint(1, 10000)
@@ -1074,7 +1002,6 @@ def test_trunk_workers_option(clear_now_and_after, purge_tasks):
         assert os.path.isfile(f"{i}.txt")
 
 
-@pytest.mark.skip(reason="temporary skip")
 @pytest.mark.skipif(not has_docker, reason="Docker container not usable")
 def test_sync_input_output_and_rerun(clear_now_and_after, purge_tasks):
     """Test sync input and output with remote host"""