Skip to content

Commit

Permalink
Refactor generic FunctionalTestCase from TwillTestCase.
Browse files Browse the repository at this point in the history 
ApiTestCase uses this newer, smaller base class since these tests do not depend on Twill.
  • Loading branch information
@jmchilton
jmchilton committed Nov 17, 2016
1 parent 2bf1cf5 commit b83cfe4
Show file tree
Hide file tree
Showing 3 changed files with 106 additions and 112 deletions.
6 changes: 2 additions & 4 deletions test/base/api.py
View file
Expand Up @@ -10,16 +10,14 @@
)
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( )
Expand Down
18 changes: 9 additions & 9 deletions test/base/interactor.py
View file
Expand Up @@ -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 ):
Expand Down Expand Up @@ -88,7 +88,7 @@ def wait_for_jobs( self, history_id, jobs, maxseconds ):

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,
Expand Down Expand Up @@ -147,7 +147,7 @@ def compare(val, expected):
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()
Expand Down Expand Up @@ -185,7 +185,7 @@ def stage_data_async( self, test_data, history_id, shed_tool_id, async=True ):
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,
Expand All @@ -195,7 +195,7 @@ def stage_data_async( self, test_data, history_id, shed_tool_id, async=True ):
})
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 )
Expand Down
194 changes: 95 additions & 99 deletions test/base/twilltestcase.py
View file
Expand Up @@ -33,7 +33,7 @@
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
Expand All @@ -59,6 +59,100 @@ def setUp( self ):
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:
Expand Down Expand Up @@ -155,14 +249,6 @@ def create( self, cntrller='user', email='test@bx.psu.edu', password='testuser',
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 ):
Expand Down Expand Up @@ -433,62 +519,6 @@ def switch_history( self, id='', name='' ):
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()
Expand All @@ -513,29 +543,6 @@ def wait( self, **kwds ):
"""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" )
Expand All @@ -552,14 +559,3 @@ def _format_stream( self, output, stream, format ):
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

0 comments on commit b83cfe4

Please sign in to comment.