working on load listing recognition for files and provenance

Author: jjkoehorst

cwltool/job.py

@@ -299,7 +299,7 @@ def _execute(
                 and isinstance(job_order, (list, dict))
             ):
                 runtimeContext.prov_obj.used_artefacts(
-                    job_order, runtimeContext.process_run_id, str(self.name)
+                    job_order, runtimeContext.process_run_id, str(self.name), load_listing=self.builder.loadListing
                 )
             else:
                 _logger.warning(
@@ -426,6 +426,7 @@ def stderr_stdout_log_path(
                 runtimeContext.process_run_id,
                 outputs,
                 datetime.datetime.now(),
+                builder.loadListing # TODO FIX THIS
             )
         if processStatus != "success":
             _logger.warning("[job %s] completed %s", self.name, processStatus)

cwltool/provenance_profile.py

@@ -89,8 +89,15 @@ def copy_job_order(
             iid = shortname(i["id"])
             # if iid in the load listing object and no_listing then....
             if iid in job_order_object:
-                if iid in load_listing and load_listing[iid] != "no_listing":
-                    customised_job[iid] = copy.deepcopy(job_order_object[iid])
+                if iid in load_listing:
+                    if load_listing[iid] == "no_listing":
+                        _logger.warning("Skip listing of " + iid)
+                        job_order_object[iid]['loadListing'] = 'no_listing'
+                        job_order_object[iid]['listing'] = []
+                        customised_job[iid] = job_order_object[iid]
+                    else:
+                        # Normal deep copy
+                        customised_job[iid] = copy.deepcopy(job_order_object[iid])
                 # TODO Other listing options here?
                 else:
                     # add the input element in dictionary for provenance
@@ -270,13 +277,14 @@ def evaluate(
             # record provenance of independent commandline tool executions
             self.prospective_prov(job)
             customised_job = copy_job_order(job, job_order_object, process)
-            self.used_artefacts(customised_job, self.workflow_run_uri)
+            self.used_artefacts(customised_job, self.workflow_run_uri, job.builder.loadListing)
             research_obj.create_job(customised_job)
         elif hasattr(job, "workflow"):
             # record provenance of workflow executions
             self.prospective_prov(job)
             customised_job = copy_job_order(job, job_order_object, process)
-            self.used_artefacts(customised_job, self.workflow_run_uri)
+            self.used_artefacts(customised_job, self.workflow_run_uri, schema=process.inputs_record_schema)
+
 
     def record_process_start(
         self, process: Process, job: JobsType, process_run_id: Optional[str] = None
@@ -355,11 +363,12 @@ def record_process_end(
         process_run_id: str,
         outputs: Union[CWLObjectType, MutableSequence[CWLObjectType], None],
         when: datetime.datetime,
+        load_listing: str = "deep_listing",
     ) -> None:
-        self.generate_output_prov(outputs, process_run_id, process_name)
+        self.generate_output_prov(outputs, process_run_id, process_name, load_listing)
         self.document.wasEndedBy(process_run_id, None, self.workflow_run_uri, when)
 
-    def declare_file(self, value: CWLObjectType) -> Tuple[ProvEntity, ProvEntity, str]:
+    def declare_file(self, value: CWLObjectType, load_listing: str = "deep_listing") -> Tuple[ProvEntity, ProvEntity, str]:
         if value["class"] != "File":
             raise ValueError("Must have class:File: %s" % value)
         # Need to determine file hash aka RO filename
@@ -436,9 +445,9 @@ def declare_file(self, value: CWLObjectType) -> Tuple[ProvEntity, ProvEntity, st
         ):
             # TODO: Record these in a specializationOf entity with UUID?
             if sec["class"] == "File":
-                (sec_entity, _, _) = self.declare_file(sec)
+                (sec_entity, _, _) = self.declare_file(sec, load_listing)
             elif sec["class"] == "Directory":
-                sec_entity = self.declare_directory(sec)
+                sec_entity = self.declare_directory(sec, load_listing)
             else:
                 raise ValueError(f"Got unexpected secondaryFiles value: {sec}")
             # We don't know how/when/where the secondary file was generated,
@@ -453,7 +462,7 @@ def declare_file(self, value: CWLObjectType) -> Tuple[ProvEntity, ProvEntity, st
 
         return file_entity, entity, checksum
 
-    def declare_directory(self, value: CWLObjectType) -> ProvEntity:
+    def declare_directory(self, value: CWLObjectType, load_listing: str = "deep_listing") -> ProvEntity:
         """Register any nested files/directories."""
         # FIXME: Calculate a hash-like identifier for directory
         # so we get same value if it's the same filenames/hashes
@@ -498,12 +507,19 @@ def declare_directory(self, value: CWLObjectType) -> ProvEntity:
         # if value['basename'] == "dirIgnore":
         #     pass
         if "listing" not in value:
-            get_listing(self.fsaccess, value)
+            if load_listing == "no_listing":
+                pass
+            elif load_listing == "deep_listing":
+                get_listing(self.fsaccess, value)
+            elif load_listing == "shallow_listing":
+                get_listing(self.fsaccess, value, False)
+            else:
+                raise ValueError("Invalid listing value: %s", load_listing)
 
         for entry in cast(MutableSequence[CWLObjectType], value.get("listing", [])):
             is_empty = False
             # Declare child-artifacts
-            entity = self.declare_artefact(entry)
+            entity = self.declare_artefact(entry, load_listing)
             self.document.membership(coll, entity)
             # Membership relation aka our ORE Proxy
             m_id = uuid.uuid4().urn
@@ -573,7 +589,7 @@ def declare_string(self, value: str) -> Tuple[ProvEntity, str]:
         )
         return entity, checksum
 
-    def declare_artefact(self, value: Any) -> ProvEntity:
+    def declare_artefact(self, value: Any, load_listing: str = "deep_listing") -> ProvEntity:
         """Create data artefact entities for all file objects."""
         if value is None:
             # FIXME: If this can happen in CWL, we'll
@@ -615,12 +631,12 @@ def declare_artefact(self, value: Any) -> ProvEntity:
 
             # Base case - we found a File we need to update
             if value.get("class") == "File":
-                (entity, _, _) = self.declare_file(value)
+                (entity, _, _) = self.declare_file(value, load_listing)
                 value["@id"] = entity.identifier.uri
                 return entity
 
             if value.get("class") == "Directory":
-                entity = self.declare_directory(value)
+                entity = self.declare_directory(value, load_listing)
                 value["@id"] = entity.identifier.uri
                 return entity
             coll_id = value.setdefault("@id", uuid.uuid4().urn)
@@ -643,7 +659,7 @@ def declare_artefact(self, value: Any) -> ProvEntity:
             # Let's iterate and recurse
             coll_attribs: List[Tuple[Union[str, Identifier], Any]] = []
             for (key, val) in value.items():
-                v_ent = self.declare_artefact(val)
+                v_ent = self.declare_artefact(val, load_listing)
                 self.document.membership(coll, v_ent)
                 m_entity = self.document.entity(uuid.uuid4().urn)
                 # Note: only support PROV-O style dictionary
@@ -664,7 +680,7 @@ def declare_artefact(self, value: Any) -> ProvEntity:
             members = []
             for each_input_obj in iter(value):
                 # Recurse and register any nested objects
-                e = self.declare_artefact(each_input_obj)
+                e = self.declare_artefact(each_input_obj, load_listing)
                 members.append(e)
 
             # If we reached this, then we were allowed to iterate
@@ -698,20 +714,31 @@ def used_artefacts(
         job_order: Union[CWLObjectType, List[CWLObjectType]],
         process_run_id: str,
         name: Optional[str] = None,
+        schema: Any = None,
+        load_listing: Optional[str] = None,
     ) -> None:
         """Add used() for each data artefact."""
         if isinstance(job_order, list):
             for entry in job_order:
-                self.used_artefacts(entry, process_run_id, name)
+                # for field in schema.fields:
+                    # if field['name'] == entry.
+                    # load_listing = schema.fields
+                self.used_artefacts(entry, process_run_id, name, load_listing)
         else:
             # FIXME: Use workflow name in packed.cwl, "main" is wrong for nested workflows
             base = "main"
             if name is not None:
                 base += "/" + name
             for key, value in job_order.items():
                 prov_role = self.wf_ns[f"{base}/{key}"]
+                if not load_listing:
+                    load_listing = "deep_listing"
+                    for field in schema['fields']:
+                        if field['name'] == key:
+                            load_listing = field['loadListing']
+                            break
                 try:
-                    entity = self.declare_artefact(value)
+                    entity = self.declare_artefact(value, load_listing)
                     self.document.used(
                         process_run_id,
                         entity,
@@ -727,11 +754,12 @@ def generate_output_prov(
         final_output: Union[CWLObjectType, MutableSequence[CWLObjectType], None],
         process_run_id: Optional[str],
         name: Optional[str],
+        load_listing: str = "deep_listing"
     ) -> None:
         """Call wasGeneratedBy() for each output,copy the files into the RO."""
         if isinstance(final_output, MutableSequence):
             for entry in final_output:
-                self.generate_output_prov(entry, process_run_id, name)
+                self.generate_output_prov(entry, process_run_id, name, load_listing)
         elif final_output is not None:
             # Timestamp should be created at the earliest
             timestamp = datetime.datetime.now()
@@ -740,7 +768,7 @@ def generate_output_prov(
             # entity (UUID) and document it as generated in
             # a role corresponding to the output
             for output, value in final_output.items():
-                entity = self.declare_artefact(value)
+                entity = self.declare_artefact(value, load_listing)
                 if name is not None:
                     name = urllib.parse.quote(str(name), safe=":/,#")
                     # FIXME: Probably not "main" in nested workflows

cwltool/utils.py

@@ -97,7 +97,7 @@
 ScatterOutputCallbackType = Callable[[Optional[ScatterDestinationsType], str], None]
 SinkType = Union[CWLOutputType, CWLObjectType]
 DirectoryType = TypedDict(
-    "DirectoryType", {"class": str, "listing": List[CWLObjectType], "basename": str}
+    "DirectoryType", {"class": str, "listing": List[CWLObjectType], "basename": str, "loadListing": str}
 )
 JSONAtomType = Union[Dict[str, Any], List[Any], str, int, float, bool, None]
 JSONType = Union[

tests/test_provenance.py

@@ -208,6 +208,10 @@ def test_directory_workflow(tmp_path: Path) -> None:
 
 @needs_docker
 def test_directory_workflow_no_listing(tmp_path: Path) -> None:
+    """
+    This test will check for 3 files that should be there and 3 files that should not be there.
+    @param tmp_path:
+    """
     dir2 = tmp_path / "dir2"
     dir2.mkdir()
     sha1 = {
@@ -253,8 +257,8 @@ def test_directory_workflow_no_listing(tmp_path: Path) -> None:
     file_list = (
         folder
         / "data"
-        / "3c"
-        / "3c363836cf4e16666669a25da280a1865c2d2874"
+        / "84"
+        / "84a516841ba77a5b4648de2cd0dfcb30ea46dbb4"
         # checksum as returned from:
         # echo -e "a\nb\nc" | sha1sum
         # 3ca69e8d6c234a469d16ac28a4a658c92267c423  -
@@ -273,8 +277,9 @@ def test_directory_workflow_no_listing(tmp_path: Path) -> None:
         # File should be empty and in the future not existing...
         # assert os.path.getsize(p.absolute()) == 0
         # To be discared when file really does not exist anymore
-        assert p.is_file(), f"Could not find {l} as {p}"
-
+        if l not in ['d', 'e', 'f']:
+            print("Analysing file %s", l)
+            assert p.is_file(), f"Could not find {l} as {p}"
 
 @needs_docker
 def test_no_data_files(tmp_path: Path) -> None:

tests/wf/directory_no_listing.cwl

@@ -13,6 +13,7 @@ hints:
 inputs:
     dir:
         type: Directory
+        loadListing: deep_listing
     ignore:
         type: Directory
         loadListing: no_listing
@@ -47,7 +48,10 @@ steps:
         run:
             class: CommandLineTool
             requirements:
-            - class: ShellCommandRequirement
+                ShellCommandRequirement: {}
+                LoadListingRequirement:
+                    loadListing: deep_listing
+
             arguments:
             - shellQuote: false
               valueFrom: >
@@ -64,10 +68,10 @@ steps:
                         glob: "dir1"
 
 outputs:
-    listing:
+    output_1:
         type: File
         outputSource: ls/listing
-    dir1:
+    output_2:
         type: Directory
         outputSource: generate/dir1