# -*- coding: utf-8 -*- """ """ import os import imp import unittest test_utils = imp.load_source( 'test_utils', os.path.join( os.path.dirname( __file__), '../unittest_utils/utility.py' ) ) from galaxy import eggs eggs.require( 'SQLAlchemy >= 0.4' ) import sqlalchemy from galaxy import model from galaxy import exceptions from base import BaseTestCase from galaxy.managers import users from galaxy.managers import histories # ============================================================================= 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 ) # ============================================================================= class UserManagerTestCase( BaseTestCase ): def test_framework( self ): self.log( "(for testing) should have admin_user, and admin_user is current" ) self.assertEqual( self.trans.user, self.admin_user ) def test_base( self ): self.log( "should be able to create a user" ) user2 = self.user_manager.create( **user2_data ) self.assertIsInstance( user2, model.User ) self.assertIsNotNone( user2.id ) self.assertEqual( user2.email, user2_data[ 'email' ] ) self.assertEqual( user2.password, default_password ) user3 = self.user_manager.create( **user3_data ) self.log( "should be able to query" ) users = self.trans.sa_session.query( model.User ).all() self.assertEqual( self.user_manager.list(), users ) self.assertEqual( self.user_manager.by_id( user2.id ), user2 ) self.assertEqual( self.user_manager.by_ids( [ user3.id, user2.id ] ), [ user3, user2 ] ) self.log( "should be able to limit and offset" ) self.assertEqual( self.user_manager.list( limit=1 ), users[0:1] ) self.assertEqual( self.user_manager.list( offset=1 ), users[1:] ) self.assertEqual( self.user_manager.list( limit=1, offset=1 ), users[1:2] ) self.assertEqual( self.user_manager.list( limit=0 ), [] ) self.assertEqual( self.user_manager.list( offset=3 ), [] ) self.log( "should be able to order" ) self.assertEqual( self.user_manager.list( order_by=( sqlalchemy.desc( model.User.create_time ) ) ), [ user3, user2, self.admin_user ] ) def test_invalid_create( self ): self.user_manager.create( **user2_data ) self.log( "emails must be unique" ) self.assertRaises( exceptions.Conflict, self.user_manager.create, **dict( email='user2@user2.user2', username='user2a', password=default_password ) ) self.log( "usernames must be unique" ) self.assertRaises( exceptions.Conflict, self.user_manager.create, **dict( email='user2a@user2.user2', username='user2', password=default_password ) ) def test_email_queries( self ): user2 = self.user_manager.create( **user2_data ) user3 = self.user_manager.create( **user3_data ) self.log( "should be able to query by email" ) self.assertEqual( self.user_manager.by_email( user2_data[ 'email' ] ), user2 ) # note: sorted by email alpha self.assertEqual( self.user_manager.by_email_like( '%@%' ), [ self.admin_user, user2, user3 ] ) def test_admin( self ): user2 = self.user_manager.create( **user2_data ) self.log( "should be able to test whether admin" ) self.assertTrue( self.user_manager.is_admin( self.admin_user ) ) self.assertFalse( self.user_manager.is_admin( user2 ) ) self.assertEqual( self.user_manager.admins(), [ self.admin_user ] ) self.assertRaises( exceptions.AdminRequiredException, self.user_manager.error_unless_admin, user2 ) self.assertEqual( self.user_manager.error_unless_admin( self.admin_user ), self.admin_user ) def test_anonymous( self ): anon = None user2 = self.user_manager.create( **user2_data ) self.log( "should be able to tell if a user is anonymous" ) self.assertRaises( exceptions.AuthenticationFailed, self.user_manager.error_if_anonymous, anon ) self.assertEqual( self.user_manager.error_if_anonymous( user2 ), user2 ) def test_current( self ): user2 = self.user_manager.create( **user2_data ) self.log( "should be able to tell if a user is the current (trans) user" ) self.assertEqual( self.user_manager.current_user( self.trans ), self.admin_user ) self.assertNotEqual( self.user_manager.current_user( self.trans ), user2 ) def test_api_keys( self ): user2 = self.user_manager.create( **user2_data ) self.log( "should return None if no APIKey has been created" ) self.assertEqual( self.user_manager.valid_api_key( user2 ), None ) self.log( "should be able to generate and retrieve valid api key" ) user2_api_key = self.user_manager.create_api_key( user2 ) self.assertIsInstance( user2_api_key, basestring ) self.assertEqual( self.user_manager.valid_api_key( user2 ).key, user2_api_key ) self.log( "should return the most recent (i.e. most valid) api key" ) user2_api_key_2 = self.user_manager.create_api_key( user2 ) self.assertEqual( self.user_manager.valid_api_key( 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_manager.create( **user2_data ) self.log( 'should have a summary view' ) summary_view = self.user_serializer.serialize_to_view( 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( user, default_view='summary' ) self.assertKeys( default_view, self.user_serializer.views[ 'summary' ] ) self.log( 'should have a serializer for all serializable keys' ) 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, self.TYPES_NEEDING_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_manager.create( **user2_data ) self.log( 'should be able to use keys with views' ) serialized = self.user_serializer.serialize_to_view( 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( user, keys=[ 'tags_used', 'is_admin' ] ) self.assertKeys( serialized, [ 'tags_used', 'is_admin' ] ) def test_serializers( self ): user = self.user_manager.create( **user2_data ) all_keys = list( self.user_serializer.serializable_keyset ) serialized = self.user_serializer.serialize( user, all_keys, trans=self.trans ) # 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.assertIsJsonifyable( serialized ) 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.log( 'should be able to serialize anonymous user' ) serialized = self.user_serializer.serialize_to_view( anonym, view='detailed', trans=self.trans ) 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.assertIsJsonifyable( serialized ) # ============================================================================= 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 unittest.main()