diff --git a/lib/galaxy/tool_shed/unittest_utils/__init__.py b/lib/galaxy/tool_shed/unittest_utils/__init__.py new file mode 100644 index 00000000000..7fb28659c83 --- /dev/null +++ b/lib/galaxy/tool_shed/unittest_utils/__init__.py @@ -0,0 +1,215 @@ +import threading +from pathlib import Path +from typing import ( + Any, + Dict, + List, + NamedTuple, + Optional, + Union, +) + +from galaxy.model.migrations import ( + DatabaseStateVerifier, + TSI, +) +from galaxy.model.orm.engine_factory import build_engine +from galaxy.model.tool_shed_install import mapping as install_mapping +from galaxy.security.idencoding import IdEncodingHelper +from galaxy.tool_shed.cache import ToolShedRepositoryCache +from galaxy.tool_shed.galaxy_install.client import ( + DataManagerInterface, + DataManagersInterface, + InstallationTarget, +) +from galaxy.tool_shed.util.repository_util import get_installed_repository +from galaxy.tool_util.data import ToolDataTableManager +from galaxy.tool_util.toolbox.base import AbstractToolBox +from galaxy.tool_util.toolbox.watcher import ( + get_tool_conf_watcher, + get_tool_watcher, +) +from galaxy.util.tool_shed.tool_shed_registry import Registry + + +class ToolShedTarget(NamedTuple): + url: str + name: str + + @property + def as_str(self) -> str: + return f""" + + + +""" + + +EMPTY_TOOL_DATA_TABLE_CONFIG = """ + + +""" + + +class Config: + tool_data_path: str + install_database_connection: str + install_database_engine_options: Dict[str, Any] = {} + update_integrated_tool_panel: bool = True + integrated_tool_panel_config: str + shed_tool_config_file: str + shed_tool_data_path: str + migrated_tools_config: Optional[str] = None + shed_tools_dir: str + edam_panel_views: list = [] + tool_configs: list = [] + shed_tool_data_table_config: str + shed_data_manager_config_file: str + + +class TestTool: + _macro_paths: List[str] = [] + params_with_missing_data_table_entry: list = [] + params_with_missing_index_file: list = [] + + def __init__(self, config_file, tool_shed_repository, guid): + self.config_file = config_file + self.tool_shed_repository = tool_shed_repository + self.guid = guid + self.id = guid + self.version = "1.0.0" + self.hidden = False + self._lineage = None + self.name = "test_tool" + + @property + def lineage(self): + return self._lineage + + +class TestToolBox(AbstractToolBox): + def create_tool(self, config_file, tool_cache_data_dir=None, **kwds): + tool = TestTool(config_file, kwds["tool_shed_repository"], kwds["guid"]) + tool._lineage = self._lineage_map.register(tool) # cleanup? + return tool + + def _get_tool_shed_repository(self, tool_shed, name, owner, installed_changeset_revision): + return get_installed_repository( + self.app, + tool_shed=tool_shed, + name=name, + owner=owner, + installed_changeset_revision=installed_changeset_revision, + from_cache=True, + ) + + +class Watchers: + def __init__(self, app): + self.app = app + self.tool_config_watcher = get_tool_conf_watcher( + reload_callback=self.app.reload_toolbox, + tool_cache=None, + ) + self.tool_watcher = get_tool_watcher(self, app.config) + + +class DummyDataManager(DataManagerInterface): + GUID_TYPE: str = "data_manager" + DEFAULT_VERSION: str = "0.0.1" + + def process_result(self, out_data): + return None + + def write_bundle(self, out) -> None: + return None + + +class StandaloneDataManagers(DataManagersInterface): + __reload_count = 0 + + def load_manager_from_elem( + self, data_manager_elem, tool_path=None, add_manager=True + ) -> Optional[DataManagerInterface]: + return DummyDataManager() + + def get_manager(self, data_manager_id: str) -> Optional[DataManagerInterface]: + return None + + def remove_manager(self, manager_ids: Union[str, List[str]]) -> None: + return None + + @property + def _reload_count(self) -> int: + self.__reload_count += 1 + return self.__reload_count + + +class StandaloneInstallationTarget(InstallationTarget): + name: str = "galaxy" + tool_shed_registry: Registry + security: IdEncodingHelper + _toolbox: TestToolBox + _toolbox_lock: threading.RLock = threading.RLock() + tool_shed_repository_cache: Optional[ToolShedRepositoryCache] = None + data_managers = StandaloneDataManagers() + + def __init__( + self, + target_directory: Path, + tool_shed_target: Optional[ToolShedTarget] = None, + ): + tool_root_dir = target_directory / "tools" + config: Config = Config() + install_db_path = str(target_directory / "install.sqlite") + config.tool_data_path = str(target_directory / "tool_data") + config.shed_tool_data_path = config.tool_data_path + config.install_database_connection = f"sqlite:///{install_db_path}?isolation_level=IMMEDIATE" + config.integrated_tool_panel_config = str(target_directory / "integrated.xml") + config.shed_tool_data_table_config = str(target_directory / "shed_tool_data_table_conf.xml") + shed_conf = target_directory / "shed_conf.xml" + shed_data_manager_config_file = target_directory / "shed_data_manager_conf.xml" + config.shed_data_manager_config_file = str(shed_data_manager_config_file) + config.shed_tool_config_file = str(shed_conf) + shed_conf.write_text(f'\n') + (target_directory / "shed_tool_data_table_conf.xml").write_text(EMPTY_TOOL_DATA_TABLE_CONFIG) + self.config = config + install_engine = build_engine(config.install_database_connection, config.install_database_engine_options) + self.security = IdEncodingHelper(id_secret="notasecretfortests") + DatabaseStateVerifier( + install_engine, + TSI, + None, + None, + True, + False, + ).run() + self.install_model = install_mapping.configure_model_mapping(install_engine) + registry_config: Optional[Path] = None + if tool_shed_target: + registry_config = target_directory / "tool_sheds_conf.xml" + with registry_config.open("w") as f: + f.write(tool_shed_target.as_str) + + self.tool_shed_registry = Registry(registry_config) + self.tool_root_dir = tool_root_dir + self.tool_root_dir.mkdir() + config.shed_tools_dir = str(tool_root_dir) + self.watchers = Watchers(self) + self.reload_toolbox() + self.tool_data_tables = ToolDataTableManager( + tool_data_path=self.config.tool_data_path, + config_filename=self.config.shed_tool_data_table_config, + other_config_dict=self.config, + ) + + def reload_toolbox(self): + self._toolbox = TestToolBox( + config_filenames=[self.config.shed_tool_config_file], + tool_root_dir=self.tool_root_dir, + app=self, + ) + + @property + def toolbox(self) -> TestToolBox: + return self._toolbox diff --git a/lib/galaxy/tools/data_manager/manager.py b/lib/galaxy/tools/data_manager/manager.py index a91512aadd6..622e7189d20 100644 --- a/lib/galaxy/tools/data_manager/manager.py +++ b/lib/galaxy/tools/data_manager/manager.py @@ -11,7 +11,8 @@ from typing import ( from typing_extensions import Protocol from galaxy import util -from galaxy.structured_app import MinimalManagerApp +from galaxy.structured_app import StructuredApp +from galaxy.tool_shed.galaxy_install.client import DataManagersInterface from galaxy.tool_util.data import ( BundleProcessingOptions, OutputDataset, @@ -21,10 +22,6 @@ from galaxy.tool_util.data.bundles.models import ( RepoInfo, ) from galaxy.util import Element -from galaxy.tool_shed.galaxy_install.client import ( - DataManagerInterface, - DataManagersInterface, -) log = logging.getLogger(__name__) @@ -34,7 +31,7 @@ class DataManagers(DataManagersInterface): managed_data_tables: Dict[str, "DataManager"] __reload_count: int - def __init__(self, app: MinimalManagerApp, xml_filename=None, reload_count: Optional[int] = None): + def __init__(self, app: StructuredApp, xml_filename=None, reload_count: Optional[int] = None): self.app = app self.data_managers = {} self.managed_data_tables = {} diff --git a/test/unit/app/test_galaxy_install.py b/test/unit/app/test_galaxy_install.py new file mode 100644 index 00000000000..f0898a79d34 --- /dev/null +++ b/test/unit/app/test_galaxy_install.py @@ -0,0 +1,60 @@ +"""Test installation using galaxy.tool_shed package. + +It should be able to quickly test installing things from the real tool shed +and from bootstrapped tool sheds. +""" +from pathlib import Path +from typing import ( + Any, + Dict, +) + +from galaxy.model.tool_shed_install import ToolShedRepository +from galaxy.tool_shed.galaxy_install.client import InstallationTarget +from galaxy.tool_shed.galaxy_install.install_manager import InstallRepositoryManager +from galaxy.tool_shed.galaxy_install.installed_repository_manager import InstalledRepositoryManager +from galaxy.tool_shed.unittest_utils import StandaloneInstallationTarget +from galaxy.tool_shed.util.repository_util import check_for_updates +from galaxy.util.tool_shed.tool_shed_registry import DEFAULT_TOOL_SHED_URL + + +def test_against_production_shed(tmp_path: Path): + repo_owner = "iuc" + repo_name = "collection_column_join" + repo_revision = "dfde09461b1e" + + install_target: InstallationTarget = StandaloneInstallationTarget(tmp_path) + install_manager = InstallRepositoryManager(install_target) + install_options: Dict[str, Any] = {} + install_manager.install( + DEFAULT_TOOL_SHED_URL, + repo_name, + repo_owner, + repo_revision, # revision 2, a known installable revision + install_options, + ) + with open(tmp_path / "shed_conf.xml") as f: + assert "toolshed.g2.bx.psu.edu/repos/iuc/collection_column_join/collection_column_join/0.0.2" in f.read() + repo_path = tmp_path / "tools" / "toolshed.g2.bx.psu.edu" / "repos" / repo_owner / repo_name / repo_revision + assert repo_path.exists() + + install_model_context = install_target.install_model.context + query = install_model_context.query(ToolShedRepository).where(ToolShedRepository.name == repo_name) + tsr = query.first() + assert tsr + message, status = check_for_updates( + install_target.tool_shed_registry, + install_model_context, + tsr.id, + ) + assert status + + irm = InstalledRepositoryManager(install_target) + errors = irm.uninstall_repository(repository=tsr, remove_from_disk=True) + assert not errors + + with open(tmp_path / "shed_conf.xml") as f: + assert "toolshed.g2.bx.psu.edu/repos/iuc/collection_column_join/collection_column_join/0.0.2" not in f.read() + + repo_path = tmp_path / "tools" / "toolshed.g2.bx.psu.edu" / "repos" / repo_owner / repo_name / repo_revision + assert not repo_path.exists()