From f08a3fe3bb354e93b962aae7e641c476eaf3709f Mon Sep 17 00:00:00 2001 From: John Chilton Date: Thu, 21 Nov 2013 19:17:12 -0600 Subject: [PATCH] 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. --- test/functional/test_toolbox.py | 170 +++++++++++++++++--------------- 1 file changed, 89 insertions(+), 81 deletions(-) diff --git a/test/functional/test_toolbox.py b/test/functional/test_toolbox.py index 68df7ff9417..009dc686152 100644 --- a/test/functional/test_toolbox.py +++ b/test/functional/test_toolbox.py @@ -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 = {}