WIP: composite uploads for workflow tests...

galaxy/tools/cwl/util.py

@@ -100,6 +100,19 @@ def upload_file(file_path, secondary_files, **kwargs):
         dataset_id = dataset["id"]
         return {"src": "hda", "id": dataset_id}
 
+    def upload_file_with_composite_data(file_path, composite_data, **kwargs):
+        if file_path is not None:
+            file_path = abs_path_or_uri(file_path, test_data_directory)
+        composite_data_resolved = []
+        for cd in composite_data:
+            composite_data_resolved.append(abs_path_or_uri(cd, test_data_directory))
+        target = FileUploadTarget(file_path, composite_data=composite_data_resolved, **kwargs)
+        upload_response = upload_func(target)
+        dataset = upload_response["outputs"][0]
+        datasets.append((dataset, target))
+        dataset_id = dataset["id"]
+        return {"src": "hda", "id": dataset_id}
+
     def upload_tar(file_path):
         file_path = abs_path_or_uri(file_path, test_data_directory)
         target = DirectoryUploadTarget(file_path)
@@ -150,6 +163,12 @@ def replacement_item(value, force_to_file=False):
 
     def replacement_file(value):
         file_path = value.get("location", None) or value.get("path", None)
+
+        composite_data = value.get("composite_data", None)
+        if composite_data:
+            filetype = value.get('filetype', None)
+            return upload_file_with_composite_data(None, composite_data, filetype=filetype)
+
         if file_path is None:
             return value
 
@@ -273,6 +292,7 @@ class FileUploadTarget(object):
     def __init__(self, path, secondary_files=None, **kwargs):
         self.path = path
         self.secondary_files = secondary_files
+        self.composite_data = kwargs.get("composite_data", [])
         self.properties = kwargs
 
     def __str__(self):
@@ -315,6 +335,7 @@ def tool_response_to_output(tool_response, history_id, output_id):
 
 
 def invocation_to_output(invocation, history_id, output_id):
+    print(invocation)
     if output_id in invocation["outputs"]:
         dataset = invocation["outputs"][output_id]
         galaxy_output = GalaxyOutput(history_id, "dataset", dataset["id"])
@@ -324,6 +345,7 @@ def invocation_to_output(invocation, history_id, output_id):
     else:
         raise Exception("Failed to find output with label [%s] in [%s]" % (output_id, invocation))
 
+    print("galaxy_output is %s " % (galaxy_output,))
     return galaxy_output
 
 
@@ -335,6 +357,7 @@ def output_to_cwl_json(
     Useful in running conformance tests and implementing the cwl-runner
     interface via Galaxy.
     """
+    print("in output_to_cwl_json")
     def element_to_cwl_json(element):
         element_output = GalaxyOutput(
             galaxy_output.history_id,
@@ -414,14 +437,17 @@ def dataset_dict_to_json_content(dataset_dict):
             return properties
 
     elif output_metadata["history_content_type"] == "dataset_collection":
-        if output_metadata["collection_type"] == "list":
+        rval = None
+        collection_type = output_metadata["collection_type"].split(":", 1)[0]
+        if collection_type in ["list", "paired"]:
             rval = []
             for element in output_metadata["elements"]:
                 rval.append(element_to_cwl_json(element))
-        elif output_metadata["collection_type"] == "record":
+        elif collection_type == "record":
             rval = {}
             for element in output_metadata["elements"]:
                 rval[element["element_identifier"]] = element_to_cwl_json(element)
+        print("rval is %s %s" % (rval, output_metadata))
         return rval
     else:
         raise NotImplementedError("Unknown history content type encountered")

galaxy/tools/verify/interactor.py

@@ -120,6 +120,57 @@ def get_tool_tests(self, tool_id, tool_version=None):
         assert response.status_code == 200, "Non 200 response from tool test API. [%s]" % response.content
         return response.json()
 
+    def verify_output_collection(self, output_collection_def, output_collection_id):
+        name = output_collection_def.name
+
+        data_collection = galaxy_interactor._get("dataset_collections/%s" % output_collection_id, data={"instance_type": "history"}).json()
+
+        def get_element(elements, id):
+            for element in elements:
+                if element["element_identifier"] == id:
+                    return element
+            return False
+
+        expected_collection_type = output_collection_def.collection_type
+        if expected_collection_type:
+            collection_type = data_collection["collection_type"]
+            if expected_collection_type != collection_type:
+                template = "Expected output collection [%s] to be of type [%s], was of type [%s]."
+                message = template % (name, expected_collection_type, collection_type)
+                raise AssertionError(message)
+
+        expected_element_count = output_collection_def.count
+        if expected_element_count:
+            actual_element_count = len(data_collection["elements"])
+            if expected_element_count != actual_element_count:
+                template = "Expected output collection [%s] to have %s elements, but it had %s."
+                message = template % (name, expected_element_count, actual_element_count)
+                raise AssertionError(message)
+
+        def verify_elements(element_objects, element_tests):
+            for element_identifier, (element_outfile, element_attrib) in element_tests.items():
+                element = get_element(element_objects, element_identifier)
+                if not element:
+                    template = "Failed to find identifier [%s] for testing, tool generated collection elements [%s]"
+                    message = template % (element_identifier, element_objects)
+                    raise AssertionError(message)
+
+                element_type = element["element_type"]
+                if element_type != "dataset_collection":
+                    hda = element["object"]
+                    galaxy_interactor.verify_output_dataset(
+                        history,
+                        hda_id=hda["id"],
+                        outfile=element_outfile,
+                        attributes=element_attrib,
+                        tool_id=tool_id
+                    )
+                if element_type == "dataset_collection":
+                    elements = element["object"]["elements"]
+                    verify_elements(elements, element_attrib.get("elements", {}))
+
+        verify_elements(data_collection["elements"], output_collection_def.element_tests)
+
     def verify_output(self, history_id, jobs, output_data, output_testdef, tool_id, maxseconds):
         outfile = output_testdef.outfile
         attributes = output_testdef.attributes
@@ -847,54 +898,8 @@ def register_exception(e):
 
             # Data collection returned from submission, elements may have been populated after
             # the job completed so re-hit the API for more information.
-            data_collection_returned = data_collection_list[name]
-            data_collection = galaxy_interactor._get("dataset_collections/%s" % data_collection_returned["id"], data={"instance_type": "history"}).json()
-
-            def get_element(elements, id):
-                for element in elements:
-                    if element["element_identifier"] == id:
-                        return element
-                return False
-
-            expected_collection_type = output_collection_def.collection_type
-            if expected_collection_type:
-                collection_type = data_collection["collection_type"]
-                if expected_collection_type != collection_type:
-                    template = "Expected output collection [%s] to be of type [%s], was of type [%s]."
-                    message = template % (name, expected_collection_type, collection_type)
-                    raise AssertionError(message)
-
-            expected_element_count = output_collection_def.count
-            if expected_element_count:
-                actual_element_count = len(data_collection["elements"])
-                if expected_element_count != actual_element_count:
-                    template = "Expected output collection [%s] to have %s elements, but it had %s."
-                    message = template % (name, expected_element_count, actual_element_count)
-                    raise AssertionError(message)
-
-            def verify_elements(element_objects, element_tests):
-                for element_identifier, (element_outfile, element_attrib) in element_tests.items():
-                    element = get_element(element_objects, element_identifier)
-                    if not element:
-                        template = "Failed to find identifier [%s] for testing, tool generated collection elements [%s]"
-                        message = template % (element_identifier, element_objects)
-                        raise AssertionError(message)
-
-                    element_type = element["element_type"]
-                    if element_type != "dataset_collection":
-                        hda = element["object"]
-                        galaxy_interactor.verify_output_dataset(
-                            history,
-                            hda_id=hda["id"],
-                            outfile=element_outfile,
-                            attributes=element_attrib,
-                            tool_id=tool_id
-                        )
-                    if element_type == "dataset_collection":
-                        elements = element["object"]["elements"]
-                        verify_elements(elements, element_attrib.get("elements", {}))
-
-            verify_elements(data_collection["elements"], output_collection_def.element_tests)
+            data_collection_id = data_collection_list[name]["id"]
+            galaxy_interactor.verify_output_collection(output_collection_def, data_collection_id)
         except Exception as e:
             register_exception(e)
 
@@ -903,7 +908,6 @@ def verify_elements(element_objects, element_tests):
     else:
         return job_stdio
 
-
 def _format_stream(output, stream, format):
     output = output or ''
     if format: