[galaxy-commits] commit/galaxy-central: 6 new changesets

15 Jan 2015

6 new commits in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/commits/6ae144fd51f7/
Changeset:   6ae144fd51f7
User:        jmchilton
Date:        2015-01-15 14:30:00+00:00
Summary:     Allow tools to output collections with static or determinable structure.
By "static" I mean tools such as a FASTQ de-interlacer that would produce a "paired" collection with two datasets everytime. By "determinable" I mean tools that perform N->N operations within the same job - such as a tool that needs to normalize a bunch of datasets all at once and not in separate jobs. (For N->N collection operations that should or can be done in N separate jobs tool authors should just write tools that operate over a dataset and produce a dataset and let the end-user 'map over' that operation.)
There are still large classes of operations where the structure of the output collection cannot be pre-determined - such as splitting files (e.g. bam files by read group) - that are not implemented in this commit.
Model:
The models have been updated to do a more thorough job of tracking collection outputs. Jobs just producing HistoryDatasetCollectionAssociations works fine for simple jobs producing collections - but you don't want to map a list over a tool that produces a pair and produce a bunch of pairs HDCAs and a list:pair HDCA- you just want a bunch of pieces and the one list:pair at that the top.
Workflow:
Workflows containing such operations can be executed - but the workflow editor has not been updated to handle this complexity (and it will require a significant overhaul) so such tools are not available in the workflow editor.
Tool Testing:
This commit also introduces a new tool XML syntax for describing tests on output collections. See files test/functional/tools/collection_creates_list.xml and test/functional/tools/collection_creates_pair.xml for examples.
Tests:
Includes two tools to test this - one that uses explicit pair output names and one that iterates over the structure of input list to produce an output list.
Includes several new tools API tests that test the tools described above via the API and implicit mapping over such tools. Includes two new workflow API tests - one that verifies a simple workflow with output collections works and one that verifies mapping over workflow steps in collections works.
Affected #:  25 files
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/jobs/runners/__init__.py

--- a/lib/galaxy/jobs/runners/__init__.py
+++ b/lib/galaxy/jobs/runners/__init__.py
@@ -216,7 +216,7 @@
         job_tool = job_wrapper.tool
         for (joda, dataset) in self._walk_dataset_outputs( job ):
             if joda and job_tool:
-                hda_tool_output = job_tool.outputs.get( joda.name, None )
+                hda_tool_output = job_tool.find_output_def( joda.name )
                 if hda_tool_output and hda_tool_output.from_work_dir:
                     # Copy from working dir to HDA.
                     # TODO: move instead of copy to save time?
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/managers/collections.py
--- a/lib/galaxy/managers/collections.py
+++ b/lib/galaxy/managers/collections.py
@@ -72,7 +72,11 @@
                 for input_name, input_collection in implicit_collection_info[ "implicit_inputs" ]:
                     dataset_collection_instance.add_implicit_input_collection( input_name, input_collection )
                 for output_dataset in implicit_collection_info.get( "outputs" ):
-                    output_dataset.hidden_beneath_collection_instance = dataset_collection_instance
+                    if isinstance( output_dataset, model.HistoryDatasetCollectionAssociation ):
+                        dataset_collection_instance.add_implicit_input_collection( input_name, input_collection )
+                    else:
+                        # dataset collection, don't need to do anything...
+                        pass
                     trans.sa_session.add( output_dataset )
dataset_collection_instance.implicit_output_name = implicit_collection_info[ "implicit_output_name" ]
@@ -92,6 +96,18 @@
return self.__persist( dataset_collection_instance )
+    def create_dataset_collection(
+        self,
+        trans,
+        collection_type,
+        elements=None,
+    ):
+        return self.__create_dataset_collection(
+            trans=trans,
+            collection_type=collection_type,
+            elements=elements,
+        )
+
     def __create_dataset_collection(
         self,
         trans,
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/model/__init__.py
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -329,6 +329,7 @@
         self.input_datasets = []
         self.output_datasets = []
         self.input_dataset_collections = []
+        self.output_dataset_collection_instances = []
         self.output_dataset_collections = []
         self.input_library_datasets = []
         self.output_library_datasets = []
@@ -472,8 +473,10 @@
         self.output_datasets.append( JobToOutputDatasetAssociation( name, dataset ) )
     def add_input_dataset_collection( self, name, dataset ):
         self.input_dataset_collections.append( JobToInputDatasetCollectionAssociation( name, dataset ) )
-    def add_output_dataset_collection( self, name, dataset ):
-        self.output_dataset_collections.append( JobToOutputDatasetCollectionAssociation( name, dataset ) )
+    def add_output_dataset_collection( self, name, dataset_collection_instance ):
+        self.output_dataset_collection_instances.append( JobToOutputDatasetCollectionAssociation( name, dataset_collection_instance ) )
+    def add_implicit_output_dataset_collection( self, name, dataset_collection ):
+        self.output_dataset_collections.append( JobToImplicitOutputDatasetCollectionAssociation( name, dataset_collection ) )
     def add_input_library_dataset( self, name, dataset ):
         self.input_library_datasets.append( JobToInputLibraryDatasetAssociation( name, dataset ) )
     def add_output_library_dataset( self, name, dataset ):
@@ -734,7 +737,18 @@
         self.dataset = dataset
+# Many jobs may map to one HistoryDatasetCollection using these for a given
+# tool output (if mapping over an input collection).
 class JobToOutputDatasetCollectionAssociation( object ):
+    def __init__( self, name, dataset_collection_instance ):
+        self.name = name
+        self.dataset_collection_instance = dataset_collection_instance
+
+
+# A DatasetCollection will be mapped to at most one job per tool output
+# using these. (You can think of many of these models as going into the
+# creation of a JobToOutputDatasetCollectionAssociation.)
+class JobToImplicitOutputDatasetCollectionAssociation( object ):
     def __init__( self, name, dataset_collection ):
         self.name = name
         self.dataset_collection = dataset_collection
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/model/mapping.py
--- a/lib/galaxy/model/mapping.py
+++ b/lib/galaxy/model/mapping.py
@@ -448,6 +448,12 @@
     Column( "dataset_collection_id", Integer, ForeignKey( "history_dataset_collection_association.id" ), index=True ),
     Column( "name", Unicode(255) ) )
+model.JobToImplicitOutputDatasetCollectionAssociation.table = Table( "job_to_implicit_output_dataset_collection", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "job_id", Integer, ForeignKey( "job.id" ), index=True ),
+    Column( "dataset_collection_id", Integer, ForeignKey( "dataset_collection.id" ), index=True ),
+    Column( "name", Unicode(255) ) )
+
 model.JobToOutputDatasetCollectionAssociation.table = Table( "job_to_output_dataset_collection", metadata,
     Column( "id", Integer, primary_key=True ),
     Column( "job_id", Integer, ForeignKey( "job.id" ), index=True ),
@@ -1809,8 +1815,13 @@
mapper( model.JobToOutputDatasetCollectionAssociation,
         model.JobToOutputDatasetCollectionAssociation.table, properties=dict(
+            job=relation( model.Job ), dataset_collection_instance=relation(
+                model.HistoryDatasetCollectionAssociation, lazy=False, backref="output_dataset_collection_instances" ) ) )
+
+mapper( model.JobToImplicitOutputDatasetCollectionAssociation,
+        model.JobToImplicitOutputDatasetCollectionAssociation.table, properties=dict(
             job=relation( model.Job ), dataset_collection=relation(
-                model.HistoryDatasetCollectionAssociation, lazy=False ) ) )
+                model.DatasetCollection, backref="output_dataset_collections" ) ) )
mapper( model.JobToInputLibraryDatasetAssociation,
         model.JobToInputLibraryDatasetAssociation.table, properties=dict(
@@ -1895,6 +1906,8 @@
                      parameters=relation( model.JobParameter, lazy=False ),
                      input_datasets=relation( model.JobToInputDatasetAssociation ),
                      output_datasets=relation( model.JobToOutputDatasetAssociation ),
+                     output_dataset_collection_instances=relation( model.JobToOutputDatasetCollectionAssociation ),
+                     output_dataset_collections=relation( model.JobToImplicitOutputDatasetCollectionAssociation ),
                      post_job_actions=relation( model.PostJobActionAssociation, lazy=False ),
                      input_library_datasets=relation( model.JobToInputLibraryDatasetAssociation ),
                      output_library_datasets=relation( model.JobToOutputLibraryDatasetAssociation ),
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/model/migrate/versions/0127_output_collection_adjustments.py
--- /dev/null
+++ b/lib/galaxy/model/migrate/versions/0127_output_collection_adjustments.py
@@ -0,0 +1,63 @@
+"""
+Migration script updating collections tables for output collections.
+"""
+
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from migrate import *
+from migrate.changeset import *
+from galaxy.model.custom_types import *
+
+import datetime
+now = datetime.datetime.utcnow
+
+import logging
+log = logging.getLogger( __name__ )
+
+metadata = MetaData()
+
+JobToImplicitOutputDatasetCollectionAssociation_table = Table(
+    "job_to_implicit_output_dataset_collection", metadata,
+    Column( "id", Integer, primary_key=True ),
+    Column( "job_id", Integer, ForeignKey( "job.id" ), index=True ),
+    Column( "dataset_collection_id", Integer, ForeignKey( "dataset_collection.id" ), index=True ),
+    Column( "name", Unicode(255) )
+)
+
+
+TABLES = [
+    JobToImplicitOutputDatasetCollectionAssociation_table,
+]
+
+
+def upgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    print __doc__
+    metadata.reflect()
+
+    for table in TABLES:
+        __create(table)
+
+
+def downgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    metadata.reflect()
+
+    for table in TABLES:
+        __drop(table)
+
+
+def __create(table):
+    try:
+        table.create()
+    except Exception as e:
+        print str(e)
+        log.exception("Creating %s table failed: %s" % (table.name, str( e ) ) )
+
+
+def __drop(table):
+    try:
+        table.drop()
+    except Exception as e:
+        print str(e)
+        log.exception("Dropping %s table failed: %s" % (table.name, str( e ) ) )
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -222,7 +222,18 @@
         return new_state
-class ToolOutput( object, Dictifiable ):
+class ToolOutputBase( object, Dictifiable ):
+
+    def __init__( self, name, label=None, filters=None, hidden=False ):
+        super( ToolOutputBase, self ).__init__()
+        self.name = name
+        self.label = label
+        self.filters = filters or []
+        self.hidden = hidden
+        self.collection = False
+
+
+class ToolOutput( ToolOutputBase ):
     """
     Represents an output datasets produced by a tool. For backward
     compatibility this behaves as if it were the tuple::
@@ -233,16 +244,18 @@
     dict_collection_visible_keys = ( 'name', 'format', 'label', 'hidden' )
def __init__( self, name, format=None, format_source=None, metadata_source=None,
-                  parent=None, label=None, filters=None, actions=None, hidden=False ):
-        self.name = name
+                  parent=None, label=None, filters=None, actions=None, hidden=False,
+                  implicit=False ):
+        super( ToolOutput, self ).__init__( name, label=label, filters=filters, hidden=hidden )
         self.format = format
         self.format_source = format_source
         self.metadata_source = metadata_source
         self.parent = parent
-        self.label = label
-        self.filters = filters or []
         self.actions = actions
-        self.hidden = hidden
+
+        # Initialize default values
+        self.change_format = []
+        self.implicit = implicit
# Tuple emulation
@@ -263,6 +276,80 @@
         return iter( ( self.format, self.metadata_source, self.parent ) )
+class ToolOutputCollection( ToolOutputBase ):
+    """
+    Represents a HistoryDatasetCollectionAssociation of  output datasets produced by a tool.
+    <outputs>
+      <dataset_collection type="list" label="${tool.name} on ${on_string} fasta">
+        <discover_datasets pattern="__name__" ext="fasta" visible="True" directory="outputFiles" />
+      </dataset_collection>
+      <dataset_collection type="paired" label="${tool.name} on ${on_string} paired reads">
+        <data name="forward" format="fastqsanger" />
+        <data name="reverse" format="fastqsanger"/>
+      </dataset_collection>
+    <outputs>
+    """
+
+    def __init__( self, name, structure, label=None, filters=None, hidden=False, default_format="data" ):
+        super( ToolOutputCollection, self ).__init__( name, label=label, filters=filters, hidden=hidden )
+        self.collection = True
+        self.default_format = default_format
+        self.structure = structure
+        self.outputs = odict()
+
+    def known_outputs( self, inputs ):
+        def to_part( ( element_identifier, output ) ):
+            return ToolOutputCollectionPart( self, element_identifier, output )
+
+        # This line is probably not right - should verify structured_like
+        # or have outputs and all outputs have name.
+        if not self.structure.structured_like and self.outputs:
+            outputs = self.outputs
+        else:
+            # TODO: Handle nested structures.
+            input_collection = inputs[ self.structure.structured_like ]
+            outputs = odict()
+            for element in input_collection.collection.elements:
+                name = element.element_identifier
+                output = ToolOutput( name, format=self.default_format, implicit=True )
+                outputs[ element.element_identifier ] = output
+
+        return map( to_part, outputs.items() )
+
+
+class ToolOutputCollectionStructure( object ):
+
+    def __init__( self, collection_type=None, structured_like=None ):
+        self.collection_type = collection_type
+        self.structured_like = structured_like
+        if collection_type is None and structured_like is None:
+            raise ValueError( "Output collection types must be specify type of structured_like" )
+
+
+class ToolOutputCollectionPart( object ):
+
+    def __init__( self, output_collection_def, element_identifier, output_def ):
+        self.output_collection_def = output_collection_def
+        self.element_identifier = element_identifier
+        self.output_def = output_def
+
+    @property
+    def effective_output_name( self ):
+        name = self.output_collection_def.name
+        part_name = self.element_identifier
+        effective_output_name = "%s|__part__|%s" % ( name, part_name )
+        return effective_output_name
+
+    @staticmethod
+    def is_named_collection_part_name( name ):
+        return "|__part__|" in name
+
+    @staticmethod
+    def split_output_name( name ):
+        assert ToolOutputCollectionPart.is_named_collection_part_name( name )
+        return name.split("|__part__|")
+
+
 class Tool( object, Dictifiable ):
     """
     Represents a computational tool that can be executed through Galaxy.
@@ -700,9 +787,7 @@
         """
         Parse <outputs> elements and fill in self.outputs (keyed by name)
         """
-        self.outputs = odict()
-        for output in tool_source.parse_outputs(self):
-            self.outputs[ output.name ] = output
+        self.outputs, self.output_collections = tool_source.parse_outputs(self)
# TODO: Include the tool's name in any parsing warnings.
     def parse_stdio( self, tool_source ):
@@ -930,6 +1015,19 @@
         while len( self.__help_by_page ) < self.npages:
             self.__help_by_page.append( self.__help )
+    def find_output_def( self, name ):
+        # name is JobToOutputDatasetAssociation name.
+        # TODO: to defensive, just throw IndexError and catch somewhere
+        # up that stack.
+        if ToolOutputCollectionPart.is_named_collection_part_name( name ):
+            collection_name, part = ToolOutputCollectionPart.split_output_name( name )
+            collection_def = self.output_collections.get( collection_name, None )
+            if not collection_def:
+                return None
+            return collection_def.outputs.get( part, None )
+        else:
+            return self.outputs.get( name, None )
+
     def check_workflow_compatible( self, tool_source ):
         """
         Determine if a tool can be used in workflows. External tools and the
@@ -1109,6 +1207,7 @@
                         num_jobs=len( execution_tracker.successful_jobs ),
                         job_errors=execution_tracker.execution_errors,
                         jobs=execution_tracker.successful_jobs,
+                        output_collections=execution_tracker.output_collections,
                         implicit_collections=execution_tracker.implicit_collections,
                     )
                 else:
@@ -1122,14 +1221,14 @@
     def __should_refresh_state( self, incoming ):
         return not( 'runtool_btn' in incoming or 'URL' in incoming or 'ajax_upload' in incoming )
-    def handle_single_execution( self, trans, rerun_remap_job_id, params, history ):
+    def handle_single_execution( self, trans, rerun_remap_job_id, params, history, mapping_over_collection ):
         """
         Return a pair with whether execution is successful as well as either
         resulting output data or an error message indicating the problem.
         """
         try:
             params = self.__remove_meta_properties( params )
-            job, out_data = self.execute( trans, incoming=params, history=history, rerun_remap_job_id=rerun_remap_job_id )
+            job, out_data = self.execute( trans, incoming=params, history=history, rerun_remap_job_id=rerun_remap_job_id, mapping_over_collection=mapping_over_collection )
         except httpexceptions.HTTPFound, e:
             #if it's a paste redirect exception, pass it up the stack
             raise e
@@ -2756,6 +2855,15 @@
         return inputs
+class TestCollectionOutputDef( object ):
+
+    def __init__( self, name, attrib, element_tests ):
+        self.name = name
+        self.collection_type = attrib.get( "type", None )
+        self.attrib = attrib
+        self.element_tests = element_tests
+
+
 def json_fix( val ):
     if isinstance( val, list ):
         return [ json_fix( v ) for v in val ]
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/actions/__init__.py
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -149,7 +149,7 @@
         tool.visit_inputs( param_values, visitor )
         return input_dataset_collections
-    def execute(self, tool, trans, incoming={}, return_job=False, set_output_hid=True, set_output_history=True, history=None, job_params=None, rerun_remap_job_id=None):
+    def execute(self, tool, trans, incoming={}, return_job=False, set_output_hid=True, set_output_history=True, history=None, job_params=None, rerun_remap_job_id=None, mapping_over_collection=False):
         """
         Executes a tool, creating job and tool outputs, associating them, and
         submitting the job to the job queue. If history is not specified, use
@@ -160,6 +160,8 @@
             history = tool.get_default_history_by_trans( trans, create=True )
out_data = odict()
+        out_collections = {}
+        out_collection_instances = {}
         # Track input dataset collections - but replace with simply lists so collect
         # input datasets can process these normally.
         inp_dataset_collections = self.collect_input_dataset_collections( tool, incoming )
@@ -264,10 +266,50 @@
                 output.actions.apply_action( data, output_action_params )
             # Store all changes to database
             trans.sa_session.flush()
+            return data
for name, output in tool.outputs.items():
             if not filter_output(output, incoming):
-                handle_output( name, output )
+                if output.collection:
+                    collections_manager = trans.app.dataset_collections_service
+
+                    # As far as I can tell - this is always true - but just verify
+                    assert set_output_history, "Cannot create dataset collection for this kind of tool."
+
+                    elements = odict()
+                    input_collections = dict( [ (k, v[0]) for k, v in inp_dataset_collections.iteritems() ] )
+                    known_outputs = output.known_outputs( input_collections )
+                    # Just to echo TODO elsewhere - this should be restructured to allow
+                    # nested collections.
+                    for output_part_def in known_outputs:
+                        effective_output_name = output_part_def.effective_output_name
+                        element = handle_output( effective_output_name, output_part_def.output_def )
+                        # Following hack causes dataset to no be added to history...
+                        child_dataset_names.add( effective_output_name )
+
+                        elements[ output_part_def.element_identifier ] = element
+
+                    if mapping_over_collection:
+                        dc = collections_manager.create_dataset_collection(
+                            trans,
+                            collection_type=output.structure.collection_type,
+                            elements=elements,
+                        )
+                        out_collections[ name ] = dc
+                    else:
+                        hdca_name = self.get_output_name( output, None, tool, on_text, trans, incoming, history, wrapped_params.params, job_params )
+                        hdca = collections_manager.create(
+                            trans,
+                            history,
+                            name=hdca_name,
+                            collection_type=output.structure.collection_type,
+                            elements=elements,
+                        )
+                        # name here is name of the output element - not name
+                        # of the hdca.
+                        out_collection_instances[ name ] = hdca
+                else:
+                    handle_output( name, output )
         # Add all the top-level (non-child) datasets to the history unless otherwise specified
         for name in out_data.keys():
             if name not in child_dataset_names and name not in incoming:  # don't add children; or already existing datasets, i.e. async created
@@ -322,6 +364,10 @@
                 job.add_input_dataset( name, None )
         for name, dataset in out_data.iteritems():
             job.add_output_dataset( name, dataset )
+        for name, dataset_collection in out_collections.iteritems():
+            job.add_implicit_output_dataset_collection( name, dataset_collection )
+        for name, dataset_collection_instance in out_collection_instances.iteritems():
+            job.add_output_dataset_collection( name, dataset_collection_instance )
         job.object_store_id = object_store_populator.object_store_id
         if job_params:
             job.params = dumps( job_params )
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/evaluation.py
--- a/lib/galaxy/tools/evaluation.py
+++ b/lib/galaxy/tools/evaluation.py
@@ -59,6 +59,9 @@
         inp_data.update( [ ( da.name, da.dataset ) for da in job.input_library_datasets ] )
         out_data.update( [ ( da.name, da.dataset ) for da in job.output_library_datasets ] )
+        out_collections = dict( [ ( obj.name, obj.dataset_collection_instance ) for obj in job.output_dataset_collection_instances ] )
+        out_collections.update( [ ( obj.name, obj.dataset_collection ) for obj in job.output_dataset_collections ] )
+
         if get_special:
# Set up output dataset association for export history jobs. Because job
@@ -84,6 +87,7 @@
             incoming,
             inp_data,
             out_data,
+            output_collections=out_collections,
             output_paths=compute_environment.output_paths(),
             job_working_directory=compute_environment.working_directory(),
             input_paths=compute_environment.input_paths()
@@ -98,7 +102,7 @@
self.param_dict = param_dict
-    def build_param_dict( self, incoming, input_datasets, output_datasets, output_paths, job_working_directory, input_paths=[] ):
+    def build_param_dict( self, incoming, input_datasets, output_datasets, output_collections, output_paths, job_working_directory, input_paths=[] ):
         """
         Build the dictionary of parameters for substituting into the command
         line. Each value is wrapped in a `InputValueWrapper`, which allows
@@ -115,6 +119,7 @@
         self.__populate_wrappers(param_dict, input_dataset_paths)
         self.__populate_input_dataset_wrappers(param_dict, input_datasets, input_dataset_paths)
         self.__populate_output_dataset_wrappers(param_dict, output_datasets, output_paths, job_working_directory)
+        self.__populate_output_collection_wrappers(param_dict, output_collections, output_paths, job_working_directory)
         self.__populate_unstructured_path_rewrites(param_dict)
         # Call param dict sanitizer, before non-job params are added, as we don't want to sanitize filenames.
         self.__sanitize_param_dict( param_dict )
@@ -269,6 +274,35 @@
                 for child in data.children:
                     param_dict[ "_CHILD___%s___%s" % ( name, child.designation ) ] = DatasetFilenameWrapper( child )
+    def __populate_output_collection_wrappers(self, param_dict, output_collections, output_paths, job_working_directory):
+        output_dataset_paths = dataset_path_rewrites( output_paths )
+        tool = self.tool
+        for name, out_collection in output_collections.items():
+            if name not in tool.output_collections:
+                continue
+                #message_template = "Name [%s] not found in tool.output_collections %s"
+                #message = message_template % ( name, tool.output_collections )
+                #raise AssertionError( message )
+
+            wrapper_kwds = dict(
+                datatypes_registry=self.app.datatypes_registry,
+                dataset_paths=output_dataset_paths,
+                tool=tool,
+                name=name
+            )
+            wrapper = DatasetCollectionWrapper(
+                out_collection,
+                **wrapper_kwds
+            )
+            param_dict[ name ] = wrapper
+            # TODO: Handle nested collections...
+            output_def = tool.output_collections[ name ]
+            for element_identifier, output_def in output_def.outputs.items():
+                if not output_def.implicit:
+                    dataset_wrapper = wrapper[ element_identifier ]
+                    param_dict[ output_def.name ] = dataset_wrapper
+                    log.info("Updating param_dict for %s with %s" % (output_def.name, dataset_wrapper) )
+
     def __populate_output_dataset_wrappers(self, param_dict, output_datasets, output_paths, job_working_directory):
         output_dataset_paths = dataset_path_rewrites( output_paths )
         for name, hda in output_datasets.items():
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/execute.py
--- a/lib/galaxy/tools/execute.py
+++ b/lib/galaxy/tools/execute.py
@@ -4,6 +4,7 @@
 collections from matched collections.
 """
 import collections
+import galaxy.tools
 from galaxy.tools.actions import on_text_for_names
import logging
@@ -23,7 +24,7 @@
             # Only workflow invocation code gets to set this, ignore user supplied
             # values or rerun parameters.
             del params[ '__workflow_invocation_uuid__' ]
-        job, result = tool.handle_single_execution( trans, rerun_remap_job_id, params, history )
+        job, result = tool.handle_single_execution( trans, rerun_remap_job_id, params, history, collection_info )
         if job:
             execution_tracker.record_success( job, result )
         else:
@@ -46,6 +47,7 @@
         self.failed_jobs = 0
         self.execution_errors = []
         self.output_datasets = []
+        self.output_collections = []
         self.outputs_by_output_name = collections.defaultdict(list)
         self.implicit_collections = {}
@@ -53,7 +55,15 @@
         self.successful_jobs.append( job )
         self.output_datasets.extend( outputs )
         for output_name, output_dataset in outputs:
+            if galaxy.tools.ToolOutputCollectionPart.is_named_collection_part_name( output_name ):
+                # Skip known collection outputs, these will be covered by
+                # output collections.
+                continue
             self.outputs_by_output_name[ output_name ].append( output_dataset )
+        for job_output in job.output_dataset_collections:
+            self.outputs_by_output_name[ job_output.name ].append( job_output.dataset_collection )
+        for job_output in job.output_dataset_collection_instances:
+            self.output_collections.append( ( job_output.name, job_output.dataset_collection_instance ) )
def record_error( self, error ):
         self.failed_jobs += 1
@@ -120,6 +130,11 @@
                 collection_type=collection_type,
                 implicit_collection_info=implicit_collection_info,
             )
+            for job in self.successful_jobs:
+                # TODO: Think through this, may only want this for output
+                # collections - or we may be already recording data in some
+                # other way.
+                job.add_output_dataset_collection( output_name, collection )
             collections[ output_name ] = collection
self.implicit_collections = collections
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/parameters/basic.py
--- a/lib/galaxy/tools/parameters/basic.py
+++ b/lib/galaxy/tools/parameters/basic.py
@@ -2088,7 +2088,6 @@
         for instance should I just be checking dynamic options).
         """
         allow = True
-        # TODO: allow should be false in some in cases...
         return allow
def _options_filter_attribute( self, value ):
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/parser/interface.py
--- a/lib/galaxy/tools/parser/interface.py
+++ b/lib/galaxy/tools/parser/interface.py
@@ -130,7 +130,8 @@
@abstractmethod
     def parse_outputs(self, tool):
-        """ Return a list of ToolOutput objects.
+        """ Return a pair of output and output collections ordered
+        dictionaries for use by Tool.
         """
@abstractmethod
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/parser/xml.py
--- a/lib/galaxy/tools/parser/xml.py
+++ b/lib/galaxy/tools/parser/xml.py
@@ -12,6 +12,7 @@
     InputSource,
 )
 from galaxy.util import string_as_bool, xml_text, xml_to_string
+from galaxy.util.odict import odict
 from galaxy.tools.deps import requirements
 import galaxy.tools
 from galaxy.tools.parameters import output_collect
@@ -133,17 +134,54 @@
def parse_outputs(self, tool):
         out_elem = self.root.find("outputs")
+        outputs = odict()
+        output_collections = odict()
         if out_elem is None:
-            return []
+            return outputs, output_collections
-        def _parse(data_elem):
-            return self._parse_output(data_elem, tool)
+        data_dict = odict()
-        return map(_parse, out_elem.findall("data"))
+        def _parse(data_elem, default_format="data"):
+            output_def = self._parse_output(data_elem, tool, default_format=default_format)
+            data_dict[output_def.name] = output_def
+            return output_def
-    def _parse_output(self, data_elem, tool):
+        map(_parse, out_elem.findall("data"))
+
+        for collection_elem in out_elem.findall("collection"):
+            name = collection_elem.get( "name" )
+            default_format = collection_elem.get( "format", "data" )
+            collection_type = collection_elem.get( "type", None )
+            structured_like = collection_elem.get( "structured_like", None )
+            structure = galaxy.tools.ToolOutputCollectionStructure(
+                collection_type=collection_type,
+                structured_like=structured_like,
+            )
+            output_collection = galaxy.tools.ToolOutputCollection(
+                name,
+                structure,
+                default_format=default_format
+            )
+            outputs[output_collection.name] = output_collection
+
+            for data_elem in collection_elem.findall("data"):
+                _parse( data_elem, default_format=default_format )
+
+            for data_elem in collection_elem.findall("data"):
+                output_name = data_elem.get("name")
+                data = data_dict[output_name]
+                assert data
+                del data_dict[output_name]
+                output_collection.outputs[output_name] = data
+            output_collections[ name ] = output_collection
+
+        for output_def in data_dict.values():
+            outputs[output_def.name] = output_def
+        return outputs, output_collections
+
+    def _parse_output(self, data_elem, tool, default_format):
         output = galaxy.tools.ToolOutput( data_elem.get("name") )
-        output.format = data_elem.get("format", "data")
+        output.format = data_elem.get("format", default_format)
         output.change_format = data_elem.findall("change_format")
         output.format_source = data_elem.get("format_source", None)
         output.metadata_source = data_elem.get("metadata_source", "")
@@ -185,6 +223,7 @@
 def _test_elem_to_dict(test_elem, i):
     rval = dict(
         outputs=__parse_output_elems(test_elem),
+        output_collections=__parse_output_collection_elems(test_elem),
         inputs=__parse_input_elems(test_elem, i),
         command=__parse_assert_list_from_elem( test_elem.find("assert_command") ),
         stdout=__parse_assert_list_from_elem( test_elem.find("assert_stdout") ),
@@ -232,6 +271,29 @@
     return __parse_assert_list_from_elem( assert_elem )
+def __parse_output_collection_elems( test_elem ):
+    output_collections = []
+    for output_collection_elem in test_elem.findall( "output_collection" ):
+        output_collection_def = __parse_output_collection_elem( output_collection_elem )
+        output_collections.append( output_collection_def )
+    return output_collections
+
+
+def __parse_output_collection_elem( output_collection_elem ):
+    attrib = dict( output_collection_elem.attrib )
+    name = attrib.pop( 'name', None )
+    if name is None:
+        raise Exception( "Test output collection does not have a 'name'" )
+    element_tests = {}
+    for element in output_collection_elem.findall("element"):
+        element_attrib = dict( element.attrib )
+        identifier = element_attrib.pop( 'name', None )
+        if identifier is None:
+            raise Exception( "Test primary dataset does not have a 'identifier'" )
+        element_tests[ identifier ] = __parse_test_attributes( element, element_attrib )
+    return galaxy.tools.TestCollectionOutputDef( name, attrib, element_tests )
+
+
 def __parse_test_attributes( output_elem, attrib ):
     assert_list = __parse_assert_list( output_elem )
     file = attrib.pop( 'file', None )
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/parser/yaml.py
--- a/lib/galaxy/tools/parser/yaml.py
+++ b/lib/galaxy/tools/parser/yaml.py
@@ -5,6 +5,7 @@
from galaxy.tools.deps import requirements
 from galaxy.tools.parameters import output_collect
+from galaxy.util.odict import odict
 import galaxy.tools
@@ -76,7 +77,11 @@
         output_defs = []
         for name, output_dict in outputs.items():
             output_defs.append(self._parse_output(tool, name, output_dict))
-        return output_defs
+        outputs = odict()
+        for output in output_defs:
+            outputs[output.name] = output
+        # TODO: parse outputs collections
+        return output_defs, odict()
def _parse_output(self, tool, name, output_dict):
         # TODO: handle filters, actions, change_format
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/test.py
--- a/lib/galaxy/tools/test.py
+++ b/lib/galaxy/tools/test.py
@@ -123,6 +123,7 @@
self.inputs = self.__process_raw_inputs( self.tool.inputs, test_dict["inputs"] )
             self.outputs = test_dict["outputs"]
+            self.output_collections = test_dict["output_collections"]
             num_outputs = test_dict.get( 'expect_num_outputs', None )
             if num_outputs:
                 num_outputs = int( num_outputs )
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/tools/wrappers.py
--- a/lib/galaxy/tools/wrappers.py
+++ b/lib/galaxy/tools/wrappers.py
@@ -292,10 +292,13 @@
             # It is a HistoryDatasetCollectionAssociation
             collection = has_collection.collection
             self.name = has_collection.name
-        else:
+        elif hasattr( has_collection, "child_collection" ):
             # It is a DatasetCollectionElement instance referencing another collection
             collection = has_collection.child_collection
             self.name = has_collection.element_identifier
+        else:
+            collection = has_collection
+            self.name = None
elements = collection.elements
         element_instances = odict.odict()
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/webapps/galaxy/api/tools.py
--- a/lib/galaxy/webapps/galaxy/api/tools.py
+++ b/lib/galaxy/webapps/galaxy/api/tools.py
@@ -192,6 +192,7 @@
         output_datasets = vars.get( 'out_data', [] )
         rval = {
             "outputs": [],
+            "output_collections": [],
             "jobs": [],
             "implicit_collections": [],
         }
@@ -214,6 +215,12 @@
         for job in vars.get('jobs', []):
             rval[ 'jobs' ].append( self.encode_all_ids( trans, job.to_dict( view='collection' ), recursive=True ) )
+        for output_name, collection_instance in vars.get('output_collections', []):
+            history = target_history or trans.history
+            output_dict = dictify_dataset_collection_instance( collection_instance, security=trans.security, parent=history )
+            output_dict[ 'output_name' ] = output_name
+            rval[ 'output_collections' ].append( output_dict )
+
         for output_name, collection_instance in vars.get( 'implicit_collections', {} ).iteritems():
             history = target_history or trans.history
             output_dict = dictify_dataset_collection_instance( collection_instance, security=trans.security, parent=history )
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b lib/galaxy/workflow/modules.py
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -821,6 +821,7 @@
             step_outputs = dict( execution_tracker.implicit_collections )
         else:
             step_outputs = dict( execution_tracker.output_datasets )
+            step_outputs.update( execution_tracker.output_collections )
         progress.set_step_outputs( step, step_outputs )
         jobs = execution_tracker.successful_jobs
         for job in jobs:
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/api/test_tools.py
--- a/test/api/test_tools.py
+++ b/test/api/test_tools.py
@@ -215,6 +215,67 @@
         contents = self.dataset_populator.get_history_dataset_content( history_id, hid=4 )
         assert contents.strip() == "123\n456", contents
+    @skip_without_tool( "collection_creates_pair" )
+    def test_paired_collection_output( self ):
+        history_id = self.dataset_populator.new_history()
+        new_dataset1 = self.dataset_populator.new_dataset( history_id, content='123\n456\n789\n0ab' )
+        inputs = {
+            "input1": {"src": "hda", "id": new_dataset1["id"]},
+        }
+        # TODO: shouldn't need this wait
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        create = self._run( "collection_creates_pair", history_id, inputs, assert_ok=True )
+        jobs = create[ 'jobs' ]
+        implicit_collections = create[ 'implicit_collections' ]
+        collections = create[ 'output_collections' ]
+
+        self.assertEquals( len( jobs ), 1 )
+        self.assertEquals( len( implicit_collections ), 0 )
+        self.assertEquals( len( collections ), 1 )
+
+        output_collection = collections[ 0 ]
+        elements = output_collection[ "elements" ]
+        assert len( elements ) == 2
+        element0, element1 = elements
+        assert element0[ "element_identifier" ] == "forward"
+        assert element1[ "element_identifier" ] == "reverse"
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        contents0 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element0["object"]["id"])
+        assert contents0 == "123\n789\n", contents0
+        contents1 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element1["object"]["id"])
+        assert contents1 == "456\n0ab\n", contents1
+
+    @skip_without_tool( "collection_creates_list" )
+    def test_list_collection_output( self ):
+        history_id = self.dataset_populator.new_history()
+        create_response = self.dataset_collection_populator.create_list_in_history( history_id, contents=["a\nb\nc\nd", "e\nf\ng\nh"] )
+        hdca_id = create_response.json()[ "id" ]
+        inputs = {
+            "input1": { "src": "hdca", "id": hdca_id },
+        }
+        # TODO: real problem here - shouldn't have to have this wait.
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        create = self._run( "collection_creates_list", history_id, inputs, assert_ok=True )
+        jobs = create[ 'jobs' ]
+        implicit_collections = create[ 'implicit_collections' ]
+        collections = create[ 'output_collections' ]
+
+        self.assertEquals( len( jobs ), 1 )
+        self.assertEquals( len( implicit_collections ), 0 )
+        self.assertEquals( len( collections ), 1 )
+
+        output_collection = collections[ 0 ]
+        elements = output_collection[ "elements" ]
+        assert len( elements ) == 2
+        element0, element1 = elements
+        assert element0[ "element_identifier" ] == "data1"
+        assert element1[ "element_identifier" ] == "data2"
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        contents0 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element0["object"]["id"])
+        assert contents0 == "0\n", contents0
+        contents1 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element1["object"]["id"])
+        assert contents1 == "1\n", contents1
+
     @skip_without_tool( "cat1" )
     def test_run_cat1_with_two_inputs( self ):
         # Run tool with an multiple data parameter and grouping (repeat)
@@ -535,6 +596,52 @@
         self.assertEquals( len( response_object[ 'jobs' ] ), 2 )
         self.assertEquals( len( response_object[ 'implicit_collections' ] ), 1 )
+    @skip_without_tool( "collection_creates_pair" )
+    def test_map_over_collection_output( self ):
+        history_id = self.dataset_populator.new_history()
+        create_response = self.dataset_collection_populator.create_list_in_history( history_id, contents=["a\nb\nc\nd", "e\nf\ng\nh"] )
+        hdca_id = create_response.json()[ "id" ]
+        inputs = {
+            "input1|__collection_multirun__": {"src": "hdca", "id": hdca_id},
+        }
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        create = self._run( "collection_creates_pair", history_id, inputs, assert_ok=True )
+        jobs = create[ 'jobs' ]
+        implicit_collections = create[ 'implicit_collections' ]
+        self.assertEquals( len( jobs ), 2 )
+        self.assertEquals( len( implicit_collections ), 1 )
+        implicit_collection = implicit_collections[ 0 ]
+        assert implicit_collection[ "collection_type" ] == "list:paired", implicit_collection
+        outer_elements = implicit_collection[ "elements" ]
+        assert len( outer_elements ) == 2
+        element0, element1 = outer_elements
+        assert element0[ "element_identifier" ] == "data1"
+        assert element1[ "element_identifier" ] == "data2"
+
+        pair0, pair1 = element0["object"], element1["object"]
+        pair00, pair01 = pair0["elements"]
+        pair10, pair11 = pair1["elements"]
+
+        for pair in pair0, pair1:
+            assert "collection_type" in pair, pair
+            assert pair["collection_type"] == "paired", pair
+
+        pair_ids = []
+        for pair_element in pair00, pair01, pair10, pair11:
+            assert "object" in pair_element
+            pair_ids.append(pair_element["object"]["id"])
+
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        expected_contents = [
+            "a\nc\n",
+            "b\nd\n",
+            "e\ng\n",
+            "f\nh\n",
+        ]
+        for i in range(4):
+            contents = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=pair_ids[i])
+            self.assertEquals(expected_contents[i], contents)
+
     @skip_without_tool( "cat1" )
     def test_cannot_map_over_incompatible_collections( self ):
         history_id = self.dataset_populator.new_history()
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/api/test_workflows.py
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -487,6 +487,57 @@
         self._assert_status_code_is( run_workflow_response, 200 )
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+    def test_workflow_run_output_collections(self):
+        workflow_id = self._upload_yaml_workflow("""
+- label: text_input
+  type: input
+- label: split_up
+  tool_id: collection_creates_pair
+  state:
+    input1:
+      $link: text_input
+- tool_id: collection_paired_test
+  state:
+    f1:
+      $link: split_up#paired_output
+""")
+        history_id = self.dataset_populator.new_history()
+        hda1 = self.dataset_populator.new_dataset( history_id, content="a\nb\nc\nd\n" )
+        inputs = {
+            '0': self._ds_entry(hda1),
+        }
+        self.__invoke_workflow( history_id, workflow_id, inputs )
+        time.sleep( 5 )
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        self.assertEquals("a\nc\nb\nd\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
+
+    def test_workflow_run_output_collection_mapping(self):
+        workflow_id = self._upload_yaml_workflow("""
+- type: input_collection
+- tool_id: collection_creates_pair
+  state:
+    input1:
+      $link: 0
+- tool_id: collection_paired_test
+  state:
+    f1:
+      $link: 1#paired_output
+- tool_id: cat_list
+  state:
+    input1:
+      $link: 2#out1
+""")
+        history_id = self.dataset_populator.new_history()
+        hdca1 = self.dataset_collection_populator.create_list_in_history( history_id, contents=["a\nb\nc\nd\n", "e\nf\ng\nh\n"] ).json()
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        inputs = {
+            '0': self._ds_entry(hdca1),
+        }
+        self.__invoke_workflow( history_id, workflow_id, inputs )
+        time.sleep( 5 )
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        self.assertEquals("a\nc\nb\nd\ne\ng\nf\nh\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
+
     def test_workflow_request( self ):
         workflow = self.workflow_populator.load_workflow( name="test_for_queue" )
         workflow_request, history_id = self._setup_workflow_run( workflow )
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/base/interactor.py
--- a/test/base/interactor.py
+++ b/test/base/interactor.py
@@ -214,6 +214,7 @@
             return Bunch(
                 inputs=inputs_tree,
                 outputs=self.__dictify_outputs( submit_response_object ),
+                output_collections=self.__dictify_output_collections( submit_response_object ),
                 jobs=submit_response_object[ 'jobs' ],
             )
         except KeyError:
@@ -247,6 +248,12 @@
             element_identifiers.append( element )
         return element_identifiers
+    def __dictify_output_collections( self, submit_response ):
+        output_collections_dict = odict()
+        for output_collection in submit_response[ 'output_collections' ]:
+            output_collections_dict[ output_collection.get("output_name") ] = output_collection
+        return output_collections_dict
+
     def __dictify_outputs( self, datasets_object ):
         ## Convert outputs list to a dictionary that can be accessed by
         ## output_name so can be more flexiable about ordering of outputs
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/functional/test_toolbox.py
--- a/test/functional/test_toolbox.py
+++ b/test/functional/test_toolbox.py
@@ -48,6 +48,7 @@
             try:
                 tool_response = galaxy_interactor.run_tool( testdef, test_history )
                 data_list, jobs, tool_inputs = tool_response.outputs, tool_response.jobs, tool_response.inputs
+                data_collection_list = tool_response.output_collections
             except RunToolException as e:
                 tool_inputs = e.inputs
                 tool_execution_exception = e
@@ -56,10 +57,10 @@
                 tool_execution_exception = e
                 raise e
-            self.assertTrue( data_list )
+            self.assertTrue( data_list or data_collection_list )
try:
-                job_stdio = self._verify_outputs( testdef, test_history, jobs, shed_tool_id, data_list, galaxy_interactor )
+                job_stdio = self._verify_outputs( testdef, test_history, jobs, shed_tool_id, data_list, data_collection_list, galaxy_interactor )
             except JobOutputsError as e:
                 job_stdio = e.job_stdio
                 job_output_exceptions = e.output_exceptions
@@ -92,7 +93,7 @@
             else:
                 raise Exception( "Test parse failure" )
-    def _verify_outputs( self, testdef, history, jobs, shed_tool_id, data_list, galaxy_interactor ):
+    def _verify_outputs( self, testdef, history, jobs, shed_tool_id, data_list, data_collection_list, galaxy_interactor ):
         assert len(jobs) == 1, "Test framework logic error, somehow tool test resulted in more than one job."
         job = jobs[ 0 ]
@@ -106,6 +107,14 @@
                 raise Exception( message )
         found_exceptions = []
+        def register_exception(e):
+            if not found_exceptions:
+                # Only print this stuff out once.
+                for stream in ['stdout', 'stderr']:
+                    if stream in job_stdio:
+                        print >>sys.stderr, self._format_stream( job_stdio[ stream ], stream=stream, format=True )
+            found_exceptions.append(e)
+
         if testdef.expect_failure:
             if testdef.outputs:
                 raise Exception("Cannot specify outputs in a test expecting failure.")
@@ -120,18 +129,18 @@
             if not testdef.expect_failure:
                 found_exceptions.append(e)
+        job_stdio = galaxy_interactor.get_job_stdio( job[ 'id' ] )
+
         if not job_failed and testdef.expect_failure:
             error = AssertionError("Expected job to fail but Galaxy indicated the job successfully completed.")
-            found_exceptions.append(error)
-
-        job_stdio = galaxy_interactor.get_job_stdio( job[ 'id' ] )
+            register_exception(error)
expect_exit_code = testdef.expect_exit_code
         if expect_exit_code is not None:
             exit_code = job_stdio["exit_code"]
             if str(expect_exit_code) != str(exit_code):
                 error = AssertionError("Expected job to complete with exit code %s, found %s" % (expect_exit_code, exit_code))
-                found_exceptions.append(error)
+                register_exception(error)
for output_index, output_tuple in enumerate(testdef.outputs):
             # Get the correct hid
@@ -151,12 +160,7 @@
             try:
                 galaxy_interactor.verify_output( history, jobs, output_data, output_testdef=output_testdef, shed_tool_id=shed_tool_id, maxseconds=maxseconds )
             except Exception as e:
-                if not found_exceptions:
-                    # Only print this stuff out once.
-                    for stream in ['stdout', 'stderr']:
-                        if stream in job_stdio:
-                            print >>sys.stderr, self._format_stream( job_stdio[ stream ], stream=stream, format=True )
-                found_exceptions.append(e)
+                register_exception(e)
other_checks = {
             "command_line": "Command produced by the job",
@@ -171,7 +175,48 @@
                 except AssertionError, err:
                     errmsg = '%s different than expected\n' % description
                     errmsg += str( err )
-                    found_exceptions.append( AssertionError( errmsg ) )
+                    register_exception( AssertionError( errmsg ) )
+
+        for output_collection_def in testdef.output_collections:
+            try:
+                name = output_collection_def.name
+                # TODO: data_collection_list is clearly a bad name for dictionary.
+                if name not in data_collection_list:
+                    template = "Failed to find output [%s], tool outputs include [%s]"
+                    message = template % (name, ",".join(data_collection_list.keys()))
+                    raise AssertionError(message)
+
+                # 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()
+                elements = data_collection[ "elements" ]
+                element_dict = dict( map(lambda e: (e["element_identifier"], e["object"]), elements) )
+
+                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)
+
+                for element_identifier, ( element_outfile, element_attrib ) in output_collection_def.element_tests.items():
+                    if element_identifier not in element_dict:
+                        template = "Failed to find identifier [%s] for testing, tool generated collection with identifiers [%s]"
+                        message = template % (element_identifier, ",".join(element_dict.keys()))
+                        raise AssertionError(message)
+                    hda = element_dict[ element_identifier ]
+
+                    galaxy_interactor.verify_output_dataset(
+                        history,
+                        hda_id=hda["id"],
+                        outfile=element_outfile,
+                        attributes=element_attrib,
+                        shed_tool_id=shed_tool_id
+                    )
+            except Exception as e:
+                register_exception(e)
if found_exceptions:
             raise JobOutputsError(found_exceptions, job_stdio)
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/functional/tools/collection_creates_list.xml
--- /dev/null
+++ b/test/functional/tools/collection_creates_list.xml
@@ -0,0 +1,37 @@
+<tool id="collection_creates_list" name="collection_creates_list" version="0.1.0">
+  <command>
+    #for $key in $list_output.keys()#
+    echo "identifier is $key" > "$list_output[$key]";
+    #end for#
+    echo 'ensure not empty';
+  </command>
+  <inputs>
+    <param name="input1" type="data_collection" collection_type="list" label="Input" help="Input collection..." format="txt" />
+  </inputs>
+  <outputs>
+    <collection name="list_output" type="list" label="Duplicate List" structured_like="input1" format="txt">
+    </collection>
+  </outputs>
+  <tests>
+    <test>
+      <param name="input1">
+        <collection type="list">
+          <element name="l11" value="simple_line.txt" />
+          <element name="l12" value="simple_line.txt" />
+        </collection>
+      </param>
+      <output_collection name="list_output" type="list">
+        <element name="l11">
+          <assert_contents>
+            <has_text_matching expression="^identifier is l11\n$" />
+          </assert_contents>
+        </element>
+        <element name="l12">
+          <assert_contents>
+            <has_text_matching expression="^identifier is l12\n$" />
+          </assert_contents>
+        </element>
+      </output_collection>
+    </test>
+  </tests>
+</tool>
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/functional/tools/collection_creates_pair.xml
--- /dev/null
+++ b/test/functional/tools/collection_creates_pair.xml
@@ -0,0 +1,33 @@
+<tool id="collection_creates_pair" name="collection_creates_pair" version="0.1.0">
+  <command>
+    sed 'n;d' $input1 > $forward ;
+    sed -n 'g;n;p' $input1 > "reverse.txt";
+  </command>
+  <inputs>
+    <param name="input1" type="data" label="Input" help="Input to be split." />
+  </inputs>
+  <outputs>
+    <collection name="paired_output" type="paired" label="Split Pair">
+      <!-- can reference parts directly or find via from_work_dir. -->
+      <data name="forward" format="txt" />
+      <data name="reverse" format="txt" from_work_dir="reverse.txt" />
+    </collection>
+  </outputs>
+  <tests>
+    <test>
+      <param name="input1" value="simple_lines_interleaved.txt" />
+      <output_collection name="paired_output" type="paired">
+        <element name="forward">
+          <assert_contents>
+            <has_text_matching expression="^This is a line of text.\nThis is a line of text.\n$" />
+          </assert_contents>
+        </element>
+        <element name="reverse">
+          <assert_contents>
+            <has_text_matching expression="^This is a different line of text.\nThis is a different line of text.\n$" />
+          </assert_contents>
+        </element>
+      </output_collection>
+    </test>
+  </tests>
+</tool>
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/functional/tools/samples_tool_conf.xml
--- a/test/functional/tools/samples_tool_conf.xml
+++ b/test/functional/tools/samples_tool_conf.xml
@@ -34,6 +34,8 @@
   <tool file="collection_nested_test.xml" /><tool file="collection_mixed_param.xml" /><tool file="collection_two_paired.xml" />
+  <tool file="collection_creates_pair.xml" />
+  <tool file="collection_creates_list.xml" /><tool file="collection_optional_param.xml" /><tool file="multiple_versions_v01.xml" />
diff -r 6d3099596550b23ed9c21384e35ffb3a7af57a64 -r 6ae144fd51f756031e1691268369dcbe5440a16b test/unit/tools/test_parsing.py
--- a/test/unit/tools/test_parsing.py
+++ b/test/unit/tools/test_parsing.py
@@ -167,8 +167,9 @@
         assert containers[0].identifier == "mycool/bwa"
def test_outputs(self):
-        outputs = self._tool_source.parse_outputs(object())
+        outputs, output_collections = self._tool_source.parse_outputs(object())
         assert len(outputs) == 1
+        assert len(output_collections) == 0
def test_stdio(self):
         exit, regexes = self._tool_source.parse_stdio()
@@ -273,8 +274,9 @@
         assert containers[0].identifier == "awesome/bowtie"
def test_outputs(self):
-        outputs = self._tool_source.parse_outputs(object())
+        outputs, output_collections = self._tool_source.parse_outputs(object())
         assert len(outputs) == 1
+        assert len(output_collections) == 0
def test_stdio(self):
         exit, regexes = self._tool_source.parse_stdio()
https://bitbucket.org/galaxy/galaxy-central/commits/99b60d99df38/
Changeset:   99b60d99df38
User:        jmchilton
Date:        2015-01-15 14:30:00+00:00
Summary:     Allow tools to output collections with a dynamic number of datasets.
Models:
Track whether dataset collections have been populated yet.
Dataset collections are still effectively immutable once populated - but dynamic output collections require them to be sort of like `final` fields in Java (analogy courtesy of JJ) - allowing them to be declared before they are initialized or populated. This is tracked by the `populated_state` field.
Tools:
Output collections can now describe `discover_datasets` elements just like datasets - except in this case instead of dynamically populating new datasets in the history - they will comprise the collection. `designation` has been reused to serve as the element_identifier for the collection element corresponding to the dataset.
See Pull Request 356 for more information on the discover_datasets tag https://bitbucket.org/galaxy/galaxy-central/pull-request/356/enhancements-fo....
Workflows:
Update workflow execution and recovery for dynamic output collections.
Galaxy workflow data flow before collections
* - * - * - * - * - *
Galaxy worfklow data flow after collections (iteration 1)
* - * - * \
           * - * - *
* - * - * /         \
                     * - * - *
* - * - * \         /
           * - * - *
* - * - * /
Galaxy worfklow data flow after this commit
/ * - * \
         * - *         * - *
        /     \ * - * /     \
       /                     \
      /                       \
     /        / * - * \        \
* - * -- * - *         * - * -- * - *
     \        \ * - * /        /
      \                       /
       \                     /
        \     / * - * \     /
         * - *         * - *
              \ * - * /
Affected #:  17 files
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/jobs/__init__.py
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -1202,6 +1202,11 @@
         out_data = dict( [ ( da.name, da.dataset ) for da in job.output_datasets ] )
         inp_data.update( [ ( da.name, da.dataset ) for da in job.input_library_datasets ] )
         out_data.update( [ ( da.name, da.dataset ) for da in job.output_library_datasets ] )
+
+        # TODO: eliminate overlap with tools/evaluation.py
+        out_collections = dict( [ ( obj.name, obj.dataset_collection_instance ) for obj in job.output_dataset_collection_instances ] )
+        out_collections.update( [ ( obj.name, obj.dataset_collection ) for obj in job.output_dataset_collections ] )
+
         input_ext = 'data'
         for _, data in inp_data.items():
             # For loop odd, but sort simulating behavior in galaxy.tools.actions
@@ -1218,6 +1223,12 @@
             'children': self.tool.collect_child_datasets(out_data, self.working_directory),
             'primary': self.tool.collect_primary_datasets(out_data, self.working_directory, input_ext)
         }
+        self.tool.collect_dynamic_collections(
+            out_collections,
+            job_working_directory=self.working_directory,
+            inp_data=inp_data,
+            job=job,
+        )
         param_dict.update({'__collected_datasets__': collected_datasets})
         # Certain tools require tasks to be completed after job execution
         # ( this used to be performed in the "exec_after_process" hook, but hooks are deprecated ).
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/managers/collections.py
--- a/lib/galaxy/managers/collections.py
+++ b/lib/galaxy/managers/collections.py
@@ -28,6 +28,7 @@
     Abstraction for interfacing with dataset collections instance - ideally abstarcts
     out model and plugin details.
     """
+    ELEMENTS_UNINITIALIZED = object()
def __init__( self, app ):
         self.type_registry = DatasetCollectionTypesRegistry( app )
@@ -129,9 +130,24 @@
             elements = self.__load_elements( trans, element_identifiers )
         # else if elements is set, it better be an ordered dict!
+        if elements is not self.ELEMENTS_UNINITIALIZED:
+            type_plugin = collection_type_description.rank_type_plugin()
+            dataset_collection = builder.build_collection( type_plugin, elements )
+        else:
+            dataset_collection = model.DatasetCollection( populated=False )
+        dataset_collection.collection_type = collection_type
+        return dataset_collection
+
+    def set_collection_elements( self, dataset_collection, dataset_instances ):
+        if dataset_collection.populated:
+            raise Exception("Cannot reset elements of an already populated dataset collection.")
+
+        collection_type = dataset_collection.collection_type
+        collection_type_description = self.collection_type_descriptions.for_collection_type( collection_type )
         type_plugin = collection_type_description.rank_type_plugin()
-        dataset_collection = builder.build_collection( type_plugin, elements )
-        dataset_collection.collection_type = collection_type
+        builder.set_collection_elements( dataset_collection, type_plugin, dataset_instances )
+        dataset_collection.mark_as_populated()
+
         return dataset_collection
def delete( self, trans, instance_type, id ):
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/managers/collections_util.py
--- a/lib/galaxy/managers/collections_util.py
+++ b/lib/galaxy/managers/collections_util.py
@@ -77,7 +77,9 @@
         # TODO: Work in progress - this end-point is not right yet...
         dict_value[ 'url' ] = web.url_for( 'library_content', library_id=encoded_library_id, id=encoded_id, folder_id=encoded_folder_id )
     if view == "element":
-        dict_value[ 'elements' ] = map( dictify_element, dataset_colleciton_instance.collection.elements )
+        collection = dataset_colleciton_instance.collection
+        dict_value[ 'elements' ] = map( dictify_element, collection.elements )
+        dict_value[ 'populated' ] = collection.populated
     security.encode_all_ids( dict_value, recursive=True )  # TODO: Use Kyle's recusrive formulation of this.
     return dict_value
@@ -87,7 +89,9 @@
     object_detials = element.element_object.to_dict()
     if element.child_collection:
         # Recursively yield elements for each nested collection...
-        object_detials[ "elements" ] = map( dictify_element, element.child_collection.elements )
+        child_collection = element.child_collection
+        object_detials[ "elements" ] = map( dictify_element, child_collection.elements )
+        object_detials[ "populated" ] = child_collection.populated
dictified[ "object" ] = object_detials
     return dictified
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/model/__init__.py
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -2676,14 +2676,37 @@
     """
     dict_collection_visible_keys = ( 'id', 'collection_type' )
     dict_element_visible_keys = ( 'id', 'collection_type' )
+    populated_states = Bunch(
+        NEW='new',  # New dataset collection, unpopulated elements
+        OK='ok',  # Collection elements populated (HDAs may or may not have errors)
+        FAILED='failed',  # some problem populating state, won't be populated
+    )
def __init__(
         self,
         id=None,
         collection_type=None,
+        populated=True,
     ):
         self.id = id
         self.collection_type = collection_type
+        if not populated:
+            self.populated_state = DatasetCollection.populated_states.NEW
+
+    @property
+    def populated( self ):
+        return self.populated_state == DatasetCollection.populated_states.OK
+
+    @property
+    def waiting_for_elements( self ):
+        return self.populated_state == DatasetCollection.populated_states.NEW
+
+    def mark_as_populated( self ):
+        self.populated_state = DatasetCollection.populated_states.OK
+
+    def handle_population_failed( self, message ):
+        self.populated_state = DatasetCollection.populated_states.FAILED
+        self.populated_state_message = message
@property
     def dataset_instances( self ):
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/model/mapping.py
--- a/lib/galaxy/model/mapping.py
+++ b/lib/galaxy/model/mapping.py
@@ -621,6 +621,8 @@
 model.DatasetCollection.table = Table( "dataset_collection", metadata,
     Column( "id", Integer, primary_key=True ),
     Column( "collection_type", Unicode(255), nullable=False ),
+    Column( "populated_state", TrimmedString( 64 ), default='ok', nullable=False ),
+    Column( "populated_state_message", TEXT ),
     Column( "create_time", DateTime, default=now ),
     Column( "update_time", DateTime, default=now, onupdate=now ),
 )
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/model/migrate/versions/0127_output_collection_adjustments.py
--- a/lib/galaxy/model/migrate/versions/0127_output_collection_adjustments.py
+++ b/lib/galaxy/model/migrate/versions/0127_output_collection_adjustments.py
@@ -38,6 +38,18 @@
     for table in TABLES:
         __create(table)
+    try:
+        dataset_collection_table = Table( "dataset_collection", metadata, autoload=True )
+        # need server_default because column in non-null
+        populated_state_column = Column( 'populated_state', TrimmedString( 64 ), default='ok', server_default="ok", nullable=False )
+        populated_state_column.create( dataset_collection_table )
+
+        populated_message_column = Column( 'populated_state_message', TEXT, nullable=True )
+        populated_message_column.create( dataset_collection_table )
+    except Exception as e:
+        print str(e)
+        log.exception( "Creating dataset collection populated column failed." )
+
def downgrade(migrate_engine):
     metadata.bind = migrate_engine
@@ -46,6 +58,16 @@
     for table in TABLES:
         __drop(table)
+    try:
+        dataset_collection_table = Table( "dataset_collection", metadata, autoload=True )
+        populated_state_column = dataset_collection_table.c.populated_state
+        populated_state_column.drop()
+        populated_message_column = dataset_collection_table.c.populated_state_message
+        populated_message_column.drop()
+    except Exception as e:
+        print str(e)
+        log.exception( "Dropping dataset collection populated_state/ column failed." )
+
def __create(table):
     try:
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -297,7 +297,13 @@
         self.structure = structure
         self.outputs = odict()
+        # TODO:
+        self.metadata_source = None
+
     def known_outputs( self, inputs ):
+        if self.dynamic_structure:
+            return []
+
         def to_part( ( element_identifier, output ) ):
             return ToolOutputCollectionPart( self, element_identifier, output )
@@ -316,14 +322,33 @@
return map( to_part, outputs.items() )
+    @property
+    def dynamic_structure(self):
+        return self.structure.dynamic
+
+    @property
+    def dataset_collectors(self):
+        if not self.dynamic_structure:
+            raise Exception("dataset_collectors called for output collection with static structure")
+        return self.structure.dataset_collectors
+
class ToolOutputCollectionStructure( object ):
-    def __init__( self, collection_type=None, structured_like=None ):
+    def __init__(
+        self,
+        collection_type,
+        structured_like,
+        dataset_collectors,
+    ):
         self.collection_type = collection_type
         self.structured_like = structured_like
-        if collection_type is None and structured_like is None:
+        self.dataset_collectors = dataset_collectors
+        if collection_type is None and structured_like is None and dataset_collectors is None:
             raise ValueError( "Output collection types must be specify type of structured_like" )
+        if dataset_collectors and structured_like:
+            raise ValueError( "Cannot specify dynamic structure (discovered_datasets) and structured_like attribute." )
+        self.dynamic = dataset_collectors is not None
class ToolOutputCollectionPart( object ):
@@ -2146,6 +2171,11 @@
         """
         return output_collect.collect_primary_datasets( self, output, job_working_directory, input_ext )
+    def collect_dynamic_collections( self, output, **kwds ):
+        """ Find files corresponding to dynamically structured collections.
+        """
+        return output_collect.collect_dynamic_collections( self, output, **kwds )
+
     def to_dict( self, trans, link_details=False, io_details=False ):
         """ Returns dict of tool. """
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/tools/actions/__init__.py
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -289,6 +289,10 @@
elements[ output_part_def.element_identifier ] = element
+                    if output.dynamic_structure:
+                        assert not elements  # known_outputs must have been empty
+                        elements = collections_manager.ELEMENTS_UNINITIALIZED
+
                     if mapping_over_collection:
                         dc = collections_manager.create_dataset_collection(
                             trans,
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/tools/parameters/dataset_matcher.py
--- a/lib/galaxy/tools/parameters/dataset_matcher.py
+++ b/lib/galaxy/tools/parameters/dataset_matcher.py
@@ -162,6 +162,11 @@
             return self.dataset_collection_match( dataset_collection )
def dataset_collection_match( self, dataset_collection ):
+        # If dataset collection not yet populated, cannot determine if it
+        # would be a valid match for this parameter.
+        if not dataset_collection.populated:
+            return False
+
         valid = True
         for element in dataset_collection.elements:
             if not self.__valid_element( element ):
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/tools/parameters/output_collect.py
--- a/lib/galaxy/tools/parameters/output_collect.py
+++ b/lib/galaxy/tools/parameters/output_collect.py
@@ -12,6 +12,176 @@
 DATASET_ID_TOKEN = "DATASET_ID"
 DEFAULT_EXTRA_FILENAME_PATTERN = r"primary_DATASET_ID_(?P<designation>[^_]+)_(?P<visible>[^_]+)_(?P<ext>[^_]+)(_(?P<dbkey>[^_]+))?"
+import logging
+log = logging.getLogger( __name__ )
+
+
+def collect_dynamic_collections(
+    tool,
+    output_collections,
+    job_working_directory,
+    inp_data={},
+    job=None,
+):
+    collections_service = tool.app.dataset_collections_service
+    job_context = JobContext(
+        tool,
+        job,
+        job_working_directory,
+        inp_data,
+    )
+
+    for name, has_collection in output_collections.items():
+        if name not in tool.output_collections:
+            continue
+        output_collection_def = tool.output_collections[ name ]
+        if not output_collection_def.dynamic_structure:
+            continue
+
+        # Could be HDCA for normal jobs or a DC for mapping
+        # jobs.
+        if hasattr(has_collection, "collection"):
+            collection = has_collection.collection
+        else:
+            collection = has_collection
+
+        try:
+            elements = job_context.build_collection_elements(
+                collection,
+                output_collection_def,
+            )
+            collections_service.set_collection_elements(
+                collection,
+                elements
+            )
+        except Exception:
+            log.info("Problem gathering output collection.")
+            collection.handle_population_failed("Problem building datasets for collection.")
+
+
+class JobContext( object ):
+
+    def __init__( self, tool, job, job_working_directory, inp_data ):
+        self.inp_data = inp_data
+        self.app = tool.app
+        self.sa_session = tool.sa_session
+        self.job = job
+        self.job_working_directory = job_working_directory
+
+    @property
+    def permissions( self ):
+        inp_data = self.inp_data
+        existing_datasets = [ inp for inp in inp_data.values() if inp ]
+        if existing_datasets:
+            permissions = self.app.security_agent.guess_derived_permissions_for_datasets( existing_datasets )
+        else:
+            # No valid inputs, we will use history defaults
+            permissions = self.app.security_agent.history_get_default_permissions( self.job.history )
+        return permissions
+
+    def find_files( self, collection, dataset_collectors ):
+        filenames = odict.odict()
+        for path, extra_file_collector in walk_over_extra_files( dataset_collectors, self.job_working_directory, collection ):
+            filenames[ path ] = extra_file_collector
+        return filenames
+
+    def build_collection_elements( self, collection, output_collection_def ):
+        datasets = self.create_datasets(
+            collection,
+            output_collection_def,
+        )
+
+        elements = odict.odict()
+        # TODO: allow configurable sorting.
+        #    <sort by="lexical" /><!-- default -->
+        #    <sort by="reverse_lexical" />
+        #    <sort regex="example.(\d+).fastq" by="1:numerical" />
+        #    <sort regex="part_(\d+)_sample_([^_]+).fastq" by="2:lexical,1:numerical" />
+        # TODO: allow nested structure
+        for designation in datasets.keys():
+            elements[ designation ] = datasets[ designation ]
+
+        return elements
+
+    def create_datasets( self, collection, output_collection_def ):
+        dataset_collectors = output_collection_def.dataset_collectors
+        filenames = self.find_files( collection, dataset_collectors )
+
+        datasets = {}
+        for filename, extra_file_collector in filenames.iteritems():
+            fields_match = extra_file_collector.match( collection, os.path.basename( filename ) )
+            if not fields_match:
+                raise Exception( "Problem parsing metadata fields for file %s" % filename )
+            designation = fields_match.designation
+            visible = fields_match.visible
+            ext = fields_match.ext
+            dbkey = fields_match.dbkey
+            # Create new primary dataset
+            name = fields_match.name or designation
+
+            dataset = self.create_dataset(
+                ext=ext,
+                designation=designation,
+                visible=visible,
+                dbkey=dbkey,
+                name=name,
+                filename=filename,
+                metadata_source_name=output_collection_def.metadata_source,
+            )
+
+            datasets[ designation ] = dataset
+        return datasets
+
+    def create_dataset(
+        self,
+        ext,
+        designation,
+        visible,
+        dbkey,
+        name,
+        filename,
+        metadata_source_name,
+    ):
+        app = self.app
+        sa_session = self.sa_session
+
+        # Copy metadata from one of the inputs if requested.
+        metadata_source = None
+        if metadata_source_name:
+            metadata_source = self.inp_data[ metadata_source_name ]
+
+        # Create new primary dataset
+        primary_data = app.model.HistoryDatasetAssociation( extension=ext,
+                                                            designation=designation,
+                                                            visible=visible,
+                                                            dbkey=dbkey,
+                                                            create_dataset=True,
+                                                            sa_session=sa_session )
+        app.security_agent.set_all_dataset_permissions( primary_data.dataset, self.permissions )
+        sa_session.add( primary_data )
+        sa_session.flush()
+        # Move data from temp location to dataset location
+        app.object_store.update_from_file(primary_data.dataset, file_name=filename, create=True)
+        primary_data.set_size()
+        # If match specified a name use otherwise generate one from
+        # designation.
+        primary_data.name = name
+
+        if metadata_source:
+            primary_data.init_meta( copy_from=metadata_source )
+        else:
+            primary_data.init_meta()
+
+        # Associate new dataset with job
+        if self.job:
+            assoc = app.model.JobToOutputDatasetAssociation( '__new_primary_file_%s|%s__' % ( name, designation ), primary_data )
+            assoc.job = self.job
+            sa_session.add( assoc )
+            sa_session.flush()
+
+        primary_data.state = 'ok'
+        return primary_data
+
def collect_primary_datasets( tool, output, job_working_directory, input_ext ):
     app = tool.app
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/tools/parser/xml.py
--- a/lib/galaxy/tools/parser/xml.py
+++ b/lib/galaxy/tools/parser/xml.py
@@ -153,9 +153,13 @@
             default_format = collection_elem.get( "format", "data" )
             collection_type = collection_elem.get( "type", None )
             structured_like = collection_elem.get( "structured_like", None )
+            dataset_collectors = None
+            if collection_elem.find( "discover_datasets" ) is not None:
+                dataset_collectors = output_collect.dataset_collectors_from_elem( collection_elem )
             structure = galaxy.tools.ToolOutputCollectionStructure(
                 collection_type=collection_type,
                 structured_like=structured_like,
+                dataset_collectors=dataset_collectors,
             )
             output_collection = galaxy.tools.ToolOutputCollection(
                 name,
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/workflow/modules.py
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -901,6 +901,11 @@
             if replacement_value.hidden_beneath_collection_instance:
                 replacement_value = replacement_value.hidden_beneath_collection_instance
             outputs[ replacement_name ] = replacement_value
+        for job_output_collection in job_0.output_dataset_collection_instances:
+            replacement_name = job_output_collection.name
+            replacement_value = job_output_collection.dataset_collection_instance
+            outputs[ replacement_name ] = replacement_value
+
         progress.set_step_outputs( step, outputs )
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 lib/galaxy/workflow/run.py
--- a/lib/galaxy/workflow/run.py
+++ b/lib/galaxy/workflow/run.py
@@ -251,7 +251,18 @@
         step_outputs = self.outputs[ connection.output_step.id ]
         if step_outputs is STEP_OUTPUT_DELAYED:
             raise modules.DelayedWorkflowEvaluation()
-        return step_outputs[ connection.output_name ]
+        replacement = step_outputs[ connection.output_name ]
+        if isinstance( replacement, model.HistoryDatasetCollectionAssociation ):
+            if not replacement.collection.populated:
+                if not replacement.collection.waiting_for_elements:
+                    # If we are not waiting for elements, there was some
+                    # problem creating the collection. Collection will never
+                    # be populated.
+                    # TODO: consider distinguish between cancelled and failed?
+                    raise modules.CancelWorkflowEvaluation()
+
+                raise modules.DelayedWorkflowEvaluation()
+        return replacement
def set_outputs_for_input( self, step, outputs={} ):
         if self.inputs_by_step_id:
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 test/api/test_tools.py
--- a/test/api/test_tools.py
+++ b/test/api/test_tools.py
@@ -276,6 +276,40 @@
         contents1 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element1["object"]["id"])
         assert contents1 == "1\n", contents1
+    @skip_without_tool( "collection_split_on_column" )
+    def test_dynamic_list_output( self ):
+        history_id = self.dataset_populator.new_history()
+        new_dataset1 = self.dataset_populator.new_dataset( history_id, content='samp1\t1\nsamp1\t3\nsamp2\t2\nsamp2\t4\n' )
+        inputs = {
+            'input1': dataset_to_param( new_dataset1 ),
+        }
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        create = self._run( "collection_split_on_column", history_id, inputs, assert_ok=True )
+
+        jobs = create[ 'jobs' ]
+        implicit_collections = create[ 'implicit_collections' ]
+        collections = create[ 'output_collections' ]
+
+        self.assertEquals( len( jobs ), 1 )
+        job_id = jobs[ 0 ][ "id" ]
+        self.assertEquals( len( implicit_collections ), 0 )
+        self.assertEquals( len( collections ), 1 )
+
+        output_collection = collections[0]
+        self._assert_has_keys( output_collection, "id", "name", "elements", "populated" )
+        assert not output_collection[ "populated" ]
+        assert len( output_collection[ "elements" ] ) == 0
+
+        self.dataset_populator.wait_for_job( job_id, assert_ok=True )
+
+        get_collection_response = self._get( "dataset_collections/%s" % output_collection[ "id" ], data={"instance_type": "history"} )
+        self._assert_status_code_is( get_collection_response, 200 )
+
+        output_collection = get_collection_response.json()
+        self._assert_has_keys( output_collection, "id", "name", "elements", "populated" )
+        assert output_collection[ "populated" ]
+        assert len( output_collection[ "elements" ] ) == 2
+
     @skip_without_tool( "cat1" )
     def test_run_cat1_with_two_inputs( self ):
         # Run tool with an multiple data parameter and grouping (repeat)
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 test/api/test_workflows.py
--- a/test/api/test_workflows.py
+++ b/test/api/test_workflows.py
@@ -538,6 +538,44 @@
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
         self.assertEquals("a\nc\nb\nd\ne\ng\nf\nh\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
+    def test_workflow_run_dynamic_output_collections(self):
+        history_id = self.dataset_populator.new_history()
+        workflow_id = self._upload_yaml_workflow("""
+- 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
+""")
+        hda1 = self.dataset_populator.new_dataset( history_id, content="samp1\t10.0\nsamp2\t20.0\n" )
+        hda2 = self.dataset_populator.new_dataset( history_id, content="samp1\t30.0\nsamp2\t40.0\n" )
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        inputs = {
+            '0': self._ds_entry(hda1),
+            '1': self._ds_entry(hda2),
+        }
+        self.__invoke_workflow( history_id, workflow_id, inputs )
+        # TODO: wait on workflow invocations
+        time.sleep(10)
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        self.assertEquals("10.0\n30.0\n20.0\n40.0\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
+
     def test_workflow_request( self ):
         workflow = self.workflow_populator.load_workflow( name="test_for_queue" )
         workflow_request, history_id = self._setup_workflow_run( workflow )
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 test/functional/tools/collection_split_on_column.xml
--- /dev/null
+++ b/test/functional/tools/collection_split_on_column.xml
@@ -0,0 +1,30 @@
+<tool id="collection_split_on_column" name="collection_split_on_column" version="0.1.0">
+  <command>
+    mkdir outputs; cd outputs; awk '{ print $2 > $1 ".tabular" }' $input1
+  </command>
+  <inputs>
+    <param name="input1" type="data" label="Input Table" help="Table to split on first column" format="tabular" />
+  </inputs>
+  <outputs>
+    <collection name="split_output" type="list" label="Table split on first column">
+      <discover_datasets pattern="__name_and_ext__" directory="outputs" />
+    </collection>
+  </outputs>
+  <tests>
+    <test>
+      <param name="input1" value="tinywga.fam" />
+      <output_collection name="split_output" type="list">
+        <element name="101">
+          <assert_contents>
+            <has_text_matching expression="^1\n2\n3\n$" />
+          </assert_contents>
+        </element>
+        <element name="1334">
+          <assert_contents>
+            <has_text_matching expression="^1\n10\n11\n12\n13\n2\n$" />
+          </assert_contents>
+        </element>
+      </output_collection>
+    </test>
+  </tests>
+</tool>
diff -r 6ae144fd51f756031e1691268369dcbe5440a16b -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 test/functional/tools/samples_tool_conf.xml
--- a/test/functional/tools/samples_tool_conf.xml
+++ b/test/functional/tools/samples_tool_conf.xml
@@ -37,6 +37,7 @@
   <tool file="collection_creates_pair.xml" /><tool file="collection_creates_list.xml" /><tool file="collection_optional_param.xml" />
+  <tool file="collection_split_on_column.xml" /><tool file="multiple_versions_v01.xml" /><tool file="multiple_versions_v02.xml" />
https://bitbucket.org/galaxy/galaxy-central/commits/9468f580cc4a/
Changeset:   9468f580cc4a
User:        jmchilton
Date:        2015-01-15 14:30:00+00:00
Summary:     More configurable format and metadata handling for output collections.
Imporvements to testing code.
Affected #:  10 files
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -290,15 +290,30 @@
     <outputs>
     """
-    def __init__( self, name, structure, label=None, filters=None, hidden=False, default_format="data" ):
+    def __init__(
+        self,
+        name,
+        structure,
+        label=None,
+        filters=None,
+        hidden=False,
+        default_format="data",
+        default_format_source=None,
+        default_metadata_source=None,
+        inherit_format=False,
+        inherit_metadata=False
+    ):
         super( ToolOutputCollection, self ).__init__( name, label=label, filters=filters, hidden=hidden )
         self.collection = True
         self.default_format = default_format
         self.structure = structure
         self.outputs = odict()
-        # TODO:
-        self.metadata_source = None
+        self.inherit_format = inherit_format
+        self.inherit_metadata = inherit_metadata
+
+        self.metadata_source = default_metadata_source
+        self.format_source = default_format_source
def known_outputs( self, inputs ):
         if self.dynamic_structure:
@@ -317,7 +332,19 @@
             outputs = odict()
             for element in input_collection.collection.elements:
                 name = element.element_identifier
-                output = ToolOutput( name, format=self.default_format, implicit=True )
+                format = self.default_format
+                if self.inherit_format:
+                    format = element.dataset_instance.ext
+                output = ToolOutput(
+                    name,
+                    format=format,
+                    format_source=self.format_source,
+                    metadata_source=self.metadata_source,
+                    implicit=True,
+                )
+                if self.inherit_metadata:
+                    output.metadata_source = element.dataset_instance
+
                 outputs[ element.element_identifier ] = output
return map( to_part, outputs.items() )
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec lib/galaxy/tools/actions/__init__.py
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -245,8 +245,16 @@
             # This may not be neccesary with the new parent/child associations
             data.designation = name
             # Copy metadata from one of the inputs if requested.
-            if output.metadata_source:
-                data.init_meta( copy_from=inp_data[output.metadata_source] )
+
+            # metadata source can be either a string referencing an input
+            # or an actual object to copy.
+            metadata_source = output.metadata_source
+            if metadata_source:
+                if isinstance( metadata_source, basestring ):
+                    metadata_source = inp_data[metadata_source]
+
+            if metadata_source is not None:
+                data.init_meta( copy_from=metadata_source )
             else:
                 data.init_meta()
             # Take dbkey from LAST input
@@ -287,6 +295,11 @@
                         # Following hack causes dataset to no be added to history...
                         child_dataset_names.add( effective_output_name )
+                        if set_output_history:
+                            history.add_dataset( element, set_hid=set_output_hid )
+                        trans.sa_session.add( element )
+                        trans.sa_session.flush()
+
                         elements[ output_part_def.element_identifier ] = element
if output.dynamic_structure:
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec lib/galaxy/tools/parameters/output_collect.py
--- a/lib/galaxy/tools/parameters/output_collect.py
+++ b/lib/galaxy/tools/parameters/output_collect.py
@@ -174,6 +174,8 @@
# Associate new dataset with job
         if self.job:
+            self.job.history.add_dataset( primary_data )
+
             assoc = app.model.JobToOutputDatasetAssociation( '__new_primary_file_%s|%s__' % ( name, designation ), primary_data )
             assoc.job = self.job
             sa_session.add( assoc )
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec lib/galaxy/tools/parser/xml.py
--- a/lib/galaxy/tools/parser/xml.py
+++ b/lib/galaxy/tools/parser/xml.py
@@ -141,8 +141,8 @@
data_dict = odict()
-        def _parse(data_elem, default_format="data"):
-            output_def = self._parse_output(data_elem, tool, default_format=default_format)
+        def _parse(data_elem, **kwds):
+            output_def = self._parse_output(data_elem, tool, **kwds)
             data_dict[output_def.name] = output_def
             return output_def
@@ -153,6 +153,14 @@
             default_format = collection_elem.get( "format", "data" )
             collection_type = collection_elem.get( "type", None )
             structured_like = collection_elem.get( "structured_like", None )
+            inherit_format = False
+            inherit_metadata = False
+            if structured_like:
+                inherit_format = string_as_bool( collection_elem.get( "inherit_format", None ) )
+                inherit_metadata = string_as_bool( collection_elem.get( "inherit_metadata", None ) )
+            default_format_source = collection_elem.get( "format_source", None )
+            default_metadata_source = collection_elem.get( "metadata_source", "" )
+
             dataset_collectors = None
             if collection_elem.find( "discover_datasets" ) is not None:
                 dataset_collectors = output_collect.dataset_collectors_from_elem( collection_elem )
@@ -164,12 +172,21 @@
             output_collection = galaxy.tools.ToolOutputCollection(
                 name,
                 structure,
-                default_format=default_format
+                default_format=default_format,
+                inherit_format=inherit_format,
+                inherit_metadata=inherit_metadata,
+                default_format_source=default_format_source,
+                default_metadata_source=default_metadata_source,
             )
             outputs[output_collection.name] = output_collection
for data_elem in collection_elem.findall("data"):
-                _parse( data_elem, default_format=default_format )
+                _parse(
+                    data_elem,
+                    default_format=default_format,
+                    default_format_source=default_format_source,
+                    default_metadata_source=default_metadata_source,
+                )
for data_elem in collection_elem.findall("data"):
                 output_name = data_elem.get("name")
@@ -183,12 +200,19 @@
             outputs[output_def.name] = output_def
         return outputs, output_collections
-    def _parse_output(self, data_elem, tool, default_format):
+    def _parse_output(
+        self,
+        data_elem,
+        tool,
+        default_format="data",
+        default_format_source=None,
+        default_metadata_source="",
+    ):
         output = galaxy.tools.ToolOutput( data_elem.get("name") )
         output.format = data_elem.get("format", default_format)
         output.change_format = data_elem.findall("change_format")
-        output.format_source = data_elem.get("format_source", None)
-        output.metadata_source = data_elem.get("metadata_source", "")
+        output.format_source = data_elem.get("format_source", default_format_source)
+        output.metadata_source = data_elem.get("metadata_source", default_metadata_source)
         output.parent = data_elem.get("parent", None)
         output.label = xml_text( data_elem, "label" )
         output.count = int( data_elem.get("count", 1) )
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec test/api/helpers.py
--- a/test/api/helpers.py
+++ b/test/api/helpers.py
@@ -134,7 +134,7 @@
def get_history_dataset_details( self, history_id, **kwds ):
         dataset_id = self.__history_dataset_id( history_id, **kwds )
-        details_response = self.__get_contents_request( history_id, "/%s" % dataset_id )
+        details_response = self.__get_contents_request( history_id, "/datasets/%s" % dataset_id )
         assert details_response.status_code == 200
         return details_response.json()
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec test/api/test_tools.py
--- a/test/api/test_tools.py
+++ b/test/api/test_tools.py
@@ -225,25 +225,11 @@
         # TODO: shouldn't need this wait
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
         create = self._run( "collection_creates_pair", history_id, inputs, assert_ok=True )
-        jobs = create[ 'jobs' ]
-        implicit_collections = create[ 'implicit_collections' ]
-        collections = create[ 'output_collections' ]
-
-        self.assertEquals( len( jobs ), 1 )
-        self.assertEquals( len( implicit_collections ), 0 )
-        self.assertEquals( len( collections ), 1 )
-
-        output_collection = collections[ 0 ]
-        elements = output_collection[ "elements" ]
-        assert len( elements ) == 2
-        element0, element1 = elements
-        assert element0[ "element_identifier" ] == "forward"
-        assert element1[ "element_identifier" ] == "reverse"
+        output_collection = self._assert_one_job_one_collection_run( create )
+        element0, element1 = self._assert_elements_are( output_collection, "forward", "reverse" )
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
-        contents0 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element0["object"]["id"])
-        assert contents0 == "123\n789\n", contents0
-        contents1 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element1["object"]["id"])
-        assert contents1 == "456\n0ab\n", contents1
+        self._verify_element( history_id, element0, contents="123\n789\n", file_ext="txt" )
+        self._verify_element( history_id, element1, contents="456\n0ab\n", file_ext="txt" )
@skip_without_tool( "collection_creates_list" )
     def test_list_collection_output( self ):
@@ -256,25 +242,31 @@
         # TODO: real problem here - shouldn't have to have this wait.
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
         create = self._run( "collection_creates_list", history_id, inputs, assert_ok=True )
-        jobs = create[ 'jobs' ]
-        implicit_collections = create[ 'implicit_collections' ]
-        collections = create[ 'output_collections' ]
+        output_collection = self._assert_one_job_one_collection_run( create )
+        element0, element1 = self._assert_elements_are( output_collection, "data1", "data2" )
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        self._verify_element( history_id, element0, contents="identifier is data1\n", file_ext="txt" )
+        self._verify_element( history_id, element1, contents="identifier is data2\n", file_ext="txt" )
-        self.assertEquals( len( jobs ), 1 )
-        self.assertEquals( len( implicit_collections ), 0 )
-        self.assertEquals( len( collections ), 1 )
-
-        output_collection = collections[ 0 ]
-        elements = output_collection[ "elements" ]
-        assert len( elements ) == 2
-        element0, element1 = elements
-        assert element0[ "element_identifier" ] == "data1"
-        assert element1[ "element_identifier" ] == "data2"
+    @skip_without_tool( "collection_creates_list_2" )
+    def test_list_collection_output_format_source( self ):
+        # test using format_source with a tool
+        history_id = self.dataset_populator.new_history()
+        new_dataset1 = self.dataset_populator.new_dataset( history_id, content='#col1\tcol2' )
+        create_response = self.dataset_collection_populator.create_list_in_history( history_id, contents=["a\tb\nc\td", "e\tf\ng\th"] )
+        hdca_id = create_response.json()[ "id" ]
+        inputs = {
+            "header": { "src": "hda", "id": new_dataset1["id"] },
+            "input_collect": { "src": "hdca", "id": hdca_id },
+        }
+        # TODO: real problem here - shouldn't have to have this wait.
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
-        contents0 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element0["object"]["id"])
-        assert contents0 == "0\n", contents0
-        contents1 = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=element1["object"]["id"])
-        assert contents1 == "1\n", contents1
+        create = self._run( "collection_creates_list_2", history_id, inputs, assert_ok=True )
+        output_collection = self._assert_one_job_one_collection_run( create )
+        element0, element1 = self._assert_elements_are( output_collection, "data1", "data2" )
+        self.dataset_populator.wait_for_history( history_id, assert_ok=True )
+        self._verify_element( history_id, element0, contents="#col1\tcol2\na\tb\nc\td\n", file_ext="txt" )
+        self._verify_element( history_id, element1, contents="#col1\tcol2\ne\tf\ng\th\n", file_ext="txt" )
@skip_without_tool( "collection_split_on_column" )
     def test_dynamic_list_output( self ):
@@ -286,21 +278,12 @@
         self.dataset_populator.wait_for_history( history_id, assert_ok=True )
         create = self._run( "collection_split_on_column", history_id, inputs, assert_ok=True )
-        jobs = create[ 'jobs' ]
-        implicit_collections = create[ 'implicit_collections' ]
-        collections = create[ 'output_collections' ]
-
-        self.assertEquals( len( jobs ), 1 )
-        job_id = jobs[ 0 ][ "id" ]
-        self.assertEquals( len( implicit_collections ), 0 )
-        self.assertEquals( len( collections ), 1 )
-
-        output_collection = collections[0]
+        output_collection = self._assert_one_job_one_collection_run( create )
         self._assert_has_keys( output_collection, "id", "name", "elements", "populated" )
         assert not output_collection[ "populated" ]
         assert len( output_collection[ "elements" ] ) == 0
-        self.dataset_populator.wait_for_job( job_id, assert_ok=True )
+        self.dataset_populator.wait_for_job( create["jobs"][0]["id"], assert_ok=True )
get_collection_response = self._get( "dataset_collections/%s" % output_collection[ "id" ], data={"instance_type": "history"} )
         self._assert_status_code_is( get_collection_response, 200 )
@@ -309,6 +292,7 @@
         self._assert_has_keys( output_collection, "id", "name", "elements", "populated" )
         assert output_collection[ "populated" ]
         assert len( output_collection[ "elements" ] ) == 2
+        # TODO: verify element identifiers
@skip_without_tool( "cat1" )
     def test_run_cat1_with_two_inputs( self ):
@@ -443,6 +427,42 @@
         assert "123\n0ab" in outputs_contents
         assert "456\n789" in outputs_contents
+    def _assert_one_job_one_collection_run( self, create ):
+        jobs = create[ 'jobs' ]
+        implicit_collections = create[ 'implicit_collections' ]
+        collections = create[ 'output_collections' ]
+
+        self.assertEquals( len( jobs ), 1 )
+        self.assertEquals( len( implicit_collections ), 0 )
+        self.assertEquals( len( collections ), 1 )
+
+        output_collection = collections[ 0 ]
+        return output_collection
+
+    def _assert_elements_are( self, collection, *args ):
+        elements = collection["elements"]
+        self.assertEquals(len(elements), len(args))
+        for index, element in enumerate(elements):
+            arg = args[index]
+            self.assertEquals(arg, element["element_identifier"])
+        return elements
+
+    def _verify_element( self, history_id, element, **props ):
+        object_id = element["object"]["id"]
+
+        if "contents" in props:
+            expected_contents = props["contents"]
+
+            contents = self.dataset_populator.get_history_dataset_content( history_id, dataset_id=object_id)
+            self.assertEquals( contents, expected_contents )
+
+            del props["contents"]
+
+        if props:
+            details = self.dataset_populator.get_history_dataset_details( history_id, dataset_id=object_id)
+            for key, value in props.items():
+                self.assertEquals( details[key], value )
+
     def _setup_repeat_multirun( self ):
         history_id = self.dataset_populator.new_history()
         new_dataset1 = self.dataset_populator.new_dataset( history_id, content='123' )
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec test/functional/tools/collection_creates_list.xml
--- a/test/functional/tools/collection_creates_list.xml
+++ b/test/functional/tools/collection_creates_list.xml
@@ -9,7 +9,9 @@
     <param name="input1" type="data_collection" collection_type="list" label="Input" help="Input collection..." format="txt" /></inputs><outputs>
-    <collection name="list_output" type="list" label="Duplicate List" structured_like="input1" format="txt">
+    <collection name="list_output" type="list" label="Duplicate List" structured_like="input1" inherit_format="true">
+      <!-- inherit_format can be used in conjunction with structured_like
+           to perserve format. --></collection></outputs><tests>
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec test/functional/tools/collection_creates_list_2.xml
--- /dev/null
+++ b/test/functional/tools/collection_creates_list_2.xml
@@ -0,0 +1,22 @@
+<tool id="collection_creates_list_2" name="collection_creates_list_2" version="0.1.0">
+  <!-- go through and a header to each item in a collection - should use implicit
+       mapping the non-collectiony add header tool to do this in a real analysis.
+  -->
+  <command>
+    #for $key in $list_output.keys()#
+    cat "$header" > "$list_output[$key]";
+    cat "$input_collect[$key]" >> "$list_output[$key]";
+    #end for#
+    echo 'ensure not empty';
+  </command>
+  <inputs>
+    <param name="header" type="data" label="Input Data" help="Input data..." />
+    <param name="input_collect" type="data_collection" collection_type="list" label="Input Collect" help="Input collection..." />
+  </inputs>
+  <outputs>
+    <collection name="list_output" type="list" label="Duplicate List" structured_like="input_collect" format_source="header">
+    </collection>
+  </outputs>
+  <tests>
+  </tests>
+</tool>
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec test/functional/tools/collection_creates_pair.xml
--- a/test/functional/tools/collection_creates_pair.xml
+++ b/test/functional/tools/collection_creates_pair.xml
@@ -8,9 +8,13 @@
   </inputs><outputs><collection name="paired_output" type="paired" label="Split Pair">
-      <!-- can reference parts directly or find via from_work_dir. -->
+      <!-- command can reference parts directly or find via from_work_dir. --><data name="forward" format="txt" />
-      <data name="reverse" format="txt" from_work_dir="reverse.txt" />
+      <data name="reverse" format_source="input1" from_work_dir="reverse.txt" />
+      <!-- data elements can use format, format_source, metadata_from,
+           from_work_dir. The format="input" idiom is not supported,
+           it should be considered deprecated and format_source is superior.
+      --></collection></outputs><tests>
diff -r 99b60d99df388fb919df5e6dadc791c0e6be90d3 -r 9468f580cc4a9c45d1770477e5175e963afd3aec test/functional/tools/samples_tool_conf.xml
--- a/test/functional/tools/samples_tool_conf.xml
+++ b/test/functional/tools/samples_tool_conf.xml
@@ -36,6 +36,7 @@
   <tool file="collection_two_paired.xml" /><tool file="collection_creates_pair.xml" /><tool file="collection_creates_list.xml" />
+  <tool file="collection_creates_list_2.xml" /><tool file="collection_optional_param.xml" /><tool file="collection_split_on_column.xml" />
https://bitbucket.org/galaxy/galaxy-central/commits/0bcbf45df979/
Changeset:   0bcbf45df979
User:        jmchilton
Date:        2015-01-15 14:30:00+00:00
Summary:     A simpler way to configure output pairs (exploiting static structure).
See example and comments in test/functional/tools/collection_creates_pair_from_type.xml.
Affected #:  6 files
diff -r 9468f580cc4a9c45d1770477e5175e963afd3aec -r 0bcbf45df97932e14522bb255f457d39aec16bf2 lib/galaxy/dataset_collections/registry.py
--- a/lib/galaxy/dataset_collections/registry.py
+++ b/lib/galaxy/dataset_collections/registry.py
@@ -1,6 +1,6 @@
 from .types import list
 from .types import paired
-
+from galaxy import model
PLUGIN_CLASSES = [list.ListDatasetCollectionType, paired.PairedDatasetCollectionType]
@@ -12,3 +12,13 @@
def get( self, plugin_type ):
         return self.__plugins[ plugin_type ]
+
+    def prototype( self, plugin_type ):
+        plugin_type_object = self.get( plugin_type )
+        if not hasattr( plugin_type_object, 'prototype_elements' ):
+            raise Exception( "Cannot pre-determine structure for collection of type %s" % plugin_type )
+
+        dataset_collection = model.DatasetCollection()
+        elements = [ e for e in plugin_type_object.prototype_elements() ]
+        dataset_collection.elements = elements
+        return dataset_collection
diff -r 9468f580cc4a9c45d1770477e5175e963afd3aec -r 0bcbf45df97932e14522bb255f457d39aec16bf2 lib/galaxy/dataset_collections/types/paired.py
--- a/lib/galaxy/dataset_collections/types/paired.py
+++ b/lib/galaxy/dataset_collections/types/paired.py
@@ -1,6 +1,6 @@
 from ..types import BaseDatasetCollectionType
-from galaxy.model import DatasetCollectionElement
+from galaxy.model import DatasetCollectionElement, HistoryDatasetAssociation
FORWARD_IDENTIFIER = "forward"
 REVERSE_IDENTIFIER = "reverse"
@@ -32,3 +32,15 @@
         )
         yield left_association
         yield right_association
+
+    def prototype_elements( self ):
+        left_association = DatasetCollectionElement(
+            element=HistoryDatasetAssociation(),
+            element_identifier=FORWARD_IDENTIFIER,
+        )
+        right_association = DatasetCollectionElement(
+            element=HistoryDatasetAssociation(),
+            element_identifier=REVERSE_IDENTIFIER,
+        )
+        yield left_association
+        yield right_association
diff -r 9468f580cc4a9c45d1770477e5175e963afd3aec -r 0bcbf45df97932e14522bb255f457d39aec16bf2 lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -315,7 +315,7 @@
         self.metadata_source = default_metadata_source
         self.format_source = default_format_source
-    def known_outputs( self, inputs ):
+    def known_outputs( self, inputs, type_registry ):
         if self.dynamic_structure:
             return []
@@ -324,13 +324,17 @@
# This line is probably not right - should verify structured_like
         # or have outputs and all outputs have name.
-        if not self.structure.structured_like and self.outputs:
+        if len( self.outputs ) > 1:
             outputs = self.outputs
         else:
+            # either must have specified structured_like or something worse
+            if self.structure.structured_like:
+                collection_prototype = inputs[ self.structure.structured_like ].collection
+            else:
+                collection_prototype = type_registry.prototype( self.structure.collection_type )
             # TODO: Handle nested structures.
-            input_collection = inputs[ self.structure.structured_like ]
             outputs = odict()
-            for element in input_collection.collection.elements:
+            for element in collection_prototype.elements:
                 name = element.element_identifier
                 format = self.default_format
                 if self.inherit_format:
diff -r 9468f580cc4a9c45d1770477e5175e963afd3aec -r 0bcbf45df97932e14522bb255f457d39aec16bf2 lib/galaxy/tools/actions/__init__.py
--- a/lib/galaxy/tools/actions/__init__.py
+++ b/lib/galaxy/tools/actions/__init__.py
@@ -286,7 +286,7 @@
elements = odict()
                     input_collections = dict( [ (k, v[0]) for k, v in inp_dataset_collections.iteritems() ] )
-                    known_outputs = output.known_outputs( input_collections )
+                    known_outputs = output.known_outputs( input_collections, collections_manager.type_registry )
                     # Just to echo TODO elsewhere - this should be restructured to allow
                     # nested collections.
                     for output_part_def in known_outputs:
diff -r 9468f580cc4a9c45d1770477e5175e963afd3aec -r 0bcbf45df97932e14522bb255f457d39aec16bf2 test/functional/tools/collection_creates_pair_from_type.xml
--- /dev/null
+++ b/test/functional/tools/collection_creates_pair_from_type.xml
@@ -0,0 +1,35 @@
+<tool id="collection_creates_pair_from_type" name="collection_creates_pair_from_type" version="0.1.0">
+  <command>
+    sed 'n;d' $input1 > $paired_output.forward;
+    sed -n 'g;n;p' $input1 > $paired_output.reverse;
+  </command>
+  <inputs>
+    <param name="input1" type="data" label="Input" help="Input to be split." />
+  </inputs>
+  <outputs>
+    <!-- unlike lists - structure of paired outputs can be predetermined
+         so we can use $output.identifier in command block, see
+         collection_creates_pair.xml for examples of actually labelling
+         pieces and allowing stuff like from_work_dir.
+    -->
+    <collection name="paired_output" type="paired" label="Split Pair" format_source="input1">
+    </collection>
+  </outputs>
+  <tests>
+    <test>
+      <param name="input1" value="simple_lines_interleaved.txt" ftype="txt" />
+      <output_collection name="paired_output" type="paired">
+        <element name="forward" ftype="txt">
+          <assert_contents>
+            <has_text_matching expression="^This is a line of text.\nThis is a line of text.\n$" />
+          </assert_contents>
+        </element>
+        <element name="reverse" ftype="txt">
+          <assert_contents>
+            <has_text_matching expression="^This is a different line of text.\nThis is a different line of text.\n$" />
+          </assert_contents>
+        </element>
+      </output_collection>
+    </test>
+  </tests>
+</tool>
diff -r 9468f580cc4a9c45d1770477e5175e963afd3aec -r 0bcbf45df97932e14522bb255f457d39aec16bf2 test/functional/tools/samples_tool_conf.xml
--- a/test/functional/tools/samples_tool_conf.xml
+++ b/test/functional/tools/samples_tool_conf.xml
@@ -35,6 +35,7 @@
   <tool file="collection_mixed_param.xml" /><tool file="collection_two_paired.xml" /><tool file="collection_creates_pair.xml" />
+  <tool file="collection_creates_pair_from_type.xml" /><tool file="collection_creates_list.xml" /><tool file="collection_creates_list_2.xml" /><tool file="collection_optional_param.xml" />
https://bitbucket.org/galaxy/galaxy-central/commits/42f2532c7e95/
Changeset:   42f2532c7e95
User:        jmchilton
Date:        2015-01-15 14:30:00+00:00
Summary:     Update workflow extraction backend for output collections.
Affected #:  5 files
diff -r 0bcbf45df97932e14522bb255f457d39aec16bf2 -r 42f2532c7e95237ce419c2f2f7cf6ff2c603170b lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -314,6 +314,7 @@
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:
diff -r 0bcbf45df97932e14522bb255f457d39aec16bf2 -r 42f2532c7e95237ce419c2f2f7cf6ff2c603170b lib/galaxy/workflow/extract.py
--- a/lib/galaxy/workflow/extract.py
+++ b/lib/galaxy/workflow/extract.py
@@ -11,6 +11,7 @@
     SelectToolParameter,
     UnvalidatedValue
 )
+from galaxy.tools import ToolOutputCollectionPart
 from galaxy.tools.parameters.grouping import (
     Conditional,
     Repeat
@@ -115,6 +116,8 @@
                 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 @@
         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 @@
         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 @@
                     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 ):
diff -r 0bcbf45df97932e14522bb255f457d39aec16bf2 -r 42f2532c7e95237ce419c2f2f7cf6ff2c603170b lib/galaxy/workflow/modules.py
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -649,7 +649,9 @@
         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()
diff -r 0bcbf45df97932e14522bb255f457d39aec16bf2 -r 42f2532c7e95237ce419c2f2f7cf6ff2c603170b test/api/test_workflow_extraction.py
--- a/test/api/test_workflow_extraction.py
+++ b/test/api/test_workflow_extraction.py
@@ -182,6 +182,97 @@
         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" ]
diff -r 0bcbf45df97932e14522bb255f457d39aec16bf2 -r 42f2532c7e95237ce419c2f2f7cf6ff2c603170b test/unit/workflows/test_extract_summary.py
--- a/test/unit/workflows/test_extract_summary.py
+++ b/test/unit/workflows/test_extract_summary.py
@@ -135,6 +135,7 @@
         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(),
https://bitbucket.org/galaxy/galaxy-central/commits/0fd00352949c/
Changeset:   0fd00352949c
User:        jmchilton
Date:        2015-01-15 14:30:00+00:00
Summary:     Hide tools producing collections in workflow editor.
Output collections will require significant overhaul to workflow editor logic.
Affected #:  1 file
diff -r 42f2532c7e95237ce419c2f2f7cf6ff2c603170b -r 0fd00352949c5d407e36f5d1ff0ffbcd699931c6 lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -496,6 +496,10 @@
                                                                                                  self.installed_changeset_revision )
         return None
+    @property
+    def produces_collections( self ):
+        return any( o.collection for o in self.outputs.values() )
+
     def __get_job_tool_configuration(self, job_params=None):
         """Generalized method for getting this tool's job configuration.
@@ -1099,10 +1103,15 @@
         if self.tool_type.startswith( 'data_source' ):
             return False
+        if self.produces_collections:
+            # Someday we will get there!
+            return False
+
         if hasattr( tool_source, "root"):
             root = tool_source.root
             if not string_as_bool( root.get( "workflow_compatible", "True" ) ):
                 return False
+
         # TODO: Anyway to capture tools that dynamically change their own
         #       outputs?
         return True
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.

    

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

[galaxy-commits] commit/galaxy-central: 6 new changesets