Unverified Commit a066a423 authored by David López's avatar David López Committed by GitHub
Browse files

Merge pull request #14355 from nsoranzo/release_22.01_fix_14348 

[22.01] Workflow download API: check if a workflow is importable (not published)
parents ce0d3a75 2ff6aa51

lib/galaxy/managers/workflows.py

+10 −4
Original line number Diff line number Diff line
@@ -27,6 +27,7 @@ from galaxy import ( 
    util

)

from galaxy.job_execution.actions.post import ActionBox

from galaxy.model import StoredWorkflow

from galaxy.model.item_attrs import UsesAnnotations

from galaxy.structured_app import MinimalManagerApp

from galaxy.tools.parameters import (
@@ -74,7 +75,7 @@ class WorkflowsManager: 
    def __init__(self, app: MinimalManagerApp):

        self.app = app



    def get_stored_workflow(self, trans, workflow_id, by_stored_id=True):

    def get_stored_workflow(self, trans, workflow_id, by_stored_id=True) -> StoredWorkflow:

        """Use a supplied ID (UUID or encoded stored workflow ID) to find

        a workflow.

        """
@@ -115,8 +116,13 @@ class WorkflowsManager: 
        stored_workflow = self.get_stored_workflow(trans, workflow_id, by_stored_id=by_stored_id)



        # check to see if user has permissions to selected workflow

        if stored_workflow.user != trans.user and not trans.user_is_admin and not stored_workflow.published:

            if trans.sa_session.query(trans.app.model.StoredWorkflowUserShareAssociation).filter_by(user=trans.user, stored_workflow=stored_workflow).count() == 0:

        if stored_workflow.user != trans.user and not trans.user_is_admin and not stored_workflow.importable:

            if (

                trans.sa_session.query(trans.app.model.StoredWorkflowUserShareAssociation)

                .filter_by(user=trans.user, stored_workflow=stored_workflow)

                .count()

                == 0

            ):

                message = "Workflow is not owned by or shared with current user"

                raise exceptions.ItemAccessibilityException(message)

lib/galaxy/structured_app.py

+2 −1
Original line number Diff line number Diff line
@@ -28,6 +28,7 @@ from galaxy.workflow.trs_proxy import TrsProxy 


if TYPE_CHECKING:

    from galaxy.jobs import JobConfiguration

    from galaxy.managers.workflows import WorkflowsManager

    from galaxy.tools.data import ToolDataTableManager
@@ -123,7 +124,7 @@ class StructuredApp(MinimalManagerApp): 
    queue_worker: Any  # 'galaxy.queue_worker.GalaxyQueueWorker'

    history_manager: Any  # 'galaxy.managers.histories.HistoryManager'

    hda_manager: Any  # 'galaxy.managers.hdas.HDAManager'

    workflow_manager: Any  # 'galaxy.managers.workflows.WorkflowsManager'

    workflow_manager: "WorkflowsManager"

    workflow_contents_manager: Any  # 'galaxy.managers.workflows.WorkflowContentsManager'

    library_folder_manager: Any  # 'galaxy.managers.folders.FolderManager'

    library_manager: Any  # 'galaxy.managers.libraries.LibraryManager'

lib/galaxy_test/api/test_workflows.py

+7 −0
Original line number Diff line number Diff line
@@ -249,6 +249,13 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase, ChangeDatatypeTestCase): 
        workflows_url = self._api_url(f"workflows/{workflow_id}/download")

        assert get(workflows_url).status_code == 403



    def test_anon_can_download_importable_workflow(self):

        workflow_id = self.workflow_populator.simple_workflow("test_downloadable", importable=True)

        workflows_url = self._api_url(f"workflows/{workflow_id}/download")

        response = get(workflows_url)

        response.raise_for_status()

        assert response.json()["a_galaxy_workflow"] == "true"



    def test_anon_can_download_public_workflow(self):

        workflow_id = self.workflow_populator.simple_workflow("test_downloadable", publish=True)

        workflows_url = self._api_url(f"workflows/{workflow_id}/download")