opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor generic FunctionalTestCase from TwillTestCase.

Browse Source 
ApiTestCase uses this newer, smaller base class since these tests do not depend on Twill.
This commit is contained in:
John Chilton
2016-11-17 08:41:52 -05:00
parent 2bf1cf50a3
commit b83cfe434a
3 changed files with 106 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/base/api.py
+2 -4
View File
@@ -10,16 +10,14 @@ from .api_asserts import (
)
from .api_util import get_master_api_key, get_user_api_key
from .interactor import GalaxyInteractorApi as BaseInteractor
# TODO: We don't need all of TwillTestCase, strip down to a common super class
# shared by API and Twill test cases.
from .twilltestcase import TwillTestCase
from .twilltestcase import FunctionalTestCase
TEST_USER = "user@bx.psu.edu"
DEFAULT_OTHER_USER = "otheruser@bx.psu.edu" # A second user for API testing.
# TODO: Allow these to point at existing Galaxy instances.
class ApiTestCase( TwillTestCase ):
class ApiTestCase( FunctionalTestCase ):
def setUp( self ):
super( ApiTestCase, self ).setUp( )
test/base/interactor.py
+9 -9
View File
@@ -46,11 +46,11 @@ def stage_data_in_history( galaxy_interactor, all_test_data, history, shed_tool_
class GalaxyInteractorApi( object ):
def __init__( self, twill_test_case, test_user=None ):
self.twill_test_case = twill_test_case
self.api_url = "%s/api" % twill_test_case.url.rstrip("/")
self.master_api_key = twill_test_case.master_api_key
self.api_key = self.__get_user_key( twill_test_case.user_api_key, twill_test_case.master_api_key, test_user=test_user )
def __init__( self, functional_test_case, test_user=None ):
self.functional_test_case = functional_test_case
self.api_url = "%s/api" % functional_test_case.url.rstrip("/")
self.master_api_key = functional_test_case.master_api_key
self.api_key = self.__get_user_key( functional_test_case.user_api_key, functional_test_case.master_api_key, test_user=test_user )
self.uploads = {}
def verify_output( self, history_id, jobs, output_data, output_testdef, shed_tool_id, maxseconds ):
@@ -88,7 +88,7 @@ class GalaxyInteractorApi( object ):
def verify_output_dataset( self, history_id, hda_id, outfile, attributes, shed_tool_id ):
fetcher = self.__dataset_fetcher( history_id )
self.twill_test_case.verify_hid(
self.functional_test_case.verify_hid(
outfile,
hda_id=hda_id,
attributes=attributes,
@@ -147,7 +147,7 @@ class GalaxyInteractorApi( object ):
raise Exception( msg )
def wait_for_job( self, job_id, history_id, maxseconds ):
self.twill_test_case.wait_for( lambda: not self.__job_ready( job_id, history_id ), maxseconds=maxseconds)
self.functional_test_case.wait_for( lambda: not self.__job_ready( job_id, history_id ), maxseconds=maxseconds)
def get_job_stdio( self, job_id ):
job_stdio = self.__get_job_stdio( job_id ).json()
@@ -185,7 +185,7 @@ class GalaxyInteractorApi( object ):
if composite_data:
files = {}
for i, composite_file in enumerate( composite_data ):
file_name = self.twill_test_case.get_filename( composite_file.get( 'value' ), shed_tool_id=shed_tool_id )
file_name = self.functional_test_case.get_filename( composite_file.get( 'value' ), shed_tool_id=shed_tool_id )
files["files_%s|file_data" % i] = open( file_name, 'rb' )
tool_input.update({
# "files_%d|NAME" % i: name,
@@ -195,7 +195,7 @@ class GalaxyInteractorApi( object ):
})
name = test_data[ 'name' ]
else:
file_name = self.twill_test_case.get_filename( fname, shed_tool_id=shed_tool_id )
file_name = self.functional_test_case.get_filename( fname, shed_tool_id=shed_tool_id )
name = test_data.get( 'name', None )
if not name:
name = os.path.basename( file_name )
test/base/twilltestcase.py
+95 -99
View File
@@ -33,7 +33,7 @@ log = logging.getLogger( __name__ )
DEFAULT_TOOL_TEST_WAIT = os.environ.get("GALAXY_TEST_DEFAULT_WAIT", 86400)
class TwillTestCase( unittest.TestCase ):
class FunctionalTestCase( unittest.TestCase ):
def setUp( self ):
# Security helper
@@ -59,6 +59,100 @@ class TwillTestCase( unittest.TestCase ):
except:
pass
def get_filename( self, filename, shed_tool_id=None ):
# For tool tests override get_filename to point at an installed tool if shed_tool_id is set.
if shed_tool_id and getattr(self, "shed_tools_dict", None):
file_dir = self.shed_tools_dict[ shed_tool_id ]
if file_dir:
return os.path.abspath( os.path.join( file_dir, filename ) )
return self.test_data_resolver.get_filename( filename )
# TODO: Make this more generic, shouldn't be related to tool stuff I guess.
def wait_for( self, func, **kwd ):
sleep_amount = 0.2
slept = 0
walltime_exceeded = kwd.get("maxseconds", None)
if walltime_exceeded is None:
walltime_exceeded = DEFAULT_TOOL_TEST_WAIT
exceeded = True
while slept <= walltime_exceeded:
result = func()
if result:
time.sleep( sleep_amount )
slept += sleep_amount
sleep_amount *= 2
else:
exceeded = False
break
if exceeded:
message = 'Tool test run exceeded walltime [total %s, max %s], terminating.' % (slept, walltime_exceeded)
log.info(message)
raise AssertionError(message)
# TODO: Move verify_xxx into GalaxyInteractor or some relevant mixin.
def verify_hid( self, filename, hda_id, attributes, shed_tool_id, hid="", dataset_fetcher=None):
assert dataset_fetcher is not None
def get_filename(test_filename):
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
def verify_extra_files(extra_files):
self._verify_extra_files_content(extra_files, hda_id, shed_tool_id=shed_tool_id, dataset_fetcher=dataset_fetcher)
data = dataset_fetcher( hda_id )
item_label = "History item %s" % hid
verify(
item_label,
data,
attributes=attributes,
filename=filename,
get_filename=get_filename,
keep_outputs_dir=self.keepOutdir,
verify_extra_files=verify_extra_files,
)
def _verify_composite_datatype_file_content( self, file_name, hda_id, base_name=None, attributes=None, dataset_fetcher=None, shed_tool_id=None ):
assert dataset_fetcher is not None
def get_filename(test_filename):
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
data = dataset_fetcher( hda_id, base_name )
item_label = "History item %s" % hda_id
try:
verify(
item_label,
data,
attributes=attributes,
filename=file_name,
get_filename=get_filename,
keep_outputs_dir=self.keepOutdir,
)
except AssertionError as err:
errmsg = 'Composite file (%s) of %s different than expected, difference:\n' % ( base_name, item_label )
errmsg += str( err )
raise AssertionError( errmsg )
def _verify_extra_files_content( self, extra_files, hda_id, dataset_fetcher, shed_tool_id=None ):
files_list = []
for extra_type, extra_value, extra_name, extra_attributes in extra_files:
if extra_type == 'file':
files_list.append( ( extra_name, extra_value, extra_attributes ) )
elif extra_type == 'directory':
for filename in os.listdir( self.get_filename( extra_value, shed_tool_id=shed_tool_id ) ):
files_list.append( ( filename, os.path.join( extra_value, filename ), extra_attributes ) )
else:
raise ValueError( 'unknown extra_files type: %s' % extra_type )
for filename, filepath, attributes in files_list:
self._verify_composite_datatype_file_content( filepath, hda_id, base_name=filename, attributes=attributes, dataset_fetcher=dataset_fetcher, shed_tool_id=shed_tool_id )
class TwillTestCase( FunctionalTestCase ):
"""Class of FunctionalTestCase geared toward HTML interactions using the Twill library."""
def check_for_strings( self, strings_displayed=[], strings_not_displayed=[] ):
if strings_displayed:
for string in strings_displayed:
@@ -155,14 +249,6 @@ class TwillTestCase( unittest.TestCase ):
def get_all_history_ids_from_api( self ):
return [ history['id'] for history in self.json_from_url( '/api/histories' ) ]
def get_filename( self, filename, shed_tool_id=None ):
# For tool tests override get_filename to point at an installed tool if shed_tool_id is set.
if shed_tool_id and getattr(self, "shed_tools_dict", None):
file_dir = self.shed_tools_dict[ shed_tool_id ]
if file_dir:
return os.path.abspath( os.path.join( file_dir, filename ) )
return self.test_data_resolver.get_filename( filename )
def get_form_controls( self, form ):
formcontrols = []
for i, control in enumerate( form.controls ):
@@ -433,62 +519,6 @@ class TwillTestCase( unittest.TestCase ):
if name:
self.check_history_for_exact_string( name )
def verify_composite_datatype_file_content( self, file_name, hda_id, base_name=None, attributes=None, dataset_fetcher=None, shed_tool_id=None ):
dataset_fetcher = dataset_fetcher or self.__default_dataset_fetcher()
def get_filename(test_filename):
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
data = dataset_fetcher( hda_id, base_name )
item_label = "History item %s" % hda_id
try:
verify(
item_label,
data,
attributes=attributes,
filename=file_name,
get_filename=get_filename,
keep_outputs_dir=self.keepOutdir,
)
except AssertionError as err:
errmsg = 'Composite file (%s) of %s different than expected, difference:\n' % ( base_name, item_label )
errmsg += str( err )
raise AssertionError( errmsg )
def verify_extra_files_content( self, extra_files, hda_id, dataset_fetcher, shed_tool_id=None ):
files_list = []
for extra_type, extra_value, extra_name, extra_attributes in extra_files:
if extra_type == 'file':
files_list.append( ( extra_name, extra_value, extra_attributes ) )
elif extra_type == 'directory':
for filename in os.listdir( self.get_filename( extra_value, shed_tool_id=shed_tool_id ) ):
files_list.append( ( filename, os.path.join( extra_value, filename ), extra_attributes ) )
else:
raise ValueError( 'unknown extra_files type: %s' % extra_type )
for filename, filepath, attributes in files_list:
self.verify_composite_datatype_file_content( filepath, hda_id, base_name=filename, attributes=attributes, dataset_fetcher=dataset_fetcher, shed_tool_id=shed_tool_id )
def verify_hid( self, filename, hda_id, attributes, shed_tool_id, hid="", dataset_fetcher=None):
dataset_fetcher = dataset_fetcher or self.__default_dataset_fetcher()
def get_filename(test_filename):
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
def verify_extra_files(extra_files):
self.verify_extra_files_content(extra_files, hda_id, shed_tool_id=shed_tool_id, dataset_fetcher=dataset_fetcher)
data = dataset_fetcher( hda_id )
item_label = "History item %s" % hid
verify(
item_label,
data,
attributes=attributes,
filename=filename,
get_filename=get_filename,
keep_outputs_dir=self.keepOutdir,
verify_extra_files=verify_extra_files,
)
def visit_url( self, url, params=None, doseq=False, allowed_codes=[ 200 ] ):
if params is None:
params = dict()
@@ -513,29 +543,6 @@ class TwillTestCase( unittest.TestCase ):
"""Waits for the tools to finish"""
return self.wait_for(lambda: self.get_running_datasets(), **kwds)
def wait_for( self, func, **kwd ):
sleep_amount = 0.2
slept = 0
walltime_exceeded = kwd.get("maxseconds", None)
if walltime_exceeded is None:
walltime_exceeded = DEFAULT_TOOL_TEST_WAIT
exceeded = True
while slept <= walltime_exceeded:
result = func()
if result:
time.sleep( sleep_amount )
slept += sleep_amount
sleep_amount *= 2
else:
exceeded = False
break
if exceeded:
message = 'Tool test run exceeded walltime [total %s, max %s], terminating.' % (slept, walltime_exceeded)
log.info(message)
raise AssertionError(message)
def write_temp_file( self, content, suffix='.html' ):
fd, fname = tempfile.mkstemp( suffix=suffix, prefix='twilltestcase-' )
f = os.fdopen( fd, "w" )
@@ -552,14 +559,3 @@ class TwillTestCase( unittest.TestCase ):
else:
msg = output
return msg
def __default_dataset_fetcher( self ):
def fetcher( hda_id, filename=None ):
if filename is None:
page_url = "/display?encoded_id=%s" % hda_id
else:
page_url = "/datasets/%s/display/%s" % ( hda_id, filename )
self.visit_url( page_url )
data = self.last_page()
return data
return fetcher