History manager: incorporate container interface; add tests

lib/galaxy/managers/containers.py

@@ -3,10 +3,9 @@
 and other (nested) containers.
 
 (e.g. DatasetCollections, Histories, LibraryFolders)
-
-Histories should be DatasetCollections.
-Libraries should be DatasetCollections.
 """
+# Histories should be DatasetCollections.
+# Libraries should be DatasetCollections.
 
 import operator
 
@@ -28,6 +27,7 @@ class ContainerManagerMixin( object ):
     each of the methods below only work on the first level of
     nesting.
     """
+    # TODO: terminology is getting a bit convoluted and silly at this point: rename three public below?
     # TODO: this should be an open mapping (not just 2)
     #: the classes that can be contained
     contained_class = None
@@ -78,6 +78,7 @@ class HistoryAsContainerManagerMixin( ContainerManagerMixin ):
     order_contents_on = operator.attrgetter( 'hid' )
 
     def _filter_to_contained( self, container, content_class ):
+        # use the backref of the container on the contained
         return content_class.history == container
 
     def _content_manager( self, content ):

lib/galaxy/managers/histories.py

@@ -4,18 +4,20 @@
 Histories are containers for datasets or dataset collections
 created (or copied) by users over the course of an analysis.
 """
+import operator
 
 from galaxy import model
 from galaxy.managers import sharable
 from galaxy.managers import deletable
+from galaxy.managers import containers
 from galaxy.managers import hdas
 from galaxy.managers import collections_util
 
 import logging
 log = logging.getLogger( __name__ )
 
 
-class HistoryManager( sharable.SharableModelManager, deletable.PurgableManagerMixin ):
+class HistoryManager( sharable.SharableModelManager, deletable.PurgableManagerMixin, containers.ContainerManagerMixin ):
 
     model_class = model.History
     foreign_key_name = 'history'
@@ -164,6 +166,24 @@ def _get_history_data( self, trans, history ):
         return { 'history': history_dictionary,
                  'contents': contents_dictionaries }
 
+# class HistoryAsContainerManagerMixin( ContainerManagerMixin ):
+
+    contained_class = model.HistoryDatasetAssociation
+    subcontainer_class = model.HistoryDatasetCollectionAssociation
+    order_contents_on = operator.attrgetter( 'hid' )
+
+    def _filter_to_contained( self, container, content_class ):
+        return content_class.history == container
+
+    def _content_manager( self, content ):
+        # type sniffing is inevitable
+        if isinstance( content, model.HistoryDatasetAssociation ):
+            return self.hda_manager
+        elif isinstance( content, model.HistoryDatasetCollectionAssociation ):
+            return self.hdca_manager
+        raise TypeError( 'Unknown contents class: ' + str( content ) )
+
+
 
 class HistorySerializer( sharable.SharableModelSerializer, deletable.PurgableSerializerMixin ):
     """

test/unit/managers/base.py

@@ -120,6 +120,21 @@ def assertIsJsonifyable( self, item ):
         self.assertIsInstance( json.dumps( item ), basestring )
 
 
+class CreatesCollectionsMixin( object ):
+
+    def build_element_identifiers( self, elements ):
+        identifier_list = []
+        for element in elements:
+            src = 'hda'
+            # if isinstance( element, model.DatasetCollection ):
+            #    src = 'collection'#?
+            # elif isinstance( element, model.LibraryDatasetDatasetAssociation ):
+            #    src = 'ldda'#?
+            encoded_id = self.trans.security.encode_id( element.id )
+            identifier_list.append( dict( src=src, name=element.name, id=encoded_id ) )
+        return identifier_list
+
+
 # =============================================================================
 if __name__ == '__main__':
     # or more generally, nosetests test_resourcemanagers.py -s -v

test/unit/managers/test_CollectionManager.py

@@ -11,6 +11,7 @@
 from galaxy import model
 
 from base import BaseTestCase
+from base import CreatesCollectionsMixin
 from galaxy.managers.datasets import DatasetManager
 from galaxy.managers.histories import HistoryManager
 from galaxy.managers.hdas import HDAManager
@@ -25,7 +26,7 @@
 
 
 # =============================================================================
-class DatasetCollectionManagerTestCase( BaseTestCase ):
+class DatasetCollectionManagerTestCase( BaseTestCase, CreatesCollectionsMixin ):
 
     def set_up_managers( self ):
         super( DatasetCollectionManagerTestCase, self ).set_up_managers()
@@ -34,18 +35,6 @@ def set_up_managers( self ):
         self.history_manager = HistoryManager( self.app )
         self.collection_manager = DatasetCollectionManager( self.app )
 
-    def build_element_identifiers( self, elements ):
-        identifier_list = []
-        for element in elements:
-            src = 'hda'
-            # if isinstance( element, model.DatasetCollection ):
-            #    src = 'collection'#?
-            # elif isinstance( element, model.LibraryDatasetDatasetAssociation ):
-            #    src = 'ldda'#?
-            encoded_id = self.trans.security.encode_id( element.id )
-            identifier_list.append( dict( src=src, name=element.name, id=encoded_id ) )
-        return identifier_list
-
     def test_create_simple_list( self ):
         owner = self.user_manager.create( **user2_data )
 

test/unit/managers/test_DatasetManager.py

@@ -112,7 +112,8 @@ def test_create_with_no_permissions( self ):
         self.assertTrue( self.dataset_manager.permissions.access.is_permitted( dataset, user3 ) )
 
     def test_create_public_dataset( self ):
-        self.log( "should be able to create a new Dataset and give it some permissions that actually, you know, might work if there's any justice in this universe" )
+        self.log( "should be able to create a new Dataset and give it some permissions that actually, you know, "
+            "might work if there's any justice in this universe" )
         owner = self.user_manager.create( **user2_data )
         owner_private_role = self.user_manager.private_role( owner )
         dataset = self.dataset_manager.create( manage_roles=[ owner_private_role ] )
@@ -134,7 +135,7 @@ def test_create_public_dataset( self ):
         self.assertTrue( self.dataset_manager.permissions.access.is_permitted( dataset, user3 ) )
 
     def test_create_private_dataset( self ):
-        self.log( "should be able to create a new Dataset and give it some permissions that actually, you know, might work if there's any justice in this universe" )
+        self.log( "should be able to create a new Dataset and give it private permissions" )
         owner = self.user_manager.create( **user2_data )
         owner_private_role = self.user_manager.private_role( owner )
         dataset = self.dataset_manager.create(

test/unit/managers/test_HistoryManager.py

@@ -4,6 +4,7 @@
 import os
 import imp
 import unittest
+import random
 
 test_utils = imp.load_source( 'test_utils',
     os.path.join( os.path.dirname( __file__), '../unittest_utils/utility.py' ) )
@@ -18,11 +19,13 @@
 from galaxy import exceptions
 
 from base import BaseTestCase
+from base import CreatesCollectionsMixin
 
 from galaxy.managers.histories import HistoryManager
 from galaxy.managers.histories import HistorySerializer
 from galaxy.managers.histories import HistoryFilters
 from galaxy.managers import hdas
+from galaxy.managers import collections
 
 default_password = '123456'
 user2_data = dict( email='user2@user2.user2', username='user2', password=default_password )
@@ -725,6 +728,69 @@ def test_list( self ):
         self.assertEqual( found, deleted_and_annotated )
 
 
+# =============================================================================
+class HistoryAsContainerTestCase( BaseTestCase, CreatesCollectionsMixin ):
+
+    def set_up_managers( self ):
+        super( HistoryAsContainerTestCase, self ).set_up_managers()
+        self.history_manager = HistoryManager( self.app )
+        self.hda_manager = hdas.HDAManager( self.app )
+        self.collection_manager = collections.DatasetCollectionManager( self.app )
+
+    def add_hda_to_history( self, history, **kwargs ):
+        dataset = self.hda_manager.dataset_manager.create()
+        hda = self.hda_manager.create( history=history, dataset=dataset, **kwargs )
+        return hda
+
+    def add_list_collection_to_history( self, history, hdas, name='test collection', **kwargs ):
+        hdca = self.collection_manager.create( self.trans, history, name, 'list',
+            element_identifiers=self.build_element_identifiers( hdas ) )
+        return hdca
+
+    def test_contents( self ):
+        user2 = self.user_manager.create( **user2_data )
+        history = self.history_manager.create( name='history', user=user2 )
+
+        self.log( "calling contents on an empty history should return an empty list" )
+        self.assertEqual( [], list( self.history_manager.contents( history ) ) )
+
+        self.log( "calling contents on an history with hdas should return those in order of their hids" )
+        hdas = [ self.add_hda_to_history( history, name=( 'hda-' + str( x ) ) ) for x in xrange( 3 ) ]
+        random.shuffle( hdas )
+        ordered_hda_contents = list( self.history_manager.contents( history ) )
+        self.assertEqual( map( lambda hda: hda.hid, ordered_hda_contents ), [ 1, 2, 3 ] )
+
+        self.log( "calling contents on an history with both hdas and collections should return both" )
+        hdca = self.add_list_collection_to_history( history, hdas )
+        all_contents = list( self.history_manager.contents( history ) )
+        self.assertEqual( all_contents, list( ordered_hda_contents ) + [ hdca ] )
+
+    def test_contained( self ):
+        user2 = self.user_manager.create( **user2_data )
+        history = self.history_manager.create( name='history', user=user2 )
+
+        self.log( "calling contained on an empty history should return an empty list" )
+        self.assertEqual( [], list( self.history_manager.contained( history ) ) )
+
+        self.log( "calling contained on an history with both hdas and collections should return only hdas" )
+        hdas = [ self.add_hda_to_history( history, name=( 'hda-' + str( x ) ) ) for x in xrange( 3 ) ]
+        self.add_list_collection_to_history( history, hdas )
+        self.assertEqual( list( self.history_manager.contained( history ) ), hdas )
+
+    def test_subcontainers( self ):
+        user2 = self.user_manager.create( **user2_data )
+        history = self.history_manager.create( name='history', user=user2 )
+
+        self.log( "calling subcontainers on an empty history should return an empty list" )
+        self.assertEqual( [], list( self.history_manager.subcontainers( history ) ) )
+
+        self.log( "calling subcontainers on an history with both hdas and collections should return only collections" )
+        hdas = [ self.add_hda_to_history( history, name=( 'hda-' + str( x ) ) ) for x in xrange( 3 ) ]
+        hdca = self.add_list_collection_to_history( history, hdas )
+        subcontainers = list( self.history_manager.subcontainers( history ) )
+        self.assertEqual( subcontainers, [ hdca ] )
+
+
 # =============================================================================
 if __name__ == '__main__':
     # or more generally, nosetests test_resourcemanagers.py -s -v