Source code for galaxy_test.api.test_roles

from typing import (
    Any,
)

from galaxy.exceptions import error_codes
from galaxy_test.base.api_asserts import (
    assert_error_code_is,
    assert_has_keys,
    assert_status_code_is,
)
from galaxy_test.base.decorators import requires_admin
from galaxy_test.base.populators import DatasetPopulator
from ._framework import ApiTestCase



[docs]
class TestRolesApi(ApiTestCase):
    dataset_populator: DatasetPopulator


[docs]
    def setUp(self):
        super().setUp()
        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)



[docs]
    @requires_admin
    def test_list_and_show(self):
        def check_roles_response(response):
            assert_status_code_is(response, 200)
            as_list = response.json()
            assert isinstance(as_list, list)
            assert len(as_list) > 0
            for role in as_list:
                self.check_role_dict(role)

        user_role_id = self.dataset_populator.user_private_role_id()
        with self._different_user():
            different_user_role_id = self.dataset_populator.user_private_role_id()

        admin_roles_response = self._get("roles", admin=True)
        user_roles_response = self._get("roles")

        check_roles_response(admin_roles_response)
        check_roles_response(user_roles_response)

        admin_roles_response_ids = [r["id"] for r in admin_roles_response.json()]
        user_roles_response_ids = [r["id"] for r in user_roles_response.json()]

        # User can see their own private role not the other users, admin can see both.
        assert user_role_id in user_roles_response_ids
        assert different_user_role_id not in user_roles_response_ids

        assert user_role_id in admin_roles_response_ids
        assert different_user_role_id in admin_roles_response_ids

        # Check showing a valid, role.
        role_response = self._get(f"roles/{user_role_id}")
        assert_status_code_is(role_response, 200)
        role = role_response.json()
        self.check_role_dict(role, assert_id=user_role_id)



[docs]
    @requires_admin
    def test_create_invalid_params(self):
        # In theory these low-level validation test cases could be handled in more
        # of a unit test style but it makes sense during the transition from wsgi to
        # asgi to have some tests that validate the whole pipeline is being integrated
        # properly in terms of exception handling.

        # Test missing description
        name = self.dataset_populator.get_random_name()
        description = "A test role."
        payload = {
            "name": name,
            "user_ids": [self.dataset_populator.user_id()],
        }
        response = self._post("roles", payload, admin=True, json=True)
        assert_status_code_is(response, 400)
        assert_error_code_is(response, error_codes.error_codes_by_name["USER_REQUEST_MISSING_PARAMETER"].code)
        assert "description" in response.json()["err_msg"]

        # Test missing name
        payload_missing_name = {
            "description": description,
            "user_ids": [self.dataset_populator.user_id()],
        }
        response = self._post("roles", payload_missing_name, admin=True, json=True)
        assert_status_code_is(response, 400)
        assert_error_code_is(response, error_codes.error_codes_by_name["USER_REQUEST_MISSING_PARAMETER"].code)
        assert "name" in response.json()["err_msg"]

        # Test invalid type for name
        payload_invalid_type = {
            "name": ["a test", "name"],
            "description": description,
            "user_ids": [self.dataset_populator.user_id()],
        }
        response = self._post("roles", payload_invalid_type, admin=True, json=True)
        assert_status_code_is(response, 400)
        assert_error_code_is(response, error_codes.error_codes_by_name["USER_REQUEST_INVALID_PARAMETER"].code)
        assert "name" in response.json()["err_msg"]
        assert "validation_errors" in response.json()



[docs]
    @requires_admin
    def test_create_valid(self):
        name = self.dataset_populator.get_random_name()
        description = "A test role."
        role = self._create_role(name=name, description=description)

        assert role["name"] == name
        assert role["description"] == description

        user_roles_response = self._get("roles")
        with self._different_user():
            different_user_roles_response = self._get("roles")

        user_roles_response_ids = [r["id"] for r in user_roles_response.json()]
        different_user_roles_response_ids = [r["id"] for r in different_user_roles_response.json()]

        # This new role is public, all users see it.
        assert role["id"] in user_roles_response_ids
        assert role["id"] in different_user_roles_response_ids



[docs]
    @requires_admin
    def test_show_error_codes(self):
        # Bad role ids are 400.
        response = self._get("roles/badroleid")
        assert_status_code_is(response, 400)

        # Trying to access others roles raise (not found) error
        with self._different_user():
            different_user_role_id = self.dataset_populator.user_private_role_id()
        response = self._get(f"roles/{different_user_role_id}")
        assert_status_code_is(response, 404)



[docs]
    @requires_admin
    def test_list_with_pagination(self):
        self._create_role()
        self._create_role()
        # Test limit
        response = self._get("roles", data={"limit": 1}, admin=True)
        assert_status_code_is(response, 200)
        data = response.json()
        assert len(data) == 1
        # Test limit + offset returns different result
        response_offset = self._get("roles", data={"limit": 1, "offset": 1}, admin=True)
        assert_status_code_is(response_offset, 200)
        data_offset = response_offset.json()
        assert len(data_offset) == 1
        assert data[0]["id"] != data_offset[0]["id"]



[docs]
    @requires_admin
    def test_list_with_search(self):
        unique = self.dataset_populator.get_random_name()
        role = self._create_role(name=f"searchable-{unique}")
        response = self._get("roles", data={"search": unique}, admin=True)
        assert_status_code_is(response, 200)
        data = response.json()
        assert any(r["id"] == role["id"] for r in data)
        # Search for non-existing returns empty
        response = self._get("roles", data={"search": "nonexistent-xyz-99999"}, admin=True)
        assert_status_code_is(response, 200)
        data = response.json()
        assert len(data) == 0



[docs]
    @requires_admin
    def test_list_exclude_private(self):
        user_role_id = self.dataset_populator.user_private_role_id()
        role = self._create_role()
        response = self._get("roles", data={"exclude_private": True}, admin=True)
        assert_status_code_is(response, 200)
        role_ids = [r["id"] for r in response.json()]
        assert role["id"] in role_ids
        assert user_role_id not in role_ids
        assert all(r["type"] != "private" for r in response.json())



[docs]
    @requires_admin
    def test_create_only_admin(self):
        response = self._post("roles", json=True)
        assert_status_code_is(response, 403)
        response_err = response.json()
        assert response_err["err_code"] == 403006
        assert "administrator" in response_err["err_msg"]



[docs]
    @requires_admin
    def test_delete(self):
        role = self._create_role()
        role_id = role["id"]
        response = self._delete(f"roles/{role_id}", admin=True)
        assert_status_code_is(response, 200)



[docs]
    @requires_admin
    def test_delete_duplicating_name_raises_409(self):
        role = self._create_role()
        role_id = role["id"]
        role_name = role["name"]

        delete_response = self._delete(f"roles/{role_id}", admin=True)
        self._assert_status_code_is_ok(delete_response)

        # Create a new role with the same name as the deleted one is not allowed
        payload = self._build_valid_role_payload(role_name)
        response = self._post("roles", payload, admin=True, json=True)
        self._assert_status_code_is(response, 409)



[docs]
    @requires_admin
    def test_purge(self):
        role = self._create_role()
        role_id = role["id"]

        # Delete and purge the role
        delete_response = self._delete(f"roles/{role_id}", admin=True)
        self._assert_status_code_is_ok(delete_response)
        purge_response = self._post(f"roles/{role_id}/purge", admin=True)
        self._assert_status_code_is_ok(purge_response)

        # The role is deleted and purged, so it cannot be found
        response = self._get(f"roles/{role_id}", admin=True)
        self._assert_status_code_is(response, 404)



[docs]
    @requires_admin
    def test_purge_can_reuse_name(self):
        role = self._create_role()
        role_id = role["id"]
        role_name = role["name"]

        # Delete and purge the role
        delete_response = self._delete(f"roles/{role_id}", admin=True)
        self._assert_status_code_is_ok(delete_response)
        purge_response = self._post(f"roles/{role_id}/purge", admin=True)
        self._assert_status_code_is_ok(purge_response)

        # Create a new role with the same name as the deleted one is allowed
        payload = self._build_valid_role_payload(role_name)
        response = self._post("roles", payload, admin=True, json=True)
        self._assert_status_code_is(response, 200)



[docs]
    @requires_admin
    def test_update(self):
        role = self._create_role()
        role_id = role["id"]
        other_user_id = self._setup_user("role-update-user@bx.psu.edu")["id"]
        group_id = self.dataset_populator.create_group()["id"]
        new_name = self.dataset_populator.get_random_name()

        payload = {
            "name": new_name,
            "description": "updated description",
            "user_ids": [other_user_id],
            "group_ids": [group_id],
        }
        response = self._put(f"roles/{role_id}", payload, admin=True, json=True)
        assert_status_code_is(response, 200)
        self.check_role_dict(response.json(), assert_id=role_id)

        role = self._get(f"roles/{role_id}", admin=True).json()
        assert role["name"] == new_name
        assert role["description"] == "updated description"
        assert self._role_user_ids(role_id) == [other_user_id]
        assert self._role_group_ids(role_id) == [group_id]



[docs]
    @requires_admin
    def test_update_omitted_associations_unchanged(self):
        group_id = self.dataset_populator.create_group()["id"]
        role = self._create_role(group_ids=[group_id])
        role_id = role["id"]
        new_name = self.dataset_populator.get_random_name()

        response = self._put(f"roles/{role_id}", {"name": new_name}, admin=True, json=True)
        assert_status_code_is(response, 200)
        assert response.json()["name"] == new_name
        assert response.json()["description"] == role["description"]
        assert self._role_user_ids(role_id) == [self.dataset_populator.user_id()]
        assert self._role_group_ids(role_id) == [group_id]



[docs]
    @requires_admin
    def test_update_clears_associations(self):
        role_id = self._create_role()["id"]
        group_id = self.dataset_populator.create_group()["id"]
        assert_status_code_is(self._put(f"roles/{role_id}", {"group_ids": [group_id]}, admin=True, json=True), 200)
        assert self._role_user_ids(role_id) == [self.dataset_populator.user_id()]
        assert self._role_group_ids(role_id) == [group_id]

        payload: dict[str, list[str]] = {"user_ids": [], "group_ids": []}
        response = self._put(f"roles/{role_id}", payload, admin=True, json=True)
        assert_status_code_is(response, 200)
        assert self._role_user_ids(role_id) == []
        assert self._role_group_ids(role_id) == []



[docs]
    @requires_admin
    def test_update_keeping_own_name(self):
        role = self._create_role()
        response = self._put(f"roles/{role['id']}", {"name": role["name"]}, admin=True, json=True)
        assert_status_code_is(response, 200)



[docs]
    @requires_admin
    def test_update_duplicating_name_raises_409(self):
        role_a = self._create_role()
        role_b = self._create_role()
        response = self._put(f"roles/{role_b['id']}", {"name": role_a["name"]}, admin=True, json=True)
        assert_status_code_is(response, 409)



[docs]
    @requires_admin
    def test_update_invalid_user_id(self):
        role = self._create_role()
        role_id = role["id"]
        invalid_user_id = self._get("configuration/encode/999999999", admin=True).json()["encoded_id"]
        payload = {"name": f"{role['name']}-renamed", "description": "renamed", "user_ids": [invalid_user_id]}
        response = self._put(f"roles/{role_id}", payload, admin=True, json=True)
        assert_status_code_is(response, 400)
        assert self._role_user_ids(role_id) == [self.dataset_populator.user_id()]
        unchanged_role = self._get(f"roles/{role_id}", admin=True).json()
        assert unchanged_role["name"] == role["name"]
        assert unchanged_role["description"] == role["description"]



[docs]
    @requires_admin
    def test_users_and_groups_leave_out_deleted(self):
        deleted_user_id = self._setup_user(f"{self.dataset_populator.get_random_name()}@bx.psu.edu")["id"]
        kept_group_id = self.dataset_populator.create_group()["id"]
        deleted_group_id = self.dataset_populator.create_group()["id"]
        role = self._create_role(extra_user_ids=[deleted_user_id], group_ids=[kept_group_id, deleted_group_id])
        assert_status_code_is(self._delete(f"users/{deleted_user_id}", admin=True), 200)
        assert_status_code_is(self._delete(f"groups/{deleted_group_id}", admin=True), 200)

        assert self._role_user_ids(role["id"]) == [self.dataset_populator.user_id()]
        assert self._role_group_ids(role["id"]) == [kept_group_id]



[docs]
    @requires_admin
    def test_update_and_associations_only_admin(self):
        role_id = self._create_role()["id"]
        assert_status_code_is(self._put(f"roles/{role_id}", {"name": "new-name"}, json=True), 403)
        assert_status_code_is(self._get(f"roles/{role_id}/users"), 403)
        assert_status_code_is(self._get(f"roles/{role_id}/groups"), 403)


    def _role_user_ids(self, role_id: str) -> list[str]:
        response = self._get(f"roles/{role_id}/users", admin=True)
        assert_status_code_is(response, 200)
        return [user["id"] for user in response.json()]

    def _role_group_ids(self, role_id: str) -> list[str]:
        response = self._get(f"roles/{role_id}/groups", admin=True)
        assert_status_code_is(response, 200)
        return [group["id"] for group in response.json()]

    def _create_role(
        self,
        name: str | None = None,
        description: str | None = None,
        extra_user_ids: list[str] | None = None,
        group_ids: list[str] | None = None,
    ) -> dict[str, Any]:
        payload = self._build_valid_role_payload(name=name, description=description)
        payload["user_ids"] += extra_user_ids or []
        payload["group_ids"] = group_ids or []
        response = self._post("roles", payload, admin=True, json=True)
        assert_status_code_is(response, 200)
        role = response.json()
        self.check_role_dict(role)
        return role

    def _build_valid_role_payload(self, name: str | None = None, description: str | None = None):
        name = name or self.dataset_populator.get_random_name()
        description = description or f"A test role with name: {name}."
        payload = {
            "name": name,
            "description": description,
            "user_ids": [self.dataset_populator.user_id()],
        }
        return payload


[docs]
    @staticmethod
    def check_role_dict(role_dict: dict[str, Any], assert_id: str | None = None) -> None:
        assert_has_keys(role_dict, "id", "name", "model_class", "url")
        assert role_dict["model_class"] == "Role"
        if assert_id is not None:
            assert role_dict["id"] == assert_id