Enable src/sink comparisons for records

The logic for comparing records from two different sections of the
workflow previous failed due to falling back to a `==` comparison that
would not pass due to differences in fully qualified record names.

This fix adds an explicit record comparison function that normalizes
field names by the record namespace and then ensures that all of the
types of each field are identical from both records.

Also adds a test to demonstrate the behavior.

Author: chapmanb

cwltool/workflow.py

@@ -96,6 +96,8 @@ def can_assign_src_to_sink(src, sink):  # type: (Any, Any) -> bool
     if isinstance(src, dict) and isinstance(sink, dict):
         if src["type"] == "array" and sink["type"] == "array":
             return can_assign_src_to_sink(src["items"], sink["items"])
+        elif src["type"] == "record" and sink["type"] == "record":
+            return _compare_records(src, sink)
     elif isinstance(src, list):
         for t in src:
             if can_assign_src_to_sink(t, sink):
@@ -108,6 +110,28 @@ def can_assign_src_to_sink(src, sink):  # type: (Any, Any) -> bool
         return src == sink
     return False
 
+def _compare_records(rec1, rec2):
+    """Compare two records, ensuring they have compatible fields.
+
+    This handles normalizing record names, which will be relative to workflow
+    step, so that they can be compared.
+    """
+    def _rec_fields(rec):
+        out = {}
+        for field in rec["fields"]:
+            name = field["name"].replace(rec["name"], "")
+            out[name] = field["type"]
+        return out
+    fields1 = _rec_fields(rec1)
+    fields2 = _rec_fields(rec2)
+    for key in set(fields1.keys() + fields2.keys()):
+        if fields1.get(key) != fields2.get(key):
+            _logger.info("Record comparison failure for %s and %s\n"
+                         "Did not match fields for %s: %s and %s" %
+                         (rec1["name"], rec2["name"], key, fields1.get(key), fields2.get(key)))
+            return False
+    return True
+
 def object_from_state(state, parms, frag_only, supportsMultipleInput, sourceField):
     # type: (Dict[unicode, WorkflowStateItem], List[Dict[unicode, Any]], bool, bool, unicode) -> Dict[unicode, Any]
     inputobj = {}  # type: Dict[unicode, Any]

tests/test_examples.py

@@ -168,6 +168,12 @@ def test_typecompare(self):
             {'items': ['string'], 'type': 'array'},
             {'items': ['int'], 'type': 'array'}))
 
+    def test_recordcompare(self):
+        src = {'fields': [{'type': {'items': 'string', 'type': 'array'}, 'name': u'file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/wf-variantcall.cwl#vc_rec/vc_rec/description'}, {'type': {'items': 'File', 'type': 'array'}, 'name': u'file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/wf-variantcall.cwl#vc_rec/vc_rec/vrn_file'}], 'type': 'record', 'name': u'file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/wf-variantcall.cwl#vc_rec/vc_rec'}
+        sink = {'fields': [{'type': {'items': 'string', 'type': 'array'}, 'name': u'file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/steps/vc_output_record.cwl#vc_rec/vc_rec/description'}, {'type': {'items': 'File', 'type': 'array'}, 'name': u'file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/steps/vc_output_record.cwl#vc_rec/vc_rec/vrn_file'}], 'type': 'record', 'name': u'file:///home/chapmanb/drive/work/cwl/test_bcbio_cwl/run_info-cwl-workflow/steps/vc_output_record.cwl#vc_rec/vc_rec'}
+        self.assertTrue(cwltool.workflow.can_assign_src_to_sink(src, sink))
+
+
 
 if __name__ == '__main__':
     unittest.main()