Unverified Commit 2ff6aa51 authored by Nicola Soranzo's avatar Nicola Soranzo
Browse files

Workflow download API: check if a workflow is importable (not published)

parent 70542612
Loading
Loading
Loading
Loading
+7 −2
Original line number Diff line number Diff line
@@ -116,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)

+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")