import os import string import unittest from galaxy.tools import ToolBox from galaxy import model from galaxy.model import tool_shed_install from galaxy.model.tool_shed_install import mapping import tools_support CONFIG_TEST_TOOL_VERSION_TEMPLATE = string.Template( """ github.com example galaxyproject ${version} github.com/galaxyproject/example/test_tool/0.${version} 0.${version} """ ) CONFIG_TEST_TOOL_VERSION_1 = CONFIG_TEST_TOOL_VERSION_TEMPLATE.safe_substitute( dict( version="1" ) ) CONFIG_TEST_TOOL_VERSION_2 = CONFIG_TEST_TOOL_VERSION_TEMPLATE.safe_substitute( dict( version="2" ) ) class BaseToolBoxTestCase( unittest.TestCase, tools_support.UsesApp, tools_support.UsesTools ): @property def integerated_tool_panel_path( self ): return os.path.join(self.test_directory, "integrated_tool_panel.xml") def assert_integerated_tool_panel( self, exists=True ): does_exist = os.path.exists( self.integerated_tool_panel_path ) if exists: assert does_exist else: assert not does_exist @property def toolbox( self ): if self.__toolbox is None: self.__toolbox = SimplifiedToolBox( self ) # wire app with this new toolbox self.app.toolbox = self.__toolbox return self.__toolbox def setUp( self ): self.reindexed = False self.setup_app( mock_model=False ) install_model = mapping.init( "sqlite:///:memory:", create_tables=True ) self.app.install_model = install_model self.app.reindex_tool_search = self.__reindex itp_config = os.path.join(self.test_directory, "integrated_tool_panel.xml") self.app.config.integrated_tool_panel_config = itp_config self.__toolbox = None self.config_files = [] def _repo_install( self, changeset ): repository = tool_shed_install.ToolShedRepository() repository.tool_shed = "github.com" repository.owner = "galaxyproject" repository.name = "example" repository.changeset_revision = changeset repository.installed_changeset_revision = changeset repository.deleted = False repository.uninstalled = False self.app.install_model.context.add( repository ) self.app.install_model.context.flush( ) return repository def _setup_two_versions( self ): repository1 = self._repo_install( changeset="1" ) version1 = tool_shed_install.ToolVersion() version1.tool_id = "github.com/galaxyproject/example/test_tool/0.1" version1.repository = repository1 self.app.install_model.context.add( version1 ) self.app.install_model.context.flush( ) repository2 = self._repo_install( changeset="2" ) version2 = tool_shed_install.ToolVersion() version2.tool_id = "github.com/galaxyproject/example/test_tool/0.2" version2.repository = repository2 self.app.install_model.context.add( version2 ) self.app.install_model.context.flush( ) version_association = tool_shed_install.ToolVersionAssociation() version_association.parent_id = version1.id version_association.tool_id = version2.id self.app.install_model.context.add( version_association ) self.app.install_model.context.flush( ) def _setup_two_versions_in_config( self, section=False ): if section: template = """
%s
%s
""" else: template = """
%s
%s
""" self._add_config( template % (self.test_directory, CONFIG_TEST_TOOL_VERSION_1, CONFIG_TEST_TOOL_VERSION_2 ) ) def _add_config( self, xml, name="tool_conf.xml" ): path = self._tool_conf_path( name=name ) with open( path, "w" ) as f: f.write( xml ) self.config_files.append( path ) def _tool_conf_path( self, name="tool_conf.xml" ): path = os.path.join( self.test_directory, name ) return path def _tool_path( self, name="tool.xml" ): path = os.path.join( self.test_directory, name ) return path def __reindex( self ): self.reindexed = True class ToolBoxTestCase( BaseToolBoxTestCase ): def test_load_file( self ): self._init_tool() self._add_config( """""" ) toolbox = self.toolbox assert toolbox.get_tool( "test_tool" ) is not None assert toolbox.get_tool( "not_a_test_tool" ) is None def test_tool_shed_properties( self ): self._init_tool() self._setup_two_versions_in_config( section=False ) self._setup_two_versions() test_tool = self.toolbox.get_tool( "test_tool" ) assert test_tool.tool_shed == "github.com" assert test_tool.repository_owner == "galaxyproject" assert test_tool.repository_name == "example" assert test_tool.installed_changeset_revision == "1" def test_tool_shed_properties_only_on_installed_tools( self ): self._init_tool() self._add_config( """""" ) toolbox = self.toolbox test_tool = toolbox.get_tool( "test_tool" ) assert test_tool.tool_shed is None assert test_tool.repository_name is None assert test_tool.repository_owner is None assert test_tool.installed_changeset_revision is None def test_load_file_in_section( self ): self._init_tool() self._add_config( """
""" ) toolbox = self.toolbox assert toolbox.get_tool( "test_tool" ) is not None assert toolbox.get_tool( "not_a_test_tool" ) is None def test_writes_integrate_tool_panel( self ): self._init_tool() self._add_config( """""" ) self.assert_integerated_tool_panel(exists=False) self.toolbox self.assert_integerated_tool_panel(exists=True) def test_groups_tools_in_section( self ): self._init_tool() self._setup_two_versions_in_config( section=True ) self._setup_two_versions() self.toolbox self.__verify_two_test_tools( ) # Assert only newer version of the tool loaded into the panel. section = self.toolbox._tool_panel["tid"] assert len(section.elems) == 1 assert section.elems.values()[0].id == "github.com/galaxyproject/example/test_tool/0.2" def test_group_tools_out_of_section( self ): self._init_tool() self._setup_two_versions_in_config( section=False ) self._setup_two_versions() self.__verify_two_test_tools( ) # Assert tools merged in tool panel. assert len( self.toolbox._tool_panel ) == 1 def test_update_shed_conf(self): self.__setup_shed_tool_conf() self.toolbox.update_shed_config( { "config_filename": "tool_conf.xml" } ) assert self.reindexed self.assert_integerated_tool_panel(exists=True) def test_update_shed_conf_deactivate_only(self): self.__setup_shed_tool_conf() self.toolbox.update_shed_config( { "config_filename": "tool_conf.xml" }, integrated_panel_changes=False ) assert self.reindexed # No changes, should be regenerated self.assert_integerated_tool_panel(exists=False) def test_get_tool_id( self ): self._init_tool() self._setup_two_versions_in_config( ) self._setup_two_versions() assert self.toolbox.get_tool_id( "test_tool" ) == "github.com/galaxyproject/example/test_tool/0.1" assert self.toolbox.get_tool_id( "github.com/galaxyproject/example/test_tool/0.1" ) == "github.com/galaxyproject/example/test_tool/0.1" assert self.toolbox.get_tool_id( "github.com/galaxyproject/example/test_tool/0.2" ) == "github.com/galaxyproject/example/test_tool/0.2" assert self.toolbox.get_tool_id( "github.com/galaxyproject/example/test_tool/0.3" ) is None def test_tool_dir( self ): self._init_tool() self._add_config( """""" % self.test_directory ) toolbox = self.toolbox assert toolbox.get_tool( "test_tool" ) is not None def test_workflow_in_panel( self ): stored_workflow = self.__test_workflow() encoded_id = self.app.security.encode_id( stored_workflow.id ) self._add_config( """""" % encoded_id ) assert len( self.toolbox._tool_panel ) == 1 panel_workflow = self.toolbox._tool_panel.values()[ 0 ] assert panel_workflow == stored_workflow.latest_workflow # TODO: test to_dict with workflows def test_workflow_in_section( self ): stored_workflow = self.__test_workflow() encoded_id = self.app.security.encode_id( stored_workflow.id ) self._add_config( """
""" % encoded_id ) assert len( self.toolbox._tool_panel ) == 1 section = self.toolbox._tool_panel[ 'tid' ] assert len( section.elems ) == 1 panel_workflow = section.elems.values()[ 0 ] assert panel_workflow == stored_workflow.latest_workflow def test_label_in_panel( self ): self._add_config( """""" ) assert len( self.toolbox._tool_panel ) == 2 self.__check_test_labels( self.toolbox._tool_panel ) def test_label_in_section( self ): self._add_config( """
""" ) assert len( self.toolbox._tool_panel ) == 1 section = self.toolbox._tool_panel[ 'tid' ] self.__check_test_labels( section.elems ) def __check_test_labels( self, panel_dict ): assert panel_dict.keys() == ["label_lab1", "label_lab2"] label1 = panel_dict.values()[ 0 ] assert label1.id == "lab1" assert label1.text == "Label 1" label2 = panel_dict[ "label_lab2" ] assert label2.id == "lab2" assert label2.text == "Label 2" def __test_workflow( self ): stored_workflow = model.StoredWorkflow() workflow = model.Workflow() workflow.stored_workflow = stored_workflow stored_workflow.latest_workflow = workflow user = model.User() user.email = "test@example.com" user.password = "passw0rD1" stored_workflow.user = user self.app.model.context.add( workflow ) self.app.model.context.add( stored_workflow ) self.app.model.context.flush() return stored_workflow def __verify_two_test_tools( self ): # Assert tool versions of the tool with simple id 'test_tool' all_versions = self.toolbox.get_tool( "test_tool", get_all_versions=True ) assert len( all_versions ) == 2 # Verify lineage_ids on both tools is correctly ordered. for version in ["0.1", "0.2"]: guid = "github.com/galaxyproject/example/test_tool/" + version lineage_ids = self.toolbox.get_tool( guid ).lineage.get_version_ids() assert lineage_ids[ 0 ] == "github.com/galaxyproject/example/test_tool/0.1" assert lineage_ids[ 1 ] == "github.com/galaxyproject/example/test_tool/0.2" # Test tool_version attribute. assert self.toolbox.get_tool( "test_tool", tool_version="0.1" ).guid == "github.com/galaxyproject/example/test_tool/0.1" assert self.toolbox.get_tool( "test_tool", tool_version="0.2" ).guid == "github.com/galaxyproject/example/test_tool/0.2" def test_default_lineage( self ): self.__init_versioned_tools() self._add_config( """""" ) self.__verify_get_tool_for_default_lineage() def test_default_lineage_reversed( self ): # Run same test as above but with entries in tool_conf reversed to # ensure versioning is at work and not order effects. self.__init_versioned_tools() self._add_config( """""" ) self.__verify_get_tool_for_default_lineage() def test_grouping_with_default_lineage( self ): self.__init_versioned_tools() self._add_config( """""" ) self.__verify_tool_panel_for_default_lineage() def test_grouping_with_default_lineage_reversed( self ): # Run same test as above but with entries in tool_conf reversed to # ensure versioning is at work and not order effects. self.__init_versioned_tools() self._add_config( """""" ) self.__verify_tool_panel_for_default_lineage() def __init_versioned_tools( self ): self._init_tool( filename="tool_v01.xml", version="0.1" ) self._init_tool( filename="tool_v02.xml", version="0.2" ) def __verify_tool_panel_for_default_lineage( self ): assert len( self.toolbox._tool_panel ) == 1 tool = self.toolbox._tool_panel["tool_test_tool"] assert tool.version == "0.2", tool.version assert tool.id == "test_tool" def __verify_get_tool_for_default_lineage( self ): tool_v01 = self.toolbox.get_tool( "test_tool", tool_version="0.1" ) tool_v02 = self.toolbox.get_tool( "test_tool", tool_version="0.2" ) assert tool_v02.id == "test_tool" assert tool_v02.version == "0.2", tool_v02.version assert tool_v01.id == "test_tool" assert tool_v01.version == "0.1" # Newer variant gets to be default for that id. default_tool = self.toolbox.get_tool( "test_tool" ) assert default_tool.id == "test_tool" assert default_tool.version == "0.2" def __remove_itp( self ): os.remove( os.path) def __setup_shed_tool_conf( self ): self._add_config( """""" ) self.toolbox # create toolbox assert not self.reindexed os.remove( self.integerated_tool_panel_path ) class SimplifiedToolBox( ToolBox ): def __init__( self, test_case ): app = test_case.app # Handle app/config stuff needed by toolbox but not by tools. app.job_config.get_tool_resource_parameters = lambda tool_id: None app.config.update_integrated_tool_panel = True config_files = test_case.config_files tool_root_dir = test_case.test_directory super( SimplifiedToolBox, self ).__init__( config_files, tool_root_dir, app, )