Merge pull request #893 from common-workflow-language/cwlprov-add_output

add output object to RO

Author: mr-c

cwltool/main.py

@@ -712,6 +712,8 @@ def my_represent_none(self, data):  # pylint: disable=unused-argument
                                      logger=_logger)
 
             if out is not None:
+                if runtimeContext.research_obj:
+                    runtimeContext.research_obj.create_job(out, None, True)
                 def loc_to_path(obj):
                     for field in ("path", "nameext", "nameroot", "dirname"):
                         if field in obj:

cwltool/provenance.py

@@ -4,7 +4,6 @@
 import copy
 import datetime
 import hashlib
-import io
 import logging
 import os
 import os.path
@@ -15,10 +14,10 @@
 import uuid
 from collections import OrderedDict
 from getpass import getuser
-from io import open
+from io import BytesIO, FileIO, TextIOWrapper, open
 from socket import getfqdn
-from typing import (IO, Any, Callable, Dict, List, MutableMapping, Optional,
-                    Set, Tuple, Union, cast)
+from typing import (IO, Any, Callable, Dict, List, Generator, MutableMapping,
+                    Optional, Set, Tuple, Union, cast)
 
 import prov.model as provM
 import six
@@ -73,7 +72,7 @@ class PermissionError(OSError):  # pylint: disable=redefined-builtin
 # 2. Bump minor number if adding resources or PROV statements
 # 3. Bump patch number for non-breaking non-adding changes,
 #    e.g. fixing broken relative paths
-CWLPROV_VERSION = "https://w3id.org/cwl/prov/0.4.0"
+CWLPROV_VERSION = "https://w3id.org/cwl/prov/0.5.0"
 
 # Research Object folders
 METADATA = "metadata"
@@ -149,7 +148,7 @@ def _whoami():
     return (username, fullname)
 
 
-class WritableBagFile(io.FileIO):
+class WritableBagFile(FileIO):
     """Writes files in research object."""
 
     def __init__(self, research_object, rel_path):
@@ -209,7 +208,7 @@ def readable(self):
 
     def truncate(self, size=None):
         # type: (Optional[int]) -> int
-        # FIXME: This breaks contract io.IOBase,
+        # FIXME: This breaks contract IOBase,
         # as it means we would have to recalculate the hash
         if size is not None:
             raise IOError("WritableBagFile can't truncate")
@@ -458,7 +457,7 @@ def evaluate(self,
             self.prospective_prov(job)
             customised_job = copy_job_order(job, job_order_object)
             self.used_artefacts(customised_job, self.workflow_run_uri)
-            research_obj.create_job(job, customised_job)
+            research_obj.create_job(customised_job, job)
             # self.used_artefacts(inputs, self.workflow_run_uri)
             name = ""
             if hasattr(job, "name"):
@@ -680,7 +679,7 @@ def declare_directory(self, value):  # type: (MutableMapping) -> ProvEntity
     def declare_string(self, value):
         # type: (Union[Text, str]) -> Tuple[ProvEntity,Text]
         """Save as string in UTF-8."""
-        byte_s = io.BytesIO(str(value).encode(ENCODING))
+        byte_s = BytesIO(str(value).encode(ENCODING))
         data_file = self.research_object.add_data_file(byte_s, content_type=TEXT_PLAIN)
         checksum = posixpath.basename(data_file)
         # FIXME: Don't naively assume add_data_file uses hash in filename!
@@ -716,7 +715,7 @@ def declare_artefact(self, value):
 
         if isinstance(value, bytes):
             # If we got here then we must be in Python 3
-            byte_s = io.BytesIO(value)
+            byte_s = BytesIO(value)
             data_file = self.research_object.add_data_file(byte_s)
             # FIXME: Don't naively assume add_data_file uses hash in filename!
             data_id = "data:%s" % posixpath.split(data_file)[1]
@@ -1051,13 +1050,13 @@ def write_bag_file(self, path, encoding=ENCODING):
         # type: (Text, Optional[str]) -> IO
         """Write the bag file into our research object."""
         # For some reason below throws BlockingIOError
-        #fp = io.BufferedWriter(WritableBagFile(self, path))
+        #fp = BufferedWriter(WritableBagFile(self, path))
         bag_file = cast(IO, WritableBagFile(self, path))
         if encoding:
             # encoding: match Tag-File-Character-Encoding: UTF-8
             # newline: ensure LF also on Windows
             return cast(IO,
-                        io.TextIOWrapper(bag_file, encoding=encoding, newline="\n"))
+                        TextIOWrapper(bag_file, encoding=encoding, newline="\n"))
         return bag_file
 
     def add_tagfile(self, path, when=None):
@@ -1490,16 +1489,22 @@ def _add_to_bagit(self, rel_path, **checksums):
         self.add_to_manifest(rel_path, checksums)
 
     def create_job(self,
-                   wf_job,
-                   builder_job   # type: Dict
+                   builder_job,  # type: Dict[Text, Any]
+                   wf_job=None,  # type: Callable[[Dict[Text, Text], Callable[[Any, Any], Any], RuntimeContext], Generator[Any, None, None]]
+                   is_output=False
                   ):  # type: (...) -> Dict
         #TODO customise the file
         """Generate the new job object with RO specific relative paths."""
         copied = copy.deepcopy(builder_job)
-        relativised_input_objecttemp = {}  # type: Dict[Any, Any]
+        relativised_input_objecttemp = {}  # type: Dict[Text, Any]
         self._relativise_files(copied)
-        rel_path = posixpath.join(_posix_path(WORKFLOW), "primary-job.json")
-        j = json_dumps(copied, indent=4, ensure_ascii=False)
+        def jdefault(o):
+            return dict(o)
+        if is_output:
+            rel_path = posixpath.join(_posix_path(WORKFLOW), "primary-output.json")
+        else:
+            rel_path = posixpath.join(_posix_path(WORKFLOW), "primary-job.json")
+        j = json_dumps(copied, indent=4, ensure_ascii=False, default=jdefault)
         with self.write_bag_file(rel_path) as file_path:
             file_path.write(j + u"\n")
         _logger.debug(u"[provenance] Generated customised job file: %s",
@@ -1520,7 +1525,7 @@ def create_job(self,
         return self.relativised_input_object
 
     def _relativise_files(self, structure):
-        # type: (Any, Dict) -> None
+        # type: (Any, Dict[Any, Any]) -> None
         """Save any file objects into the RO and update the local paths."""
         # Base case - we found a File we need to update
         _logger.debug(u"[provenance] Relativising: %s", structure)

cwltool/workflow.py

@@ -568,7 +568,7 @@ def job(self,
                 runtimeContext.research_obj.make_fs_access = runtimeContext.make_fs_access
             if runtimeContext.toplevel:
                 # Record primary-job.json
-                runtimeContext.research_obj.create_job(self.job, builder.job)
+                runtimeContext.research_obj.create_job(builder.job, self.job)
 
         job = WorkflowJob(self, runtimeContext)
         yield job

tests/test_provenance.py

@@ -13,8 +13,7 @@
 import arcp
 import pytest
 from rdflib import Graph, Literal, Namespace, URIRef
-from rdflib.namespace import DC, DCTERMS, FOAF, RDF, RDFS, SKOS, XSD
-from six import StringIO
+from rdflib.namespace import DC, DCTERMS, RDF
 from six.moves import urllib
 
 import bagit
@@ -82,6 +81,7 @@ def test_hello_single_tool(self):
     def test_revsort_workflow(self):
         self.cwltool(get_data('tests/wf/revsort.cwl'),
             get_data('tests/wf/revsort-job.json'))
+        self.check_output_object()
         self.check_provenance()
 
     def test_nested_workflow(self):
@@ -182,6 +182,17 @@ def test_directory_workflow(self):
             self.assertTrue(os.path.isfile(p),
                 "Could not find %s as %s" % (l, p))
 
+    def check_output_object(self):
+        output_obj = os.path.join(self.folder, "workflow", "primary-output.json")
+        compare_checksum = "sha1$b9214658cc453331b62c2282b772a5c063dbd284"
+        compare_location = "../data/b9/b9214658cc453331b62c2282b772a5c063dbd284"
+        with open(output_obj) as fp:
+            out_json = json.load(fp)
+        f1 = out_json["output"]
+        self.assertEqual(f1["checksum"], compare_checksum)
+        self.assertEqual(f1["location"], compare_location)
+
+
     def check_secondary_files(self):
         foo_data = os.path.join(self.folder, "data",
             # checksum as returned from:
@@ -316,7 +327,7 @@ def check_ro(self, nested=False):
             f = "metadata/provenance/primary.cwlprov.%s" % ext
             self.assertTrue(f in paths, "provenance file missing " + f)
 
-        for f in ["workflow/primary-job.json", "workflow/packed.cwl"]:
+        for f in ["workflow/primary-job.json", "workflow/packed.cwl", "workflow/primary-output.json"]:
             self.assertTrue(f in paths, "workflow file missing " + f)
         # Can't test snapshot/ files directly as their name varies
 
@@ -325,6 +336,7 @@ def check_ro(self, nested=False):
 
         packed = urllib.parse.urljoin(arcp_root, "/workflow/packed.cwl")
         primary_job = urllib.parse.urljoin(arcp_root, "/workflow/primary-job.json")
+        output_object = urllib.parse.urljoin(arcp_root, "/workflow/primary-output.json")
         primary_prov_nt = urllib.parse.urljoin(arcp_root, "/metadata/provenance/primary.cwlprov.nt")
         uuid = arcp.parse_arcp(arcp_root).uuid