### Initiate Repository Installation Process Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/tool_shed/galaxy_install/install_manager Starts the installation process for tool shed repositories. This function orchestrates the handling of tool panel sections, including creating new sections or assigning tools to existing ones, based on the provided installation dictionary. ```python def initiate_repository_installation(self, installation_dict): # The following installation_dict entries are all required. created_or_updated_tool_shed_repositories = installation_dict["created_or_updated_tool_shed_repositories"] filtered_repo_info_dicts = installation_dict["filtered_repo_info_dicts"] has_repository_dependencies = installation_dict["has_repository_dependencies"] includes_tool_dependencies = installation_dict["includes_tool_dependencies"] includes_tools = installation_dict["includes_tools"] includes_tools_for_display_in_tool_panel = installation_dict["includes_tools_for_display_in_tool_panel"] install_repository_dependencies = installation_dict["install_repository_dependencies"] install_resolver_dependencies = installation_dict["install_resolver_dependencies"] install_tool_dependencies = installation_dict["install_tool_dependencies"] message = installation_dict["message"] new_tool_panel_section_label = installation_dict["new_tool_panel_section_label"] shed_tool_conf = installation_dict["shed_tool_conf"] status = installation_dict["status"] tool_panel_section_id = installation_dict["tool_panel_section_id"] tool_panel_section_keys = installation_dict["tool_panel_section_keys"] tool_panel_section_mapping = installation_dict.get("tool_panel_section_mapping", {}) tool_path = installation_dict["tool_path"] tool_shed_url = installation_dict["tool_shed_url"] # Handle contained tools. if includes_tools_for_display_in_tool_panel and (new_tool_panel_section_label or tool_panel_section_id): self.tpm.handle_tool_panel_section( self.app.toolbox, tool_panel_section_id=tool_panel_section_id, new_tool_panel_section_label=new_tool_panel_section_label, ) if includes_tools_for_display_in_tool_panel and (tool_panel_section_mapping is not None): for tool_guid in tool_panel_section_mapping: if tool_panel_section_mapping[tool_guid]["action"] == "create": new_tool_panel_section_name = tool_panel_section_mapping[tool_guid]["tool_panel_section"] log.debug(f'Creating tool panel section "{new_tool_panel_section_name}" for tool {tool_guid}') self.tpm.handle_tool_panel_section( ``` -------------------------------- ### Get All Installed Repositories Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Retrieves a list of all installed repositories from the installation client. This function requires an initialized installation client. ```python def get_all_installed_repositories(self) -> list[galaxy_model.ToolShedRepository]: assert self._installation_client return self._installation_client.get_all_installed_repositories() ``` -------------------------------- ### Test Tool Repository Installation with Dependencies (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/functional/test_1010_install_repository_with_tool_dependencies This Python test class, `TestToolWithToolDependencies`, verifies the process of installing a Galaxy tool repository that includes tool dependencies. It covers user authentication, repository creation and setup, browsing the tool shed, installing the repository (optionally with dependencies), and validating the installed tool and its metadata. ```python import logging from ..base import common from ..base.twilltestcase import ShedTwillTestCase repository_name = "freebayes_0010" repository_description = "Galaxy's freebayes tool" repository_long_description = "Long description of Galaxy's freebayes tool" category_name = "Test 0010 Repository With Tool Dependencies" log = logging.getLogger(__name__) class TestToolWithToolDependencies(ShedTwillTestCase): """Test installing a repository with tool dependencies.""" requires_galaxy = True def test_0000_initiate_users(self): """Create necessary user accounts.""" self.login(email=common.test_user_1_email, username=common.test_user_1_name) self.login(email=common.admin_email, username=common.admin_username) def test_0005_ensure_repositories_and_categories_exist(self): """Create the 0010 category and upload the freebayes repository to it, if necessary.""" category = self.create_category( name=category_name, description="Tests for a repository with tool dependencies." ) self.login(email=common.test_user_1_email, username=common.test_user_1_name) repository = self.get_or_create_repository( name=repository_name, description=repository_description, long_description=repository_long_description, owner=common.test_user_1_name, category=category, ) if self.repository_is_new(repository): self.setup_freebayes_0010_repo(repository) def test_0010_browse_tool_shed(self): """Browse the available tool sheds in this Galaxy instance and preview the freebayes tool.""" self.browse_tool_shed(url=self.url, strings_displayed=[category_name]) category = self.populator.get_category_with_name(category_name) self.browse_category(category, strings_displayed=[repository_name]) if not self.is_v2: strings_displayed = [repository_name, "Valid tools", "Tool dependencies"] self.preview_repository_in_tool_shed( repository_name, common.test_user_1_name, strings_displayed=strings_displayed ) def test_0015_install_freebayes_repository(self): """Install the freebayes repository without installing tool dependencies.""" self._install_repository( repository_name, common.test_user_1_name, category_name, install_tool_dependencies=False, new_tool_panel_section_label="test_1010", ) installed_repository = self._get_installed_repository_by_name_owner(repository_name, common.test_user_1_name) assert self._get_installed_repository_for( common.test_user_1, repository_name, installed_repository.installed_changeset_revision ) self._assert_has_valid_tool_with_name("FreeBayes") self._assert_repo_has_tool_with_id(installed_repository, "freebayes") def test_0020_verify_installed_repository_metadata(self): """Verify that resetting the metadata on an installed repository does not change the metadata.""" self.verify_installed_repository_metadata_unchanged(repository_name, common.test_user_1_name) def test_0025_verify_sample_files(self): """Verify that the installed repository populated shed_tool_data_table.xml and the sample files.""" self.verify_installed_repository_data_table_entries(required_data_table_entries=["sam_fa_indexes"]) ``` -------------------------------- ### Example Conda Resolver Configuration with Multiple Prefixes Source: https://docs.galaxyproject.org/en/latest/admin/dependency_resolvers Illustrates how to configure the Conda dependency resolver to use multiple installation prefixes. This example shows a read-only administrator-maintained installation followed by a Galaxy-maintained writable installation where missing dependencies are automatically installed. ```yaml - type: conda auto_init: false auto_install: false prefix: /hpc/conda - type: conda auto_init: true auto_install: true prefix: /galaxy/conda ``` -------------------------------- ### Setup Workflow Run with Various Inputs Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/base/populators Prepares all necessary components for running a Galaxy workflow, including creating datasets, histories, and formatting inputs based on the 'inputs_by' strategy. ```python def setup_workflow_run( self, workflow: Optional[dict[str, Any]] = None, inputs_by: str = "step_id", history_id: Optional[str] = None, workflow_id: Optional[str] = None, ) -> tuple[dict[str, Any], str, str]: ds_entry = self.dataset_populator.ds_entry if not workflow_id: assert workflow, "If workflow_id not specified, must specify a workflow dictionary to load" workflow_id = self.create_workflow(workflow) if not history_id: history_id = self.dataset_populator.new_history() hda1: Optional[dict[str, Any]] = None hda2: Optional[dict[str, Any]] = None label_map: Optional[dict[str, Any]] = None if inputs_by != "url": hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3", wait=True) hda2 = self.dataset_populator.new_dataset(history_id, content="4 5 6", wait=True) label_map = {"WorkflowInput1": ds_entry(hda1), "WorkflowInput2": ds_entry(hda2)} workflow_request = dict( history=f"hist_id={history_id}", ) if inputs_by == "step_id": assert label_map ds_map = self.build_ds_map(workflow_id, label_map) workflow_request["ds_map"] = ds_map elif inputs_by == "step_index": assert hda1 assert hda2 index_map = {"0": ds_entry(hda1), "1": ds_entry(hda2)} workflow_request["inputs"] = json.dumps(index_map) workflow_request["inputs_by"] = "step_index" elif inputs_by == "name": assert label_map workflow_request["inputs"] = json.dumps(label_map) workflow_request["inputs_by"] = "name" elif inputs_by in ["step_uuid", "uuid_implicitly"]: assert hda1 assert hda2 assert workflow, f"Must specify workflow for this inputs_by {inputs_by} parameter value" uuid_map = { workflow["steps"]["0"]["uuid"]: ds_entry(hda1), workflow["steps"]["1"]["uuid"]: ds_entry(hda2), } workflow_request["inputs"] = json.dumps(uuid_map) if inputs_by == "step_uuid": workflow_request["inputs_by"] = "step_uuid" elif inputs_by in ["url", "deferred_url"]: input_b64_1 = self.dataset_populator.base64_url_for_string("1 2 3") input_b64_2 = self.dataset_populator.base64_url_for_string("4 5 6") deferred = inputs_by == "deferred_url" inputs = { "WorkflowInput1": {"src": "url", "url": input_b64_1, "ext": "txt", "deferred": deferred}, "WorkflowInput2": {"src": "url", "url": input_b64_2, "ext": "txt", "deferred": deferred}, } workflow_request["inputs"] = json.dumps(inputs) ``` -------------------------------- ### Generate Tool GUID (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/tool_shed/util/shed_util_common Creates a unique identifier (GUID) for an installed tool. The generated GUID must match the one used in the Galaxy Tool Shed for proper identification during installation. ```python def generate_tool_guid(repository_clone_url, tool): """ Generate a guid for the installed tool. It is critical that this guid matches the guid for the tool in the Galaxy tool shed from which it is being installed. The form of the guid is /repos//// """ tmp_url = common_util.remove_protocol_and_user_from_clone_url(repository_clone_url) return f"{tmp_url}/{tool.id}/{tool.version}" ``` -------------------------------- ### Setup and Launch Galaxy Server Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/driver/driver_util Sets up a Galaxy server instance for functional testing. It first calls `_configure` to prepare the necessary settings and then `_register_and_run_servers` to instantiate and start the Galaxy server process. ```python def setup(self, config_object=None): """Setup a Galaxy server for functional test (if needed). Configuration options can be specified as attributes on the supplied ```config_object``` (defaults to self). """ self._saved_galaxy_config = None self._configure(config_object) self._register_and_run_servers(config_object) ``` -------------------------------- ### Initiate and Process Repository Installation Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/tool_shed/galaxy_install/install_manager Handles the core logic for initiating and processing the installation of repositories, including checking for and setting up various types of dependencies (repository, resolver, and tool dependencies). It validates the response from the tool shed for essential parameters. ```python def __initiate_and_install_repositories( self, tool_shed_url: str, repository_revision_dict: RepositoryMetadataInstallInfoDict, repo_info_dicts: list[ExtraRepoInfo], install_options: dict[str, Any], ): try: has_repository_dependencies = repository_revision_dict["has_repository_dependencies"] except KeyError: raise exceptions.InternalServerError( "Tool shed response missing required parameter 'has_repository_dependencies'." ) try: includes_tools = repository_revision_dict["includes_tools"] except KeyError: raise exceptions.InternalServerError("Tool shed response missing required parameter 'includes_tools'.") try: includes_tool_dependencies = repository_revision_dict["includes_tool_dependencies"] except KeyError: raise exceptions.InternalServerError( "Tool shed response missing required parameter 'includes_tool_dependencies'." ) try: includes_tools_for_display_in_tool_panel = repository_revision_dict[ "includes_tools_for_display_in_tool_panel" ] except KeyError: raise exceptions.InternalServerError( "Tool shed response missing required parameter 'includes_tools_for_display_in_tool_panel'." ) # Get the information about the Galaxy components (e.g., tool panel section, tool config file, etc) that will contain the repository information. install_repository_dependencies = install_options.get("install_repository_dependencies", False) install_resolver_dependencies = install_options.get("install_resolver_dependencies", False) install_tool_dependencies = install_options.get("install_tool_dependencies", False) ``` -------------------------------- ### Automatic Bulk Datatype Change in Galaxy Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/api/test_history_contents Initiates an automatic bulk datatype change for datasets within a Galaxy history. This example demonstrates the setup for changing the datatype of multiple datasets, starting with tabular data. ```python [docs] def test_bulk_datatype_change_auto(self): with self.dataset_populator.test_history() as history_id: tabular_contents = "1\t2\t3\na\tb\tc\n" dataset_ids = [ self.dataset_populator.new_dataset(history_id, content=tabular_contents)("id"), ``` -------------------------------- ### Setup Fetch to Folder Destination (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/base/populators Sets up necessary components for fetching data to a specific folder within a library. It creates a new history, a private library, and defines the destination configuration. Returns history ID, library object, and destination dictionary. ```python def setup_fetch_to_folder(self, test_name): history_id = self.dataset_populator.new_history() library = self.new_private_library(test_name) folder_id = library["root_folder_id"][1:] destination = {"type": "library_folder", "library_folder_id": folder_id} return history_id, library, destination ``` -------------------------------- ### Install Repository (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Initiates the installation of a new repository from the ToolShed into Galaxy. It takes repository details, dependency installation flags, and tool panel section information as input. The function constructs a payload and sends a POST request to the Galaxy API, then waits for the installation to complete. ```python def install_repository( self, name: str, owner: str, changeset_revision: str, install_tool_dependencies: bool, install_repository_dependencies: bool, new_tool_panel_section_label: Optional[str], ): payload = { "tool_shed_url": self.testcase.url, "name": name, "owner": owner, "changeset_revision": changeset_revision, "install_tool_dependencies": install_tool_dependencies, "install_repository_dependencies": install_repository_dependencies, "install_resolver_dependencies": False, } if new_tool_panel_section_label: payload["new_tool_panel_section_label"] = new_tool_panel_section_label create_response = self.testcase.galaxy_interactor._post( "tool_shed_repositories/new/install_repository_revision", data=payload, admin=True ) assert_status_code_is_ok(create_response) create_response_object = create_response.json() if isinstance(create_response_object, dict): assert "status" in create_response_object assert "ok" == create_response_object["status"] # repo already installed... return assert isinstance(create_response_object, list) repository_ids = [repo["id"] for repo in create_response.json()] log.debug(f"Waiting for the installation of repository IDs: {repository_ids}") self._wait_for_repository_installation(repository_ids) ``` -------------------------------- ### Get Repository Install Information Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/managers/repositories Retrieves installation information for a specific repository revision. This includes repository details and metadata required for installation. ```APIDOC ## GET /api/repositories/get_repository_revision_install_info ### Description Retrieves installation information for a specific repository and changeset revision. ### Method GET ### Endpoint /api/repositories/get_repository_revision_install_info ### Parameters #### Query Parameters - **name** (string) - Required - The name of the repository. - **owner** (string) - Required - The owner of the repository. - **changeset_revision** (string) - Required - The changeset revision of the repository. ### Request Example GET /api/repositories/get_repository_revision_install_info?name=myrepo&owner=myowner&changeset_revision=abc123xyz ### Response #### Success Response (200) - **repository_dict** (object) - Dictionary containing repository details. - **repository_metadata** (object) - Dictionary containing repository metadata. - **changeset_revision** (string) - The resolved changeset revision. #### Response Example { "repository_dict": { "id": "repo1_id", "name": "myrepo", "owner": "myowner", "url": "/api/repositories/repo1_id" }, "repository_metadata": { "tool_dependencies": {}, "tools": [] }, "changeset_revision": "abc123xyz" } ``` -------------------------------- ### Download History to Store (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/base/populators Prepares a history for download to a store. It sends a POST request to 'histories/{history_id}/prepare_store_download'. After asserting the download request is OK and waiting for it to be ready, it either returns a temporary file path if `serve_file` is True, or the storage request ID. Dependencies: `_post`, `assert_download_request_ok`, `wait_for_download_ready`, `_get_to_tempfile`. ```python def download_history_to_store(self, history_id: str, extension: str = "tgz", serve_file: bool = False): url = f"histories/{history_id}/prepare_store_download" download_response = self._post(url, dict(include_files=False, model_store_format=extension), json=True) storage_request_id = self.assert_download_request_ok(download_response) self.wait_for_download_ready(storage_request_id) if serve_file: return self._get_to_tempfile(f"short_term_storage/{storage_request_id}") else: return storage_request_id ``` -------------------------------- ### Get Tool Dependencies Being Installed (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install Filters and returns a list of tool dependencies that are currently in the 'INSTALLING' status. ```python @property def tool_dependencies_being_installed(self): dependencies_being_installed = [] for tool_dependency in self.tool_dependencies: if tool_dependency.status == ToolDependency.installation_status.INSTALLING: dependencies_being_installed.append(tool_dependency) return dependencies_being_installed ``` -------------------------------- ### Handle Instance Access, Resources, Help, Support, and Citation Links (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/managers/markdown_util These snippets handle the generation of various instance-specific links, including access, resources, help, support, and citation links. They retrieve the corresponding URLs from `trans.app.config`. Each function takes the `line` and the retrieved URL as input. ```python if container == "instance_access_link": url = trans.app.config.instance_access_url rval = self.handle_instance_access_link(line, url) ``` ```python if container == "instance_resources_link": url = trans.app.config.instance_resource_url rval = self.handle_instance_resources_link(line, url) ``` ```python if container == "instance_help_link": url = trans.app.config.helpsite_url rval = self.handle_instance_help_link(line, url) ``` ```python if container == "instance_support_link": url = trans.app.config.support_url rval = self.handle_instance_support_link(line, url) ``` ```python if container == "instance_citation_link": url = trans.app.config.citation_url rval = self.handle_instance_citation_link(line, url) ``` -------------------------------- ### Perform Upload and Rule Configuration in Galaxy Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/selenium/test_uploads Demonstrates the setup process for a Uniprot example within the Galaxy rule builder. It involves performing an upload, waiting for history panel updates, starting rule uploads, and configuring data types and datasets. ```python def _setup_uniprot_example(self): self.perform_upload(self.get_filename("rules/uniprot.tsv")) self.history_panel_wait_for_hid_ok(1) self.upload_rule_start() self.upload_rule_set_data_type("Collections") self.upload_rule_dataset_dialog() self.upload_rule_set_dataset(1) ``` -------------------------------- ### Get Installed Repository Dependencies (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install This method retrieves a list of repository dependencies that are currently marked as 'INSTALLED'. It iterates through all repository dependencies and filters them based on their installation status. ```Python @property def installed_repository_dependencies(self): """Return the repository's repository dependencies that are currently installed.""" installed_required_repositories = [] for required_repository in self.repository_dependencies: if required_repository.status == self.installation_status.INSTALLED: installed_required_repositories.append(required_repository) return installed_required_repositories ``` -------------------------------- ### Start Webless Handler Processes Source: https://docs.galaxyproject.org/en/latest/_sources/admin/scaling These console commands demonstrate how to start webless handler processes manually using the galaxy-main script. They include options for specifying the server name, attaching to specific pools, and daemonizing the process. ```bash $ cd /srv/galaxy/server ./scripts/galaxy-main -c config/galaxy.yml --server-name handler_0 --attach-to-pool job-handlers --attach-to-pool workflow-scheduler --daemonize ./scripts/galaxy-main -c config/galaxy.yml --server-name handler_1 --attach-to-pool job-handlers --attach-to-pool workflow-scheduler --daemonize ./scripts/galaxy-main -c config/galaxy.yml --server-name handler_3 --attach-to-pool job-handlers --attach-to-pool workflow-scheduler --daemonize ./scripts/galaxy-main -c config/galaxy.yml --server-name special_0 --attach-to-pool job-handlers.special --daemonize ``` -------------------------------- ### Get Tool Dependencies Missing or Being Installed (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install Identifies and returns tool dependencies that are either missing, currently being installed, in an error state, or have never been installed. This provides a comprehensive view of dependencies needing installation or recovery. ```python @property def tool_dependencies_missing_or_being_installed(self): dependencies_missing_or_being_installed = [] for tool_dependency in self.tool_dependencies: if tool_dependency.status in [ ToolDependency.installation_status.ERROR, ToolDependency.installation_status.INSTALLING, ToolDependency.installation_status.NEVER_INSTALLED, ToolDependency.installation_status.UNINSTALLED, ]: dependencies_missing_or_being_installed.append(tool_dependency) return dependencies_missing_or_being_installed ``` -------------------------------- ### Get Installable Revisions Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/webapp/api/repositories Retrieves a list of installable revisions for a given repository. ```APIDOC ## GET /api/repositories/get_installable_revisions ### Description Retrieves a list of installable revisions for a given repository identified by its toolshed ID. ### Method GET ### Endpoint /api/repositories/get_installable_revisions ### Parameters #### Query Parameters - **tsr_id** (string) - Required - The encoded toolshed ID of the repository. ### Response #### Success Response (200) - **revisions** (list of lists) - A list of lists, where each inner list contains revision information. For example: `[ [ 0, "fbb391dc803c" ], [ 1, "9d9ec4d9c03e" ] ]`. #### Response Example ```json [ [ 0, "fbb391dc803c" ], [ 1, "9d9ec4d9c03e" ], [ 2, "9b5b20673b89" ], [ 3, "e8c99ce51292" ] ] ``` #### Error Response (Empty List) If the `tsr_id` is missing or invalid, an empty list is returned. ```json [] ``` ``` -------------------------------- ### Set Up Python Virtual Environment and Run Galaxy Source: https://docs.galaxyproject.org/en/latest/_sources/admin/production This snippet demonstrates how to install virtualenv, create a new virtual environment for Galaxy, activate it, navigate to the Galaxy distribution directory, and start the Galaxy server using the provided run script. This ensures Galaxy runs with isolated dependencies. ```shell nate@weyerbacher% pip install virtualenv nate@weyerbacher% virtualenv --no-site-packages galaxy_env nate@weyerbacher% . ./galaxy_env/bin/activate nate@weyerbacher% cd galaxy-dist nate@weyerbacher% sh run.sh ``` -------------------------------- ### Install and Start Munge for Slurm Authentication Source: https://docs.galaxyproject.org/en/latest/_sources/dev/debugging_galaxy_slurm Installs and starts the Munge service, which is used for authentication between Slurm components. It also ensures that the munge key is properly generated and has restricted permissions, owned by the munge user and group. ```bash sudo apt install munge sudo service munge start sudo ls -l /etc/munge/munge.key ``` -------------------------------- ### Create Bismark Repository and Setup (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/functional/test_0050_circular_dependencies_4_levels This function initializes a 'bismark' repository by creating a category, retrieving or creating the repository, and then setting up the repository content using a user populator. It's used for Bismark tool setup. ```python def test_0035_create_bismark_repository(self): """Create and populate bismark_0050.""" category = self.create_category(name=category_name, description=category_description) repository = self.get_or_create_repository( name=bismark_repository_name, description=bismark_repository_description, long_description=bismark_repository_long_description, owner=common.test_user_1_name, category=category, strings_displayed=[], ) self.user_populator().setup_bismark_repo(repository, end=1) ``` -------------------------------- ### Configure and Start Application Stack and Services in Python Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/app This snippet demonstrates the initialization and registration of various application components and services within the Galaxy framework. It includes setting up tasks, managers, signal handlers, and message queues, ensuring a robust application startup process. Dependencies include configuration objects, application stack utilities, and various Galaxy model and service classes. ```python self.config.fixed_delegated_auth = \ (len(list(self.config.oidc)) == 1 and len(list(self.auth_manager.authenticators)) == 0) if not self.config.enable_celery_tasks and self.config.history_audit_table_prune_interval > 0: self.prune_history_audit_task = IntervalTask( func=lambda: galaxy.model.HistoryAudit.prune(self.model.session), name="HistoryAuditTablePruneTask", interval=self.config.history_audit_table_prune_interval, immediate_start=False, time_execution=True, ) self.application_stack.register_postfork_function(self.prune_history_audit_task.start) self.haltables.append(("HistoryAuditTablePruneTask", self.prune_history_audit_task.shutdown)) self.proxy_manager = ProxyManager(self.config) self.workflow_scheduling_manager = scheduling_manager.WorkflowSchedulingManager(self) self.application_stack.register_postfork_function(self.job_manager.start) self.application_stack.init_late_prefork() handlers = {} if self.heartbeat: handlers[signal.SIGUSR1] = self.heartbeat.dump_signal_handler self._configure_signal_handlers(handlers) self.database_heartbeat = DatabaseHeartbeat(application_stack=self.application_stack) self.database_heartbeat.add_change_callback(self.watchers.change_state) self.application_stack.register_postfork_function(self.database_heartbeat.start) self.application_stack.register_postfork_function(self.application_stack.start) self.application_stack.register_postfork_function(self.queue_worker.bind_and_start) self.application_stack.register_postfork_function( lambda: reload_toolbox(self, save_integrated_tool_panel=False), post_fork_only=True ) self.application_stack.register_postfork_function( lambda: send_local_control_task(self, "rebuild_toolbox_search_index") ) self.url_for = url_for self.server_starttime = server_starttime # used for cachebusting self.tool_shed_repository_cache = None self.api_spec = None self.legacy_mapper = None self.application_stack.register_postfork_function(self.object_store.start) log.info(f"Galaxy app startup finished {startup_timer}") ``` -------------------------------- ### Get Installed Tool Dependencies (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install This method returns a list of tool dependencies that are currently installed, including those that might be in an error state. It filters the repository's tool dependencies, selecting only those with the status 'INSTALLED'. ```Python @property def installed_tool_dependencies(self): """Return the repository's tool dependencies that are currently installed, but possibly in an error state.""" installed_dependencies = [] for tool_dependency in self.tool_dependencies: if tool_dependency.status in [ToolDependency.installation_status.INSTALLED]: installed_dependencies.append(tool_dependency) return installed_dependencies ``` -------------------------------- ### Get All Installed Repositories (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Retrieves a list of all installed tool shed repositories from the Galaxy installation. It ensures that each repository object is refreshed in the context before returning the list. ```python repositories = test_db_util.get_all_installed_repositories( session=self._installation_target.install_model.context ) for repository in repositories: self._installation_target.install_model.context.refresh(repository) return repositories ``` -------------------------------- ### Create and Setup Bismark Repository Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/functional/test_0300_reset_all_metadata Creates a category, logs in an admin user, then a test user, and gets or creates a repository for the Bismark tool. It then uses a `user_populator` to set up the repository. This function is conditional on `running_standalone`. ```python if running_standalone: self.login(email=common.admin_email, username=common.admin_username) category = self.create_category(name=category_0050_name, description=category_0050_description) self.login(email=common.test_user_1_email, username=common.test_user_1_name) repository = self.get_or_create_repository( name=bismark_repository_name, description=bismark_repository_description, long_description=bismark_repository_long_description, owner=common.test_user_1_name, category=category, strings_displayed=[], ) self.user_populator().setup_bismark_repo(repository, end=1) ``` -------------------------------- ### Browse Tool Shed and Preview Repository Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/functional/test_1020_install_repository_with_repository_dependencies This test function browses the Galaxy Tool Shed, specifically looking for the 'Test 0020 Basic Repository Dependencies' category and previewing the 'emboss' repository. It checks for expected display strings. ```Python def test_0010_browse_tool_shed(self): """Browse the available tool sheds in this Galaxy instance and preview the emboss tool.""" self.browse_tool_shed(url=self.url, strings_displayed=["Test 0020 Basic Repository Dependencies"]) category = self.populator.get_category_with_name("Test 0020 Basic Repository Dependencies") self.browse_category(category, strings_displayed=[emboss_repository_name]) if not self.is_v2: self.preview_repository_in_tool_shed( emboss_repository_name, common.test_user_1_name, strings_displayed=[emboss_repository_name, "Valid tools"], ) ``` -------------------------------- ### Get Conda Target Installed Path Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/tool_util/deps/conda_util Retrieves the installation path for a given Conda target, checking both the regular and capitalized versions of the install environment name. Returns the path if found, otherwise None. ```python def get_conda_target_installed_path(self, conda_target: "CondaTarget") -> Optional[str]: for env_name in (conda_target.install_environment, conda_target.capitalized_install_environment): if self.has_env(env_name): return self.env_path(env_name) return None ``` -------------------------------- ### Get Tool Dependencies With Installation Errors (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install Filters and returns a list of tool dependencies that are specifically in the 'ERROR' status, indicating a failed installation attempt. ```python @property def tool_dependencies_with_installation_errors(self): dependencies_with_installation_errors = [] for tool_dependency in self.tool_dependencies: if tool_dependency.status == ToolDependency.installation_status.ERROR: dependencies_with_installation_errors.append(tool_dependency) ``` -------------------------------- ### Initialize StandaloneToolShedInstallationClient (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Initializes the StandaloneToolShedInstallationClient, which manages the installation of tools from a Galaxy Tool Shed. It sets up a temporary directory for installations and configures the tool shed target URL. This constructor is essential for preparing the client for subsequent installation operations. ```python def __init__(self, testcase: "ShedTwillTestCase"): self.testcase = testcase self.temp_directory = Path(tempfile.mkdtemp(prefix="toolshedtestinstalltarget")) tool_shed_target = ToolShedTarget( self.testcase.url, "Tool Shed for Testing", ) self._installation_target = StandaloneInstallationTarget(self.temp_directory, tool_shed_target=tool_shed_target) ``` -------------------------------- ### CommunityWebApplication Setup Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/webapp/buildapp Defines a web application class for the Tool Shed, enabling injection awareness and customizing the transaction chooser. It integrates with the base Galaxy web application framework. ```python class CommunityWebApplication(galaxy.webapps.base.webapp.WebApplication): injection_aware: bool = True def transaction_chooser(self, environ, galaxy_app: BasicSharedApp, session_cookie: str): return ToolShedGalaxyWebTransaction(environ, galaxy_app, self, session_cookie) ``` -------------------------------- ### Get Tool Shed Status for Installed Repository Source: https://docs.galaxyproject.org/en/latest/lib/galaxy.tool_shed Fetches the latest information from the Tool Shed for an installed repository, including update status and deprecation. ```APIDOC ## GET /api/tool_shed/repository/status ### Description Send a request to the tool shed to retrieve information about newer installable repository revisions, current revision updates, whether the repository revision is the latest downloadable revision, and whether the repository has been deprecated in the tool shed. The received repository is a ToolShedRepository object from Galaxy. ### Method GET ### Endpoint `/api/tool_shed/repository/status` ### Parameters #### Query Parameters - **repository** (object) - Required - The Tool Shed repository object from Galaxy. - **id** (integer) - The repository's unique ID. - **name** (string) - The name of the repository. - **owner** (string) - The owner of the repository. - **tool_shed** (string) - The URL of the Tool Shed. - **changeset_revision** (string) - The changeset revision of the repository. ### Response #### Success Response (200) - **status_info** (object) - Information about the repository's status in the Tool Shed. - **has_update** (boolean) - Indicates if an update is available. - **is_latest_revision** (boolean) - Indicates if the current revision is the latest. - **is_deprecated** (boolean) - Indicates if the repository is deprecated. - **message** (string) - A message providing details about the status. #### Response Example ```json { "status_info": { "has_update": true, "is_latest_revision": false, "is_deprecated": false, "message": "A newer revision is available." } } ``` ``` -------------------------------- ### Get Installed Repository by Name and Owner in Galaxy Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Retrieves a specific installed repository from the Galaxy Tool Shed using its name and owner. Requires an initialized installation client. ```python def _get_installed_repository_by_name_owner( self, repository_name: str, repository_owner: str, ) -> galaxy_model.ToolShedRepository: assert self._installation_client return self._installation_client.get_installed_repository_by_name_owner(repository_name, repository_owner) ``` -------------------------------- ### Download Contents to Store Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/base/populators Initiates and prepares for downloading history content to a store. It constructs a URL to request download preparation, makes a POST request with download parameters, asserts the request is okay, waits for the download to be ready, and returns the path to the downloaded content in a temporary file. ```Python def download_contents_to_store(self, history_id: str, history_content: dict[str, Any], extension=".tgz") -> str: url = f"histories/{history_id}/contents/{history_content['history_content_type']}s/{history_content['id']}/prepare_store_download" download_response = self._post(url, dict(include_files=False, model_store_format=extension), json=True) storage_request_id = self.assert_download_request_ok(download_response) self.wait_for_download_ready(storage_request_id) return self._get_to_tempfile(f"short_term_storage/{storage_request_id}") ``` -------------------------------- ### Register and Run Galaxy Servers Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/driver/driver_util This internal method handles the instantiation and launching of Galaxy server processes. It manages temporary directory creation for the database and configuration, and uses provided configuration objects or defaults to set up the Galaxy environment. ```python def _register_and_run_servers(self, config_object=None, handle_config=None) -> None: config_object = self._ensure_config_object(config_object) self.app: Optional[GalaxyUniverseApplication] = None if self.external_galaxy is None: if self._saved_galaxy_config is not None: galaxy_config = self._saved_galaxy_config else: tempdir = tempfile.mkdtemp(dir=self.galaxy_test_tmp_dir) # Configure the database path. galaxy_db_path = database_files_path(tempdir) # Allow config object to specify a config dict or a method to produce # one - other just read the properties above and use the default # implementation from this file. galaxy_config = getattr(config_object, "galaxy_config", None) if callable(galaxy_config): galaxy_config = galaxy_config() if galaxy_config is None: galaxy_config = setup_galaxy_config( galaxy_db_path, use_test_file_dir=not self.testing_shed_tools, default_install_db_merged=True, default_tool_conf=self.default_tool_conf, ``` -------------------------------- ### Get Missing Repository Dependencies (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install This method retrieves a list of repository dependencies that are not currently installed. It iterates through all repository dependencies and includes those whose status is not 'INSTALLED'. ```Python @property def missing_repository_dependencies(self): """Return the repository's repository dependencies that are not currently installed, and may not ever have been installed.""" missing_required_repositories = [] for required_repository in self.repository_dependencies: if required_repository.status not in [self.installation_status.INSTALLED]: missing_required_repositories.append(required_repository) return missing_required_repositories ``` -------------------------------- ### Python: Get Message for No Shed Tool Config Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/webapps/galaxy/api/tool_shed_repositories Generates a user-friendly message explaining the requirement for a shed-related tool panel configuration file in Galaxy's configuration. This message guides users on how to set up the necessary 'tool_config_file' in 'galaxy.ini' for automatic tool installation from a tool shed. ```python def get_message_for_no_shed_tool_config(): # This Galaxy instance is not configured with a shed-related tool panel configuration file. message = "The tool_config_file setting in galaxy.ini must include at least one shed tool configuration file name with a " message += "tag that includes a tool_path attribute value which is a directory relative to the Galaxy installation directory in order to " message += "automatically install tools from a tool shed into Galaxy (e.g., the file name shed_tool_conf.xml whose tag is " message += '). For details, see the "Installation of Galaxy tool shed repository tools into a ' message += 'local Galaxy instance" section of the Galaxy tool shed wiki at https://galaxyproject.org/installing-repositories-to-galaxy/' return message ``` -------------------------------- ### DescribeToolExecution: Initialize and Configure Tool Execution Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy_test/base/populators Initializes DescribeToolExecution with a dataset populator and tool ID. It allows setting history IDs, inputs (in various formats), and preparing for tool execution. ```python class DescribeToolExecution: _history_id: Optional[str] = None _execute_response: Optional[Response] = None _input_format: Optional[INPUT_FORMAT_T] = None _inputs: dict[str, Any] _tool_request_id: Optional[str] = None def __init__(self, dataset_populator: BaseDatasetPopulator, tool_id: str): self._dataset_populator = dataset_populator self._tool_id = tool_id self._inputs = {} def in_history(self, has_history_id: Union[str, "TargetHistory"]) -> Self: if isinstance(has_history_id, str): self._history_id = has_history_id else: self._history_id = has_history_id._history_id return self def with_inputs(self, inputs: Union[DescribeToolInputs, dict[str, Any]]) -> Self: if isinstance(inputs, DescribeToolInputs): self._inputs = inputs._inputs or {} self._input_format = inputs._input_format else: self._inputs = inputs self._input_format = "legacy" return self def with_nested_inputs(self, inputs: dict[str, Any]) -> Self: self._inputs = inputs self._input_format = "21.01" return self def with_request(self, inputs: dict[str, Any]) -> Self: self._inputs = inputs self._input_format = "request" return self ``` -------------------------------- ### Initialize Users and Login Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/functional/test_0430_browse_utilities Sets up necessary user accounts and logs in an admin user for the test suite. This ensures that the test can run independently by creating or confirming the existence of required user credentials. ```Python import logging from ..base import common from ..base.api import skip_if_api_v2 from ..base.twilltestcase import ShedTwillTestCase log = logging.getLogger(__name__) emboss_repository_name = "emboss_0430" emboss_repository_description = "EMBOSS tools for test 0430" emboss_repository_long_description = "Long description of EMBOSS tools for test 0430" freebayes_repository_name = "freebayes_0430" freebayes_repository_description = "Freebayes tool for test 0430" freebayes_repository_long_description = "Long description of Freebayes tool for test 0430" """ 1. Create and populate repositories. 2. Browse Custom Datatypes. 3. Browse Tools. 4. Browse Repository Dependencies. 5. Browse Tool Dependencies. """ [docs] class TestToolShedBrowseUtilities(ShedTwillTestCase): """Test browsing for Galaxy utilities.""" [docs] def test_0000_initiate_users(self): """Create necessary user accounts and login as an admin user. Create all the user accounts that are needed for this test script to run independently of other tests. Previously created accounts will not be re-created. """ self.login(email=common.test_user_1_email, username=common.test_user_1_name) self.login(email=common.admin_email, username=common.admin_username) ``` -------------------------------- ### Get Installed Repositories by Name and Owner in Galaxy Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Retrieves a list of installed repositories matching the given name and owner from the Galaxy Tool Shed. Requires an initialized installation client. ```python def _get_installed_repositories_by_name_owner( self, repository_name: str, repository_owner: str, ) -> list[galaxy_model.ToolShedRepository]: assert self._installation_client return self._installation_client.get_installed_repositories_by_name_owner(repository_name, repository_owner) ``` -------------------------------- ### Install Repository from Tool Shed Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/tool_shed/galaxy_install/install_manager Initiates the installation of a repository from a specified tool shed. This function retrieves necessary information about the repository, checks if it's already installed, and if not, proceeds with the installation process. It handles various installation options and dependencies. ```python def install( self, tool_shed_url: str, name: str, owner: str, changeset_revision: str, install_options: dict[str, Any] ): # Get all of the information necessary for installing the repository from the specified tool shed. repository_revision_dict, repo_info_dicts = self.__get_install_info_from_tool_shed( tool_shed_url, name, owner, changeset_revision ) if changeset_revision != repository_revision_dict["changeset_revision"]: # Demanded installation of a non-installable revision. Stop here if repository already installed. repo = repository_util.get_installed_repository( app=self.app, tool_shed=tool_shed_url, name=name, owner=owner, changeset_revision=repository_revision_dict["changeset_revision"], ) if repo and repo.is_installed: # Repo installed. Returning empty list indicated repo already installed. return [] installed_tool_shed_repositories = self.__initiate_and_install_repositories( tool_shed_url, repository_revision_dict, repo_info_dicts, install_options ) return installed_tool_shed_repositories ``` -------------------------------- ### Get Repository Dependencies Being Installed (Python) Source: https://docs.galaxyproject.org/en/latest/_modules/galaxy/model/tool_shed_install Filters the repository's dependencies to return only those currently undergoing installation. It checks the status of each repository dependency against a list of installation-related statuses. ```python @property def repository_dependencies_being_installed(self): """Return the repository's repository dependencies that are currently being installed.""" required_repositories_being_installed = [] for required_repository in self.repository_dependencies: if required_repository.status in [ self.installation_status.CLONING, self.installation_status.INSTALLING_REPOSITORY_DEPENDENCIES, self.installation_status.INSTALLING_TOOL_DEPENDENCIES, self.installation_status.LOADING_PROPRIETARY_DATATYPES, self.installation_status.SETTING_TOOL_VERSIONS, ]: required_repositories_being_installed.append(required_repository) return required_repositories_being_installed ``` -------------------------------- ### Get Installed Repository for Galaxy Tool Shed Source: https://docs.galaxyproject.org/en/latest/_modules/tool_shed/test/base/twilltestcase Fetches an installed repository from the Galaxy Tool Shed based on optional owner, name, or changeset revision. Requires an initialized installation client. ```python def _get_installed_repository_for( self, owner: Optional[str] = None, name: Optional[str] = None, changeset: Optional[str] = None, ): assert self._installation_client return self._installation_client.get_installed_repository_for(owner=owner, name=name, changeset=changeset) ```