mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Refactor data staging and tool running logic out of test/functional/test_toolbox.py.
In a new twill specific interactor class, idea here is to proceed by implementing a matching API interactor class.
This commit is contained in:
John Chilton
@@ -20,16 +20,20 @@ class ToolTestCase( TwillTestCase ):
|
||||
|
||||
self.__handle_test_def_errors( testdef )
|
||||
|
||||
latest_history = self.__setup_test_history()
|
||||
galaxy_interactor = GalaxyInteractorTwill( self )
|
||||
|
||||
self.__setup_test_data( testdef, shed_tool_id )
|
||||
test_history = galaxy_interactor.new_history()
|
||||
|
||||
data_list = self.__run_tool( testdef )
|
||||
# Upload any needed files
|
||||
for test_data in testdef.test_data():
|
||||
galaxy_interactor.stage_data( test_data, shed_tool_id )
|
||||
|
||||
data_list = galaxy_interactor.run_tool( testdef )
|
||||
self.assertTrue( data_list )
|
||||
|
||||
self.__verify_outputs( testdef, shed_tool_id, data_list )
|
||||
|
||||
self.__delete_history( latest_history )
|
||||
galaxy_interactor.delete_history( test_history )
|
||||
|
||||
def __handle_test_def_errors(self, testdef):
|
||||
# If the test generation had an error, raise
|
||||
@@ -39,81 +43,6 @@ class ToolTestCase( TwillTestCase ):
|
||||
else:
|
||||
raise Exception( "Test parse failure" )
|
||||
|
||||
def __setup_test_history( self ):
|
||||
# Start with a new history
|
||||
self.logout()
|
||||
self.login( email='test@bx.psu.edu' )
|
||||
admin_user = sa_session.query( galaxy.model.User ).filter( galaxy.model.User.table.c.email == 'test@bx.psu.edu' ).one()
|
||||
self.new_history()
|
||||
latest_history = sa_session.query( galaxy.model.History ) \
|
||||
.filter( and_( galaxy.model.History.table.c.deleted == False,
|
||||
galaxy.model.History.table.c.user_id == admin_user.id ) ) \
|
||||
.order_by( desc( galaxy.model.History.table.c.create_time ) ) \
|
||||
.first()
|
||||
assert latest_history is not None, "Problem retrieving latest_history from database"
|
||||
if len( self.get_history_as_data_list() ) > 0:
|
||||
raise AssertionError("ToolTestCase.do_it failed")
|
||||
return latest_history
|
||||
|
||||
def __setup_test_data( self, testdef, shed_tool_id ):
|
||||
# Upload any needed files
|
||||
for test_data in testdef.test_data():
|
||||
self.upload_file( test_data['fname'],
|
||||
ftype=test_data['ftype'],
|
||||
dbkey=test_data['dbkey'],
|
||||
metadata=test_data['metadata'],
|
||||
composite_data=test_data['composite_data'],
|
||||
shed_tool_id=shed_tool_id )
|
||||
name = test_data.get('name', None)
|
||||
if name:
|
||||
hda_id = self.get_history_as_data_list()[-1].get( 'id' )
|
||||
try:
|
||||
self.edit_hda_attribute_info( hda_id=str(hda_id), new_name=name )
|
||||
except:
|
||||
print "### call to edit_hda failed for hda_id %s, new_name=%s" % (hda_id, name)
|
||||
|
||||
def __run_tool( self, testdef ):
|
||||
# We need to handle the case where we've uploaded a valid compressed file since the upload
|
||||
# tool will have uncompressed it on the fly.
|
||||
all_inputs = {}
|
||||
for name, value, _ in testdef.inputs:
|
||||
all_inputs[ name ] = value
|
||||
|
||||
# See if we have a grouping.Repeat element
|
||||
repeat_name = None
|
||||
for input_name, input_value in testdef.tool.inputs_by_page[0].items():
|
||||
if isinstance( input_value, grouping.Repeat ) and all_inputs.get( input_name, 1 ) not in [ 0, "0" ]: # default behavior is to test 1 repeat, for backwards compatibility
|
||||
if not input_value.min: # If input_value.min == 1, the element is already on the page don't add new element.
|
||||
repeat_name = input_name
|
||||
break
|
||||
|
||||
#check if we need to verify number of outputs created dynamically by tool
|
||||
if testdef.tool.force_history_refresh:
|
||||
job_finish_by_output_count = len( self.get_history_as_data_list() )
|
||||
else:
|
||||
job_finish_by_output_count = False
|
||||
|
||||
# Do the first page
|
||||
page_inputs = self.__expand_grouping(testdef.tool.inputs_by_page[0], all_inputs)
|
||||
|
||||
# Run the tool
|
||||
self.run_tool( testdef.tool.id, repeat_name=repeat_name, **page_inputs )
|
||||
print "page_inputs (0)", page_inputs
|
||||
# Do other pages if they exist
|
||||
for i in range( 1, testdef.tool.npages ):
|
||||
page_inputs = self.__expand_grouping(testdef.tool.inputs_by_page[i], all_inputs)
|
||||
self.submit_form( **page_inputs )
|
||||
print "page_inputs (%i)" % i, page_inputs
|
||||
|
||||
# Check the results ( handles single or multiple tool outputs ). Make sure to pass the correct hid.
|
||||
# The output datasets from the tool should be in the same order as the testdef.outputs.
|
||||
data_list = None
|
||||
while data_list is None:
|
||||
data_list = self.get_history_as_data_list()
|
||||
if job_finish_by_output_count and len( testdef.outputs ) > ( len( data_list ) - job_finish_by_output_count ):
|
||||
data_list = None
|
||||
return data_list
|
||||
|
||||
def __verify_outputs( self, testdef, shed_tool_id, data_list ):
|
||||
maxseconds = testdef.maxseconds
|
||||
|
||||
@@ -135,8 +64,87 @@ class ToolTestCase( TwillTestCase ):
|
||||
print >>sys.stderr, self.get_job_stderr( elem.get( 'id' ), format=True )
|
||||
raise
|
||||
|
||||
def __delete_history( self, latest_history ):
|
||||
self.delete_history( id=self.security.encode_id( latest_history.id ) )
|
||||
|
||||
class GalaxyInteractorTwill( object ):
|
||||
|
||||
def __init__( self, twill_test_case ):
|
||||
self.twill_test_case = twill_test_case
|
||||
|
||||
def stage_data( self, test_data, shed_tool_id ):
|
||||
self.twill_test_case.upload_file( test_data['fname'],
|
||||
ftype=test_data['ftype'],
|
||||
dbkey=test_data['dbkey'],
|
||||
metadata=test_data['metadata'],
|
||||
composite_data=test_data['composite_data'],
|
||||
shed_tool_id=shed_tool_id )
|
||||
name = test_data.get('name', None)
|
||||
if name:
|
||||
hda_id = self.twill_test_case.get_history_as_data_list()[-1].get( 'id' )
|
||||
try:
|
||||
self.twill_test_case.edit_hda_attribute_info( hda_id=str(hda_id), new_name=name )
|
||||
except:
|
||||
print "### call to edit_hda failed for hda_id %s, new_name=%s" % (hda_id, name)
|
||||
|
||||
def run_tool( self, testdef ):
|
||||
# We need to handle the case where we've uploaded a valid compressed file since the upload
|
||||
# tool will have uncompressed it on the fly.
|
||||
all_inputs = {}
|
||||
for name, value, _ in testdef.inputs:
|
||||
all_inputs[ name ] = value
|
||||
|
||||
# See if we have a grouping.Repeat element
|
||||
repeat_name = None
|
||||
for input_name, input_value in testdef.tool.inputs_by_page[0].items():
|
||||
if isinstance( input_value, grouping.Repeat ) and all_inputs.get( input_name, 1 ) not in [ 0, "0" ]: # default behavior is to test 1 repeat, for backwards compatibility
|
||||
if not input_value.min: # If input_value.min == 1, the element is already on the page don't add new element.
|
||||
repeat_name = input_name
|
||||
break
|
||||
|
||||
#check if we need to verify number of outputs created dynamically by tool
|
||||
if testdef.tool.force_history_refresh:
|
||||
job_finish_by_output_count = len( self.twill_test_case.get_history_as_data_list() )
|
||||
else:
|
||||
job_finish_by_output_count = False
|
||||
|
||||
# Do the first page
|
||||
page_inputs = self.__expand_grouping(testdef.tool.inputs_by_page[0], all_inputs)
|
||||
|
||||
# Run the tool
|
||||
self.twill_test_case.run_tool( testdef.tool.id, repeat_name=repeat_name, **page_inputs )
|
||||
print "page_inputs (0)", page_inputs
|
||||
# Do other pages if they exist
|
||||
for i in range( 1, testdef.tool.npages ):
|
||||
page_inputs = self.__expand_grouping(testdef.tool.inputs_by_page[i], all_inputs)
|
||||
self.twill_test_case.submit_form( **page_inputs )
|
||||
print "page_inputs (%i)" % i, page_inputs
|
||||
|
||||
# Check the results ( handles single or multiple tool outputs ). Make sure to pass the correct hid.
|
||||
# The output datasets from the tool should be in the same order as the testdef.outputs.
|
||||
data_list = None
|
||||
while data_list is None:
|
||||
data_list = self.twill_test_case.get_history_as_data_list()
|
||||
if job_finish_by_output_count and len( testdef.outputs ) > ( len( data_list ) - job_finish_by_output_count ):
|
||||
data_list = None
|
||||
return data_list
|
||||
|
||||
def new_history( self ):
|
||||
# Start with a new history
|
||||
self.twill_test_case.logout()
|
||||
self.twill_test_case.login( email='test@bx.psu.edu' )
|
||||
admin_user = sa_session.query( galaxy.model.User ).filter( galaxy.model.User.table.c.email == 'test@bx.psu.edu' ).one()
|
||||
self.twill_test_case.new_history()
|
||||
latest_history = sa_session.query( galaxy.model.History ) \
|
||||
.filter( and_( galaxy.model.History.table.c.deleted == False,
|
||||
galaxy.model.History.table.c.user_id == admin_user.id ) ) \
|
||||
.order_by( desc( galaxy.model.History.table.c.create_time ) ) \
|
||||
.first()
|
||||
assert latest_history is not None, "Problem retrieving latest_history from database"
|
||||
if len( self.twill_test_case.get_history_as_data_list() ) > 0:
|
||||
raise AssertionError("ToolTestCase.do_it failed")
|
||||
return latest_history
|
||||
|
||||
def delete_history( self, latest_history ):
|
||||
self.twill_test_case.delete_history( id=self.twill_test_case.security.encode_id( latest_history.id ) )
|
||||
|
||||
def __expand_grouping( self, tool_inputs, declared_inputs, prefix='' ):
|
||||
expanded_inputs = {}
|
||||
|
||||
Reference in New Issue
Block a user