Source code for galaxy_test.api.sharable

from typing import Any, Dict, List
from unittest import SkipTest

from galaxy_test.base.api import UsesApiTestCaseMixin


[docs]class SharingApiTests(UsesApiTestCaseMixin):
    """ Includes some tests for the sharing functionality of a particular resource type."""

    # The api_name has to be set to the appropriate value in the class using these tests.
    api_name: str

[docs]    def create(self, name: str) -> str:
        """Creates a shareable resource with the given name and returns it's ID.

        :param name: The name of the shareable resource to create.
        :return: The ID of the resource.
        """
        raise SkipTest("Abstract")

[docs]    def test_sharing_get_status(self):
        resource_id = self.create("resource-to-share")
        sharing_response = self._get_resource_sharing_status(resource_id)
        self._assert_has_keys(sharing_response, "title", "importable", "id", "username_and_slug", "published", "users_shared_with")

[docs]    def test_sharing_access(self):
        resource_id = self.create("resource-to-enable-link-access")

        sharing_response = self._get_resource_sharing_status(resource_id)
        assert sharing_response["importable"] is False

        sharing_response = self._set_resource_sharing(resource_id, "enable_link_access")
        assert sharing_response["importable"] is True

        sharing_response = self._set_resource_sharing(resource_id, "disable_link_access")
        assert sharing_response["importable"] is False

[docs]    def test_sharing_publish(self):
        resource_id = self.create("resource-to-publish")

        sharing_response = self._get_resource_sharing_status(resource_id)
        assert sharing_response["importable"] is False
        assert sharing_response["published"] is False

        sharing_response = self._set_resource_sharing(resource_id, "publish")
        assert sharing_response["importable"] is True
        assert sharing_response["published"] is True

        sharing_response = self._set_resource_sharing(resource_id, "unpublish")
        assert sharing_response["importable"] is True
        assert sharing_response["published"] is False

[docs]    def test_sharing_without_user(self):
        resource_id = self.create("resource-to-share-with-empty")

        sharing_response = self._get_resource_sharing_status(resource_id)
        assert not sharing_response["users_shared_with"]

        payload: Dict[str, List[str]] = {"user_ids": []}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert not sharing_response["users_shared_with"]

[docs]    def test_sharing_with_user_id(self):
        target_user = self._setup_user("target@user.com")
        target_user_id = target_user["id"]

        resource_id = self.create("resource-to-share-user-id")

        sharing_response = self._get_resource_sharing_status(resource_id)
        assert not sharing_response["users_shared_with"]

        payload = {"user_ids": [target_user_id]}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert sharing_response["users_shared_with"]
        assert sharing_response["users_shared_with"][0]["id"] == target_user_id

        payload = {"user_ids": []}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert not sharing_response["users_shared_with"]

[docs]    def test_sharing_with_user_email(self):
        target_user = self._setup_user("target@user.com")
        target_user_email = target_user["email"]

        resource_id = self.create("resource-to-share-user-email")

        sharing_response = self._get_resource_sharing_status(resource_id)
        assert not sharing_response["users_shared_with"]

        payload = {"user_ids": [target_user_email]}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert sharing_response["users_shared_with"]
        assert sharing_response["users_shared_with"][0]["email"] == target_user_email

        payload = {"user_ids": []}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert not sharing_response["users_shared_with"]

[docs]    def test_update_sharing_with_users(self):
        target_user_list = ["target01@user.com", "target02@user.com"]
        additional_user_list = ["add01@user.com", "add02@user.com"]
        all_user_emails = target_user_list + additional_user_list
        for email in all_user_emails:
            self._setup_user(email)

        resource_id = self.create("resource-to-share-and-update")

        payload = {"user_ids": all_user_emails}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert sharing_response["users_shared_with"]
        assert len(sharing_response["users_shared_with"]) == len(all_user_emails)

        # We just keep target users so additional users should be removed
        payload = {"user_ids": target_user_list}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert len(sharing_response["users_shared_with"]) == len(target_user_list)
        assert additional_user_list not in sharing_response["users_shared_with"]

[docs]    def test_sharing_with_invalid_user(self):
        invalid_user_email = "unknown@user.com"

        resource_id = self.create("resource-to-share-user-unknown")

        sharing_response = self._get_resource_sharing_status(resource_id)
        assert not sharing_response["users_shared_with"]

        payload = {"user_ids": [invalid_user_email]}
        sharing_response = self._set_resource_sharing(resource_id, action="share_with_users", payload=payload)
        assert not sharing_response["users_shared_with"]
        assert sharing_response["errors"]
        assert invalid_user_email in sharing_response["errors"][0]

[docs]    def test_set_slug(self):
        resource_id = self.create("resource-to-set-slug")
        other_resource_id = self.create("other-resource-to-set-slug")

        response = self._set_slug(resource_id, "new-slug")
        self._assert_status_code_is_ok(response)

        # Slugs must be unique for the same user/resource
        response = self._set_slug(other_resource_id, "new-slug")
        self._assert_status_code_is(response, 409)

        # Other users cannot change the slug if they don't own the resource
        with self._different_user():
            response = self._set_slug(resource_id, "another-slug")
            self._assert_status_code_is(response, 403)

    def _get_resource_sharing_status(self, resource_id: str):
        sharing_response = self._get(f"{self.api_name}/{resource_id}/sharing")
        self._assert_status_code_is(sharing_response, 200)
        return sharing_response.json()

    def _set_resource_sharing(self, resource_id: str, action: str, payload: Any = None, expect_response_status: int = 200):
        sharing_response = self._put(f"{self.api_name}/{resource_id}/{action}", data=payload, json=True)
        self._assert_status_code_is(sharing_response, expect_response_status)
        return sharing_response.json()

    def _set_slug(self, resource_id: str, new_slug: str):
        payload = {"new_slug": new_slug}
        response = self._put(f"{self.api_name}/{resource_id}/slug", data=payload, json=True)
        return response