diff --git a/lib/galaxy/managers/containers.py b/lib/galaxy/managers/containers.py index 95f9bab00ba..7f38d99f50f 100644 --- a/lib/galaxy/managers/containers.py +++ b/lib/galaxy/managers/containers.py @@ -3,10 +3,9 @@ Manager mixins to unify the interface into things that can contain: Datasets 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 ): diff --git a/lib/galaxy/managers/histories.py b/lib/galaxy/managers/histories.py index d2371fa8732..6fe37c35aa4 100644 --- a/lib/galaxy/managers/histories.py +++ b/lib/galaxy/managers/histories.py @@ -4,10 +4,12 @@ Manager and Serializer for histories. 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 @@ -15,7 +17,7 @@ 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 @@ class HistoryManager( sharable.SharableModelManager, deletable.PurgableManagerMi 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 ): """ diff --git a/test/unit/managers/base.py b/test/unit/managers/base.py index 6284ded7d63..b51f7976a2b 100644 --- a/test/unit/managers/base.py +++ b/test/unit/managers/base.py @@ -120,6 +120,21 @@ class BaseTestCase( test_utils.unittest.TestCase ): 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 diff --git a/test/unit/managers/test_CollectionManager.py b/test/unit/managers/test_CollectionManager.py index 6d5f3da4ffe..655f5d20a30 100644 --- a/test/unit/managers/test_CollectionManager.py +++ b/test/unit/managers/test_CollectionManager.py @@ -11,6 +11,7 @@ test_utils = imp.load_source( 'test_utils', 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 @@ user3_data = dict( email='user3@user3.user3', username='user3', password=default # ============================================================================= -class DatasetCollectionManagerTestCase( BaseTestCase ): +class DatasetCollectionManagerTestCase( BaseTestCase, CreatesCollectionsMixin ): def set_up_managers( self ): super( DatasetCollectionManagerTestCase, self ).set_up_managers() @@ -34,18 +35,6 @@ class DatasetCollectionManagerTestCase( BaseTestCase ): 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 ) diff --git a/test/unit/managers/test_DatasetManager.py b/test/unit/managers/test_DatasetManager.py index 140ed02f2d5..8402321db7e 100644 --- a/test/unit/managers/test_DatasetManager.py +++ b/test/unit/managers/test_DatasetManager.py @@ -112,7 +112,8 @@ class DatasetManagerTestCase( BaseTestCase ): 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 @@ class DatasetManagerTestCase( BaseTestCase ): 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( diff --git a/test/unit/managers/test_HistoryManager.py b/test/unit/managers/test_HistoryManager.py index 4bb30b74bf6..e5254d8e106 100644 --- a/test/unit/managers/test_HistoryManager.py +++ b/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 model 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 @@ class HistoryFiltersTestCase( BaseTestCase ): 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