Source code for galaxy.tool_shed.tools.tool_validator

import logging

from galaxy.tool_shed.tools.data_table_manager import ShedToolDataTableManager
from galaxy.tool_shed.util import (
    basic_util,
    hg_util
)
from galaxy.tool_util.fetcher import ToolLocationFetcher
from galaxy.tools import (
    create_tool_from_source,
    get_tool_source,
    parameters,
)
from galaxy.tools.parameters import dynamic_options

log = logging.getLogger(__name__)


[docs]class ToolValidator:

[docs]    def __init__(self, app):
        self.app = app
        self.stdtm = ShedToolDataTableManager(self.app)

[docs]    def check_tool_input_params(self, repo_dir, tool_config_name, tool, sample_files):
        """
        Check all of the tool's input parameters, looking for any that are dynamically
        generated using external data files to make sure the files exist.
        """
        invalid_files_and_errors_tups = []
        for input_param in tool.input_params:
            if isinstance(input_param, parameters.basic.SelectToolParameter) and input_param.is_dynamic:
                # If the tool refers to .loc files or requires an entry in the tool_data_table_conf.xml,
                # make sure all requirements exist.
                options = input_param.dynamic_options or input_param.options
                if options and isinstance(options, dynamic_options.DynamicOptions):
                    if options.tool_data_table or options.missing_tool_data_table_name:
                        # Make sure the repository contains a tool_data_table_conf.xml.sample file.
                        sample_tool_data_table_conf = hg_util.get_config_from_disk('tool_data_table_conf.xml.sample', repo_dir)
                        if sample_tool_data_table_conf:
                            error, correction_msg = \
                                self.stdtm.handle_sample_tool_data_table_conf_file(sample_tool_data_table_conf,
                                                                                  persist=False)
                            if error:
                                invalid_files_and_errors_tups.append(('tool_data_table_conf.xml.sample', correction_msg))
                        else:
                            correction_msg = "This file requires an entry in the tool_data_table_conf.xml file.  "
                            correction_msg += "Upload a file named tool_data_table_conf.xml.sample to the repository "
                            correction_msg += "that includes the required entry to correct this error.<br/>"
                            invalid_tup = (tool_config_name, correction_msg)
                            if invalid_tup not in invalid_files_and_errors_tups:
                                invalid_files_and_errors_tups.append(invalid_tup)
                    if options.index_file or options.tool_data_table and options.tool_data_table.missing_index_file:
                        # Make sure the repository contains the required xxx.loc.sample file.
                        index_file = options.index_file or options.tool_data_table.missing_index_file
                        index_file_name = basic_util.strip_path(index_file)
                        sample_found = False
                        for sample_file in sample_files:
                            sample_file_name = basic_util.strip_path(sample_file)
                            if sample_file_name == '%s.sample' % index_file_name:
                                options.index_file = index_file_name
                                if options.tool_data_table:
                                    options.tool_data_table.missing_index_file = None
                                sample_found = True
                                break
                        if not sample_found:
                            correction_msg = "This file refers to a file named <b>%s</b>.  " % str(index_file_name)
                            correction_msg += "Upload a file named <b>%s.sample</b> to the repository to correct this error." % \
                                str(index_file_name)
                            invalid_files_and_errors_tups.append((tool_config_name, correction_msg))
        return invalid_files_and_errors_tups

[docs]    def load_tool_from_config(self, repository_id, full_path):
        tool_source = get_tool_source(
            full_path,
            enable_beta_formats=getattr(self.app.config, "enable_beta_tool_formats", False),
            tool_location_fetcher=ToolLocationFetcher(),
        )
        try:
            tool = create_tool_from_source(config_file=full_path, app=self.app, tool_source=tool_source, repository_id=repository_id, allow_code_files=False)
            tool.assert_finalized(raise_if_invalid=True)
            valid = True
            error_message = None
        except KeyError as e:
            tool = None
            valid = False
            error_message = 'This file requires an entry for "%s" in the tool_data_table_conf.xml file.  Upload a file ' % str(e)
            error_message += 'named tool_data_table_conf.xml.sample to the repository that includes the required entry to correct '
            error_message += 'this error.  '
            log.exception(error_message)
        except Exception as e:
            tool = None
            valid = False
            error_message = str(e)
            log.exception('Caught exception loading tool from %s:', full_path)
        return tool, valid, error_message