Merge pull request #1194 from common-workflow-language/toil-prov

3rd party cwltprov support (toil-cwl-runner, will help others)

Author: mr-c

cwltool/executors.py

@@ -159,7 +159,8 @@ def run_jobs(self,
                 user_provenance=False,
                 orcid=runtime_context.orcid,
                 # single tool execution, so RO UUID = wf UUID = tool UUID
-                run_uuid=runtime_context.research_obj.ro_uuid)
+                run_uuid=runtime_context.research_obj.ro_uuid,
+                fsaccess=runtime_context.make_fs_access(''))
             process.parent_wf = process.provenance_object
         jobiter = process.job(job_order_object, self.output_callback,
                               runtime_context)
@@ -178,6 +179,7 @@ def run_jobs(self,
                             prov_obj = job.prov_obj
                         if prov_obj:
                             runtime_context.prov_obj = prov_obj
+                            prov_obj.fsaccess = runtime_context.make_fs_access('')
                             prov_obj.evaluate(
                                 process, job, job_order_object,
                                 runtime_context.research_obj)

cwltool/main.py

@@ -600,6 +600,7 @@ def main(argsl=None,                   # type: Optional[List[str]]
                 _logger.error("--provenance incompatible with --no-compute-checksum")
                 return 1
             ro = ResearchObject(
+                getdefault(runtimeContext.make_fs_access, StdFsAccess),
                 temp_prefix_ro=args.tmpdir_prefix, orcid=args.orcid,
                 full_name=args.cwl_full_name)
             runtimeContext.research_obj = ro

cwltool/provenance.py

@@ -305,9 +305,11 @@ def __init__(self,
                  host_provenance,        # type: bool
                  user_provenance,        # type: bool
                  orcid,                  # type: str
+                 fsaccess,               # type: StdFsAccess
                  run_uuid=None           # type: Optional[uuid.UUID]
                 ):  # type: (...) -> None
         """Initialize the provenance profile."""
+        self.fsaccess = fsaccess
         self.orcid = orcid
         self.research_object = research_object
         self.folder = self.research_object.folder
@@ -504,8 +506,7 @@ def declare_file(self, value):
         if not entity and 'location' in value:
             location = str(value['location'])
             # If we made it here, we'll have to add it to the RO
-            fsaccess = StdFsAccess("")
-            with fsaccess.open(location, "rb") as fhandle:
+            with self.fsaccess.open(location, "rb") as fhandle:
                 relative_path = self.research_object.add_data_file(fhandle)
                 # FIXME: This naively relies on add_data_file setting hash as filename
                 checksum = PurePath(relative_path).name
@@ -597,8 +598,7 @@ def declare_directory(self, value):  # type: (MutableMapping[Text, Any]) -> Prov
         is_empty = True
 
         if "listing" not in value:
-            fsaccess = StdFsAccess("")
-            get_listing(fsaccess, value)
+            get_listing(self.fsaccess, value)
         for entry in value.get("listing", []):
             is_empty = False
             # Declare child-artifacts
@@ -945,8 +945,8 @@ def finalize_prov_profile(self, name):
 class ResearchObject():
     """CWLProv Research Object."""
 
-    def __init__(self, temp_prefix_ro="tmp", orcid='', full_name=''):
-        # type: (str, Text, Text) -> None
+    def __init__(self, fsaccess, temp_prefix_ro="tmp", orcid='', full_name=''):
+        # type: (StdFsAccess, str, Text, Text) -> None
         """Initialize the ResearchObject."""
         self.temp_prefix = temp_prefix_ro
         self.orcid = '' if not orcid else _valid_orcid(orcid)
@@ -962,7 +962,7 @@ def __init__(self, temp_prefix_ro="tmp", orcid='', full_name=''):
         self._external_aggregates = []  # type: List[Dict[Text, Text]]
         self.annotations = []  # type: List[Dict[Text, Any]]
         self._content_types = {}  # type: Dict[Text,str]
-
+        self.fsaccess = fsaccess
         # These should be replaced by generate_prov_doc when workflow/run IDs are known:
         self.engine_uuid = "urn:uuid:%s" % uuid.uuid4()
         self.ro_uuid = uuid.uuid4()
@@ -1561,8 +1561,7 @@ def _relativise_files(self, structure):
                     # Register in RO; but why was this not picked
                     # up by used_artefacts?
                     _logger.info("[provenance] Adding to RO %s", structure["location"])
-                    fsaccess = StdFsAccess("")
-                    with fsaccess.open(structure["location"], "rb") as fp:
+                    with self.fsaccess.open(structure["location"], "rb") as fp:
                         relative_path = self.add_data_file(fp)
                         checksum = PurePosixPath(relative_path).name
                         structure["checksum"] = "%s$%s" % (SHA1, checksum)

cwltool/workflow.py

@@ -522,7 +522,8 @@ def __init__(self,
                 host_provenance=loadingContext.host_provenance,
                 user_provenance=loadingContext.user_provenance,
                 orcid=loadingContext.orcid,
-                run_uuid=run_uuid)  # inherit RO UUID for master wf run
+                run_uuid=run_uuid,
+                fsaccess=loadingContext.research_obj.fsaccess)  # inherit RO UUID for master wf run
             # TODO: Is Workflow(..) only called when we are the master workflow?
             self.parent_wf = self.provenance_object
 
@@ -582,6 +583,7 @@ def job(self,
         if runtimeContext.research_obj is not None:
             if runtimeContext.toplevel:
                 # Record primary-job.json
+                runtimeContext.research_obj.fsaccess = runtimeContext.make_fs_access('')
                 runtimeContext.research_obj.create_job(builder.job, self.job)
 
         job = WorkflowJob(self, runtimeContext)

tests/test_provenance.py

@@ -24,6 +24,7 @@
 from cwltool.main import main
 from cwltool.resolver import Path
 from cwltool.context import RuntimeContext
+from cwltool.stdfsaccess import StdFsAccess
 
 from .util import get_data, needs_docker, temp_dir, working_directory
 
@@ -513,7 +514,7 @@ def check_prov(base_path, nested=False, single_tool=False, directory=False,
 
 @pytest.fixture
 def research_object():
-    re_ob = provenance.ResearchObject()
+    re_ob = provenance.ResearchObject(StdFsAccess(''))
     yield re_ob
     re_ob.close()