mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Fixes and functional tests:
- move admin method for creating new user account to admin controller - fixes and functional tests for initial settings of DefaultUserPermissions and DefaultHistoryPermissions when accounts and histories created - fixes and functional tests for changing DefaultHistoryPermissions for the current history - fixes and functional tests for Changing DefaultHistoryPermissions for new histories
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 += ' <a href="%s">Click here</a> 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
|
||||
|
||||
@@ -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 ):
|
||||
"""
|
||||
|
||||
@@ -42,14 +42,14 @@
|
||||
|
||||
<a name="TOP"><h2>Users</h2></a>
|
||||
|
||||
<ul class="manage-table-actions">
|
||||
<li><a class="action-button" href="${h.url_for( controller='admin', action='create_new_user' )}">Create a new user</a></li>
|
||||
</ul>
|
||||
|
||||
%if msg:
|
||||
${render_msg( msg, messagetype )}
|
||||
%endif
|
||||
|
||||
<ul class="manage-table-actions">
|
||||
<li><a class="action-button" href="${h.url_for( controller='user', action='create' )}">Create a new user</a></li>
|
||||
</ul>
|
||||
|
||||
%if len( users_groups_roles ) == 0:
|
||||
There are no Galaxy users
|
||||
%else:
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
</%def>
|
||||
|
||||
<%
|
||||
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 ):
|
||||
|
||||
@@ -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"""
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user