diff --git a/lib/galaxy/managers/tools.py b/lib/galaxy/managers/tools.py index 1fc27ded3739..d0bde86a4770 100644 --- a/lib/galaxy/managers/tools.py +++ b/lib/galaxy/managers/tools.py @@ -154,9 +154,13 @@ def create_unprivileged_tool(self, user: model.User, tool_payload: DynamicUnpriv tool_version=tool_payload.representation.version, active=tool_payload.active, hidden=tool_payload.hidden, - value=tool_payload.representation, + value=tool_payload.representation.model_dump(), public=False, + flush=True, ) + session = self.session() + session.add(UserDynamicToolAssociation(user_id=user.id, dynamic_tool_id=dynamic_tool.id)) + session.commit() return dynamic_tool def list_tools(self, active=True): diff --git a/lib/galaxy_test/api/test_unprivileged_tools.py b/lib/galaxy_test/api/test_unprivileged_tools.py new file mode 100644 index 000000000000..75a2cc6f3897 --- /dev/null +++ b/lib/galaxy_test/api/test_unprivileged_tools.py @@ -0,0 +1,74 @@ +# Test tools API. +import contextlib +import json +import os +import zipfile +from io import BytesIO +from typing import ( + Any, + Dict, + List, + Optional, +) +from uuid import uuid4 + +import pytest +from requests import ( + get, + put, +) + +from galaxy.tool_util.verify.interactor import ValidToolTestDict +from galaxy.util import galaxy_root_path +from galaxy.util.unittest_utils import skip_if_github_down +from galaxy.schema.tools import UserToolSource +from galaxy_test.base import rules_test_data +from galaxy_test.base.api_asserts import ( + assert_has_keys, + assert_status_code_is, +) +from galaxy_test.base.decorators import requires_new_history +from galaxy_test.base.populators import ( + BaseDatasetCollectionPopulator, + DatasetCollectionPopulator, + DatasetPopulator, + skip_without_tool, + stage_rules_example, +) +from ._framework import ApiTestCase + +from .test_tools import TOOL_WITH_SHELL_COMMAND + + +class TestUnprivilegedToolsApi(ApiTestCase): + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor) + + def test_create_unprivileged(self): + response = self.dataset_populator.create_unprivileged_tool(UserToolSource(**TOOL_WITH_SHELL_COMMAND)) + assert response.status_code == 200, response.text + dynamic_tool = response.json() + assert dynamic_tool["uuid"] + + def test_list_unprivileged(self): + response = self.dataset_populator.create_unprivileged_tool(UserToolSource(**TOOL_WITH_SHELL_COMMAND)) + assert response.status_code == 200, response.text + response = self.dataset_populator.get_unprivileged_tools() + assert response.status_code == 200, response.text + assert response.json() + + def test_show(self): + response = self.dataset_populator.create_unprivileged_tool(UserToolSource(**TOOL_WITH_SHELL_COMMAND)) + assert response.status_code == 200, response.text + response = self.dataset_populator.show_unprivileged_tool(TOOL_WITH_SHELL_COMMAND["id"]) + assert response.status_code == 200, response.text + assert response.json() + + def test_deactivate(self): + pass + + def test_run(self): + pass diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index 3e4f7a989d5f..fa765e8e23d3 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -93,6 +93,10 @@ ToolLandingRequest, WorkflowLandingRequest, ) +from galaxy.schema.tools import ( + DynamicUnprivilegedToolCreatePayload, + UserToolSource, +) from galaxy.tool_util.client.staging import InteractorStaging from galaxy.tool_util.cwl.util import ( download_output, @@ -837,6 +841,18 @@ def create_tool_from_path(self, tool_path: str) -> Dict[str, Any]: ) return self._create_tool_raw(payload) + def create_unprivileged_tool(self, representation: UserToolSource, active=True, hidden=False, uuid=None): + data = DynamicUnprivilegedToolCreatePayload( + active=active, hidden=hidden, uuid=uuid, src="representation", representation=representation + ).model_dump(by_alias=True, exclude_unset=True) + return self._post("unprivileged_tools", data=data, json=True) + + def get_unprivileged_tools(self, active=True): + return self._get("unprivileged_tools", data={"active": active}) + + def show_unprivileged_tool(self, tool_id: str): + return self._get(f"unprivileged_tools/{tool_id}") + def create_tool(self, representation, tool_directory: Optional[str] = None) -> Dict[str, Any]: payload = dict( representation=representation,