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

Managers, Users: corrections and refactoring, expansion of tests

Browse Source
This commit is contained in:
Carl Eberhard
2015-03-30 18:00:13 -04:00
parent e5bf7822d1
commit 188ff0076f
5 changed files with 153 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/users.py
+16 -14
View File
@@ -260,7 +260,7 @@ class UserSerializer( base.ModelSerializer, deletable.PurgableSerializerMixin ):
'total_disk_usage',
'nice_total_disk_usage',
'quota_percent'
'quota_percent',
#'deleted',
#'purged',
@@ -284,13 +284,24 @@ class UserSerializer( base.ModelSerializer, deletable.PurgableSerializerMixin ):
'is_admin' : lambda t, i, k: self.user_manager.is_admin( t, i ),
'total_disk_usage' : lambda t, i, k: float( i.total_disk_usage ),
'quota_percent' : lambda t, i, k: self.user_manager.quota( t, i ),
'quota_percent' : lambda t, i, k: self.user_manager.quota( i ),
'tags_used' : lambda t, i, k: self.user_manager.tags_used( t, i ),
#TODO: 'has_requests' is more apt
'requests' : lambda t, i, k: self.user_manager.has_requests( t, i )
})
class CurrentUserSerializer( UserSerializer ):
def serialize( self, trans, user, keys ):
"""
Override to return at least some usage info if user is anonymous.
"""
if self.user_manager.is_anonymous( user ):
return self.serialize_current_anonymous_user( trans, user, keys )
return super( UserSerializer, self ).serialize( trans, user, keys )
def serialize_current_anonymous_user( self, trans, user, keys ):
# use the current history if any to get usage stats for trans' anonymous user
#TODO: might be better as sep. Serializer class
@@ -304,7 +315,7 @@ class UserSerializer( base.ModelSerializer, deletable.PurgableSerializerMixin ):
# a very small subset of keys available
values = {
'id' : None,
'total_disk_usage' : int( usage ),
'total_disk_usage' : float( usage ),
'nice_total_disk_usage' : util.nice_size( usage ),
'quota_percent' : percent,
}
@@ -314,25 +325,16 @@ class UserSerializer( base.ModelSerializer, deletable.PurgableSerializerMixin ):
serialized[ key ] = values[ key ]
return serialized
def serialize( self, trans, user, keys ):
"""
Override to return at least some usage info if user is anonymous.
"""
if self.user_manager.is_anonymous( user ):
return self.serialize_current_anonymous_user( trans, user, keys )
return super( UserSerializer, self ).serialize( trans, user, keys )
class AdminUserFilters( base.ModelFilterParser, deletable.PurgableFiltersMixin ):
class AdminUserFilterParser( base.ModelFilterParser, deletable.PurgableFiltersMixin ):
model_class = model.User
def _add_parsers( self ):
super( AdminUserFilters, self )._add_parsers()
super( AdminUserFilterParser, self )._add_parsers()
deletable.PurgableFiltersMixin._add_parsers( self )
#PRECONDITION: user making the query has been verified as an admin
self.orm_filter_parsers.update({
'name' : { 'op': ( 'eq', 'contains', 'like' ) },
'email' : { 'op': ( 'eq', 'contains', 'like' ) },
'username' : { 'op': ( 'eq', 'contains', 'like' ) },
'active' : { 'op': ( 'eq' ) },
test/unit/managers/mock.py
+9 -4
View File
@@ -16,6 +16,8 @@ from galaxy import objectstore
from galaxy.model import mapping
from galaxy.util.bunch import Bunch
from galaxy.managers import tags
from galaxy import quota
# =============================================================================
class OpenObject( object ):
@@ -29,6 +31,8 @@ class MockApp( object ):
self.model = mapping.init( "/tmp", "sqlite:///:memory:", create_tables=True, object_store=self.object_store )
self.security_agent = self.model.security_agent
self.visualizations_registry = MockVisualizationsRegistry()
self.tag_handler = tags.GalaxyTagManager( self )
self.quota_agent = quota.QuotaAgent( self.model )
class MockAppConfig( Bunch ):
def __init__( self, **kwargs ):
@@ -53,10 +57,15 @@ class MockWebapp( object ):
def __init__( self, **kwargs ):
self.name = kwargs.get( 'name', 'galaxy' )
class MockVisualizationsRegistry( object ):
def get_visualizations( self, trans, target ):
return []
class MockTrans( object ):
def __init__( self, user=None, history=None, **kwargs ):
self.app = MockApp( **kwargs )
self.model = self.app.model
self.webapp = MockWebapp( **kwargs )
self.sa_session = self.app.model.session
@@ -91,7 +100,3 @@ class MockTrans( object ):
template = template_lookup.get_template( filename )
template.output_encoding = 'utf-8'
return template.render( **kwargs )
class MockVisualizationsRegistry( object ):
def get_visualizations( self, trans, target ):
return []
test/unit/managers/test_HistoryManager.py
-1
View File
@@ -316,7 +316,6 @@ testable_url_for = lambda *a, **k: '(fake url): %s, %s' % ( a, k )
HistorySerializer.url_for = staticmethod( testable_url_for )
hdas.HDASerializer.url_for = staticmethod( testable_url_for )
class HistorySerializerTestCase( BaseTestCase ):
def set_up_managers( self ):
test/unit/managers/test_ModelManager.py
+9 -3
View File
@@ -88,23 +88,29 @@ class BaseTestCase( unittest.TestCase ):
def assertEncodedId( self, item ):
if not isinstance( item, basestring ):
self.fail( 'Non-string: ' + type( item ) )
self.fail( 'Non-string: ' + str( type( item ) ) )
# TODO: len mod 8 and hex re
self.assertTrue( True, 'is id: ' + item )
def assertDate( self, item ):
if not isinstance( item, basestring ):
self.fail( 'Non-string: ' + type( item ) )
self.fail( 'Non-string: ' + str( type( item ) ) )
# TODO: no great way to parse this fully (w/o python-dateutil)
# TODO: re?
self.assertTrue( True, 'is date: ' + item )
def assertUUID( self, item ):
if not isinstance( item, basestring ):
self.fail( 'Non-string: ' + type( item ) )
self.fail( 'Non-string: ' + str( type( item ) ) )
# TODO: re for d4d76d69-80d4-4ed7-80c7-211ebcc1a358
self.assertTrue( True, 'is uuid: ' + item )
def assertORMFilter( self, item, msg=None ):
if not isinstance( item, sqlalchemy.sql.elements.BinaryExpression ):
self.fail( 'Not an orm filter: ' + str( type( item ) ) )
# TODO: re for d4d76d69-80d4-4ed7-80c7-211ebcc1a358
self.assertTrue( True, msg or ( 'is an orm filter: ' + item ) )
# =============================================================================
test/unit/managers/test_UserManager.py
+119 -2
View File
@@ -5,6 +5,7 @@ import sys
import os
import pprint
import unittest
import json
__GALAXY_ROOT__ = os.getcwd() + '/../../../'
sys.path.insert( 1, __GALAXY_ROOT__ + 'lib' )
@@ -19,14 +20,15 @@ from galaxy.util.bunch import Bunch
import mock
from test_ModelManager import BaseTestCase
from galaxy.managers import users
from galaxy.managers import histories
# =============================================================================
admin_email = 'admin@admin.admin'
admin_users = admin_email
default_password = '123456'
user2_data = dict( email='user2@user2.user2', username='user2', password=default_password )
user3_data = dict( email='user3@user3.user3', username='user3', password=default_password )
user4_data = dict( email='user4@user4.user4', username='user4', password=default_password )
# =============================================================================
@@ -130,6 +132,121 @@ class UserManagerTestCase( BaseTestCase ):
self.assertEqual( self.user_mgr.valid_api_key( self.trans, user2 ).key, user2_api_key_2 )
# =============================================================================
class UserSerializerTestCase( BaseTestCase ):
def set_up_managers( self ):
super( UserSerializerTestCase, self ).set_up_managers()
self.user_serializer = users.UserSerializer( self.app )
def test_views( self ):
user = self.user_mgr.create( self.trans, **user2_data )
self.log( 'should have a summary view' )
summary_view = self.user_serializer.serialize_to_view( self.trans, user, view='summary' )
self.assertKeys( summary_view, self.user_serializer.views[ 'summary' ] )
self.log( 'should have the summary view as default view' )
default_view = self.user_serializer.serialize_to_view( self.trans, user, default_view='summary' )
self.assertKeys( summary_view, self.user_serializer.views[ 'summary' ] )
self.log( 'should have a serializer for all serializable keys' )
need_no_serializers = ( basestring, bool, type( None ) )
for key in self.user_serializer.serializable_keyset:
instantiated_attribute = getattr( user, key, None )
if not ( ( key in self.user_serializer.serializers )
or ( isinstance( instantiated_attribute, need_no_serializers ) ) ):
self.fail( 'no serializer for: %s (%s)' % ( key, instantiated_attribute ) )
else:
self.assertTrue( True, 'all serializable keys have a serializer' )
def test_views_and_keys( self ):
user = self.user_mgr.create( self.trans, **user2_data )
self.log( 'should be able to use keys with views' )
serialized = self.user_serializer.serialize_to_view( self.trans, user,
view='summary', keys=[ 'create_time' ] )
self.assertKeys( serialized,
self.user_serializer.views[ 'summary' ] + [ 'create_time' ] )
self.log( 'should be able to use keys on their own' )
serialized = self.user_serializer.serialize_to_view( self.trans, user,
keys=[ 'tags_used', 'is_admin' ] )
self.assertKeys( serialized, [ 'tags_used', 'is_admin' ] )
def test_serializers( self ):
user = self.user_mgr.create( self.trans, **user2_data )
all_keys = self.user_serializer.serializable_keyset
serialized = self.user_serializer.serialize( self.trans, user, all_keys )
# pprint.pprint( serialized )
self.log( 'everything serialized should be of the proper type' )
self.assertEncodedId( serialized[ 'id' ] )
self.assertDate( serialized[ 'create_time' ] )
self.assertDate( serialized[ 'update_time' ] )
self.assertIsInstance( serialized[ 'deleted' ], bool )
self.assertIsInstance( serialized[ 'purged' ], bool )
# self.assertIsInstance( serialized[ 'active' ], bool )
self.assertIsInstance( serialized[ 'is_admin' ], bool )
self.assertIsInstance( serialized[ 'total_disk_usage' ], float )
self.assertIsInstance( serialized[ 'nice_total_disk_usage' ], basestring )
self.assertIsInstance( serialized[ 'quota_percent' ], ( type( None ), float ) )
self.assertIsInstance( serialized[ 'tags_used' ], list )
self.assertIsInstance( serialized[ 'requests' ], list )
self.log( 'serialized should jsonify well' )
self.assertIsInstance( json.dumps( serialized ), basestring )
class CurrentUserSerializerTestCase( BaseTestCase ):
def set_up_managers( self ):
super( CurrentUserSerializerTestCase, self ).set_up_managers()
self.history_manager = histories.HistoryManager( self.app )
self.user_serializer = users.CurrentUserSerializer( self.app )
def test_anonymous( self ):
anonym = None
# need a history here for total_disk_usage
self.trans.set_history( self.history_manager.create( self.trans ) )
self.log( 'should be able to serialize anonymous user' )
serialized = self.user_serializer.serialize_to_view( self.trans, anonym, view='detailed' )
self.assertKeys( serialized,
[ 'id', 'total_disk_usage', 'nice_total_disk_usage', 'quota_percent' ] )
self.log( 'anonymous\'s id should be None' )
self.assertEqual( serialized[ 'id' ], None )
self.log( 'everything serialized should be of the proper type' )
self.assertIsInstance( serialized[ 'total_disk_usage' ], float )
self.assertIsInstance( serialized[ 'nice_total_disk_usage' ], basestring )
self.assertIsInstance( serialized[ 'quota_percent' ], ( type( None ), float ) )
self.log( 'serialized should jsonify well' )
self.assertIsInstance( json.dumps( serialized ), basestring )
# =============================================================================
class AdminUserFilterParserTestCase( BaseTestCase ):
def set_up_managers( self ):
super( AdminUserFilterParserTestCase, self ).set_up_managers()
self.filter_parser = users.AdminUserFilterParser( self.app )
def test_parsable( self ):
self.log( 'the following filters should be parsable' )
self.assertORMFilter( self.filter_parser.parse_filter( 'email', 'eq', 'wot' ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'email', 'contains', 'wot' ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'email', 'like', 'wot' ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'username', 'eq', 'wot' ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'username', 'contains', 'wot' ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'username', 'like', 'wot' ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'active', 'eq', True ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'disk_usage', 'le', 500000.00 ) )
self.assertORMFilter( self.filter_parser.parse_filter( 'disk_usage', 'ge', 500000.00 ) )
# =============================================================================
if __name__ == '__main__':
# or more generally, nosetests test_resourcemanagers.py -s -v