Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy_test.api.test_users
from urllib.parse import (
quote,
urljoin,
)
import requests
from galaxy.exceptions import error_codes
from galaxy_test.api._framework import ApiTestCase
from galaxy_test.base.api_asserts import (
assert_error_message_contains,
assert_object_id_error,
)
from galaxy_test.base.api_util import baseauth_headers
from galaxy_test.base.decorators import (
requires_admin,
requires_new_history,
requires_new_user,
)
from galaxy_test.base.populators import (
DatasetPopulator,
PRIVATE_ROLE_TYPE,
skip_without_tool,
)
TEST_USER_EMAIL = "user_for_users_index_test@bx.psu.edu"
TEST_USER_EMAIL_INDEX_DELETED = "user_for_users_index_deleted_test@bx.psu.edu"
TEST_USER_EMAIL_DELETE = "user_for_delete_test@bx.psu.edu"
TEST_USER_EMAIL_DELETE_CANCEL_JOBS = "user_for_delete_cancel_jobs_test@bx.psu.edu"
TEST_USER_EMAIL_PURGE = "user_for_purge_test@bx.psu.edu"
TEST_USER_EMAIL_UNDELETE = "user_for_undelete_test@bx.psu.edu"
TEST_USER_EMAIL_SHOW = "user_for_show_test@bx.psu.edu"
TEST_USER_EMAIL_UPDATE = "user_for_email_update_test@bx.psu.edu"
TEST_USER_EMAIL_UPDATED = "user_for_email_update_test_changed@bx.psu.edu"
TEST_USER_EMAIL_ROLES_AND_GROUPS = "user_for_roles_and_groups_test@bx.psu.edu"
INVALID_PARAMETER = error_codes.error_codes_by_name["USER_REQUEST_INVALID_PARAMETER"]
[docs]
class TestUsersApi(ApiTestCase):
[docs]
@requires_admin
@requires_new_user
def test_index(self):
user = self._setup_user(TEST_USER_EMAIL_INDEX_DELETED)
all_users_response = self._get("users", admin=True)
self._assert_status_code_is(all_users_response, 200)
all_users = all_users_response.json()
# New user is in list
assert len([u for u in all_users if u["email"] == TEST_USER_EMAIL_INDEX_DELETED]) == 1
# Request made from admin user, so should at least self and this
# new user.
assert len(all_users) > 1
# index of deleted users
self._delete(f"users/{user['id']}", admin=True)
all_deleted_users_response_1 = self._get("users/deleted", admin=True)
self._assert_status_code_is(all_deleted_users_response_1, 200)
payload = {"deleted": "True"}
all_deleted_users_response_2 = self._get("users", data=payload, admin=True)
self._assert_status_code_is(all_deleted_users_response_2, 200)
# user is in list of deleted users
all_deleted_users = all_deleted_users_response_1.json()
assert len([u for u in all_deleted_users if u["email"] == TEST_USER_EMAIL_INDEX_DELETED]) == 1
all_deleted_users = all_deleted_users_response_2.json()
assert len([u for u in all_deleted_users if u["email"] == TEST_USER_EMAIL_INDEX_DELETED]) == 1
[docs]
@requires_admin
def test_index_with_pagination(self):
dataset_populator = DatasetPopulator(self.galaxy_interactor)
self._setup_user(f"pagination_test_{dataset_populator.get_random_name()}@bx.psu.edu")
self._setup_user(f"pagination_test_{dataset_populator.get_random_name()}@bx.psu.edu")
response = self._get("users", data={"limit": 1}, admin=True)
self._assert_status_code_is(response, 200)
data = response.json()
assert len(data) == 1
# With offset
response_offset = self._get("users", data={"limit": 1, "offset": 1}, admin=True)
self._assert_status_code_is(response_offset, 200)
data_offset = response_offset.json()
assert len(data_offset) == 1
# Different results
assert data[0]["id"] != data_offset[0]["id"]
[docs]
def test_index_anon(self):
with self._different_user(anon=True):
all_users_response = self._get("users")
self._assert_status_code_is(all_users_response, 403)
[docs]
@requires_new_user
def test_index_only_self_for_nonadmins(self):
self._setup_user(TEST_USER_EMAIL)
with self._different_user():
all_users_response = self._get("users")
# Non admin users can only see themselves
assert len(all_users_response.json()) == 1
[docs]
@requires_new_user
def test_show(self):
user = self._setup_user(TEST_USER_EMAIL)
with self._different_user(email=TEST_USER_EMAIL):
show_response = self.__show(user)
self._assert_status_code_is(show_response, 200)
self.__assert_matches_user(user, show_response.json())
[docs]
@requires_new_user
def test_update(self):
payload = {"username": "linnaeus"}
user = self._setup_user(TEST_USER_EMAIL)
not_the_user = self._setup_user("email@example.com")
with self._different_user(email=TEST_USER_EMAIL):
# working
update_response = self.__update(user, data=payload)
self._assert_status_code_is(update_response, 200)
update_json = update_response.json()
assert update_json["username"] == payload["username"]
# too short
update_response = self.__update(user, data={"username": ""})
self._assert_status_code_is(update_response, 400)
# the username rule applies to a new username
update_response = self.__update(user, data={"username": "Carl Linnaeus"})
self._assert_status_code_is(update_response, 400)
# the current username may be sent back alongside other changes
data = {"username": payload["username"], "display_name": "Carl Linnaeus"}
update_response = self.__update(user, data=data)
self._assert_status_code_is(update_response, 200)
assert update_response.json()["username"] == payload["username"]
# the field may be left out, but null is not a username
update_response = self.__update(user, data={"username": None})
self._assert_status_code_is(update_response, 400)
self._assert_error_code_is(update_response, INVALID_PARAMETER)
# fields the route does not update are refused, not dropped
for refused in ({"is_admin": True}, {"password": "new-password"}, {"usrname": "linnaeus"}):
update_response = self.__update(user, data=refused)
self._assert_status_code_is(update_response, 400)
self._assert_error_code_is(update_response, INVALID_PARAMETER)
# not them
update_response = self.__update(not_the_user, data=payload)
self._assert_status_code_is(update_response, 400)
# non-existent
no_user_id = "5d7db0757a2eb7ef"
update_url = self._api_url(f"users/{no_user_id}")
update_response = self._put(update_url, data=payload, json=True)
assert_object_id_error(update_response)
[docs]
@requires_new_user
def test_update_display_name(self):
user = self._setup_user(TEST_USER_EMAIL)
with self._different_user(email=TEST_USER_EMAIL):
update_response = self.__update(user, data={"display_name": "Carl von Linné"})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["display_name"] == "Carl von Linné"
# surrounding whitespace is stripped rather than rejected
update_response = self.__update(user, data={"display_name": " Carl von Linné "})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["display_name"] == "Carl von Linné"
# an empty display name clears the field
update_response = self.__update(user, data={"display_name": ""})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["display_name"] is None
# text-direction characters would let the name render as other text
update_response = self.__update(user, data={"display_name": "Carl\u202evon Linné"})
self._assert_status_code_is(update_response, 400)
self._assert_error_code_is(update_response, INVALID_PARAMETER)
# over the column length
update_response = self.__update(user, data={"display_name": "N" * 256})
self._assert_status_code_is(update_response, 400)
# the length limit applies once surrounding whitespace is stripped
update_response = self.__update(user, data={"display_name": " " + "N" * 255 + " "})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["display_name"] == "N" * 255
update_response = self.__update(user, data={"display_name": " " * 256})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["display_name"] is None
[docs]
@requires_admin
@requires_new_user
def test_admin_update(self):
payload = {"username": "flexo"}
user = self._setup_user(TEST_USER_EMAIL)
update_url = self._api_url(f"users/{user['id']}")
update_response = self._put(update_url, data=payload, admin=True, json=True)
self._assert_status_code_is(update_response, 200)
update_json = update_response.json()
assert update_json["username"] == payload["username"]
[docs]
@requires_new_user
def test_update_email(self):
user = self._setup_user(TEST_USER_EMAIL_UPDATE)
with self._different_user(email=TEST_USER_EMAIL_UPDATE):
update_response = self.__update(user, data={"email": TEST_USER_EMAIL_UPDATED})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["email"] == TEST_USER_EMAIL_UPDATED
# surrounding whitespace is stripped rather than rejected
update_response = self.__update(user, data={"email": f" {TEST_USER_EMAIL_UPDATE} "})
self._assert_status_code_is(update_response, 200)
assert update_response.json()["email"] == TEST_USER_EMAIL_UPDATE
[docs]
@requires_admin
@requires_new_user
def test_create_email(self):
email = "user_for_create_padded_email_test@bx.psu.edu"
payload = {"email": f" {email} ", "username": "padded-email", "password": "testpass"}
create_response = self._post("users", data=payload, admin=True, json=True)
self._assert_status_code_is(create_response, 200)
assert create_response.json()["email"] == email
payload = {"email": "not-an-email", "username": "malformed-email", "password": "testpass"}
create_response = self._post("users", data=payload, admin=True, json=True)
self._assert_status_code_is(create_response, 400)
self._assert_error_code_is(create_response, INVALID_PARAMETER)
[docs]
@requires_admin
@requires_new_user
def test_create_username_invalid(self):
payload = {"email": "user_for_create_bad_username@bx.psu.edu", "username": "Not Valid", "password": "testpass"}
create_response = self._post("users", data=payload, admin=True, json=True)
self._assert_status_code_is(create_response, 400)
self._assert_error_code_is(create_response, INVALID_PARAMETER)
[docs]
@requires_new_user
def test_update_email_invalid(self):
user = self._setup_user(TEST_USER_EMAIL)
other_user = self._setup_user("email_update_conflict@bx.psu.edu")
with self._different_user(email=TEST_USER_EMAIL):
# malformed
update_response = self.__update(user, data={"email": "not-an-email"})
self._assert_status_code_is(update_response, 400)
# already taken by another account
update_response = self.__update(user, data={"email": other_user["email"]})
self._assert_status_code_is(update_response, 400)
# null is not an email address
update_response = self.__update(user, data={"email": None})
self._assert_status_code_is(update_response, 400)
self._assert_error_code_is(update_response, INVALID_PARAMETER)
[docs]
@requires_admin
@requires_new_user
def test_update_active_requires_admin(self):
email = "active_flag_update@bx.psu.edu"
user = self._setup_user(email)
with self._different_user(email=email):
# activation is the email-verification gate, so users cannot flip it on themselves
update_response = self.__update(user, data={"active": True})
self._assert_status_code_is(update_response, 403)
update_url = self._api_url(f"users/{user['id']}")
for active in (False, True):
update_response = self._put(update_url, data={"active": active}, admin=True, json=True)
self._assert_status_code_is(update_response, 200)
# `active` is not on DetailedUserModel, so it is read back from the index.
listed = self._get("users", data={"f_email": email}, admin=True).json()
assert [u for u in listed if u["id"] == user["id"]][0]["active"] is active
# null is not a bool; reject it before it reaches the NOT NULL column
update_response = self._put(update_url, data={"active": None}, admin=True, json=True)
self._assert_status_code_is(update_response, 400)
self._assert_error_code_is(update_response, INVALID_PARAMETER)
[docs]
@requires_new_user
def test_extra_preferences(self):
user = self._setup_user(TEST_USER_EMAIL)
with self._different_user(email=TEST_USER_EMAIL):
# The default test instance configures no extra preferences.
response = self._get("configuration/extra_preferences")
self._assert_status_code_is(response, 200)
assert response.json() == []
response = self._get(f"users/{user['id']}/extra_preferences")
self._assert_status_code_is(response, 200)
assert response.json() == {}
response = self._put(f"users/{user['id']}/extra_preferences", data={"no_such_section": {}}, json=True)
self._assert_status_code_is(response, 400)
[docs]
@requires_new_user
def test_extra_preferences_other_user_forbidden(self):
user = self._setup_user(TEST_USER_EMAIL)
self._setup_user("extra_prefs_other@bx.psu.edu")
with self._different_user(email="extra_prefs_other@bx.psu.edu"):
response = self._get(f"users/{user['id']}/extra_preferences")
self._assert_status_code_is(response, 403)
[docs]
@requires_admin
@requires_new_user
def test_delete_user(self):
user = self._setup_user(TEST_USER_EMAIL_DELETE)
self._delete(f"users/{user['id']}", admin=True)
updated_user = self._get(f"users/deleted/{user['id']}", admin=True).json()
assert updated_user["deleted"] is True, updated_user
[docs]
@requires_admin
@requires_new_user
def test_purge_user(self):
"""Delete user and then purge them."""
user = self._setup_user(TEST_USER_EMAIL_PURGE)
response = self._delete(f"users/{user['id']}", admin=True)
self._assert_status_code_is_ok(response)
params = dict(purge="True")
response = self._delete(f"users/{user['id']}", params=params, admin=True, json=True)
self._assert_status_code_is_ok(response)
params = {"deleted": "True"}
purged_user = self._get(f"users/{user['id']}", params, admin=True).json()
assert purged_user["deleted"] is True, purged_user
assert purged_user["purged"] is True, purged_user
[docs]
@requires_admin
@requires_new_user
def test_undelete_user(self):
"""Delete user and then undelete them."""
user = self._setup_user(TEST_USER_EMAIL_UNDELETE)
self._delete(f"users/{user['id']}", admin=True)
payload = {"deleted": "True"}
deleted_user = self._get(f"users/{user['id']}", payload, admin=True).json()
assert deleted_user["deleted"] is True, deleted_user
self._post(f"users/deleted/{user['id']}/undelete", admin=True)
undeleted_user = self._get(f"users/{user['id']}", admin=True).json()
assert undeleted_user["deleted"] is False, undeleted_user
[docs]
@requires_admin
@requires_new_user
@requires_new_history
@skip_without_tool("cat_data_and_sleep")
def test_delete_user_cancel_all_jobs(self):
dataset_populator = DatasetPopulator(self.galaxy_interactor)
with self._different_user(TEST_USER_EMAIL_DELETE_CANCEL_JOBS):
user_id = self._get_current_user_id()
history_id = dataset_populator.new_history()
hda_id = dataset_populator.new_dataset(history_id)["id"]
inputs = {
"input1": {"src": "hda", "id": hda_id},
"sleep_time": 6000,
}
run_response = dataset_populator.run_tool_raw(
"cat_data_and_sleep",
inputs,
history_id,
)
self._assert_status_code_is_ok(run_response)
job_id = run_response.json()["jobs"][0]["id"]
# Wait a bit for the job to be ready
expected_job_states = ["new", "queued", "running"]
dataset_populator.wait_for_job(job_id, ok_states=expected_job_states)
# Get the job state
job_response = self._get(f"jobs/{job_id}").json()
assert job_response["state"] in expected_job_states, job_response
# Delete user will cancel all jobs
self._delete(f"users/{user_id}", admin=True)
# Get the job state again (this time as admin), it should be deleting or deleted
job_response = self._get(f"jobs/{job_id}", admin=True).json()
assert job_response["state"] in ["deleting", "deleted"], job_response
[docs]
@requires_new_user
def test_information(self):
user = self._setup_user(TEST_USER_EMAIL)
url = self.__url("information/inputs", user)
response = self._get(url).json()
assert response["username"] == user["username"]
assert response["email"] == TEST_USER_EMAIL
payload = {"username": "newname", "email": "new@email.email"}
self._put(url, data=payload, json=True)
response = self._get(url).json()
assert response["username"] == "newname"
assert response["email"] == "new@email.email"
# surrounding whitespace is stripped here too, not only on PUT /api/users/{id}
payload = {"username": user["username"], "email": f" {TEST_USER_EMAIL} "}
self._put(url, data=payload, json=True)
response = self._get(url).json()
assert response["username"] == user["username"]
assert response["email"] == TEST_USER_EMAIL
self._put(url, data={"address_0|desc": "_desc"}, json=True)
response = self._get(url).json()
assert len(response["addresses"]) == 1
assert response["addresses"][0]["desc"] == "_desc"
[docs]
@requires_new_user
def test_information_addresses_of_another_user(self):
owner_email = "address_owner@bx.psu.edu"
owner = self._setup_user(owner_email)
owner_url = f"users/{owner['id']}/information/inputs"
with self._different_user(email=owner_email):
self._assert_status_code_is(self._put(owner_url, data={"address_0|desc": "owner"}, json=True), 200)
address_id = self._get(owner_url).json()["addresses"][0]["id"]
response = self._put(owner_url, data={"address_0|id": address_id, "address_0|desc": "owner2"}, json=True)
self._assert_status_code_is(response, 200)
other_email = "address_other@bx.psu.edu"
other = self._setup_user(other_email)
other_url = f"users/{other['id']}/information/inputs"
with self._different_user(email=other_email):
response = self._put(other_url, data={"address_0|id": address_id, "address_0|desc": "other"}, json=True)
self._assert_status_code_is(response, 404)
response = self._put(other_url, data={"address_0|id": "notanid", "address_0|desc": "other"}, json=True)
self._assert_status_code_is(response, 400)
owner_addresses = self._get(owner_url, admin=True).json()["addresses"]
assert [(address["id"], address["desc"]) for address in owner_addresses] == [(address_id, "owner2")]
[docs]
@requires_new_user
def test_manage_api_key(self):
with self._different_user("manage-api-key-test@user.com"):
user_id = self._get_current_user_id()
# Initially we have an API key because it is bootstrapped for tests
response = self._get(f"users/{user_id}/api_key")
user_api_key = response.json()
assert user_api_key
# Test detailed endpoint
response = self._get(f"users/{user_id}/api_key/detailed")
api_key = response.json()
assert api_key["key"] == user_api_key
# Delete user API key
response = self._delete(f"users/{user_id}/api_key")
self._assert_status_code_is_ok(response)
# No API key anymore, so the detailed request returns no content 204 with admin key
response = self._get(f"users/{user_id}/api_key/detailed", admin=True)
self._assert_status_code_is_ok(response)
# No API key anymore, so the detailed request returns unauthorized 401 with user key
response = self._get(f"users/{user_id}/api_key/detailed")
self._assert_status_code_is(response, 401)
# create new as admin
response = self._post(f"users/{user_id}/api_key", admin=True)
self._assert_status_code_is_ok(response)
new_api_key = response.json()
assert new_api_key
assert new_api_key != user_api_key
[docs]
@requires_new_user
def test_only_admin_can_manage_other_users_api_key(self):
with self._different_user():
other_user_id = self._get_current_user_id()
current_user_id = self._get_current_user_id()
# Users cannot access other users API keys
assert current_user_id != other_user_id
response = self._get(f"users/{other_user_id}/api_key")
self._assert_status_code_is(response, 403)
response = self._post(f"users/{other_user_id}/api_key")
self._assert_status_code_is(response, 403)
response = self._delete(f"users/{other_user_id}/api_key")
self._assert_status_code_is(response, 403)
# Admins can access other users API keys
response = self._get(f"users/{other_user_id}/api_key", admin=True)
self._assert_status_code_is_ok(response)
response = self._post(f"users/{other_user_id}/api_key", admin=True)
self._assert_status_code_is_ok(response)
response = self._delete(f"users/{other_user_id}/api_key", admin=True)
self._assert_status_code_is_ok(response)
[docs]
@requires_admin
@requires_new_user
@skip_without_tool("cat1")
def test_favorites(self):
user = self._setup_user(TEST_USER_EMAIL)
# adding a tool to favorites
url = self._api_url(f"users/{user['id']}/favorites/tools")
put_response = self._put(url, data={"object_id": "cat1"}, admin=True, json=True)
self._assert_status_code_is_ok(put_response)
assert put_response.json()["tools"][0] == "cat1"
# not implemented for workflows yet
url = self._api_url(f"users/{user['id']}/favorites/workflows")
put_response = self._put(url, data={"object_id": "14ds68f4sda68gf46dsag4"}, admin=True, json=True)
self._assert_status_code_is(put_response, 400)
# delete existing tool favorite
url = self._api_url(f"users/{user['id']}/favorites/tools/cat1")
delete_response = self._delete(url, admin=True)
self._assert_status_code_is_ok(delete_response)
assert delete_response.json()["tools"] == []
# delete non-existing tool favorite
url = self._api_url(f"users/{user['id']}/favorites/tools/madeuptoolthatdoes/not/exist/in/favs")
delete_response = self._delete(url, admin=True)
self._assert_status_code_is(delete_response, 404)
# delete non existing workflow favorite
url = self._api_url(f"users/{user['id']}/favorites/workflows/1as5das5das56d465")
delete_response = self._delete(url, admin=True)
self._assert_status_code_is(delete_response, 400)
[docs]
@requires_admin
@requires_new_user
@skip_without_tool("Remove beginning1")
def test_favorites_whitespace_tool_id(self):
user = self._setup_user(TEST_USER_EMAIL)
tool_id = "Remove beginning1"
add_url = self._api_url(f"users/{user['id']}/favorites/tools")
add_response = self._put(add_url, data={"object_id": tool_id}, admin=True, json=True)
self._assert_status_code_is_ok(add_response)
assert add_response.json()["tools"] == [tool_id]
remove_url = self._api_url(f"users/{user['id']}/favorites/tools/{tool_id}")
remove_response = self._delete(remove_url, admin=True)
self._assert_status_code_is_ok(remove_response)
assert remove_response.json()["tools"] == []
[docs]
@requires_admin
@requires_new_user
@skip_without_tool("cat1")
@skip_without_tool("__ZIP_COLLECTION__")
def test_favorite_tags(self):
user = self._setup_user(TEST_USER_EMAIL)
tool_favorites_url = self._api_url(f"users/{user['id']}/favorites/tools")
tool_response = self._put(tool_favorites_url, data={"object_id": "cat1"}, admin=True, json=True)
self._assert_status_code_is_ok(tool_response)
assert tool_response.json()["tools"] == ["cat1"]
assert tool_response.json()["tags"] == []
tag_name = "Collection Operations"
tag_favorites_url = self._api_url(f"users/{user['id']}/favorites/tags")
tag_response = self._put(tag_favorites_url, data={"object_id": tag_name}, admin=True, json=True)
self._assert_status_code_is_ok(tag_response)
assert tag_response.json()["tools"] == ["cat1"]
assert tag_response.json()["tags"] == [tag_name]
remove_tag_url = self._api_url(f"users/{user['id']}/favorites/tags/{quote(tag_name)}")
remove_tag_response = self._delete(remove_tag_url, admin=True)
self._assert_status_code_is_ok(remove_tag_response)
assert remove_tag_response.json()["tools"] == ["cat1"]
assert remove_tag_response.json()["tags"] == []
[docs]
@requires_admin
@requires_new_user
@skip_without_tool("cat1")
def test_reorder_favorites(self):
user = self._setup_user(TEST_USER_EMAIL)
tool_favorites_url = self._api_url(f"users/{user['id']}/favorites/tools")
tool_response = self._put(tool_favorites_url, data={"object_id": "cat1"}, admin=True, json=True)
self._assert_status_code_is_ok(tool_response)
tag_name = "Collection Operations"
tag_favorites_url = self._api_url(f"users/{user['id']}/favorites/tags")
tag_response = self._put(tag_favorites_url, data={"object_id": tag_name}, admin=True, json=True)
self._assert_status_code_is_ok(tag_response)
order_url = self._api_url(f"users/{user['id']}/favorites/order")
reorder_response = self._put(
order_url,
data={
"order": [
{"object_type": "tags", "object_id": tag_name},
{"object_type": "tools", "object_id": "cat1"},
]
},
admin=True,
json=True,
)
self._assert_status_code_is_ok(reorder_response)
assert reorder_response.json()["order"] == [
{"object_type": "tags", "object_id": tag_name},
{"object_type": "tools", "object_id": "cat1"},
]
assert reorder_response.json()["tools"] == ["cat1"]
assert reorder_response.json()["tags"] == [tag_name]
[docs]
@requires_admin
@requires_new_user
def test_favorite_edam_operations(self):
user = self._setup_user(TEST_USER_EMAIL)
operations_panel = self._get("tool_panels/ontology:edam_operations", admin=True).json()
operation_id = next(
(item["id"] for item in operations_panel.values() if item.get("model_class") == "ToolSection"), None
)
assert operation_id is not None
operation_favorites_url = self._api_url(f"users/{user['id']}/favorites/edam_operations")
operation_response = self._put(operation_favorites_url, data={"object_id": operation_id}, admin=True, json=True)
self._assert_status_code_is_ok(operation_response)
assert operation_response.json()["edam_operations"] == [operation_id]
remove_operation_url = self._api_url(f"users/{user['id']}/favorites/edam_operations/{operation_id}")
remove_operation_response = self._delete(remove_operation_url, admin=True)
self._assert_status_code_is_ok(remove_operation_response)
assert remove_operation_response.json()["edam_operations"] == []
[docs]
@skip_without_tool("cat1")
def test_search_favorites(self):
user, user_key = self._setup_user_get_key(TEST_USER_EMAIL)
url = self._api_url(f"users/{user['id']}/favorites/tools", params=dict(key=user_key))
fav_response = self._put(url, data={"object_id": "cat1"}, json=True)
self._assert_status_code_is_ok(fav_response)
assert "cat1" in fav_response.json()["tools"]
url = self._api_url("tools", params=dict(q="#favs", key=user_key))
search_response = self._get(url).json()
assert "cat1" in search_response
[docs]
@requires_new_user
def test_set_theme(self):
user = self._setup_user(TEST_USER_EMAIL)
with self._different_user(email=TEST_USER_EMAIL):
url = self._api_url(f"users/{user['id']}/theme/test_theme")
theme_response = self._put(url)
self._assert_status_code_is_ok(theme_response)
url = self._api_url("users/current")
updated_theme = self._get(url).json()["preferences"]["theme"]
assert updated_theme == "test_theme"
[docs]
def test_custom_build_from_unconvertible_dataset(self):
user_id = self._get("users/current").json()["id"]
dataset_populator = DatasetPopulator(self.galaxy_interactor)
with dataset_populator.test_history() as history_id:
hda = dataset_populator.new_dataset(history_id, content="chr1\t0\t100\n", file_type="bed", wait=True)
response = self._put(
f"users/{user_id}/custom_builds/unconvertible_build",
data={"name": "Unconvertible build", "len|type": "fasta", "len|value": hda["id"]},
json=True,
)
self._assert_status_code_is(response, 400)
self._assert_error_code_is(response, error_codes.error_codes_by_name["USER_REQUEST_INVALID_PARAMETER"])
assert_error_message_contains(response, "Conversion from 'bed' to 'len' not possible")
builds = self._get(f"users/{user_id}/custom_builds").json()
assert "unconvertible_build" not in [build["id"] for build in builds]
[docs]
@requires_admin
@requires_new_user
def test_show_delete(self):
user = self._setup_user(TEST_USER_EMAIL_SHOW)
url = self._api_url(f"users/{user['id']}")
response_1 = self._get(url, admin=True).json()
self._delete(f"users/{user['id']}", admin=True)
# Both request should return the same user
response_2 = self._get(f"users/deleted/{user['id']}", admin=True).json()
payload = {"deleted": "True"}
response_3 = self._get(f"users/{user['id']}", payload, admin=True).json()
assert response_1["id"] == response_2["id"] == response_3["id"]
assert response_2 == response_3
[docs]
def test_show_current(self):
user_id = self._get_current_user_id()
url = self._api_url(f"users/{user_id}")
specified_user = self._get(url).json()
url = self._api_url("users/current")
current_user = self._get(url).json()
assert specified_user == current_user
def __url(self, action, user):
return self._api_url(f"users/{user['id']}/{action}", params=dict(key=self.master_api_key))
def __show(self, user):
return self._get(f"users/{user['id']}")
def __update(self, user, data):
update_url = self._api_url(f"users/{user['id']}")
return self._put(update_url, data=data, json=True)
def __assert_matches_user(self, userA, userB):
self._assert_has_keys(userB, "id", "username", "total_disk_usage")
assert userA["id"] == userB["id"]
assert userA["username"] == userB["username"]
def _get_current_user_id(self):
users_response = self._get("users")
users = users_response.json()
assert len(users) == 1
return users[0]["id"]
[docs]
def test_manage_beacon_settings(self):
user_id = self._get_current_user_id()
# Assert that beacon sharing is initially disabled
response = self._get(f"users/{user_id}/beacon")
user_beacon_settings = response.json()
assert not user_beacon_settings["enabled"]
# Check if post request is successful
response = self._post(f"users/{user_id}/beacon", data={"enabled": True}, json=True)
user_beacon_settings = response.json()
assert user_beacon_settings["enabled"]
# Check if the setting has been persisted
response = self._get(f"users/{user_id}/beacon")
user_beacon_settings = response.json()
assert user_beacon_settings["enabled"]
[docs]
@requires_admin
@requires_new_user
def test_user_roles(self):
user = self._setup_user(TEST_USER_EMAIL)
response = self._get(f"users/{user['id']}/roles", admin=True)
user_roles = response.json()
assert len(user_roles) == 1
assert user_roles[0]["type"] == PRIVATE_ROLE_TYPE
[docs]
@requires_admin
@requires_new_user
def test_user_roles_leave_out_deleted_roles(self):
user = self._setup_user(TEST_USER_EMAIL_ROLES_AND_GROUPS)
populator = DatasetPopulator(self.galaxy_interactor)
deleted_role_id = populator.create_role([user["id"]])["id"]
self._assert_status_code_is(self._delete(f"roles/{deleted_role_id}", admin=True), 200)
response = self._get(f"users/{user['id']}/roles", admin=True)
self._assert_status_code_is(response, 200)
assert deleted_role_id not in [role["id"] for role in response.json()]
[docs]
@requires_admin
def test_user_roles_404_for_unknown_user(self):
unknown_user_id = self._unknown_user_id()
self._assert_status_code_is(self._get(f"users/{unknown_user_id}/roles", admin=True), 404)
[docs]
@requires_admin
@requires_new_user
def test_set_user_roles_keeps_private_role(self):
user = self._setup_user(TEST_USER_EMAIL_ROLES_AND_GROUPS)
role_id = DatasetPopulator(self.galaxy_interactor).create_role([])["id"]
response = self._put(f"users/{user['id']}/roles", {"role_ids": [role_id]}, admin=True, json=True)
self._assert_status_code_is(response, 200)
types_by_id = {role["id"]: role["type"] for role in response.json()}
assert len(types_by_id) == 2
assert types_by_id[role_id] == "admin"
assert PRIVATE_ROLE_TYPE in types_by_id.values()
response = self._put(f"users/{user['id']}/roles", {"role_ids": []}, admin=True, json=True)
self._assert_status_code_is(response, 200)
assert [role["type"] for role in response.json()] == [PRIVATE_ROLE_TYPE]
assert [role["type"] for role in self._get(f"users/{user['id']}/roles", admin=True).json()] == [
PRIVATE_ROLE_TYPE
]
[docs]
@requires_admin
@requires_new_user
def test_user_groups(self):
user = self._setup_user(TEST_USER_EMAIL_ROLES_AND_GROUPS)
group_id = DatasetPopulator(self.galaxy_interactor).create_group()["id"]
response = self._put(f"users/{user['id']}/groups", {"group_ids": [group_id]}, admin=True, json=True)
self._assert_status_code_is(response, 200)
assert [group["id"] for group in response.json()] == [group_id]
assert [group["id"] for group in self._get(f"users/{user['id']}/groups", admin=True).json()] == [group_id]
group_users = self._get(f"groups/{group_id}/users", admin=True).json()
assert [group_user["id"] for group_user in group_users] == [user["id"]]
response = self._put(f"users/{user['id']}/groups", {"group_ids": []}, admin=True, json=True)
self._assert_status_code_is(response, 200)
assert response.json() == []
[docs]
@requires_admin
@requires_new_user
def test_user_groups_leave_out_deleted_groups(self):
user = self._setup_user(TEST_USER_EMAIL_ROLES_AND_GROUPS)
populator = DatasetPopulator(self.galaxy_interactor)
kept_group_id = populator.create_group(user_ids=[user["id"]])["id"]
deleted_group_id = populator.create_group(user_ids=[user["id"]])["id"]
self._assert_status_code_is(self._delete(f"groups/{deleted_group_id}", admin=True), 200)
response = self._get(f"users/{user['id']}/groups", admin=True)
self._assert_status_code_is(response, 200)
assert [group["id"] for group in response.json()] == [kept_group_id]
[docs]
@requires_admin
def test_set_user_roles_and_groups_404_for_unknown_user(self):
unknown_user_id = self._unknown_user_id()
self._assert_status_code_is(self._get(f"users/{unknown_user_id}/groups", admin=True), 404)
roles_response = self._put(f"users/{unknown_user_id}/roles", {"role_ids": []}, admin=True, json=True)
self._assert_status_code_is(roles_response, 404)
groups_response = self._put(f"users/{unknown_user_id}/groups", {"group_ids": []}, admin=True, json=True)
self._assert_status_code_is(groups_response, 404)
[docs]
@requires_admin
@requires_new_user
def test_reset_password(self):
email = f"{DatasetPopulator(self.galaxy_interactor).get_random_name()}@bx.psu.edu"
user = self._setup_user(email, password="oldpassword1")
baseauth_url = self._api_url("authenticate/baseauth", use_key=False)
self._assert_status_code_is(requests.get(baseauth_url, headers=baseauth_headers(email, "oldpassword1")), 200)
with requests.Session() as browser_session:
self._login_browser_session(browser_session, email, "oldpassword1")
current_user_response = browser_session.get(urljoin(self.url, "api/users/current"))
self._assert_status_code_is(current_user_response, 200)
assert current_user_response.json()["email"] == email
response = self._put(f"users/{user['id']}/password", {"password": "newpassword1"}, admin=True, json=True)
self._assert_status_code_is(response, 204)
# The reset logged the browser session out, so it now gets the anonymous user.
current_user_response = browser_session.get(urljoin(self.url, "api/users/current"))
self._assert_status_code_is(current_user_response, 200)
current_user = current_user_response.json()
assert "id" not in current_user
assert "email" not in current_user
new_auth = requests.get(baseauth_url, headers=baseauth_headers(email, "newpassword1"))
self._assert_status_code_is(new_auth, 200)
old_auth = requests.get(baseauth_url, headers=baseauth_headers(email, "oldpassword1"))
self._assert_status_code_is(old_auth, 401)
[docs]
@requires_admin
@requires_new_user
def test_reset_password_rejects_invalid_password(self):
user = self._setup_user(f"{DatasetPopulator(self.galaxy_interactor).get_random_name()}@bx.psu.edu")
response = self._put(f"users/{user['id']}/password", {"password": "short"}, admin=True, json=True)
self._assert_status_code_is(response, 400)
[docs]
@requires_admin
def test_reset_password_404_for_unknown_user(self):
payload = {"password": "newpassword1"}
response = self._put(f"users/{self._unknown_user_id()}/password", payload, admin=True, json=True)
self._assert_status_code_is(response, 404)
[docs]
@requires_new_user
def test_reset_password_only_admin(self):
user_id = DatasetPopulator(self.galaxy_interactor).user_id()
response = self._put(f"users/{user_id}/password", {"password": "newpassword1"}, json=True)
self._assert_status_code_is(response, 403)
[docs]
@requires_new_user
def test_user_roles_and_groups_only_admin(self):
user_id = DatasetPopulator(self.galaxy_interactor).user_id()
self._assert_status_code_is(self._put(f"users/{user_id}/roles", {"role_ids": []}, json=True), 403)
self._assert_status_code_is(self._get(f"users/{user_id}/groups"), 403)
self._assert_status_code_is(self._put(f"users/{user_id}/groups", {"group_ids": []}, json=True), 403)
def _unknown_user_id(self) -> str:
return self._get("configuration/encode/999999999", admin=True).json()["encoded_id"]
[docs]
@requires_admin
@requires_new_user
def test_custom_build_from_inaccessible_dataset(self):
dataset_populator = DatasetPopulator(self.galaxy_interactor)
with dataset_populator.test_history() as history_id:
hda = dataset_populator.new_dataset(history_id, content=">chr1\nACGT\n", file_type="fasta", wait=True)
dataset_populator.make_private(history_id, hda["id"])
with self._different_user(email=TEST_USER_EMAIL):
user_id = self._get("users/current").json()["id"]
response = self._put(
f"users/{user_id}/custom_builds/inaccessible_build",
data={"name": "Inaccessible build", "len|type": "fasta", "len|value": hda["id"]},
json=True,
)
self._assert_status_code_is(response, 403)
builds = self._get(f"users/{user_id}/custom_builds").json()
assert "inaccessible_build" not in [build["id"] for build in builds]