mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
John Chilton
44f7317fa5
By "static" I mean tools such as a FASTQ de-interlacer that would produce a "paired" collection with two datasets everytime. By "determinable" I mean tools that perform N->N operations within the same job - such as a tool that needs to normalize a bunch of datasets all at once and not in separate jobs. (For N->N collection operations that should or can be done in N separate jobs tool authors should just write tools that operate over a dataset and produce a dataset and let the end-user 'map over' that operation.) There are still large classes of operations where the structure of the output collection cannot be pre-determined - such as splitting files (e.g. bam files by read group) - that are not implemented in this commit. Model: The models have been updated to do a more thorough job of tracking collection outputs. Jobs just producing HistoryDatasetCollectionAssociations works fine for simple jobs producing collections - but you don't want to map a list over a tool that produces a pair and produce a bunch of pairs HDCAs and a list:pair HDCA- you just want a bunch of pieces and the one list:pair at that the top. Workflow: Workflows containing such operations can be executed - but the workflow editor has not been updated to handle this complexity (and it will require a significant overhaul) so such tools are not available in the workflow editor. Tool Testing: This commit also introduces a new tool XML syntax for describing tests on output collections. See files test/functional/tools/collection_creates_list.xml and test/functional/tools/collection_creates_pair.xml for examples. Tests: Includes two tools to test this - one that uses explicit pair output names and one that iterates over the structure of input list to produce an output list. Includes several new tools API tests that test the tools described above via the API and implicit mapping over such tools. Includes two new workflow API tests - one that verifies a simple workflow with output collections works and one that verifies mapping over workflow steps in collections works.
280 lines
12 KiB
Python
280 lines
12 KiB
Python
import new
|
|
import sys
|
|
from base.twilltestcase import TwillTestCase
|
|
from base.asserts import verify_assertions
|
|
from base.interactor import build_interactor, stage_data_in_history, RunToolException
|
|
from base.instrument import register_job_data
|
|
from galaxy.tools import DataManagerTool
|
|
from galaxy.util import bunch
|
|
import logging
|
|
try:
|
|
from nose.tools import nottest
|
|
except ImportError:
|
|
nottest = lambda x: x
|
|
|
|
log = logging.getLogger( __name__ )
|
|
|
|
toolbox = None
|
|
|
|
#Do not test Data Managers as part of the standard Tool Test Framework.
|
|
TOOL_TYPES_NO_TEST = ( DataManagerTool, )
|
|
|
|
class ToolTestCase( TwillTestCase ):
|
|
"""Abstract test case that runs tests based on a `galaxy.tools.test.ToolTest`"""
|
|
|
|
def do_it( self, testdef ):
|
|
"""
|
|
Run through a tool test case.
|
|
"""
|
|
shed_tool_id = self.shed_tool_id
|
|
|
|
self._handle_test_def_errors( testdef )
|
|
|
|
galaxy_interactor = self._galaxy_interactor( testdef )
|
|
|
|
test_history = galaxy_interactor.new_history()
|
|
|
|
stage_data_in_history( galaxy_interactor, testdef.test_data(), test_history, shed_tool_id )
|
|
|
|
# Once data is ready, run the tool and check the outputs - record API
|
|
# input, job info, tool run exception, as well as exceptions related to
|
|
# job output checking and register they with the test plugin so it can
|
|
# record structured information.
|
|
tool_inputs = None
|
|
job_stdio = None
|
|
job_output_exceptions = None
|
|
tool_execution_exception = None
|
|
try:
|
|
try:
|
|
tool_response = galaxy_interactor.run_tool( testdef, test_history )
|
|
data_list, jobs, tool_inputs = tool_response.outputs, tool_response.jobs, tool_response.inputs
|
|
data_collection_list = tool_response.output_collections
|
|
except RunToolException as e:
|
|
tool_inputs = e.inputs
|
|
tool_execution_exception = e
|
|
raise e
|
|
except Exception as e:
|
|
tool_execution_exception = e
|
|
raise e
|
|
|
|
self.assertTrue( data_list or data_collection_list )
|
|
|
|
try:
|
|
job_stdio = self._verify_outputs( testdef, test_history, jobs, shed_tool_id, data_list, data_collection_list, galaxy_interactor )
|
|
except JobOutputsError as e:
|
|
job_stdio = e.job_stdio
|
|
job_output_exceptions = e.output_exceptions
|
|
raise e
|
|
except Exception as e:
|
|
job_output_exceptions = [e]
|
|
raise e
|
|
finally:
|
|
job_data = {}
|
|
if tool_inputs is not None:
|
|
job_data["inputs"] = tool_inputs
|
|
if job_stdio is not None:
|
|
job_data["job"] = job_stdio
|
|
if job_output_exceptions:
|
|
job_data["output_problems"] = map(str, job_output_exceptions)
|
|
if tool_execution_exception:
|
|
job_data["execution_problem"] = str(tool_execution_exception)
|
|
register_job_data(job_data)
|
|
|
|
galaxy_interactor.delete_history( test_history )
|
|
|
|
def _galaxy_interactor( self, testdef ):
|
|
return build_interactor( self, testdef.interactor )
|
|
|
|
def _handle_test_def_errors(self, testdef):
|
|
# If the test generation had an error, raise
|
|
if testdef.error:
|
|
if testdef.exception:
|
|
raise testdef.exception
|
|
else:
|
|
raise Exception( "Test parse failure" )
|
|
|
|
def _verify_outputs( self, testdef, history, jobs, shed_tool_id, data_list, data_collection_list, galaxy_interactor ):
|
|
assert len(jobs) == 1, "Test framework logic error, somehow tool test resulted in more than one job."
|
|
job = jobs[ 0 ]
|
|
|
|
maxseconds = testdef.maxseconds
|
|
if testdef.num_outputs is not None:
|
|
expected = testdef.num_outputs
|
|
actual = len( data_list )
|
|
if expected != actual:
|
|
messaage_template = "Incorrect number of outputs - expected %d, found %s."
|
|
message = messaage_template % ( expected, actual )
|
|
raise Exception( message )
|
|
found_exceptions = []
|
|
|
|
def register_exception(e):
|
|
if not found_exceptions:
|
|
# Only print this stuff out once.
|
|
for stream in ['stdout', 'stderr']:
|
|
if stream in job_stdio:
|
|
print >>sys.stderr, self._format_stream( job_stdio[ stream ], stream=stream, format=True )
|
|
found_exceptions.append(e)
|
|
|
|
if testdef.expect_failure:
|
|
if testdef.outputs:
|
|
raise Exception("Cannot specify outputs in a test expecting failure.")
|
|
|
|
# Wait for the job to complete and register expections if the final
|
|
# status was not what test was expecting.
|
|
job_failed = False
|
|
try:
|
|
galaxy_interactor.wait_for_job( job[ 'id' ], history, maxseconds )
|
|
except Exception as e:
|
|
job_failed = True
|
|
if not testdef.expect_failure:
|
|
found_exceptions.append(e)
|
|
|
|
job_stdio = galaxy_interactor.get_job_stdio( job[ 'id' ] )
|
|
|
|
if not job_failed and testdef.expect_failure:
|
|
error = AssertionError("Expected job to fail but Galaxy indicated the job successfully completed.")
|
|
register_exception(error)
|
|
|
|
expect_exit_code = testdef.expect_exit_code
|
|
if expect_exit_code is not None:
|
|
exit_code = job_stdio["exit_code"]
|
|
if str(expect_exit_code) != str(exit_code):
|
|
error = AssertionError("Expected job to complete with exit code %s, found %s" % (expect_exit_code, exit_code))
|
|
register_exception(error)
|
|
|
|
for output_index, output_tuple in enumerate(testdef.outputs):
|
|
# Get the correct hid
|
|
name, outfile, attributes = output_tuple
|
|
output_testdef = bunch.Bunch( name=name, outfile=outfile, attributes=attributes )
|
|
try:
|
|
output_data = data_list[ name ]
|
|
except (TypeError, KeyError):
|
|
# Legacy - fall back on ordered data list access if data_list is
|
|
# just a list (case with twill variant or if output changes its
|
|
# name).
|
|
if hasattr(data_list, "values"):
|
|
output_data = data_list.values()[ output_index ]
|
|
else:
|
|
output_data = data_list[ len(data_list) - len(testdef.outputs) + output_index ]
|
|
self.assertTrue( output_data is not None )
|
|
try:
|
|
galaxy_interactor.verify_output( history, jobs, output_data, output_testdef=output_testdef, shed_tool_id=shed_tool_id, maxseconds=maxseconds )
|
|
except Exception as e:
|
|
register_exception(e)
|
|
|
|
other_checks = {
|
|
"command_line": "Command produced by the job",
|
|
"stdout": "Standard output of the job",
|
|
"stderr": "Standard error of the job",
|
|
}
|
|
for what, description in other_checks.items():
|
|
if getattr( testdef, what, None ) is not None:
|
|
try:
|
|
data = job_stdio[what]
|
|
verify_assertions( data, getattr( testdef, what ) )
|
|
except AssertionError, err:
|
|
errmsg = '%s different than expected\n' % description
|
|
errmsg += str( err )
|
|
register_exception( AssertionError( errmsg ) )
|
|
|
|
for output_collection_def in testdef.output_collections:
|
|
try:
|
|
name = output_collection_def.name
|
|
# TODO: data_collection_list is clearly a bad name for dictionary.
|
|
if name not in data_collection_list:
|
|
template = "Failed to find output [%s], tool outputs include [%s]"
|
|
message = template % (name, ",".join(data_collection_list.keys()))
|
|
raise AssertionError(message)
|
|
|
|
# Data collection returned from submission, elements may have been populated after
|
|
# the job completed so re-hit the API for more information.
|
|
data_collection_returned = data_collection_list[ name ]
|
|
data_collection = galaxy_interactor._get( "dataset_collections/%s" % data_collection_returned[ "id" ], data={"instance_type": "history"} ).json()
|
|
elements = data_collection[ "elements" ]
|
|
element_dict = dict( map(lambda e: (e["element_identifier"], e["object"]), elements) )
|
|
|
|
expected_collection_type = output_collection_def.collection_type
|
|
if expected_collection_type:
|
|
collection_type = data_collection[ "collection_type"]
|
|
if expected_collection_type != collection_type:
|
|
template = "Expected output collection [%s] to be of type [%s], was of type [%s]."
|
|
message = template % (name, expected_collection_type, collection_type)
|
|
raise AssertionError(message)
|
|
|
|
for element_identifier, ( element_outfile, element_attrib ) in output_collection_def.element_tests.items():
|
|
if element_identifier not in element_dict:
|
|
template = "Failed to find identifier [%s] for testing, tool generated collection with identifiers [%s]"
|
|
message = template % (element_identifier, ",".join(element_dict.keys()))
|
|
raise AssertionError(message)
|
|
hda = element_dict[ element_identifier ]
|
|
|
|
galaxy_interactor.verify_output_dataset(
|
|
history,
|
|
hda_id=hda["id"],
|
|
outfile=element_outfile,
|
|
attributes=element_attrib,
|
|
shed_tool_id=shed_tool_id
|
|
)
|
|
except Exception as e:
|
|
register_exception(e)
|
|
|
|
if found_exceptions:
|
|
raise JobOutputsError(found_exceptions, job_stdio)
|
|
else:
|
|
return job_stdio
|
|
|
|
|
|
class JobOutputsError(AssertionError):
|
|
|
|
def __init__(self, output_exceptions, job_stdio):
|
|
big_message = "\n".join(map(str, output_exceptions))
|
|
super(JobOutputsError, self).__init__(big_message)
|
|
self.job_stdio = job_stdio
|
|
self.output_exceptions = output_exceptions
|
|
|
|
|
|
@nottest
|
|
def build_tests( app=None, testing_shed_tools=False, master_api_key=None, user_api_key=None ):
|
|
"""
|
|
If the module level variable `toolbox` is set, generate `ToolTestCase`
|
|
classes for all of its tests and put them into this modules globals() so
|
|
they can be discovered by nose.
|
|
"""
|
|
if app is None:
|
|
return
|
|
|
|
# Push all the toolbox tests to module level
|
|
G = globals()
|
|
|
|
# Eliminate all previous tests from G.
|
|
for key, val in G.items():
|
|
if key.startswith( 'TestForTool_' ):
|
|
del G[ key ]
|
|
for i, tool_id in enumerate( app.toolbox.tools_by_id ):
|
|
tool = app.toolbox.get_tool( tool_id )
|
|
if isinstance( tool, TOOL_TYPES_NO_TEST ):
|
|
#We do not test certain types of tools (e.g. Data Manager tools) as part of ToolTestCase
|
|
continue
|
|
if tool.tests:
|
|
shed_tool_id = None if not testing_shed_tools else tool.id
|
|
# Create a new subclass of ToolTestCase, dynamically adding methods
|
|
# named test_tool_XXX that run each test defined in the tool config.
|
|
name = "TestForTool_" + tool.id.replace( ' ', '_' )
|
|
baseclasses = ( ToolTestCase, )
|
|
namespace = dict()
|
|
for j, testdef in enumerate( tool.tests ):
|
|
def make_test_method( td ):
|
|
def test_tool( self ):
|
|
self.do_it( td )
|
|
return test_tool
|
|
test_method = make_test_method( testdef )
|
|
test_method.__doc__ = "%s ( %s ) > %s" % ( tool.name, tool.id, testdef.name )
|
|
namespace[ 'test_tool_%06d' % j ] = test_method
|
|
namespace[ 'shed_tool_id' ] = shed_tool_id
|
|
namespace[ 'master_api_key' ] = master_api_key
|
|
namespace[ 'user_api_key' ] = user_api_key
|
|
# The new.classobj function returns a new class object, with name name, derived
|
|
# from baseclasses (which should be a tuple of classes) and with namespace dict.
|
|
new_class_obj = new.classobj( name, baseclasses, namespace )
|
|
G[ name ] = new_class_obj
|