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_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"]