diff --git a/lib/galaxy/security/__init__.py b/lib/galaxy/security/__init__.py
index afdb673c090..e21aa24c759 100644
--- a/lib/galaxy/security/__init__.py
+++ b/lib/galaxy/security/__init__.py
@@ -47,12 +47,8 @@ class RBACAgent:
raise "Unimplemented Method"
def get_private_user_role( self, user ):
raise "Unimplemented Method"
- def user_set_default_permissions( self, user, permissions = None, history = False, dataset = False ):
+ def user_set_default_permissions( self, user, permissions={}, history=False, dataset=False ):
raise "Unimplemented Method"
- def setup_new_user( self, user ):
- self.create_private_user_role( user )
- self.user_set_default_permissions( user, history = True, dataset = True )
- #self.associate_components( user=user, group=self.get_public_group() )
def history_set_default_permissions( self, history, permissions=None, dataset=False, bypass_manage_permission=False ):
raise "Unimplemented Method"
def set_dataset_permissions( self, dataset, permissions ):
@@ -169,7 +165,7 @@ class GalaxyRBACAgent( RBACAgent ):
else:
return None
return role
- def user_set_default_permissions( self, user, permissions = {}, history = False, dataset = False ):
+ def user_set_default_permissions( self, user, permissions = {}, history=False, dataset=False ):
if user is None:
return None
if not permissions:
@@ -195,7 +191,7 @@ class GalaxyRBACAgent( RBACAgent ):
for dup in user.default_permissions:
perms[ self.get_action( dup.action ) ].append( dup.role )
return perms
- def history_set_default_permissions( self, history, permissions = {}, dataset = False, bypass_manage_permission = False ):
+ def history_set_default_permissions( self, history, permissions={}, dataset=False, bypass_manage_permission=False ):
if not history.user:
return None # default permissions on a userless history are none
if not permissions:
diff --git a/lib/galaxy/web/controllers/admin.py b/lib/galaxy/web/controllers/admin.py
index 8593d0f65f5..dc72c913abe 100644
--- a/lib/galaxy/web/controllers/admin.py
+++ b/lib/galaxy/web/controllers/admin.py
@@ -485,6 +485,47 @@ class Admin( BaseController ):
trans.response.send_redirect( web.url_for( action='deleted_groups', msg=msg, messagetype='done' ) )
# Galaxy User Stuff
+ @web.expose
+ @web.require_admin
+ def create_new_user( self, trans, email='', password='', confirm='', subscribe=False ):
+ email_error = password_error = confirm_error = None
+ if email:
+ if len( email ) == 0 or "@" not in email or "." not in email:
+ email_error = "Please enter a real email address"
+ elif len( email) > 255:
+ email_error = "Email address exceeds maximum allowable length"
+ elif trans.app.model.User.filter_by( email=email ).first():
+ email_error = "User with that email already exists"
+ elif len( password ) < 6:
+ password_error = "Please use a password of at least 6 characters"
+ elif password != confirm:
+ confirm_error = "Passwords do not match"
+ else:
+ user = trans.app.model.User( email=email )
+ user.set_password_cleartext( password )
+ user.flush()
+ trans.app.security_agent.create_private_user_role( user )
+ trans.app.security_agent.user_set_default_permissions( user, history=False, dataset=False )
+ trans.log_event( "Admin created a new account for user %s" % email )
+ msg = 'Created new account'
+ messagetype = 'done'
+ #subscribe user to email list
+ if subscribe:
+ mail = os.popen( "%s -t" % trans.app.config.sendmail_path, 'w' )
+ mail.write( "To: %s\nFrom: %s\nSubject: Join Mailing List\n\nJoin Mailing list." % ( trans.app.config.mailing_join_addr,email ) )
+ if mail.close():
+ msg + ". However, subscribing to the mailing list has failed."
+ messagetype = 'error'
+ trans.response.send_redirect( web.url_for( action='users', msg=msg, messagetype=messagetype ) )
+ # TODO: make this a mako template
+ return trans.show_form(
+ web.FormBuilder( web.url_for(), "Create account", submit_text="Create" )
+ .add_text( "email", "Email address", value=email, error=email_error )
+ .add_password( "password", "Password", value='', error=password_error )
+ .add_password( "confirm", "Confirm password", value='', error=confirm_error )
+ .add_input( "checkbox","Subscribe To Mailing List","subscribe", value='subscribe' ) )
+
+
@web.expose
@web.require_admin
def users( self, trans, **kwd ):
@@ -960,7 +1001,7 @@ class Admin( BaseController ):
yield build_name, dbkey, ( dbkey==last_used_build )
dbkeys = get_dbkey_options( last_used_build )
# Send list of roles to the form so the dataset can be associated with 1 or more of them.
- roles = trans.app.model.Role.filter( trans.app.model.Role.c.type != trans.app.model.Role.types.PRIVATE ).order_by( trans.app.model.Role.c.name ).all()
+ roles = trans.app.model.Role.query().order_by( trans.app.model.Role.c.name ).all()
return trans.fill_template( '/admin/library/new_dataset.mako',
folder_id=folder_id,
file_formats=file_formats,
diff --git a/lib/galaxy/web/controllers/root.py b/lib/galaxy/web/controllers/root.py
index 5d7949ed347..beb32d121fa 100644
--- a/lib/galaxy/web/controllers/root.py
+++ b/lib/galaxy/web/controllers/root.py
@@ -800,7 +800,9 @@ class RootController( BaseController ):
in_roles = [ in_roles ]
in_roles = [ trans.app.model.Role.get( x ) for x in in_roles ]
permissions[ trans.app.security_agent.get_action( v.action ) ] = in_roles
- trans.app.security_agent.history_set_default_permissions( history, permissions )
+ dataset = 'dataset' in kwd
+ bypass_manage_permission = 'bypass_manage_permission' in kwd
+ trans.app.security_agent.history_set_default_permissions( history, permissions, dataset=dataset, bypass_manage_permission=bypass_manage_permission )
return trans.show_ok_message( 'Default history permissions have been changed.' )
return trans.fill_template( 'history/permissions.mako' )
else:
diff --git a/lib/galaxy/web/controllers/user.py b/lib/galaxy/web/controllers/user.py
index f8251c3057e..9e197c021dd 100644
--- a/lib/galaxy/web/controllers/user.py
+++ b/lib/galaxy/web/controllers/user.py
@@ -81,9 +81,6 @@ class User( BaseController ):
@web.expose
def login( self, trans, email='', password='' ):
- if trans.app.memory_usage:
- # Keep track of memory usage
- m0 = trans.app.memory_usage.memory()
email_error = password_error = None
# Attempt login
if trans.app.config.require_login:
@@ -102,13 +99,11 @@ class User( BaseController ):
else:
trans.handle_user_login( user )
trans.log_event( "User logged in" )
+ return trans.show_ok_message( "Now logged in as " + user.email, refresh_frames=refresh_frames )
msg = "Now logged in as " + user.email + "."
if trans.app.config.require_login:
msg += ' Click here to continue to the front page.' % web.url_for( '/static/welcome.html' )
return trans.show_ok_message( msg, refresh_frames=refresh_frames )
- if trans.app.memory_usage:
- m1 = trans.app.memory_usage.memory( m0, pretty=True )
- log.info( "End of user/login, memory used increased by %s" % m1 )
form = web.FormBuilder( web.url_for(), "Login", submit_text="Login" ) \
.add_text( "email", "Email address", value=email, error=email_error ) \
.add_password( "password", "Password", value='', error=password_error,
@@ -143,15 +138,12 @@ class User( BaseController ):
@web.expose
def create( self, trans, email='', password='', confirm='', subscribe=False ):
- if trans.app.memory_usage:
- # Keep track of memory usage
- m0 = trans.app.memory_usage.memory()
if trans.app.config.require_login:
refresh_frames = [ 'masthead', 'history', 'tools' ]
else:
refresh_frames = [ 'masthead', 'history' ]
if not trans.app.config.allow_user_creation and not trans.user_is_admin():
- return trans.show_error_message( 'User registration is disabled. Please contact your local Galaxy administrator for an account.' )
+ return trans.show_error_message( 'User registration is disabled. Please contact your Galaxy administrator for an account.' )
email_error = password_error = confirm_error = None
if email:
if len( email ) == 0 or "@" not in email or "." not in email:
@@ -168,32 +160,23 @@ class User( BaseController ):
user = trans.app.model.User( email=email )
user.set_password_cleartext( password )
user.flush()
- if trans.user_is_admin():
- trans.app.security_agent.create_private_user_role( user )
- trans.app.security_agent.user_set_default_permissions( user )
- trans.log_event( "Admin created a new account" )
- msg = 'Created account ' + user.email
- else:
- trans.app.security_agent.setup_new_user( user )
- trans.handle_user_login( user )
- trans.log_event( "User created a new account" )
- trans.log_event( "User logged in" )
- msg = 'Now logged in as ' + user.email
+ trans.app.security_agent.create_private_user_role( user )
+ trans.handle_user_login( user )
+ trans.app.security_agent.user_set_default_permissions( user, history=True, dataset=True )
+ trans.log_event( "User created a new account" )
+ trans.log_event( "User logged in" )
#subscribe user to email list
if subscribe:
mail = os.popen("%s -t" % trans.app.config.sendmail_path, 'w')
mail.write("To: %s\nFrom: %s\nSubject: Join Mailing List\n\nJoin Mailing list." % (trans.app.config.mailing_join_addr,email) )
if mail.close():
- return trans.show_warn_message( msg + ". However, subscribing to the mailing list has failed.", refresh_frames=refresh_frames )
- if trans.app.memory_usage:
- m1 = trans.app.memory_usage.memory( m0, pretty=True )
- log.info( "End of user/create, memory used increased by %s" % m1 )
- return trans.show_ok_message( msg, refresh_frames=refresh_frames )
+ return trans.show_warn_message( "Now logged in as " + user.email+". However, subscribing to the mailing list has failed.", refresh_frames=['masthead', 'history'] )
+ return trans.show_ok_message( "Now logged in as " + user.email, refresh_frames=['masthead', 'history'] )
return trans.show_form(
web.FormBuilder( web.url_for(), "Create account", submit_text="Create" )
.add_text( "email", "Email address", value=email, error=email_error )
- .add_password( "password", "Password", value='', error=password_error )
- .add_password( "confirm", "Confirm password", value='', error=confirm_error )
+ .add_password( "password", "Password", value='', error=password_error )
+ .add_password( "confirm", "Confirm password", value='', error=confirm_error )
.add_input( "checkbox","Subscribe To Mailing List","subscribe", value='subscribe' ) )
@web.expose
diff --git a/lib/galaxy/web/framework/__init__.py b/lib/galaxy/web/framework/__init__.py
index c982464ad10..af93111117f 100644
--- a/lib/galaxy/web/framework/__init__.py
+++ b/lib/galaxy/web/framework/__init__.py
@@ -316,20 +316,25 @@ class UniverseWebTransaction( base.DefaultWebTransaction ):
Login a new user (possibly newly created)
- create a new session
- associate new session with user
- - if old session had a history and it was not associated with a user, associate it with the new session.
+ - if old session had a history and it was not associated with a user, associate it with the new session,
+ otherwise associate the current session's history with the user
"""
prev_galaxy_session = self.galaxy_session
prev_galaxy_session.is_valid = False
self.galaxy_session = self.__create_new_session( prev_galaxy_session, user )
if prev_galaxy_session.current_history:
history = prev_galaxy_session.current_history
- if history.user is None:
- self.galaxy_session.add_history( history )
- self.galaxy_session.current_history = history
- history.user = user
- self.sa_session.flush( [ prev_galaxy_session, self.galaxy_session, history ] )
+ elif self.galaxy_session.current_history:
+ history = self.galaxy_session.current_history
else:
- self.sa_session.flush( [ prev_galaxy_session, self.galaxy_session ] )
+ history = self.history
+ if history not in self.galaxy_session.histories:
+ self.galaxy_session.add_history( history )
+ if history.user is None:
+ history.user = user
+ self.galaxy_session.current_history = history
+ self.app.security_agent.history_set_default_permissions( history, dataset=True )
+ self.sa_session.flush( [ prev_galaxy_session, self.galaxy_session, history ] )
self.__update_session_cookie()
def handle_user_logout( self ):
"""
diff --git a/templates/admin/dataset_security/users.mako b/templates/admin/dataset_security/users.mako
index 8dc7f58367a..39dad937182 100644
--- a/templates/admin/dataset_security/users.mako
+++ b/templates/admin/dataset_security/users.mako
@@ -42,14 +42,14 @@
Users
+
+
%if msg:
${render_msg( msg, messagetype )}
%endif
-
-
%if len( users_groups_roles ) == 0:
There are no Galaxy users
%else:
diff --git a/templates/admin/library/dataset.mako b/templates/admin/library/dataset.mako
index d910e01b362..01f33a65836 100644
--- a/templates/admin/library/dataset.mako
+++ b/templates/admin/library/dataset.mako
@@ -18,7 +18,7 @@
<%
- roles = trans.app.model.Role.filter( trans.app.model.Role.table.c.type != trans.app.model.Role.types.PRIVATE ).all()
+ roles = trans.app.model.Role.query().all()
%>
%if isinstance( dataset, list ):
diff --git a/test/base/twilltestcase.py b/test/base/twilltestcase.py
index 08da005df79..dff00121fc0 100644
--- a/test/base/twilltestcase.py
+++ b/test/base/twilltestcase.py
@@ -373,19 +373,48 @@ class TwillTestCase( unittest.TestCase ):
self.visit_page( "user/set_default_permissions" )
self.check_page_for_string( email )
self.home()
-
+ def user_set_default_permissions( self, permissions_out=[], permissions_in=[], role_id=2 ): # role.id = 2 is Private Role for test2@bx.psu.edu
+ # NOTE: Twill has a bug that requires the ~/user/permissions page to contain at least 1 option value
+ # in each select list or twill throws an exception, which is: ParseError: OPTION outside of SELECT
+ # Due to this bug, we'll bypass visiting the page, and simply pass the permissions on to the
+ # /user/set_default_permissions method.
+ url = "user/set_default_permissions?update_roles=Save&id=None"
+ for po in permissions_out:
+ key = '%s_out' % po
+ url ="%s&%s=%s" % ( url, key, str( role_id ) )
+ for pi in permissions_in:
+ key = '%s_in' % pi
+ url ="%s&%s=%s" % ( url, key, str( role_id ) )
+ self.visit_url( "%s/%s" % ( self.url, url ) )
+ self.last_page()
+ self.check_page_for_string( 'Default new history permissions have been changed.' )
+ self.home()
+ def history_set_default_permissions( self, permissions_out=[], permissions_in=[], role_id=3 ): # role.id = 3 is Private Role for test3@bx.psu.edu
+ # NOTE: Twill has a bug that requires the ~/user/permissions page to contain at least 1 option value
+ # in each select list or twill throws an exception, which is: ParseError: OPTION outside of SELECT
+ # Due to this bug, we'll bypass visiting the page, and simply pass the permissions on to the
+ # /user/set_default_permissions method.
+ url = "root/history_set_default_permissions?update_roles=Save&id=None&dataset=True"
+ for po in permissions_out:
+ key = '%s_out' % po
+ url ="%s&%s=%s" % ( url, key, str( role_id ) )
+ for pi in permissions_in:
+ key = '%s_in' % pi
+ url ="%s&%s=%s" % ( url, key, str( role_id ) )
+ self.visit_url( "%s/%s" % ( self.url, url ) )
+ self.last_page()
+ self.check_page_for_string( 'Default history permissions have been changed.' )
+ self.home()
def login( self, email='test@bx.psu.edu', password='testuser'):
# test@bx.psu.edu is configured as an admin user
self.create( email=email, password=password, confirm=password )
self.visit_page( "user/login?email=%s&password=%s" % (email, password) )
self.check_page_for_string( "Now logged in as %s" %email )
self.home()
-
def logout( self ):
self.visit_page( "user/logout" )
self.check_page_for_string( "You are no longer logged in" )
self.home()
-
# Functions associated with browsers, cookies, HTML forms and page visits
def check_page_for_string( self, patt ):
"""Looks for 'patt' in the current browser page"""
diff --git a/test/functional/test_security_and_libraries.py b/test/functional/test_security_and_libraries.py
index df0af3dedf8..8a95da8122e 100644
--- a/test/functional/test_security_and_libraries.py
+++ b/test/functional/test_security_and_libraries.py
@@ -1,3 +1,4 @@
+import sys
import galaxy.model
from galaxy.model.orm import *
from base.twilltestcase import *
@@ -41,50 +42,162 @@ class TestHistory( TwillTestCase ):
self.check_page_for_string( not_logged_in_security_msg )
self.visit_url( "%s/admin/dataset" % self.url )
self.check_page_for_string( not_logged_in_security_msg )
- def test_03_login_as_admin( self ):
- """Testing logging in as an admin user"""
- self.login( email='test@bx.psu.edu' ) #This is configured as our admin user
+ def test_03_login_as_admin_user( self ):
+ """Testing logging in as an admin user - tests initial settings for DefaultUserPermissions and DefaultHistoryPermissions"""
+ self.login( email='test@bx.psu.edu' ) # test@bx.psu.edu is configured as our admin user
self.visit_page( "admin" )
self.check_page_for_string( 'Administration' )
global testuser1
testuser1 = galaxy.model.User.filter( galaxy.model.User.table.c.email=='test@bx.psu.edu' ).first()
- # Make sure a private role exists for the user
- private_role_found = False
- for role in testuser1.all_roles():
- if role.name == testuser1.email and role.description == 'Private Role for %s' % testuser1.email:
- private_role_found = True
- break
- if not private_role_found:
- raise AssertionError( "Private role not found for user '%s'" % testuser1.email )
- # Make sure a DefaultUserPermission exists for the user
+ # Make sure DefaultUserPermissions are correct
if not testuser1.default_permissions:
raise AssertionError( 'No DefaultUserPermissions were created for %s when their account was created' % testuser1.email )
if len( testuser1.default_permissions ) > 1:
raise AssertionError( 'More than 1 DefaultUserPermissions were created for %s when their account was created' % testuser1.email )
dup = galaxy.model.DefaultUserPermissions.filter( galaxy.model.DefaultUserPermissions.table.c.user_id==testuser1.id ).first()
- if not dup.action == 'manage permissions':
- raise AssertionError( 'The DefaultUserPermission.action for user "%s" is "%s", but it should be "manage permissions"' % ( testuser1.email, dup.action ) )
+ if not dup.action == galaxy.model.Dataset.permitted_actions.DATASET_MANAGE_PERMISSIONS.action:
+ raise AssertionError( 'The DefaultUserPermission.action for user "%s" is "%s", but it should be "%s"' \
+ % ( testuser1.email, dup.action, galaxy.model.Dataset.permitted_actions.DATASET_MANAGE_PERMISSIONS.action ) )
+ # Make sure DefaultHistoryPermissions are correct
+ latest_history = galaxy.model.History.query().order_by( desc( galaxy.model.History.table.c.create_time ) ).first()
+ if not latest_history.default_permissions:
+ raise AssertionError( 'No DefaultHistoryPermissions were created for history id %d when it was created' % latest_history.id )
+ if len( latest_history.default_permissions ) > 1:
+ raise AssertionError( 'More than 1 DefaultHistoryPermissions were created for history id %d when it was created' % latest_history.id )
+ dhp = galaxy.model.DefaultHistoryPermissions.filter( galaxy.model.DefaultHistoryPermissions.table.c.history_id==latest_history.id ).first()
+ if not dhp.action == galaxy.model.Dataset.permitted_actions.DATASET_MANAGE_PERMISSIONS.action:
+ raise AssertionError( 'The DefaultHistoryPermission.action for history id %d is "%s", but it should be "%s"' \
+ % ( latest_history.id, dhp.action, galaxy.model.Dataset.permitted_actions.DATASET_MANAGE_PERMISSIONS.action ) )
self.visit_url( "%s/admin/user?user_id=%s" % ( self.url, testuser1.id ) )
self.check_page_for_string( testuser1.email )
self.home()
self.logout()
- # Make sure that we have 3 users
- self.login( email='test2@bx.psu.edu' ) # This will not be an admin user
+ def test_06_login_as_non_admin_user1( self ):
+ """Testing logging in as non-admin user1 - tests private role creation, changing DefaultHistoryPermissions for new histories"""
+ self.login( email='test2@bx.psu.edu' ) # test2@bx.psu.edu is not an admin user
global testuser2
testuser2 = galaxy.model.User.filter( galaxy.model.User.table.c.email=='test2@bx.psu.edu' ).first()
self.visit_page( "admin" )
self.check_page_for_string( logged_in_security_msg )
- # NOTE: we cannot currently Set DefaultHistoryPermissions for this user
- # because of a bug in twill where it is not able to handle select lists that
- # include no options...
+ # Make sure a private role exists for testuser2
+ private_role = None
+ for role in testuser2.all_roles():
+ if role.name == testuser2.email and role.description == 'Private Role for %s' % testuser2.email:
+ private_role = role
+ break
+ if not private_role:
+ raise AssertionError( "Private role not found for user '%s'" % testuser2.email )
+ # Add a dataset to the history
+ self.upload_file( '1.bed' )
+ latest_dataset = galaxy.model.Dataset.query().order_by( desc( galaxy.model.Dataset.table.c.create_time ) ).first()
+ # Make sure ActionDatasetRoleAssociation is correct
+ if not latest_dataset.actions:
+ raise AssertionError( 'No ActionDatasetRoleAssociations were created for dataset id %d when it was created' % latest_dataset.id )
+ if len( latest_dataset.actions ) > 1:
+ raise AssertionError( 'More than 1 ActionDatasetRoleAssociations were created for dataset id %d when it was created' % latest_dataset.id )
+ adra = galaxy.model.ActionDatasetRoleAssociation.filter( galaxy.model.ActionDatasetRoleAssociation.table.c.dataset_id==latest_dataset.id ).first()
+ if not adra.action == galaxy.model.Dataset.permitted_actions.DATASET_MANAGE_PERMISSIONS.action:
+ raise AssertionError( 'The ActionDatasetRoleAssociation.action for dataset id %d is "%s", but it should be "%s"' \
+ % ( latest_dataset.id, adra.action, galaxy.model.Dataset.permitted_actions.DATASET_MANAGE_PERMISSIONS.action ) )
+ # Change DefaultHistoryPermissions for testuser2
+ permissions_in = []
+ actions_in = []
+ for key, value in galaxy.model.Dataset.permitted_actions.items():
+ permissions_in.append( key )
+ actions_in.append( value.action )
+ # Sort actions for later comparison
+ actions_in.sort()
+ role_id = str( private_role.id )
+ self.user_set_default_permissions( permissions_in=permissions_in, role_id=role_id )
+ # Make sure the default permissions are changed for new histories
+ self.new_history()
+ latest_history = galaxy.model.History.query().order_by( desc( galaxy.model.History.table.c.create_time ) ).first()
+ if not latest_history.default_permissions:
+ raise AssertionError( 'No DefaultHistoryPermissions were created for history id %d when DefaultHistoryPermissions were changed' % latest_history.id )
+ if len( latest_history.default_permissions ) != len( galaxy.model.Dataset.permitted_actions.items() ):
+ raise AssertionError( '%d DefaultHistoryPermissions were created for history id %d, should have been %d' % ( len( latest_history.default_permissions ), latest_history.id, len( galaxy.model.Dataset.permitted_actions ) ) )
+ dhps = []
+ for dhp in latest_history.default_permissions:
+ dhps.append( dhp.action )
+ # Sort actions for later comparison
+ dhps.sort()
+ for key, value in galaxy.model.Dataset.permitted_actions.items():
+ if value.action not in dhps:
+ raise AssertionError( '%s not in history id %d default_permissions after they were changed' % ( value.action, latest_history.id ) )
+ # Add a dataset to the history
+ self.upload_file( '1.bed' )
+ latest_dataset = galaxy.model.Dataset.query().order_by( desc( galaxy.model.Dataset.table.c.create_time ) ).first()
+ # Make sure ActionDatasetRoleAssociations are correct
+ if not latest_dataset.actions:
+ raise AssertionError( 'No ActionDatasetRoleAssociations were created for dataset id %d when it was created' % latest_dataset.id )
+ if len( latest_dataset.actions ) != len( latest_history.default_permissions ):
+ raise AssertionError( '%d ActionDatasetRoleAssociations were created for dataset id %d when it was created ( should have been %d )' % ( len( latest_dataset.actions ), latest_dataset.id, len( latest_history.default_permissions ) ) )
+ adras = []
+ for adra in latest_dataset.actions:
+ adras.append( adra.action )
+ # Sort actions for later comparison
+ adras.sort()
+ # Compare ActionDatasetRoleAssociations with permissions_in - shouuld be the same
+ if adras != actions_in:
+ raise AssertionError( 'ActionDatasetRoleAssociations "%s" for dataset id %d differ from changed default permissions "%s"' \
+ % ( str( adras ), latest_dataset.id, str( actions_in ) ) )
+ # Compare DefaultHistoryPermissions and ActionDatasetRoleAssociations - should be the same
+ if adras != dhps:
+ raise AssertionError( 'ActionDatasetRoleAssociations "%s" for dataset id %d differ from DefaultHistoryPermissions "%s" for history id %d' \
+ % ( str( adras ), latest_dataset.id, str( dhps ), latest_history.id ) )
+ self.home()
self.logout()
+ def test_09_login_as_non_admin_user2( self ):
+ """Testing logging in as non-admin user2 - tests changing DefaultHistoryPermissions for the current history"""
self.login( email='test3@bx.psu.edu' ) # This will not be an admin user
global testuser3
testuser3 = galaxy.model.User.filter( galaxy.model.User.table.c.email=='test3@bx.psu.edu' ).first()
- self.visit_page( "admin" )
- self.check_page_for_string( logged_in_security_msg )
+ latest_history = galaxy.model.History.query().order_by( desc( galaxy.model.History.table.c.create_time ) ).first()
+ self.upload_file( '1.bed' )
+ latest_dataset = galaxy.model.Dataset.query().order_by( desc( galaxy.model.Dataset.table.c.create_time ) ).first()
+ permissions_in = [ 'DATASET_EDIT_METADATA', 'DATASET_MANAGE_PERMISSIONS' ]
+ # Make sure these are in sorted order for later comparison
+ actions_in = [ 'edit metadata', 'manage permissions' ]
+ permissions_out = [ 'DATASET_ACCESS' ]
+ actions_out = [ 'access' ]
+ private_role = None
+ for role in testuser3.all_roles():
+ if role.name == testuser3.email and role.description == 'Private Role for %s' % testuser3.email:
+ private_role = role
+ break
+ if not private_role:
+ raise AssertionError( "Private role not found for user '%s'" % testuser3.email )
+ role_id = str( private_role.id )
+ # Change DefaultHistoryPermissions for the current history
+ self.history_set_default_permissions( permissions_out=permissions_out, permissions_in=permissions_in, role_id=role_id )
+ if not latest_history.default_permissions:
+ raise AssertionError( 'No DefaultHistoryPermissions were created for history id %d when DefaultHistoryPermissions were changed' % latest_history.id )
+ if len( latest_history.default_permissions ) != len( actions_in ):
+ raise AssertionError( '%d DefaultHistoryPermissions were created for history id %d, should have been %d' \
+ % ( len( latest_history.default_permissions ), latest_history.id, len( permissions_in ) ) )
+ # Make sure DefaultHistoryPermissions were correctly changed for the current history
+ dhps = []
+ for dhp in latest_history.default_permissions:
+ dhps.append( dhp.action )
+ # Sort actions for later comparison
+ dhps.sort()
+ # Compare DefaultHistoryPermissions and actions_in - should be the same
+ if dhps != actions_in:
+ raise AssertionError( 'DefaultHistoryPermissions "%s" for history id %d differ from actions "%s" passed for changing' \
+ % ( str( dhps ), latest_history.id, str( actions_in ) ) )
+ # Make sure ActionDatasetRoleAssociations are correct
+ if not latest_dataset.actions:
+ raise AssertionError( 'No ActionDatasetRoleAssociations were created for dataset id %d when it was created' % latest_dataset.id )
+ if len( latest_dataset.actions ) != len( latest_history.default_permissions ):
+ raise AssertionError( '%d ActionDatasetRoleAssociations were created for dataset id %d when it was created ( should have been %d )' % ( len( latest_dataset.actions ), latest_dataset.id, len( latest_history.default_permissions ) ) )
+ adras = []
+ for adra in latest_dataset.actions:
+ adras.append( adra.action )
+ # Sort actions for later comparison
+ adras.sort()
+ self.home()
self.logout()
- def test_06_create_role( self ):
+ def test_12_create_role( self ):
"""Testing creating new non-private role with 2 members"""
self.login( email=testuser1.email )
name = 'New Test Role'
@@ -93,14 +206,14 @@ class TestHistory( TwillTestCase ):
# Get the role object for later tests
global new_test_role
new_test_role = galaxy.model.Role.filter( galaxy.model.Role.table.c.name==name ).first()
- def test_09_create_group( self ):
+ def test_15_create_group( self ):
"""Testing creating new group with 2 members and 1 associated role"""
name = 'New Test Group'
self.create_group( name=name, user_ids=[ str( testuser1.id ), str( testuser2.id ) ], role_ids=[ str( new_test_role.id ) ] )
# Get the group object for later tests
global new_test_group
new_test_group = galaxy.model.Group.filter( galaxy.model.Group.table.c.name==name ).first()
- def test_12_add_group_member( self ):
+ def test_18_add_group_member( self ):
"""Testing editing membership of an existing group"""
name = 'Another Test Group'
self.create_group( name=name )
@@ -110,15 +223,11 @@ class TestHistory( TwillTestCase ):
self.add_group_members( str( another_test_group.id ), [ str( testuser3.id ) ] )
self.visit_url( "%s/admin/group_members_edit?group_id=%s" % ( self.url, str( another_test_group.id ) ) )
self.check_page_for_string( testuser3.email )
- def test_15_associate_groups_with_role( self ):
+ def test_21_associate_groups_with_role( self ):
"""Testing adding existing groups to an existing role"""
# NOTE: To get this to work with twill, all select lists on the ~/admin/role page must contain at least
# 1 option value or twill throws an exception, which is: ParseError: OPTION outside of SELECT
# Due to this bug in twill, we create the role, associating it with at least 1 user and 1 group...
- #
- # TODO: need to enhance this test to associate DefaultUserPermissions and DefaultHistoryPermissions
- # with the role, then add tests in test_55_purge_role to make sure the association records are deleted
- # when the role is purged.
name = 'Another Test Role'
description = 'Another cool new test role'
self.create_role( name=name,
@@ -133,7 +242,7 @@ class TestHistory( TwillTestCase ):
self.associate_groups_with_role( str( another_test_role.id ), group_ids=[ str( new_test_group.id ) ] )
self.visit_page( 'admin/roles' )
self.check_page_for_string( new_test_group.name )
- def test_18_create_library( self ):
+ def test_24_create_library( self ):
"""Testing creating new library"""
name = 'New Test Library'
description = 'New Test Library Description'
@@ -145,20 +254,20 @@ class TestHistory( TwillTestCase ):
library = galaxy.model.Library.filter( and_( galaxy.model.Library.table.c.name==name,
galaxy.model.Library.table.c.description==description,
galaxy.model.Library.table.c.deleted==False ) ).first()
- def test_21_rename_library( self ):
+ def test_27_rename_library( self ):
"""Testing renaming a library"""
self.rename_library( str( library.id ), name='New Test Library Renamed', description='New Test Library Description Re-described', root_folder='on' )
self.visit_page( 'admin/libraries' )
self.check_page_for_string( "New Test Library Renamed" )
# Rename it back to what it was originally
self.rename_library( str( library.id ), name='New Test Library', description='New Test Library Description', root_folder='on' )
- def test_24_rename_root_folder( self ):
+ def test_30_rename_root_folder( self ):
"""Testing renaming a library root folder"""
folder = library.root_folder
self.rename_folder( str( folder.id ), name='New Test Library Root Folder', description='New Test Library Root Folder Description' )
self.visit_page( 'admin/libraries' )
self.check_page_for_string( "New Test Library Root Folder" )
- def test_27_add_public_dataset_to_root_folder( self ):
+ def test_33_add_public_dataset_to_root_folder( self ):
"""Testing adding a public dataset to a library root folder"""
folder = library.root_folder
self.add_dataset( '1.bed', str( folder.id ), extension='bed', dbkey='hg18', roles=[] )
@@ -166,7 +275,7 @@ class TestHistory( TwillTestCase ):
self.check_page_for_string( "1.bed" )
self.check_page_for_string( "bed" )
self.check_page_for_string( "hg18" )
- def test_30_copy_dataset_from_history_to_root_folder( self ):
+ def test_36_copy_dataset_from_history_to_root_folder( self ):
"""Testing copying a dataset from the current history to a library root folder"""
folder = library.root_folder
self.add_dataset_to_folder_from_history( str( folder.id ) )
@@ -181,7 +290,7 @@ class TestHistory( TwillTestCase ):
for adra in adras:
if not adra.action == 'manage permissions':
raise AssertionError( 'ActionDatasetRoleAssociation.action "%s" is not the DefaultHistoryPermission setting, which is "manage permissions"' % str( adra.action ) )
- def test_33_add_new_folder( self ):
+ def test_39_add_new_folder( self ):
"""Testing adding a folder to a library root folder"""
root_folder = library.root_folder
name = 'New Test Folder'
@@ -193,27 +302,27 @@ class TestHistory( TwillTestCase ):
galaxy.model.LibraryFolder.table.c.description==description ) ).first()
self.visit_page( 'admin/libraries' )
self.check_page_for_string( "New Test Folder" )
- def test_36_add_datasets_from_library_dir( self ):
+ def test_42_add_datasets_from_library_dir( self ):
"""Testing adding several datasets from library directory to sub-folder"""
roles_tuple = [ ( str( new_test_role.id ), new_test_role.description ) ]
self.add_datasets_from_library_dir( str( new_test_folder.id ), roles_tuple=roles_tuple )
- def test_39_mark_group_deleted( self ):
+ def test_45_mark_group_deleted( self ):
"""Testing marking a group as deleted"""
self.visit_page( "admin/groups" )
self.check_page_for_string( another_test_group.name )
self.mark_group_deleted( str( another_test_group.id ) )
- def test_42_undelete_group( self ):
+ def test_48_undelete_group( self ):
"""Testing undeleting a deleted group"""
self.undelete_group( str( another_test_group.id ) )
- def test_45_mark_role_deleted( self ):
+ def test_51_mark_role_deleted( self ):
"""Testing marking a role as deleted"""
self.visit_page( "admin/roles" )
self.check_page_for_string( another_test_role.description )
self.mark_role_deleted( str( another_test_role.id ) )
- def test_48_undelete_role( self ):
+ def test_54_undelete_role( self ):
"""Testing undeleting a deleted role"""
self.undelete_role( str( another_test_role.id ) )
- def test_51_mark_library_deleted( self ):
+ def test_57_mark_library_deleted( self ):
"""Testing marking a library as deleted"""
self.mark_library_deleted( str( library.id ) )
# Make sure the library was deleted
@@ -237,7 +346,7 @@ class TestHistory( TwillTestCase ):
if lfda.dataset.deleted:
raise AssertionError( 'The dataset with id "%s" has been marked as deleted when it should not have been.' % lfda.dataset.id )
check_folder( library.root_folder )
- def test_54_mark_library_undeleted( self ):
+ def test_60_mark_library_undeleted( self ):
"""Testing marking a library as not deleted"""
self.mark_library_undeleted( str( library.id ) )
# Make sure the library is undeleted
@@ -266,7 +375,7 @@ class TestHistory( TwillTestCase ):
library.refresh()
if not library.deleted:
raise AssertionError( 'The library id %s named "%s" has not been marked as deleted after it was undeleted.' % ( str( library.id ), library.name ) )
- def test_57_purge_group( self ):
+ def test_63_purge_group( self ):
"""Testing purging a group"""
group_id = str( another_test_group.id )
self.purge_group( group_id )
@@ -278,7 +387,7 @@ class TestHistory( TwillTestCase ):
gra = galaxy.model.GroupRoleAssociation.filter( galaxy.model.GroupRoleAssociation.table.c.group_id == group_id ).all()
if gra:
raise AssertionError( "Purging the group did not delete the GroupRoleAssociations for group_id '%s'" % group_id )
- def test_60_purge_role( self ):
+ def test_66_purge_role( self ):
"""Testing purging a role"""
role_id = str( another_test_role.id )
self.purge_role( role_id )
@@ -290,7 +399,7 @@ class TestHistory( TwillTestCase ):
adra = galaxy.model.ActionDatasetRoleAssociation.filter( galaxy.model.ActionDatasetRoleAssociation.table.c.role_id == role_id ).all()
if adra:
raise AssertionError( "Purging the role did not delete the ActionDatasetRoleAssociations for role_id '%s'" % role_id )
- def test_63_purge_library( self ):
+ def test_69_purge_library( self ):
"""Testing purging a library"""
self.purge_library( str( library.id ) )
# Make sure the library was purged