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] 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 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
)