Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy_test.base.uses_shed
import os
import shutil
import string
import tempfile
from typing import ClassVar
import pytest
from galaxy.app import UniverseApplication
from galaxy.tool_util.verify.interactor import GalaxyInteractorApi
from galaxy.util import unicodify
from galaxy_test.base.api_asserts import assert_status_code_is
from galaxy_test.base.populators import DEFAULT_TIMEOUT
from galaxy_test.driver.driver_util import FRAMEWORK_UPLOAD_TOOL_CONF, GalaxyTestDriver
# Needs a longer timeout because of the conda_auto_install.
CONDA_AUTO_INSTALL_JOB_TIMEOUT = DEFAULT_TIMEOUT * 3
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
TOOL_SHEDS_CONF = os.path.join(SCRIPT_DIRECTORY, "tool_sheds_conf.xml")
SHED_TOOL_CONF = string.Template("""<?xml version="1.0"?>
<toolbox tool_path="$shed_tools_path">
</toolbox>""")
SHED_DATA_MANAGER_CONF = """<?xml version="1.0"?>
<data_managers>
</data_managers>"""
SHED_DATA_TABLES = """<?xml version="1.0"?>
<tables>
</tables>"""
[docs]class UsesShed:
@property
def _app(self) -> UniverseApplication:
...
shed_tools_dir: ClassVar[str]
shed_tool_data_dir: ClassVar[str]
conda_tmp_prefix: ClassVar[str]
galaxy_interactor: GalaxyInteractorApi
_test_driver: GalaxyTestDriver
[docs] @classmethod
def configure_shed(cls, config):
cls.shed_tools_dir = tempfile.mkdtemp()
cls.shed_tool_data_dir = tempfile.mkdtemp()
cls._test_driver.temp_directories.extend([cls.shed_tool_data_dir, cls.shed_tools_dir])
shed_tool_config = os.path.join(cls.shed_tools_dir, 'shed_tool_conf.xml')
config["tool_sheds_config_file"] = TOOL_SHEDS_CONF
config["tool_config_file"] = FRAMEWORK_UPLOAD_TOOL_CONF
config["shed_tool_config_file"] = shed_tool_config
config["shed_data_manager_config_file"] = os.path.join(cls.shed_tool_data_dir, 'shed_data_manager_config_file')
config["shed_tool_data_table_config"] = os.path.join(cls.shed_tool_data_dir, 'shed_data_table_conf.xml')
config["shed_tool_data_path"] = cls.shed_tool_data_dir
with open(shed_tool_config, 'w') as tool_conf_file:
tool_conf_file.write(SHED_TOOL_CONF.substitute(shed_tools_path=cls.shed_tools_dir))
with open(config["shed_data_manager_config_file"], 'w') as shed_data_config:
shed_data_config.write(SHED_DATA_MANAGER_CONF)
with open(config["shed_tool_data_table_config"], 'w') as shed_data_table_config:
shed_data_table_config.write(SHED_DATA_TABLES)
[docs] @classmethod
def configure_shed_and_conda(cls, config):
cls.configure_shed(config)
cls.conda_tmp_prefix = tempfile.mkdtemp()
cls._test_driver.temp_directories.append(cls.conda_tmp_prefix)
config["conda_auto_init"] = True
config["conda_auto_install"] = True
config["conda_prefix"] = os.environ.get('GALAXY_TEST_CONDA_PREFIX') or os.path.join(cls.conda_tmp_prefix, 'conda')
[docs] def setup_shed_config(self):
shutil.rmtree(self._app.config.shed_tools_dir, ignore_errors=True)
os.makedirs(self._app.config.shed_tools_dir)
self._app.config.shed_tools_dir = self.shed_tools_dir
with open(self._app.config.shed_tool_config_file, "w") as tool_conf_file:
tool_conf_file.write(SHED_TOOL_CONF.substitute(shed_tools_path=self._app.config.shed_tools_dir))
# deleting the containing folder doesn't trigger a toolbox reload, so signal it now and wait until it's done
self._app.queue_worker.send_control_task('reload_toolbox', get_response=True)
[docs] def reset_shed_tools(self):
self.setup_shed_config()
model = self._app.install_model
models_to_delete = [
model.RepositoryRepositoryDependencyAssociation,
model.RepositoryDependency,
model.ToolVersion,
model.ToolVersionAssociation,
model.ToolDependency,
model.ToolShedRepository,
]
for item in models_to_delete:
model.context.query(item).delete()
model.context.flush()
[docs] def delete_repo_request(self, payload):
return self.galaxy_interactor._delete('tool_shed_repositories', data=payload, admin=True)
[docs] def install_repo_request(self, payload):
return self.galaxy_interactor._post('tool_shed_repositories/new/install_repository_revision', data=payload, admin=True)
[docs] def repository_operation(self, operation, owner, name, changeset, tool_shed_url='https://toolshed.g2.bx.psu.edu'):
payload = {
'tool_shed_url': tool_shed_url,
'name': name,
'owner': owner,
'changeset_revision': changeset
}
create_response = operation(payload)
assert_status_code_is(create_response, 200)
return create_response.json()
[docs] def install_repository(self, owner, name, changeset, tool_shed_url='https://toolshed.g2.bx.psu.edu'):
try:
return self.repository_operation(operation=self.install_repo_request, owner=owner, name=name, changeset=changeset, tool_shed_url=tool_shed_url)
except AssertionError as e:
if "Error attempting to retrieve installation information from tool shed" in unicodify(e):
pytest.skip("Toolshed '%s' unavailable" % tool_shed_url)
raise
[docs] def uninstall_repository(self, owner, name, changeset, tool_shed_url='https://toolshed.g2.bx.psu.edu'):
return self.repository_operation(operation=self.delete_repo_request, owner=owner, name=name, changeset=changeset, tool_shed_url=tool_shed_url)