Update workflow extraction backend for output collections.

lib/galaxy/tools/__init__.py

@@ -314,6 +314,7 @@ class ToolOutputCollection( ToolOutputBase ):
 
         self.metadata_source = default_metadata_source
         self.format_source = default_format_source
+        self.change_format = []  # TODO
 
     def known_outputs( self, inputs, type_registry ):
         if self.dynamic_structure:

lib/galaxy/workflow/extract.py

@@ -11,6 +11,7 @@ from galaxy.tools.parameters.basic import (
     SelectToolParameter,
     UnvalidatedValue
 )
+from galaxy.tools import ToolOutputCollectionPart
 from galaxy.tools.parameters.grouping import (
     Conditional,
     Repeat
@@ -115,6 +116,8 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_
                 input_collection = an_implicit_output_collection.find_implicit_input_collection( input_name )
                 if input_collection:
                     other_hid = input_collection.hid
+                else:
+                    log.info("Cannot find implicit input collection for %s" % input_name)
             if other_hid in hid_to_output_pair:
                 other_step, other_name = hid_to_output_pair[ other_hid ]
                 conn = model.WorkflowStepConnection()
@@ -126,18 +129,26 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_
         steps.append( step )
         steps_by_job_id[ job_id ] = step
         # Store created dataset hids
-        for assoc in job.output_datasets:
+        for assoc in (job.output_datasets + job.output_dataset_collection_instances):
+            assoc_name = assoc.name
+            if ToolOutputCollectionPart.is_named_collection_part_name( assoc_name ):
+                continue
             if job in summary.implicit_map_jobs:
                 hid = None
                 for implicit_pair in jobs[ job ]:
                     query_assoc_name, dataset_collection = implicit_pair
-                    if query_assoc_name == assoc.name:
+                    if query_assoc_name == assoc_name:
                         hid = dataset_collection.hid
                 if hid is None:
-                    log.warn("Failed to find matching implicit job.")
+                    template = "Failed to find matching implicit job - job is %s, jobs are %s, assoc_name is %s."
+                    message = template % ( job.id, jobs, assoc.name )
+                    log.warn( message )
                     raise Exception( "Failed to extract job." )
             else:
-                hid = assoc.dataset.hid
+                if hasattr( assoc, "dataset" ):
+                    hid = assoc.dataset.hid
+                else:
+                    hid = assoc.dataset_collection_instance.hid
             hid_to_output_pair[ hid ] = ( step, assoc.name )
     return steps
 
@@ -210,11 +221,18 @@ class WorkflowSummary( object ):
         dataset_collection = content
         hid = content.hid
         self.collection_types[ hid ] = content.collection.collection_type
-        if not content.implicit_output_name:
-            job = DatasetCollectionCreationJob( content )
-            self.jobs[ job ] = [ ( None, content ) ]
-        else:
-            dataset_collection = content
+        if content.creating_job_associations:
+            for assoc in content.creating_job_associations:
+                job = assoc.job
+                if job not in self.jobs or self.jobs[ job ][ 0 ][ 1 ].history_content_type == "dataset":
+                    self.jobs[ job ] = [ ( assoc.name, dataset_collection ) ]
+                    if content.implicit_output_name:
+                        self.implicit_map_jobs.append( job )
+                else:
+                    self.jobs[ job ].append( ( assoc.name, dataset_collection ) )
+        # This whole elif condition may no longer be needed do to additional
+        # tracking with creating_job_associations. Will delete at some point.
+        elif content.implicit_output_name:
             # TODO: Optimize db call
             dataset_instance = dataset_collection.collection.dataset_instances[ 0 ]
             if not self.__check_state( dataset_instance ):
@@ -235,6 +253,9 @@ class WorkflowSummary( object ):
                     self.implicit_map_jobs.append( job )
                 else:
                     self.jobs[ job ].append( ( assoc.name, dataset_collection ) )
+        else:
+            job = DatasetCollectionCreationJob( content )
+            self.jobs[ job ] = [ ( None, content ) ]
 
     def __summarize_dataset( self, dataset ):
         if not self.__check_state( dataset ):

lib/galaxy/workflow/modules.py

@@ -649,7 +649,9 @@ class ToolModule( WorkflowModule ):
         data_outputs = []
         data_inputs = None
         for name, tool_output in self.tool.outputs.iteritems():
-            if tool_output.format_source != None:
+            if tool_output.collection:
+                formats = [ 'input' ]
+            elif tool_output.format_source != None:
                 formats = [ 'input' ]  # default to special name "input" which remove restrictions on connections
                 if data_inputs == None:
                     data_inputs = self.get_data_inputs()

test/api/test_workflow_extraction.py

@@ -182,6 +182,97 @@ test_data:
         collection_step_state = loads( collection_step[ "tool_state" ] )
         self.assertEquals( collection_step_state[ "collection_type" ], u"list:paired" )
 
+    def test_extract_workflow_with_output_collections( self ):
+        jobs_summary = self._run_jobs("""
+steps:
+  - label: text_input1
+    type: input
+  - label: text_input2
+    type: input
+  - label: cat_inputs
+    tool_id: cat1
+    state:
+      input1:
+        $link: text_input1
+      queries:
+        - input2:
+            $link: text_input2
+  - label: split_up
+    tool_id: collection_split_on_column
+    state:
+      input1:
+        $link: cat_inputs#out_file1
+  - tool_id: cat_list
+    state:
+      input1:
+        $link: split_up#split_output
+test_data:
+  text_input1: "samp1\t10.0\nsamp2\t20.0\n"
+  text_input2: "samp1\t30.0\nsamp2\t40.0\n"
+""")
+        tool_ids = [ "cat1", "collection_split_on_column", "cat_list" ]
+        job_ids = map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids )
+        downloaded_workflow = self._extract_and_download_workflow(
+            dataset_ids=[ "1", "2" ],
+            job_ids=job_ids,
+        )
+        self.__check_workflow(
+            downloaded_workflow,
+            step_count=5,
+            verify_connected=True,
+            data_input_count=2,
+            data_collection_input_count=0,
+            tool_ids=tool_ids,
+        )
+
+    def test_extract_with_mapped_output_collections( self ):
+        jobs_summary = self._run_jobs("""
+steps:
+  - label: text_input1
+    type: input_collection
+  - label: cat_inputs
+    tool_id: cat1
+    state:
+      input1:
+        $link: text_input1
+  - label: pair_off
+    tool_id: collection_creates_pair
+    state:
+      input1:
+        $link: cat_inputs#out_file1
+  - label: cat_pairs
+    tool_id: cat_collection
+    state:
+      input1:
+        $link: pair_off#paired_output
+  - tool_id: cat_list
+    state:
+      input1:
+        $link: cat_pairs#out_file1
+test_data:
+  text_input1:
+    type: list
+    elements:
+      - identifier: samp1
+        content: "samp1\t10.0\nsamp2\t20.0\n"
+      - identifier: samp2
+        content: "samp1\t30.0\nsamp2\t40.0\n"
+""")
+        tool_ids = [ "cat1", "collection_creates_pair", "cat_collection", "cat_list" ]
+        job_ids = map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids )
+        downloaded_workflow = self._extract_and_download_workflow(
+            dataset_collection_ids=[ "3" ],
+            job_ids=job_ids,
+        )
+        self.__check_workflow(
+            downloaded_workflow,
+            step_count=5,
+            verify_connected=True,
+            data_input_count=0,
+            data_collection_input_count=1,
+            tool_ids=tool_ids,
+        )
+
     def _job_id_for_tool( self, jobs, tool_id ):
         return self._job_for_tool( jobs, tool_id )[ "id" ]
 

test/unit/workflows/test_extract_summary.py

@@ -135,6 +135,7 @@ class MockHdca( object ):
         self.implicit_output_name = implicit_output_name
         self.hid = 1
         self.collection = model.DatasetCollection()
+        self.creating_job_associations = []
         element = model.DatasetCollectionElement(
             collection=self.collection,
             element=model.HistoryDatasetAssociation(),