import os from xml.etree.ElementTree import XML from unittest import TestCase from galaxy.model import Job from galaxy.model import History from galaxy.model import Dataset from galaxy.model import JobParameter from galaxy.model import HistoryDatasetAssociation from galaxy.model import JobToInputDatasetAssociation from galaxy.tools.evaluation import ToolEvaluator from galaxy.jobs import SimpleComputeEnvironment from galaxy.jobs.datasets import DatasetPath from galaxy.util.bunch import Bunch # For MockTool from galaxy.tools.parameters import params_from_strings from galaxy.tools import ToolOutput from galaxy.tools.parameters.grouping import Repeat from galaxy.tools.parameters.grouping import Conditional from galaxy.tools.parameters.grouping import ConditionalWhen from galaxy.tools.parameters.basic import IntegerToolParameter from galaxy.tools.parameters.basic import SelectToolParameter from galaxy.tools.parameters.basic import DataToolParameter # Test fixtures for Galaxy infrastructure. from tools_support import UsesApp # To Test: # - param_file handling. TEST_TOOL_DIRECTORY = "/path/to/the/tool" class ToolEvaluatorTestCase(TestCase, UsesApp): def setUp(self): self.setup_app() self.tool = MockTool(self.app) self.job = Job() self.job.history = History() self.job.parameters = [ JobParameter( name="thresh", value="4" ) ] self.evaluator = ToolEvaluator( self.app, self.tool, self.job, self.test_directory ) def tearDown(self): self.tear_down_app() def test_simple_evaluation( self ): self._setup_test_bwa_job() self._set_compute_environment() command_line, extra_filenames = self.evaluator.build( ) self.assertEquals( command_line, "bwa --thresh=4 --in=/galaxy/files/dataset_1.dat --out=/galaxy/files/dataset_2.dat" ) def test_repeat_evaluation( self ): repeat = Repeat() repeat.name = "r" repeat.inputs = { "thresh": self.tool.test_thresh_param() } self.tool.set_params( { "r": repeat } ) self.job.parameters = [ JobParameter( name="r", value='''[{"thresh": 4, "__index__": 0},{"thresh": 5, "__index__": 1}]''' ) ] self.tool._command_line = "prog1 #for $r_i in $r # $r_i.thresh#end for#" self._set_compute_environment() command_line, extra_filenames = self.evaluator.build( ) self.assertEquals( command_line, "prog1 4 5" ) def test_conditional_evaluation( self ): select_xml = XML('''''') parameter = SelectToolParameter( self.tool, select_xml ) conditional = Conditional() conditional.name = "c" conditional.test_param = parameter when = ConditionalWhen() when.inputs = { "thresh": self.tool.test_thresh_param() } when.value = "true" conditional.cases = [ when ] self.tool.set_params( { "c": conditional } ) self.job.parameters = [ JobParameter( name="c", value='''{"thresh": 4, "always_true": "true", "__current_case__": 0}''' ) ] self.tool._command_line = "prog1 --thresh=${c.thresh} --test_param=${c.always_true}" self._set_compute_environment() command_line, extra_filenames = self.evaluator.build( ) self.assertEquals( command_line, "prog1 --thresh=4 --test_param=true" ) def test_evaluation_of_optional_datasets( self ): # Make sure optional dataset don't cause evaluation to break and # evaluate in cheetah templates as 'None'. select_xml = XML('''''') parameter = DataToolParameter( self.tool, select_xml ) self.job.parameters = [ JobParameter( name="input1", value=u'null' ) ] self.tool.set_params( { "input1": parameter } ) self.tool._command_line = "prog1 --opt_input='${input1}'" self._set_compute_environment() command_line, extra_filenames = self.evaluator.build( ) self.assertEquals( command_line, "prog1 --opt_input='None'" ) def test_evaluation_with_path_rewrites_wrapped( self ): self.tool.check_values = True self.__test_evaluation_with_path_rewrites() def test_evaluation_with_path_rewrites_unwrapped( self ): self.tool.check_values = False self.__test_evaluation_with_path_rewrites() def __test_evaluation_with_path_rewrites( self ): # Various things can cause dataset paths to be rewritten (Task # splitting, config.outputs_to_working_directory). This tests that #functionality. self._setup_test_bwa_job() job_path_1 = "%s/dataset_1.dat" % self.test_directory job_path_2 = "%s/dataset_2.dat" % self.test_directory self._set_compute_environment( input_paths=[DatasetPath(1, '/galaxy/files/dataset_1.dat', false_path=job_path_1)], output_paths=[DatasetPath(2, '/galaxy/files/dataset_2.dat', false_path=job_path_2)], ) command_line, extra_filenames = self.evaluator.build( ) self.assertEquals( command_line, "bwa --thresh=4 --in=%s --out=%s" % (job_path_1, job_path_2) ) def test_configfiles_evaluation( self ): self.tool.config_files.append( ( "conf1", None, "$thresh" ) ) self.tool._command_line = "prog1 $conf1" self._set_compute_environment() command_line, extra_filenames = self.evaluator.build( ) self.assertEquals( len( extra_filenames ), 1) config_filename = extra_filenames[ 0 ] config_basename = os.path.basename( config_filename ) # Verify config file written into working directory. self.assertEquals( os.path.join( self.test_directory, config_basename ), config_filename ) # Verify config file contents are evaluated against parameters. assert open( config_filename, "r").read() == "4" self.assertEquals(command_line, "prog1 %s" % config_filename) def test_arbitrary_path_rewriting_wrapped( self ): self.tool.check_values = True self.__test_arbitrary_path_rewriting() def test_arbitrary_path_rewriting_unwrapped( self ): self.tool.check_values = False self.__test_arbitrary_path_rewriting() def __test_arbitrary_path_rewriting( self ): self.job.parameters = [ JobParameter( name="index_path", value="\"/old/path/human\"" ) ] xml = XML(''' ''') parameter = SelectToolParameter( self.tool, xml ) def get_field_by_name_for_value( name, value, trans, other_values ): assert value == "/old/path/human" assert name == "path" return ["/old/path/human"] parameter.options = Bunch(get_field_by_name_for_value=get_field_by_name_for_value) self.tool.set_params( { "index_path": parameter } ) self.tool._command_line = "prog1 $index_path.fields.path" def test_path_rewriter(v): if v: v = v.replace("/old", "/new") return v self._set_compute_environment(path_rewriter=test_path_rewriter) command_line, extra_filenames = self.evaluator.build( ) self.assertEquals(command_line, "prog1 /new/path/human") def test_template_property_app( self ): self._assert_template_property_is("$__app__.config.new_file_path", self.app.config.new_file_path) def test_template_property_new_file_path( self ): self._assert_template_property_is("$__new_file_path__", self.app.config.new_file_path) def test_template_property_root_dir( self ): self._assert_template_property_is("$__root_dir__", self.app.config.root) def test_template_property_admin_users( self ): self._assert_template_property_is("$__admin_users__", "mary@example.com") def _assert_template_property_is(self, expression, value): self.tool._command_line = "test.exe" self.tool.config_files.append( ( "conf1", None, """%s""" % expression) ) self._set_compute_environment() _, extra_filenames = self.evaluator.build( ) config_filename = extra_filenames[ 0 ] self.assertEquals(open( config_filename, "r").read(), value) def _set_compute_environment(self, **kwds): if "working_directory" not in kwds: kwds[ "working_directory" ] = self.test_directory if "new_file_path" not in kwds: kwds[ "new_file_path" ] = self.app.config.new_file_path self.evaluator.set_compute_environment( TestComputeEnviornment( **kwds ) ) assert "exec_before_job" in self.tool.hooks_called def _setup_test_bwa_job( self ): self.job.input_datasets = [ self._job_dataset( 'input1', '/galaxy/files/dataset_1.dat' ) ] self.job.output_datasets = [ self._job_dataset( 'output1', '/galaxy/files/dataset_2.dat' ) ] def _job_dataset( self, name, path ): metadata = dict( ) hda = HistoryDatasetAssociation( name=name, metadata=metadata ) hda.dataset = Dataset( id=123, external_filename=path ) hda.dataset.metadata = dict() hda.children = [] jida = JobToInputDatasetAssociation( name=name, dataset=hda ) return jida class MockHistoryDatasetAssociation( HistoryDatasetAssociation ): def __init__( self, **kwds ): self._metadata = dict() super( MockHistoryDatasetAssociation, self ).__init__( **kwds ) class TestComputeEnviornment( SimpleComputeEnvironment ): def __init__( self, new_file_path, working_directory, input_paths=[ '/galaxy/files/dataset_1.dat' ], output_paths=[ '/galaxy/files/dataset_2.dat' ], path_rewriter=None ): self._new_file_path = new_file_path self._working_directory = working_directory self._input_paths = input_paths self._output_paths = output_paths self._path_rewriter = path_rewriter def input_paths( self ): return self._input_paths def output_paths( self ): return self._output_paths def working_directory( self ): return self._working_directory def new_file_path(self): return self._new_file_path def unstructured_path_rewriter(self): if self._path_rewriter: return self._path_rewriter else: return super(TestComputeEnviornment, self).unstructured_path_rewriter() def tool_directory( self ): return TEST_TOOL_DIRECTORY class MockTool( object ): def __init__( self, app ): self.app = app self.hooks_called = [] self._config_files = [] self._command_line = "bwa --thresh=$thresh --in=$input1 --out=$output1" self._params = { "thresh": self.test_thresh_param() } self.options = Bunch(sanitize=False) self.check_values = True def test_thresh_param( self ): elem = XML( '' ) return IntegerToolParameter( self, elem ) def params_from_strings( self, params, app, ignore_errors=False ): return params_from_strings( self.inputs, params, app, ignore_errors ) @property def template_macro_params( self ): return {} @property def inputs( self ): return self._params def set_params( self, params ): self._params = params @property def outputs( self ): #elem_output1 = XML( '' ) # DataToolParameter( self, elem_output1 ), return dict( output1=ToolOutput( "output1" ), ) @property def config_files( self ): return self._config_files @property def command( self ): return self._command_line @property def interpreter( self ): return None def handle_unvalidated_param_values( self, input_values, app ): pass def build_param_dict( self, incoming, *args, **kwds ): return incoming def call_hook( self, hook_name, *args, **kwargs ): self.hooks_called.append( hook_name ) def exec_before_job( self, *args, **kwargs ): pass