opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-24 02:04:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

More documentation; minor fixes

Browse Source
This commit is contained in:
Carl Eberhard
2015-01-07 14:55:37 -05:00
parent 9e25551161
commit 2806fbe83e
8 changed files with 291 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/base.py
+35 -7
View File
@@ -131,14 +131,15 @@ class ModelManager( object ):
foreign_key_name = None
def __init__( self, app ):
"""
"""
self.app = app
#NOTE: at this layer, all ids are expected to be decoded and in int form
# ------------------------------------------------------------------------- get/read/query
def query( self, trans, eagerloads=True, filters=None, order_by=None, limit=None, offset=None, **kwargs ):
"""
Return a basic query from model_class, filters, order_by, and limit and offset.
Set eagerloads to False to disable them for this query.
"""
#print 'query:', eagerloads, filters, order_by, limit, offset, kwargs
query = trans.sa_session.query( self.model_class )
@@ -154,6 +155,7 @@ class ModelManager( object ):
def _apply_filters( self, query, filters ):
"""
Add any filters to the given query.
"""
if filters is None:
return query
@@ -167,9 +169,11 @@ class ModelManager( object ):
def _munge_filters( self, filtersA, filtersB ):
"""
Combine two lists into a single list.
(While allowing them to be None, non-lists, or lists.)
"""
# ... which sounds like a euphamism for something
final_filters = [ self.model_class.id == id ]
if filtersA is None:
return filtersB
if filtersB is None:
@@ -182,12 +186,16 @@ class ModelManager( object ):
def _apply_order_by_limit_offset( self, query, order_by, limit, offset ):
"""
Return the query after adding the order_by, limit, and offset clauses.
"""
query = self._apply_order_by( query, order_by )
return self._apply_limit_offset( query, limit, offset )
def _apply_order_by( self, query, order_by ):
"""
Return the query after adding the order_by clauses.
Use the manager's default_order_by if order_by is None.
"""
if order_by is None:
#TODO: allow non-tuple default_order_by
@@ -199,6 +207,7 @@ class ModelManager( object ):
def _apply_limit_offset( self, query, limit, offset ):
"""
Return the query after applying the given limit and offset (if not None).
"""
if limit is not None:
query = query.limit( limit )
@@ -207,37 +216,47 @@ class ModelManager( object ):
return query
# ......................................................................... common queries
#TODO: needed?
#TODO: needed?
def one( self, trans, **kwargs ):
"""
Sends kwargs to build the query and returns one and only one model.
:raises exceptions.ObjectNotFound: if no model is found
:raises exceptions.Conflict: if more than one model is found
"""
# overridden to raise serializable errors
try:
return self.query( trans, **kwargs ).one()
# re-raise as serializable errors
except sqlalchemy.orm.exc.NoResultFound, not_found:
raise exceptions.ObjectNotFound( self.model_class.__name__ + ' not found' )
#TODO: exceptions.InconsistentDatabase?
except sqlalchemy.orm.exc.MultipleResultsFound, multi_found:
raise exceptions.Conflict( 'found more than one ' + self.model_class.__name__ )
def list( self, trans, **kwargs ):
def list( self, trans, query=None, **kwargs ):
"""
Sends kwargs to build the query return all models found.
"""
return self.query( trans, **kwargs ).all()
query = query or self.query( trans, **kwargs )
return query.all()
def by_id( self, trans, id, **kwargs ):
"""
Gets a model by primary id.
"""
id_filter = self.model_class.id == id
return self.one( trans, filters=id_filter, **kwargs )
def _query_by_ids( self, trans, ids, filters=None, **kwargs ):
"""
Builds a query to find a list of models with the given list of `ids`.
"""
ids_filter = self.model_class.id.in_( ids )
return self.query( trans, filters=self._munge_filters( ids_filter, filters ), **kwargs )
def by_ids( self, trans, ids, **kwargs ):
"""
Returns an in-order list of models with the matching ids in `ids`.
"""
found = self._query_by_ids( trans, ids, **kwargs ).all()
#TODO: this does not order by the original 'ids' array
@@ -277,7 +296,9 @@ class ModelManager( object ):
# ------------------------------------------------------------------------- create
def create( self, trans, flush=True, *args, **kwargs ):
"""
Generically create a new model.
"""
# override in subclasses
item = self.model_class( *args, **kwargs )
trans.sa_session.add( item )
if flush:
@@ -286,12 +307,16 @@ class ModelManager( object ):
def copy( self, trans, item, **kwargs ):
"""
Clone or copy an item.
"""
raise exceptions.NotImplemented( 'Abstract method' )
# ------------------------------------------------------------------------- update
def update( self, trans, item, new_values, flush=True, **kwargs ):
"""
Given a dictionary of new values, update `item` and return it.
..note: NO validation or deserialization occurs here.
"""
trans.sa_session.add( item )
for key, value in new_values.items():
@@ -301,8 +326,10 @@ class ModelManager( object ):
trans.sa_session.flush()
return item
#TODO: yagni?
def associate( self, trans, associate_with, item, foreign_key_name=None ):
"""
Generically associate `item` with `associate_with` based on `foreign_key_name`.
"""
foreign_key_name = foreign_key_name or self.foreign_key_name
setattr( associate_with, foreign_key_name, item )
@@ -310,6 +337,7 @@ class ModelManager( object ):
def query_associated( self, trans, associated_model_class, item, foreign_key_name=None ):
"""
Generically query other items that have been associated with this `item`.
"""
foreign_key_name = foreign_key_name or self.foreign_key_name
foreign_key = getattr( associated_model_class, foreign_key_name )
lib/galaxy/managers/datasets.py
+86 -28
View File
@@ -32,6 +32,7 @@ class DatasetManager( base.ModelManager, base.AccessibleModelInterface, base.Pur
# default to NEW state on new datasets
kwargs.update( dict( state=( kwargs.get( 'state', model.Dataset.states.NEW ) ) ) )
dataset = model.Dataset( **kwargs )
trans.sa_session.add( dataset )
if flush:
trans.sa_session.flush()
@@ -54,6 +55,12 @@ class DatasetManager( base.ModelManager, base.AccessibleModelInterface, base.Pur
pass
# ......................................................................... delete
def error_unless_dataset_purge_allowed( self, trans, item, msg=None ):
if not trans.app.config.allow_user_dataset_purge:
msg = msg or 'This instance does not allow user dataset purging'
raise exceptions.ConfigDoesNotAllowException( msg )
return item
def purge( self, trans, dataset, flush=True ):
"""
"""
@@ -78,42 +85,91 @@ class DatasetManager( base.ModelManager, base.AccessibleModelInterface, base.Pur
return True
return False
def _access_permission( self, trans, dataset, user=None, role=None ):
"""
"""
security_agent = trans.app.security_agent
access_action = security_agent.permitted_actions.DATASET_ACCESS.action
# get a list of role ids to check access for (defaulting to all_roles)
user_roles = [ role ] if role else user.all_roles()
user_role_ids = [ role.id for role in user_roles ]
query = ( trans.sa_session.query( model.DatasetPermissions )
.filter( model.DatasetPermissions.action == access_action )
.filter( model.DatasetPermissions.dataset == dataset )
.filter( model.DatasetPermissions.role_id.in_( user_role_ids ) ) )
perms = query.all()
if len( perms ):
return perms[0]
return None
def _create_access_permission( self, trans, dataset, role, flush=True ):
"""
"""
access_action = trans.app.security_agent.permitted_actions.DATASET_ACCESS.action
permission = model.DatasetPermissions( access_action, dataset, role )
trans.sa_session.add( permission )
if flush:
trans.sa_session.flush()
return permission
def has_access_permission( self, trans, dataset, user ):
"""
"""
roles = user.roles if user else []
roles = user.all_roles() if user else []
return trans.app.security_agent.can_access_dataset( roles, dataset )
def give_access_permission( self, trans, dataset, user ):
"""
"""
#permission = {}
##TODO:?? is this correct and reliable?
#user_private_role = self.user_mgr.private_role( trans, user )
#permission[ trans.app.security_agent.permitted_actions.DATASET_ACCESS ] = [ user_private_role ]
#trans.app.security_agent.set_dataset_permission( dataset, permission )
pass
#TODO: this needs work
#def give_access_permission( self, trans, dataset, user, flush=True ):
# """
# """
# # for now, use the user's private role
# security_agent = trans.app.security_agent
# user_role = security_agent.get_private_user_role( user )
#
# existing_permission = self._access_permission( trans, dataset, role=user_role )
# print 'existing access_perm:', existing_permission
# if existing_permission:
# return dataset
#
# permission = self._create_access_permission( trans, dataset, user_role )
# print 'access_roles:', [ role.name for role in dataset.get_access_roles( trans ) ]
#
# #access_action = security_agent.permitted_actions.DATASET_ACCESS.action
# #access_action = security_agent.get_action( access_action )
# #permissions = { access_action : [ user_role ] }
# #security_agent.set_dataset_permission( dataset, permissions )
#
# #trans.sa_session.add( dataset )
# #if flush:
# # trans.sa_session.flush()
#
# dbl_chk = self._access_permission( trans, dataset, role=user_role )
# print 'access_perm:', dbl_chk
#
# return dataset
def remove_access_permission( self, trans, dataset, user ):
"""
"""
pass
#def remove_access_permission( self, trans, dataset, user ):
# """
# """
# pass
# ......................................................................... manage/modify
def has_manage_permission( self, trans, dataset, user ):
"""
"""
pass
def give_manage_permission( self, trans, dataset, user ):
"""
"""
pass
def remove_manage_permission( self, trans, dataset, user ):
"""
"""
pass
#def has_manage_permission( self, trans, dataset, user ):
# """
# """
# pass
#
#def give_manage_permission( self, trans, dataset, user ):
# """
# """
# pass
#
#def remove_manage_permission( self, trans, dataset, user ):
# """
# """
# pass
#TODO: implement above for groups
#TODO: datatypes?
@@ -183,9 +239,11 @@ class DatasetAssociationManager( base.ModelManager, base.AccessibleModelInterfac
def metadata( self, trans, dataset_assoc ):
"""
"""
# get metadata
pass
def is_being_used( self, trans, dataset_assoc ):
"""
"""
#TODO: check history_associations, library_associations
pass
lib/galaxy/managers/hdas.py
+5 -3
View File
@@ -37,11 +37,11 @@ class HDAManager( datasets.DatasetAssociationManager, base.OwnableModelInterface
def is_accessible( self, trans, hda, user ):
"""
"""
if self.is_owned( trans, hda, user ):
if self.is_owner( trans, hda, user ):
return True
return super( HDAManager, self ).is_accessible( trans, hda, user )
def is_owned( self, trans, hda, user ):
def is_owner( self, trans, hda, user ):
"""
Use history to see if current user owns HDA.
"""
@@ -116,7 +116,9 @@ class HDAManager( datasets.DatasetAssociationManager, base.OwnableModelInterface
def purge( self, trans, hda ):
"""
"""
self.error_unless_dataset_purge_allowed( trans, hda )
# error here if disallowed - before jobs are stopped
#TODO: poss. move to DatasetAssociationManager
self.dataset_mgr.error_unless_dataset_purge_allowed( trans, hda )
super( HDAManager, self ).purge( trans, hda, flush=True )
# signal to stop the creating job?
lib/galaxy/managers/histories.py
+2 -2
View File
@@ -225,7 +225,7 @@ class HistoryManager( sharable.SharableModelManager, base.PurgableModelInterface
## =============================================================================
class HistorySerializer( base.ModelSerializer ):
class HistorySerializer( sharable.SharableModelSerializer ):
"""
Interface/service object for serializing histories into dictionaries.
"""
@@ -321,7 +321,7 @@ class HistorySerializer( base.ModelSerializer ):
# =============================================================================
class HistoryDeserializer( base.ModelDeserializer ):
class HistoryDeserializer( sharable.SharableModelDeserializer ):
"""
Interface/service object for validating and deserializing dictionaries into histories.
"""
lib/galaxy/managers/sharable.py
+64 -26
View File
@@ -22,12 +22,12 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
can be published effectively making it available to all other Users
can be rated
"""
# histories, pages, stored workflows, visualizations
# e.g. histories, pages, stored workflows, visualizations
# the model used for UserShareAssociations with this model
user_share_model = None
def __init__( self, app ):
"""
"""
super( SharableModelManager, self ).__init__( app )
# user manager is needed to check access/ownership/admin
self.user_mgr = users.UserManager( app )
@@ -40,22 +40,21 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
filters=self._munge_filters( user_filter, filters )
return self.query( trans, filters=filters, **kwargs )
def by_user( self, trans, user, filters=None, **kwargs ):
def by_user( self, trans, user, **kwargs ):
"""
"""
user_filter = self.model_class.user_id == user.id
filters=self._munge_filters( user_filter, filters )
#note: no ownership check needed
return self.list( trans, filters=filters, **kwargs )
query = self._query_by_user( trans, user, **kwargs )
return self.list( trans, query=query, **kwargs )
# ......................................................................... owned model interface
# this really is more of a 'check_*editable/write*' - in essence that's what check_ownership boils down to often
def is_owner( self, trans, item, user ):
"""
Return true if this sharable belongs to `user` (or `user` is an admin).
"""
if self.user_mgr.is_anonymous( user ):
# note: this needs to be overridden in the case of anon users and session.current_history
return False
# ... effectively a good fit to have this here, but not semantically
if self.user_mgr.is_admin( trans, user ):
return True
return item.user == user
@@ -63,6 +62,8 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
# ......................................................................... accessible interface
def is_accessible( self, trans, item, user ):
"""
If the item is importable, is owned by `user`, or (the valid) `user`
is in 'users shared with' list for the item: return True.
"""
if item.importable:
return True
@@ -76,6 +77,8 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
return False
# ......................................................................... importable
#TODO: the relationship between importable and published is ill defined
#??: can a published item be non-importable?
def make_importable( self, trans, item, flush=True ):
"""
Makes item accessible--viewable and importable--and sets item's slug.
@@ -95,7 +98,6 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
Does not flush/commit changes, however. Item must have name, user,
importable, and slug attributes.
"""
#??: can a published item be non-importable?
#if item.published:
# if unpublish:
# self.unpublish( trans, item, flush=False )
@@ -107,13 +109,21 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
trans.sa_session.flush()
return item
def list_importable( self, trans, **kwargs ):
#TODO:
"""
"""
pass
#def _query_importable( self, trans, filters=None, **kwargs ):
# """
# """
# importable_filter = self.model_class.importable == True
# filters = self._munge_filters( importable_filter, filters )
# return self.list( trans, filters=filters, **kwargs )
#
#def list_importable( self, trans, **kwargs ):
# """
# """
# query = self._query_importable( trans, user, **kwargs )
# return self.list( trans, query=query, **kwargs )
# ......................................................................... published
#TODO: the relationship between importable and published is ill defined
#def publish( self, trans, item, make_accessible=True, flush=True ):
# if not item.importable:
# if make_accessible:
@@ -124,6 +134,7 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
# a published item is also/already importable(=True) implicitly
def publish( self, trans, item, flush=True ):
"""
Set both the importable and published flags on `item` to True.
"""
if not item.importable:
self.make_importable( trans, item, flush=False )
@@ -135,6 +146,7 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
def unpublish( self, trans, item, flush=True ):
"""
Set the published flag on `item` to False.
"""
trans.sa_session.add( item )
item.published = False
@@ -142,11 +154,18 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
trans.sa_session.flush()
return item
def list_published( self, trans, **kwargs ):
#TODO:
"""
"""
pass
#def _query_published( self, trans, filters=None, **kwargs ):
# """
# """
# published_filter = self.model_class.published == True
# filters = self._munge_filters( published_filter, filters )
# return self.list( trans, filters=filters, **kwargs )
#
#def list_published( self, trans, **kwargs ):
# """
# """
# query = self._query_published( trans, user, **kwargs )
# return self.list( trans, query=query, **kwargs )
# ......................................................................... user sharing
# sharing is often done via a 3rd table btwn a User and an item -> a <Item>UserShareAssociation
@@ -186,23 +205,31 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
trans.sa_session.flush()
return item
def list_shared_with( self, trans, item, user, **kwargs ):
#TODO:
"""
"""
pass
#def _query_shared_with( self, trans, user, filters=None, **kwargs ):
##TODO:
# """
# """
# pass
#def list_published( self, trans, user, **kwargs ):
# """
# """
# query = self._query_shared_with( trans, user, **kwargs )
# return self.list( trans, query=query, **kwargs )
# ......................................................................... slugs
# slugs are human readable strings often used to link to sharable resources (replacing ids)
#TODO: as validator, deserializer, etc. (maybe another object entirely?)
def is_valid_slug( self, slug ):
"""
Returns true if slug is valid.
Returns true if `slug` is valid.
"""
VALID_SLUG_RE = re.compile( "^[a-z0-9\-]+$" )
return VALID_SLUG_RE.match( slug )
def set_slug( self, trans, item, new_slug, flush=True ):
"""
Validate and set the new slug for `item`.
"""
if not self.is_valid_slug( new_slug ):
raise exceptions.RequestParameterInvalidException( "Invalid slug", slug=new_slug )
@@ -228,6 +255,7 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
# Setup slug base.
if cur_slug is None or cur_slug == "":
# Item can have either a name or a title.
#TODO: this depends on the model having name or title
if hasattr( item, 'name' ):
item_name = item.name
elif hasattr( item, 'title' ):
@@ -282,6 +310,16 @@ class SharableModelManager( base.ModelManager, base.OwnableModelInterface, base.
# """ Returns item's name and link. """
# =============================================================================
class SharableModelSerializer( base.ModelSerializer ):
pass
# the only ones that needs any fns:
# user/user_id
# user_shares?
# username_and_slug?
# =============================================================================
class SharableModelDeserializer( base.ModelDeserializer ):
lib/galaxy/managers/users.py
+3 -3
View File
@@ -161,9 +161,6 @@ class UserManager( base.ModelManager ):
#TODO: catch NoResultFound
return trans.sa_session.query( model.APIKeys ).filter_by( user=user ).one()
#TODO: there is quite a bit of functionality around the user (authentication, quotas, groups/roles)
# most of which it may be unneccessary to have here
# ------------------------------------------------------------------------- roles
def private_role( self, trans, user ):
"""
@@ -171,6 +168,9 @@ class UserManager( base.ModelManager ):
#TODO: not sure we need to go through sec agent... it's just the first role of type private
return trans.app.security_agent.get_private_user_role( user )
#TODO: there is quite a bit of functionality around the user (authentication, quotas, groups/roles)
# most of which it may be unneccessary to have here
# =============================================================================
class UserSerializer( base.ModelSerializer ):
test/unit/managers/test_DatasetManager.py
+24 -5
View File
@@ -95,6 +95,30 @@ class DatasetManagerTestCase( BaseTestCase ):
self.assertRaises( exceptions.ConfigDoesNotAllowException, self.dataset_mgr.purge, self.trans, item1 )
self.assertFalse( item1.purged )
##TODO: I'm unclear as to how these work, so I'm kicking this down the road a bit....
#def test_access_permission( self ):
# owner = self.user_mgr.create( self.trans, **user2_data )
# dataset = self.dataset_mgr.create( self.trans )
# # giving one user access permission makes it non-public, removing access for anyone else
# self.dataset_mgr.give_access_permission( self.trans, dataset, owner )
#
# user3 = self.user_mgr.create( self.trans, **user3_data )
# user4 = self.user_mgr.create( self.trans,
# email='user4@user4.user4', username='user4', password=default_password )
#
# self.assertTrue( self.dataset_mgr.has_access_permission( self.trans, dataset, owner ) )
# self.assertFalse( self.dataset_mgr.has_access_permission( self.trans, dataset, user3 ) )
# self.assertFalse( self.dataset_mgr.has_access_permission( self.trans, dataset, user4 ) )
#
# # should be able to progressively add more roles without removing the previous
# self.dataset_mgr.give_access_permission( self.trans, dataset, user3 )
# self.assertTrue( self.dataset_mgr.has_access_permission( self.trans, dataset, user3 ) )
# self.assertTrue( self.dataset_mgr.has_access_permission( self.trans, dataset, owner ) )
# self.assertFalse( self.dataset_mgr.has_access_permission( self.trans, dataset, user4 ) )
#
# #self.assertTrue( self.dataset_mgr.is_accessible( self.trans, dataset, owner ) )
# #self.assertFalse( self.dataset_mgr.is_accessible( self.trans, dataset, non_owner ) )
def test_accessible( self ):
owner = self.user_mgr.create( self.trans, **user2_data )
non_owner = self.user_mgr.create( self.trans, **user3_data )
@@ -105,11 +129,6 @@ class DatasetManagerTestCase( BaseTestCase ):
for user in self.user_mgr.list( self.trans ):
self.assertTrue( self.dataset_mgr.is_accessible( self.trans, dataset, user ) )
#TODO: this is ... not working...
#self.dataset_mgr.give_access_permission( self.trans, dataset, owner )
#self.assertTrue( self.dataset_mgr.is_accessible( self.trans, dataset, owner ) )
#self.assertFalse( self.dataset_mgr.is_accessible( self.trans, dataset, non_owner ) )
# =============================================================================
class DatasetAssociationManagerTestCase( BaseTestCase ):
test/unit/managers/test_HDAManager.py
+72 -50
View File
@@ -36,6 +36,33 @@ class HDAManagerTestCase( BaseTestCase ):
self.dataset_mgr = DatasetManager( self.app )
self.hda_mgr = HDAManager( self.app )
def test_base( self ):
hda_model = model.HistoryDatasetAssociation
owner = self.user_mgr.create( self.trans, **user2_data )
history1 = self.history_mgr.create( self.trans, name='history1', user=owner )
hda1 = self.hda_mgr.create( self.trans, history=history1, hid=1 )
hda2 = self.hda_mgr.create( self.trans, history=history1, hid=2 )
hda3 = self.hda_mgr.create( self.trans, history=history1, hid=3 )
self.log( "should be able to query" )
hdas = self.trans.sa_session.query( hda_model ).all()
self.assertEqual( self.hda_mgr.list( self.trans ), hdas )
self.assertEqual( self.hda_mgr.one( self.trans, filters=( hda_model.id == hda1.id ) ), hda1 )
self.assertEqual( self.hda_mgr.by_id( self.trans, hda1.id ), hda1 )
self.assertEqual( self.hda_mgr.by_ids( self.trans, [ hda2.id, hda1.id ] ), [ hda2, hda1 ] )
self.log( "should be able to limit and offset" )
self.assertEqual( self.hda_mgr.list( self.trans, limit=1 ), hdas[0:1] )
self.assertEqual( self.hda_mgr.list( self.trans, offset=1 ), hdas[1:] )
self.assertEqual( self.hda_mgr.list( self.trans, limit=1, offset=1 ), hdas[1:2] )
self.assertEqual( self.hda_mgr.list( self.trans, limit=0 ), [] )
self.assertEqual( self.hda_mgr.list( self.trans, offset=3 ), [] )
self.log( "should be able to order" )
self.assertEqual( self.hda_mgr.list( self.trans, order_by=sqlalchemy.desc( hda_model.create_time ) ),
[ hda3, hda2, hda1 ] )
def test_create( self ):
owner = self.user_mgr.create( self.trans, **user2_data )
non_owner = self.user_mgr.create( self.trans, **user3_data )
@@ -87,33 +114,6 @@ class HDAManagerTestCase( BaseTestCase ):
# self.log( "should be able to copy an HDA" )
# hda2 = self.hda_mgr.copy_ldda( self.trans, history1, hda1 )
def test_base( self ):
hda_model = model.HistoryDatasetAssociation
owner = self.user_mgr.create( self.trans, **user2_data )
history1 = self.history_mgr.create( self.trans, name='history1', user=owner )
hda1 = self.hda_mgr.create( self.trans, history=history1, hid=1 )
hda2 = self.hda_mgr.create( self.trans, history=history1, hid=2 )
hda3 = self.hda_mgr.create( self.trans, history=history1, hid=3 )
self.log( "should be able to query" )
hdas = self.trans.sa_session.query( hda_model ).all()
self.assertEqual( self.hda_mgr.list( self.trans ), hdas )
self.assertEqual( self.hda_mgr.one( self.trans, filters=( hda_model.id == hda1.id ) ), hda1 )
self.assertEqual( self.hda_mgr.by_id( self.trans, hda1.id ), hda1 )
self.assertEqual( self.hda_mgr.by_ids( self.trans, [ hda2.id, hda1.id ] ), [ hda2, hda1 ] )
self.log( "should be able to limit and offset" )
self.assertEqual( self.hda_mgr.list( self.trans, limit=1 ), hdas[0:1] )
self.assertEqual( self.hda_mgr.list( self.trans, offset=1 ), hdas[1:] )
self.assertEqual( self.hda_mgr.list( self.trans, limit=1, offset=1 ), hdas[1:2] )
self.assertEqual( self.hda_mgr.list( self.trans, limit=0 ), [] )
self.assertEqual( self.hda_mgr.list( self.trans, offset=3 ), [] )
self.log( "should be able to order" )
self.assertEqual( self.hda_mgr.list( self.trans, order_by=sqlalchemy.desc( hda_model.create_time ) ),
[ hda3, hda2, hda1 ] )
def test_delete( self ):
owner = self.user_mgr.create( self.trans, **user2_data )
history1 = self.history_mgr.create( self.trans, name='history1', user=owner )
@@ -142,8 +142,6 @@ class HDAManagerTestCase( BaseTestCase ):
def test_purge_not_allowed( self ):
self.trans.app.config.allow_user_dataset_purge = False
self.assertRaises( exceptions.ConfigDoesNotAllowException,
self.hda_mgr.error_unless_dataset_purge_allowed, self.trans, None )
owner = self.user_mgr.create( self.trans, **user2_data )
history1 = self.history_mgr.create( self.trans, name='history1', user=owner )
@@ -155,27 +153,31 @@ class HDAManagerTestCase( BaseTestCase ):
self.assertRaises( exceptions.ConfigDoesNotAllowException, self.hda_mgr.purge, self.trans, item1 )
self.assertFalse( item1.purged )
#def test_ownable( self ):
# owner = self.user_mgr.create( self.trans, **user2_data )
# non_owner = self.user_mgr.create( self.trans, **user3_data )
#
# history1 = self.history_mgr.create( self.trans, name='history1', user=owner )
# dataset1 = self.dataset_mgr.create( self.trans )
# item1 = self.hda_mgr.create( self.trans, history1, dataset1 )
#
# self.log( "should be able to poll whether a given user owns an item" )
# self.assertTrue( self.hda_mgr.is_owner( self.trans, item1, owner ) )
# self.assertFalse( self.hda_mgr.is_owner( self.trans, item1, non_owner ) )
#
# self.log( "should raise an error when checking ownership with non-owner" )
# self.assertRaises( exceptions.ItemOwnershipException,
# self.hda_mgr.check_ownership, self.trans, item1, non_owner )
#
# self.log( "should not raise an error when checking ownership with owner" )
# self.assertEqual( self.hda_mgr.check_ownership( self.trans, item1, owner ), item1 )
#
# #self.log( "should not raise an error when checking ownership with admin" )
# #self.assertEqual( self.hda_mgr.check_ownership( self.trans, item1, self.admin_user ), item1 )
def test_ownable( self ):
owner = self.user_mgr.create( self.trans, **user2_data )
non_owner = self.user_mgr.create( self.trans, **user3_data )
history1 = self.history_mgr.create( self.trans, name='history1', user=owner )
dataset1 = self.dataset_mgr.create( self.trans )
item1 = self.hda_mgr.create( self.trans, history1, dataset1 )
self.log( "should be able to poll whether a given user owns an item" )
self.assertTrue( self.hda_mgr.is_owner( self.trans, item1, owner ) )
self.assertFalse( self.hda_mgr.is_owner( self.trans, item1, non_owner ) )
self.log( "should raise an error when checking ownership with non-owner" )
self.assertRaises( exceptions.ItemOwnershipException,
self.hda_mgr.error_unless_owner, self.trans, item1, non_owner )
self.log( "should raise an error when checking ownership with anonymous" )
self.assertRaises( exceptions.ItemOwnershipException,
self.hda_mgr.error_unless_owner, self.trans, item1, None )
self.log( "should not raise an error when checking ownership with owner" )
self.assertEqual( self.hda_mgr.error_unless_owner( self.trans, item1, owner ), item1 )
self.log( "should not raise an error when checking ownership with admin" )
self.assertEqual( self.hda_mgr.error_unless_owner( self.trans, item1, self.admin_user ), item1 )
def test_accessible( self ):
owner = self.user_mgr.create( self.trans, **user2_data )
@@ -189,6 +191,26 @@ class HDAManagerTestCase( BaseTestCase ):
for user in self.user_mgr.list( self.trans ):
self.assertTrue( self.hda_mgr.is_accessible( self.trans, item1, user ) )
#TODO: set perms on underlying dataset and then test accessible
def test_anon( self ):
anon_user = None
self.trans.set_user( anon_user )
history1 = self.history_mgr.create( self.trans, name='anon_history', user=anon_user )
self.trans.set_history( history1 )
dataset1 = self.dataset_mgr.create( self.trans )
item1 = self.hda_mgr.create( self.trans, history1, dataset1 )
self.log( "should not raise an error when checking ownership/access on anonymous' own dataset" )
self.assertTrue( self.hda_mgr.is_accessible( self.trans, item1, anon_user ) )
self.assertEqual( self.hda_mgr.error_unless_owner( self.trans, item1, anon_user ), item1 )
self.log( "should raise an error when checking ownership on anonymous' dataset with other user" )
non_owner = self.user_mgr.create( self.trans, **user3_data )
self.assertRaises( exceptions.ItemOwnershipException,
self.hda_mgr.error_unless_owner, self.trans, item1, non_owner )
# =============================================================================
if __name__ == '__main__':