Warning
This document is for an old release 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
import json
from requests import (
delete,
get,
put,
)
from galaxy_test.base.api_asserts import assert_object_id_error
from galaxy_test.base.populators import skip_without_tool
from ._framework import ApiTestCase
TEST_USER_EMAIL = "user_for_users_index_test@bx.psu.edu"
TEST_USER_EMAIL_DELETE = "user_for_delete_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"
[docs]class UsersApiTestCase(ApiTestCase):
[docs] def test_index(self):
self._setup_user(TEST_USER_EMAIL)
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]) == 1
# Request made from admin user, so should at least self and this
# new user.
assert len(all_users) > 1
[docs] 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] 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] def test_update(self):
new_name = "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, username=new_name)
self._assert_status_code_is(update_response, 200)
update_json = update_response.json()
self.assertEqual(update_json["username"], new_name)
# too short
update_response = self.__update(user, username="")
self._assert_status_code_is(update_response, 400)
# not them
update_response = self.__update(not_the_user, username=new_name)
self._assert_status_code_is(update_response, 403)
# non-existent
no_user_id = "5d7db0757a2eb7ef"
update_url = self._api_url(f"users/{no_user_id}", use_key=True)
update_response = put(update_url, data=json.dumps(dict(username=new_name)))
assert_object_id_error(update_response)
[docs] def test_admin_update(self):
new_name = "flexo"
user = self._setup_user(TEST_USER_EMAIL)
update_url = self._api_url(f"users/{user['id']}", params=dict(key=self.master_api_key))
update_response = put(update_url, data=json.dumps(dict(username=new_name)))
self._assert_status_code_is(update_response, 200)
update_json = update_response.json()
self.assertEqual(update_json["username"], new_name)
[docs] 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] 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)
data = dict(purge="True")
response = self._delete(f"users/{user['id']}", data=data, admin=True)
self._assert_status_code_is_ok(response)
payload = {"deleted": "True"}
purged_user = self._get(f"users/{user['id']}", payload, admin=True).json()
assert purged_user["deleted"] is True, purged_user
assert purged_user["purged"] is True, purged_user
[docs] 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] def test_information(self):
user = self._setup_user(TEST_USER_EMAIL)
url = self.__url("information/inputs", user)
response = get(url).json()
self.assertEqual(response["username"], user["username"])
self.assertEqual(response["email"], TEST_USER_EMAIL)
put(url, data=json.dumps(dict(username="newname", email="new@email.email")))
response = get(url).json()
self.assertEqual(response["username"], "newname")
self.assertEqual(response["email"], "new@email.email")
put(url, data=json.dumps(dict(username=user["username"], email=TEST_USER_EMAIL)))
response = get(url).json()
self.assertEqual(response["username"], user["username"])
self.assertEqual(response["email"], TEST_USER_EMAIL)
put(url, data=json.dumps({"address_0|desc": "_desc"}))
response = get(url).json()
self.assertEqual(len(response["addresses"]), 1)
self.assertEqual(response["addresses"][0]["desc"], "_desc")
[docs] def test_create_api_key(self):
user = self._setup_user(TEST_USER_EMAIL)
user_id = user["id"]
response = self._put(f"users/{user_id}/api_key/inputs", admin=True)
self._assert_status_code_is_ok(response)
self.assertEqual(response.json()["inputs"][0]["name"], "api-key")
[docs] @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", params=dict(key=self.master_api_key))
put_response = put(url, data=json.dumps({"object_id": "cat1"}))
self._assert_status_code_is_ok(put_response)
self.assertEqual(put_response.json()["tools"][0], "cat1")
# not implemented for workflows yet
url = self._api_url(f"users/{user['id']}/favorites/workflows", params=dict(key=self.master_api_key))
put_response = put(url, data=json.dumps({"object_id": "14ds68f4sda68gf46dsag4"}))
self._assert_status_code_is(put_response, 400)
# delete existing tool favorite
url = self._api_url(f"users/{user['id']}/favorites/tools/cat1", params=dict(key=self.master_api_key))
delete_response = delete(url)
self._assert_status_code_is_ok(delete_response)
self.assertEqual(delete_response.json()["tools"], [])
# delete non-existing tool favorite
url = self._api_url(
f"users/{user['id']}/favorites/tools/madeuptoolthatdoes/not/exist/in/favs",
params=dict(key=self.master_api_key),
)
delete_response = delete(url)
self._assert_status_code_is(delete_response, 404)
# delete non existing workflow favorite
url = self._api_url(
f"users/{user['id']}/favorites/workflows/1as5das5das56d465", params=dict(key=self.master_api_key)
)
delete_response = delete(url)
self._assert_status_code_is(delete_response, 400)
[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 = put(url, data=json.dumps({"object_id": "cat1"}))
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 = get(url).json()
assert "cat1" in search_response
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, **new_data):
update_url = self._api_url(f"users/{user['id']}", use_key=True)
return put(update_url, data=new_data)
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"]