Merge pull request #155 from common-workflow-language/fsaccess-refactor

Refactor fs_access abstraction

Author: mr-c

cwltool/cwltest.py

@@ -123,6 +123,9 @@ def run_test(args, i, t):  # type: (argparse.Namespace, Any, Dict[str,str]) -> i
         _logger.error(u"""Test failed: %s""", " ".join([pipes.quote(tc) for tc in test_command]))
         _logger.error(outstr)
         _logger.error(u"Parse error %s", str(e))
+    except KeyboardInterrupt:
+        _logger.error(u"""Test interrupted: %s""", " ".join([pipes.quote(tc) for tc in test_command]))
+        return 1
 
     failed = False
 

cwltool/draft2tool.py

@@ -365,16 +365,17 @@ def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
         # type: (Set[Dict[str,Any]], Builder, str) -> Dict[unicode, Union[unicode, List[Any], Dict[unicode, Any]]]
         try:
             ret = {}  # type: Dict[unicode, Union[unicode, List[Any], Dict[unicode, Any]]]
-            custom_output = builder.fs_access.join(outdir, "cwl.output.json")
-            if builder.fs_access.exists(custom_output):
-                with builder.fs_access.open(custom_output, "r") as f:
+            fs_access = builder.make_fs_access(outdir)
+            custom_output = fs_access.join(outdir, "cwl.output.json")
+            if fs_access.exists(custom_output):
+                with fs_access.open(custom_output, "r") as f:
                     ret = json.load(f)
                 _logger.debug(u"Raw output from %s: %s", custom_output, json.dumps(ret, indent=4))
             else:
                 for port in ports:
                     fragment = shortname(port["id"])
                     try:
-                        ret[fragment] = self.collect_output(port, builder, outdir, compute_checksum)
+                        ret[fragment] = self.collect_output(port, builder, outdir, fs_access, compute_checksum=compute_checksum)
                     except Exception as e:
                         _logger.debug(u"Error collecting output for parameter '%s'" % shortname(port["id"]), exc_info=e)
                         raise WorkflowException(u"Error collecting output for parameter '%s': %s" % (shortname(port["id"]), e))
@@ -388,14 +389,14 @@ def collect_output_ports(self, ports, builder, outdir, compute_checksum=True):
                 adjustDirObjs(ret, remove_path)
                 normalizeFilesDirs(ret)
                 if compute_checksum:
-                    adjustFileObjs(ret, partial(compute_checksums, builder.fs_access))
+                    adjustFileObjs(ret, partial(compute_checksums, fs_access))
 
             validate.validate_ex(self.names.get_name("outputs_record_schema", ""), ret)
             return ret if ret is not None else {}
         except validate.ValidationException as e:
             raise WorkflowException("Error validating output record, " + str(e) + "\n in " + json.dumps(ret, indent=4))
 
-    def collect_output(self, schema, builder, outdir, compute_checksum=True):
+    def collect_output(self, schema, builder, outdir, fs_access, compute_checksum=True):
         # type: (Dict[str,Any], Builder, str) -> Union[Dict[unicode, Any], List[Union[Dict[unicode, Any], unicode]]]
         r = []  # type: List[Any]
         if "outputBinding" in schema:
@@ -419,16 +420,16 @@ def collect_output(self, schema, builder, outdir, compute_checksum=True):
                         raise WorkflowException("glob patterns must not start with '/'")
                     try:
                         r.extend([{"location": g,
-                                   "class": "File" if builder.fs_access.isfile(g) else "Directory"}
-                                  for g in builder.fs_access.glob(builder.fs_access.join(outdir, gb))])
+                                   "class": "File" if fs_access.isfile(g) else "Directory"}
+                                  for g in fs_access.glob(fs_access.join(outdir, gb))])
                     except (OSError, IOError) as e:
                         _logger.warn(str(e))
 
                 for files in r:
                     if files["class"] == "Directory" and "listing" not in files:
-                        getListing(builder.fs_access, files)
+                        getListing(fs_access, files)
                     else:
-                        with builder.fs_access.open(files["location"], "rb") as f:
+                        with fs_access.open(files["location"], "rb") as f:
                             contents = ""
                             if binding.get("loadContents") or compute_checksum:
                                 contents = f.read(CONTENT_LIMIT)
@@ -488,7 +489,7 @@ def collect_output(self, schema, builder, outdir, compute_checksum=True):
                                 sfpath = {"location": substitute(primary["location"], sf), "class": "File"}
 
                             for sfitem in aslist(sfpath):
-                                if builder.fs_access.exists(sfitem["location"]):
+                                if fs_access.exists(sfitem["location"]):
                                     primary["secondaryFiles"].append(sfitem)
 
             if not r and optional:
@@ -499,6 +500,6 @@ def collect_output(self, schema, builder, outdir, compute_checksum=True):
             out = {}
             for f in schema["type"]["fields"]:
                 out[shortname(f["name"])] = self.collect_output(  # type: ignore
-                        f, builder, outdir, compute_checksum)
+                        f, builder, outdir, fs_access, compute_checksum=compute_checksum)
             return out
         return r

cwltool/main.py

@@ -351,7 +351,8 @@ def generate_parser(toolparser, tool, namemap):
     return toolparser
 
 
-def load_job_order(args, t, stdin, print_input_deps=False, relative_deps=False, stdout=sys.stdout):
+def load_job_order(args, t, stdin, print_input_deps=False, relative_deps=False,
+                   stdout=sys.stdout, make_fs_access=None):
     # type: (argparse.Namespace, Process, IO[Any], bool, bool, IO[Any]) -> Union[int,Tuple[Dict[str,Any],str]]
 
     job_order_object = None
@@ -439,7 +440,7 @@ def pathToLoc(p):
     adjustFileObjs(job_order_object, pathToLoc)
     normalizeFilesDirs(job_order_object)
     adjustDirObjs(job_order_object, cast(Callable[..., Any],
-        functools.partial(getListing, StdFsAccess(input_basedir))))
+        functools.partial(getListing, make_fs_access(input_basedir))))
 
     if "cwl:tool" in job_order_object:
         del job_order_object["cwl:tool"]
@@ -572,7 +573,8 @@ def main(argsl=None,
          stdout=sys.stdout,
          stderr=sys.stderr,
          versionfunc=versionstring,
-         job_order_object=None):
+         job_order_object=None,
+         make_fs_access=StdFsAccess):
     # type: (List[str], argparse.Namespace, Callable[..., Union[str, Dict[str, str]]], Callable[..., Process], Callable[[Dict[str, int]], Dict[str, int]], IO[Any], IO[Any], IO[Any], Callable[[], unicode], Union[int, Tuple[Dict[str, Any], str]]) -> int
 
     _logger.removeHandler(defaultStreamHandler)
@@ -695,7 +697,8 @@ def main(argsl=None,
             job_order_object = load_job_order(args, tool, stdin,
                                               print_input_deps=args.print_input_deps,
                                               relative_deps=args.relative_deps,
-                                              stdout=stdout)
+                                              stdout=stdout,
+                                              make_fs_access=make_fs_access)
 
         if isinstance(job_order_object, int):
             return job_order_object
@@ -714,6 +717,7 @@ def main(argsl=None,
             out = executor(tool, job_order_object[0],
                            makeTool=makeTool,
                            select_resources=selectResources,
+                           make_fs_access=make_fs_access,
                            **vars(args))
 
             # This is the workflow output, it needs to be written

cwltool/process.py

@@ -428,7 +428,8 @@ def _init_job(self, joborder, **kwargs):
             builder.tmpdir = os.path.realpath(kwargs.get("tmpdir") or tempfile.mkdtemp())
             builder.stagedir = os.path.realpath(kwargs.get("stagedir") or tempfile.mkdtemp())
 
-        builder.fs_access = kwargs.get("fs_access") or StdFsAccess(kwargs["basedir"])
+        builder.make_fs_access = kwargs.get("make_fs_access") or StdFsAccess
+        builder.fs_access = builder.make_fs_access(kwargs["basedir"])
 
         if self.formatgraph:
             for i in self.tool["inputs"]: