opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #11315 from davelopez/prepare_fastapi_route_quotas

Browse Source 
Prepare FastAPI route for quotas
This commit is contained in:
John Chilton
2021-09-08 09:14:49 -04:00
committed by GitHub
parent 35ebd2d79f 9a45c2ad1b
commit 62295efeb5
10 changed files with 971 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/actions/admin.py
-211
View File
@@ -1,211 +0,0 @@
"""
Contains administrative functions
"""
import logging
from galaxy import util
from galaxy.exceptions import ActionInputError
log = logging.getLogger(__name__)
class AdminActions:
"""
Mixin for controllers that provide administrative functionality.
"""
def _create_quota(self, params, decode_id=None):
if params.amount.lower() in ('unlimited', 'none', 'no limit'):
create_amount = None
else:
try:
create_amount = util.size_to_bytes(params.amount)
except AssertionError:
create_amount = False
if not params.name or not params.description:
raise ActionInputError("Enter a valid name and a description.")
elif self.sa_session.query(self.app.model.Quota).filter(self.app.model.Quota.name == params.name).first():
raise ActionInputError("Quota names must be unique and a quota with that name already exists, so choose another name.")
elif not params.get('amount', None):
raise ActionInputError("Enter a valid quota amount.")
elif create_amount is False:
raise ActionInputError("Unable to parse the provided amount.")
elif params.operation not in self.app.model.Quota.valid_operations:
raise ActionInputError("Enter a valid operation.")
elif params.default != 'no' and params.default not in self.app.model.DefaultQuotaAssociation.types.__members__.values():
raise ActionInputError("Enter a valid default type.")
elif params.default != 'no' and params.operation != '=':
raise ActionInputError("Operation for a default quota must be '='.")
elif create_amount is None and params.operation != '=':
raise ActionInputError("Operation for an unlimited quota must be '='.")
else:
# Create the quota
quota = self.app.model.Quota(name=params.name, description=params.description, amount=create_amount, operation=params.operation)
self.sa_session.add(quota)
# If this is a default quota, create the DefaultQuotaAssociation
if params.default != 'no':
self.app.quota_agent.set_default_quota(params.default, quota)
message = f"Default quota '{quota.name}' has been created."
else:
# Create the UserQuotaAssociations
in_users = [self.sa_session.query(self.app.model.User).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_users)]
in_groups = [self.sa_session.query(self.app.model.Group).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_groups)]
if None in in_users:
raise ActionInputError("One or more invalid user id has been provided.")
for user in in_users:
uqa = self.app.model.UserQuotaAssociation(user, quota)
self.sa_session.add(uqa)
# Create the GroupQuotaAssociations
if None in in_groups:
raise ActionInputError("One or more invalid group id has been provided.")
for group in in_groups:
gqa = self.app.model.GroupQuotaAssociation(group, quota)
self.sa_session.add(gqa)
message = "Quota '%s' has been created with %d associated users and %d associated groups." % (quota.name, len(in_users), len(in_groups))
self.sa_session.flush()
return quota, message
def _rename_quota(self, quota, params):
if not params.name:
raise ActionInputError('Enter a valid name.')
elif params.name != quota.name and self.sa_session.query(self.app.model.Quota).filter(self.app.model.Quota.name == params.name).first():
raise ActionInputError('A quota with that name already exists.')
else:
old_name = quota.name
quota.name = params.name
quota.description = params.description
self.sa_session.add(quota)
self.sa_session.flush()
message = f"Quota '{old_name}' has been renamed to '{params.name}'."
return message
def _manage_users_and_groups_for_quota(self, quota, params, decode_id=None):
if quota.default:
raise ActionInputError('Default quotas cannot be associated with specific users and groups.')
else:
in_users = [self.sa_session.query(self.app.model.User).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_users)]
if None in in_users:
raise ActionInputError("One or more invalid user id has been provided.")
in_groups = [self.sa_session.query(self.app.model.Group).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_groups)]
if None in in_groups:
raise ActionInputError("One or more invalid group id has been provided.")
self.app.quota_agent.set_entity_quota_associations(quotas=[quota], users=in_users, groups=in_groups)
self.sa_session.refresh(quota)
message = "Quota '%s' has been updated with %d associated users and %d associated groups." % (quota.name, len(in_users), len(in_groups))
return message
def _edit_quota(self, quota, params):
if params.amount.lower() in ('unlimited', 'none', 'no limit'):
new_amount = None
else:
try:
new_amount = util.size_to_bytes(params.amount)
except (AssertionError, ValueError):
new_amount = False
if not params.amount:
raise ActionInputError('Enter a valid amount.')
elif new_amount is False:
raise ActionInputError('Unable to parse the provided amount.')
elif params.operation not in self.app.model.Quota.valid_operations:
raise ActionInputError('Enter a valid operation.')
else:
quota.amount = new_amount
quota.operation = params.operation
self.sa_session.add(quota)
self.sa_session.flush()
message = f"Quota '{quota.name}' is now '{quota.operation + quota.display_amount}'."
return message
def _set_quota_default(self, quota, params):
if params.default != 'no' and params.default not in self.app.model.DefaultQuotaAssociation.types.__members__.values():
raise ActionInputError('Enter a valid default type.')
else:
if params.default != 'no':
self.app.quota_agent.set_default_quota(params.default, quota)
message = f"Quota '{quota.name}' is now the default for {params.default} users."
else:
if quota.default:
message = f"Quota '{quota.name}' is no longer the default for {quota.default[0].type} users."
for dqa in quota.default:
self.sa_session.delete(dqa)
self.sa_session.flush()
else:
message = f"Quota '{quota.name}' is not a default."
return message
def _unset_quota_default(self, quota, params=None):
if not quota.default:
raise ActionInputError(f"Quota '{quota.name}' is not a default.")
else:
message = f"Quota '{quota.name}' is no longer the default for {quota.default[0].type} users."
for dqa in quota.default:
self.sa_session.delete(dqa)
self.sa_session.flush()
return message
def _delete_quota(self, quota, params=None):
quotas = util.listify(quota)
names = []
for q in quotas:
if q.default:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' is a default, please unset it as a default before deleting it.")
elif len(names) > 1:
raise ActionInputError(f"Quotas are defaults, please unset them as defaults before deleting them: {', '.join(names)}")
message = "Deleted %d quotas: " % len(quotas)
for q in quotas:
q.deleted = True
self.sa_session.add(q)
names.append(q.name)
self.sa_session.flush()
message += ', '.join(names)
return message
def _undelete_quota(self, quota, params=None):
quotas = util.listify(quota)
names = []
for q in quotas:
if not q.deleted:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' has not been deleted, so it cannot be undeleted.")
elif len(names) > 1:
raise ActionInputError(f"Quotas have not been deleted so they cannot be undeleted: {', '.join(names)}")
message = "Undeleted %d quotas: " % len(quotas)
for q in quotas:
q.deleted = False
self.sa_session.add(q)
names.append(q.name)
self.sa_session.flush()
message += ', '.join(names)
return message
def _purge_quota(self, quota, params=None):
"""
This method should only be called for a Quota that has previously been deleted.
Purging a deleted Quota deletes all of the following from the database:
- UserQuotaAssociations where quota_id == Quota.id
- GroupQuotaAssociations where quota_id == Quota.id
"""
quotas = util.listify(quota)
names = []
for q in quotas:
if not q.deleted:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' has not been deleted, so it cannot be purged.")
elif len(names) > 1:
raise ActionInputError(f"Quotas have not been deleted so they cannot be undeleted: {', '.join(names)}")
message = "Purged %d quotas: " % len(quotas)
for q in quotas:
# Delete UserQuotaAssociations
for uqa in q.users:
self.sa_session.delete(uqa)
# Delete GroupQuotaAssociations
for gqa in q.groups:
self.sa_session.delete(gqa)
names.append(q.name)
self.sa_session.flush()
message += ', '.join(names)
return message
lib/galaxy/managers/quotas.py
+365
View File
@@ -0,0 +1,365 @@
"""
Manager and Serializers for Quotas.
For more information about quotas: https://galaxyproject.org/admin/disk-quotas/
"""
import logging
from typing import (
cast,
Optional,
Tuple,
Union,
)
from sqlalchemy import (
false,
true
)
from galaxy import model, util
from galaxy.app import StructuredApp
from galaxy.exceptions import ActionInputError
from galaxy.managers import base
from galaxy.managers.context import ProvidesUserContext
from galaxy.quota import DatabaseQuotaAgent
from galaxy.quota._schema import (
CreateQuotaParams,
CreateQuotaResult,
DefaultQuotaValues,
DeleteQuotaPayload,
QuotaDetails,
QuotaOperation,
QuotaSummaryList,
UpdateQuotaParams,
)
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.web import url_for
log = logging.getLogger(__name__)
class QuotaManager:
"""Interface/service object to interact with Quotas."""
def __init__(self, app: StructuredApp):
self.app = app
@property
def sa_session(self):
return self.app.model.context
@property
def quota_agent(self) -> DatabaseQuotaAgent:
return cast(DatabaseQuotaAgent, self.app.quota_agent)
def create_quota(self, payload: dict, decode_id=None) -> Tuple[model.Quota, str]:
params = CreateQuotaParams.parse_obj(payload)
create_amount = self._parse_amount(params.amount)
if self.sa_session.query(model.Quota).filter(model.Quota.name == params.name).first():
raise ActionInputError("Quota names must be unique and a quota with that name already exists, please choose another name.")
elif create_amount is False:
raise ActionInputError("Unable to parse the provided amount.")
elif params.operation not in model.Quota.valid_operations:
raise ActionInputError("Enter a valid operation.")
elif params.default != DefaultQuotaValues.NO and params.operation != QuotaOperation.EXACT:
raise ActionInputError("Operation for a default quota must be '='.")
elif create_amount is None and params.operation != QuotaOperation.EXACT:
raise ActionInputError("Operation for an unlimited quota must be '='.")
# Create the quota
quota = model.Quota(name=params.name, description=params.description, amount=create_amount, operation=params.operation)
self.sa_session.add(quota)
# If this is a default quota, create the DefaultQuotaAssociation
if params.default != DefaultQuotaValues.NO:
self.quota_agent.set_default_quota(params.default, quota)
message = f"Default quota '{quota.name}' has been created."
else:
# Create the UserQuotaAssociations
in_users = [self.sa_session.query(model.User).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_users)]
in_groups = [self.sa_session.query(model.Group).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_groups)]
if None in in_users:
raise ActionInputError("One or more invalid user id has been provided.")
for user in in_users:
uqa = model.UserQuotaAssociation(user, quota)
self.sa_session.add(uqa)
# Create the GroupQuotaAssociations
if None in in_groups:
raise ActionInputError("One or more invalid group id has been provided.")
for group in in_groups:
gqa = model.GroupQuotaAssociation(group, quota)
self.sa_session.add(gqa)
message = f"Quota '{quota.name}' has been created with {len(in_users)} associated users and {len(in_groups)} associated groups."
self.sa_session.flush()
return quota, message
def _parse_amount(self, amount: str) -> Optional[Union[int, bool]]:
if amount.lower() in ('unlimited', 'none', 'no limit'):
return None
try:
return util.size_to_bytes(amount)
except AssertionError:
return False
def rename_quota(self, quota, params) -> str:
if not params.name:
raise ActionInputError('Enter a valid name.')
elif params.name != quota.name and self.sa_session.query(model.Quota).filter(model.Quota.name == params.name).first():
raise ActionInputError('A quota with that name already exists.')
else:
old_name = quota.name
quota.name = params.name
if params.description:
quota.description = params.description
self.sa_session.add(quota)
self.sa_session.flush()
message = f"Quota '{old_name}' has been renamed to '{params.name}'."
return message
def manage_users_and_groups_for_quota(self, quota, params, decode_id=None) -> str:
if quota.default:
raise ActionInputError('Default quotas cannot be associated with specific users and groups.')
else:
in_users = [self.sa_session.query(model.User).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_users)]
if None in in_users:
raise ActionInputError("One or more invalid user id has been provided.")
in_groups = [self.sa_session.query(model.Group).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_groups)]
if None in in_groups:
raise ActionInputError("One or more invalid group id has been provided.")
self.quota_agent.set_entity_quota_associations(quotas=[quota], users=in_users, groups=in_groups)
self.sa_session.refresh(quota)
message = f"Quota '{quota.name}' has been updated with {len(in_users)} associated users and {len(in_groups)} associated groups."
return message
def edit_quota(self, quota, params) -> str:
if params.amount.lower() in ('unlimited', 'none', 'no limit'):
new_amount = None
else:
try:
new_amount = util.size_to_bytes(params.amount)
except (AssertionError, ValueError):
new_amount = False
if not params.amount:
raise ActionInputError('Enter a valid amount.')
elif new_amount is False:
raise ActionInputError('Unable to parse the provided amount.')
elif params.operation not in model.Quota.valid_operations:
raise ActionInputError('Enter a valid operation.')
else:
quota.amount = new_amount
quota.operation = params.operation
self.sa_session.add(quota)
self.sa_session.flush()
message = f"Quota '{quota.name}' is now '{quota.operation}{quota.display_amount}'."
return message
def set_quota_default(self, quota, params) -> str:
if params.default != 'no' and params.default not in model.DefaultQuotaAssociation.types.__members__.values():
raise ActionInputError('Enter a valid default type.')
else:
if params.default != 'no':
self.quota_agent.set_default_quota(params.default, quota)
message = f"Quota '{quota.name}' is now the default for {params.default} users."
else:
if quota.default:
message = f"Quota '{quota.name}' is no longer the default for {quota.default[0].type} users."
for dqa in quota.default:
self.sa_session.delete(dqa)
self.sa_session.flush()
else:
message = f"Quota '{quota.name}' is not a default."
return message
def unset_quota_default(self, quota, params=None) -> str:
if not quota.default:
raise ActionInputError(f"Quota '{quota.name}' is not a default.")
else:
message = f"Quota '{quota.name}' is no longer the default for {quota.default[0].type} users."
for dqa in quota.default:
self.sa_session.delete(dqa)
self.sa_session.flush()
return message
def delete_quota(self, quota, params=None) -> str:
quotas = util.listify(quota)
names = []
for q in quotas:
if q.default:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' is a default, please unset it as a default before deleting it.")
elif len(names) > 1:
raise ActionInputError(f"Quotas are defaults, please unset them as defaults before deleting them: {', '.join(names)}")
message = f"Deleted {len(quotas)} quotas: "
for q in quotas:
q.deleted = True
self.sa_session.add(q)
names.append(q.name)
self.sa_session.flush()
message += ', '.join(names)
return message
def undelete_quota(self, quota, params=None) -> str:
quotas = util.listify(quota)
names = []
for q in quotas:
if not q.deleted:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' has not been deleted, so it cannot be undeleted.")
elif len(names) > 1:
raise ActionInputError(f"Quotas have not been deleted so they cannot be undeleted: {', '.join(names)}")
message = f"Undeleted {len(quotas)} quotas: "
for q in quotas:
q.deleted = False
self.sa_session.add(q)
names.append(q.name)
self.sa_session.flush()
message += ', '.join(names)
return message
def purge_quota(self, quota, params=None):
"""
This method should only be called for a Quota that has previously been deleted.
Purging a deleted Quota deletes all of the following from the database:
- UserQuotaAssociations where quota_id == Quota.id
- GroupQuotaAssociations where quota_id == Quota.id
"""
quotas = util.listify(quota)
names = []
for q in quotas:
if not q.deleted:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' has not been deleted, so it cannot be purged.")
elif len(names) > 1:
raise ActionInputError(f"Quotas have not been deleted so they cannot be undeleted: {', '.join(names)}")
message = f"Purged {len(quotas)} quotas: "
for q in quotas:
# Delete UserQuotaAssociations
for uqa in q.users:
self.sa_session.delete(uqa)
# Delete GroupQuotaAssociations
for gqa in q.groups:
self.sa_session.delete(gqa)
names.append(q.name)
self.sa_session.flush()
message += ', '.join(names)
return message
def get_quota(self, trans, id: EncodedDatabaseIdField, deleted: Optional[bool] = None) -> model.Quota:
return base.get_object(trans, id, 'Quota', check_ownership=False, check_accessible=False, deleted=deleted)
class QuotasService:
"""Interface/service object shared by controllers for interacting with quotas."""
def __init__(self, app: StructuredApp):
self.quota_manager: QuotaManager = QuotaManager(app)
def index(self, trans: ProvidesUserContext, deleted: bool = False) -> QuotaSummaryList:
"""Displays a collection (list) of quotas."""
rval = []
query = trans.sa_session.query(model.Quota)
if deleted:
route = 'deleted_quota'
query = query.filter(model.Quota.deleted == true())
else:
route = 'quota'
query = query.filter(model.Quota.deleted == false())
for quota in query:
item = quota.to_dict(value_mapper={'id': trans.security.encode_id})
encoded_id = trans.security.encode_id(quota.id)
item['url'] = self._url_for(route, id=encoded_id)
rval.append(item)
return QuotaSummaryList.parse_obj(rval)
def show(self, trans: ProvidesUserContext, id: EncodedDatabaseIdField, deleted: bool = False) -> QuotaDetails:
"""Displays information about a quota."""
quota = self.quota_manager.get_quota(trans, id, deleted=deleted)
rval = quota.to_dict(view='element', value_mapper={'id': trans.security.encode_id, 'total_disk_usage': float})
return QuotaDetails.parse_obj(rval)
def create(self, trans: ProvidesUserContext, params: CreateQuotaParams) -> CreateQuotaResult:
"""Creates a new quota."""
payload = params.dict()
self.validate_in_users_and_groups(trans, payload)
quota, message = self.quota_manager.create_quota(payload)
item = quota.to_dict(value_mapper={'id': trans.security.encode_id})
item['url'] = self._url_for('quota', id=trans.security.encode_id(quota.id))
item['message'] = message
return CreateQuotaResult.parse_obj(item)
def update(self, trans: ProvidesUserContext, id: EncodedDatabaseIdField, params: UpdateQuotaParams) -> str:
"""Modifies a quota."""
payload = params.dict()
self.validate_in_users_and_groups(trans, payload)
quota = self.quota_manager.get_quota(trans, id, deleted=False)
params = UpdateQuotaParams(**payload)
# FIXME: Doing it this way makes the update non-atomic if a method fails after an earlier one has succeeded.
methods = []
if params.name or params.description:
methods.append(self.quota_manager.rename_quota)
if params.amount:
methods.append(self.quota_manager.edit_quota)
if params.default == DefaultQuotaValues.NO:
methods.append(self.quota_manager.unset_quota_default)
elif params.default:
methods.append(self.quota_manager.set_quota_default)
if params.in_users or params.in_groups:
methods.append(self.quota_manager.manage_users_and_groups_for_quota)
messages = []
for method in methods:
message = method(quota, params)
messages.append(message)
return '; '.join(messages)
def delete(self, trans: ProvidesUserContext, id: EncodedDatabaseIdField, payload: Optional[DeleteQuotaPayload] = None) -> str:
"""Marks a quota as deleted."""
quota = self.quota_manager.get_quota(trans, id, deleted=False) # deleted quotas are not technically members of this collection
message = self.quota_manager.delete_quota(quota)
if payload and payload.purge:
message += self.quota_manager.purge_quota(quota)
return message
def undelete(self, trans: ProvidesUserContext, id: EncodedDatabaseIdField) -> str:
"""Restores a previously deleted quota."""
quota = self.quota_manager.get_quota(trans, id, deleted=True)
return self.quota_manager.undelete_quota(quota)
def validate_in_users_and_groups(self, trans, payload):
"""
For convenience, in_users and in_groups can be encoded IDs or emails/group names in the API.
"""
def get_id(item, model_class, column):
try:
return trans.security.decode_id(item)
except Exception:
pass # maybe an email/group name
# this will raise if the item is invalid
return trans.sa_session.query(model_class).filter(column == item).first().id
new_in_users = []
new_in_groups = []
invalid = []
for item in util.listify(payload.get('in_users', [])):
try:
new_in_users.append(get_id(item, model.User, model.User.email))
except Exception:
invalid.append(item)
for item in util.listify(payload.get('in_groups', [])):
try:
new_in_groups.append(get_id(item, model.Group, model.Group.name))
except Exception:
invalid.append(item)
if invalid:
msg = f"The following value(s) for associated users and/or groups could not be parsed: {', '.join(invalid)}."
msg += " Valid values are email addresses of users, names of groups, or IDs of both."
raise Exception(msg)
payload['in_users'] = list(map(str, new_in_users))
payload['in_groups'] = list(map(str, new_in_groups))
def _url_for(self, *args, **kargs):
try:
return url_for(*args, **kargs)
except AttributeError:
return "*deprecated attribute not filled in by FastAPI server*"
lib/galaxy/quota/_schema.py
+247
View File
@@ -0,0 +1,247 @@
from enum import Enum
from typing import (
List,
Optional,
)
from pydantic import (
BaseModel,
Field,
)
from galaxy.schema.fields import (
EncodedDatabaseIdField,
ModelClassField,
)
from galaxy.schema.schema import (
GroupModel,
UserModel,
)
QUOTA_MODEL_CLASS_NAME = "Quota"
USER_QUOTA_ASSOCIATION_MODEL_CLASS_NAME = "UserQuotaAssociation"
GROUP_QUOTA_ASSOCIATION_MODEL_CLASS_NAME = "GroupQuotaAssociation"
DEFAULT_QUOTA_ASSOCIATION_MODEL_CLASS_NAME = "DefaultQuotaAssociation"
class QuotaOperation(str, Enum):
EXACT = "="
ADD = "+"
SUBTRACT = "-"
class DefaultQuotaTypes(str, Enum): # TODO: should this replace lib.galaxy.model.DefaultQuotaAssociation.types at some point?
UNREGISTERED = 'unregistered'
REGISTERED = 'registered'
class DefaultQuotaValues(str, Enum):
UNREGISTERED = 'unregistered'
REGISTERED = 'registered'
NO = "no"
QuotaNameField = Field(
...,
title="Name",
description="The name of the quota. This must be unique within a Galaxy instance.",
)
QuotaDescriptionField = Field(
...,
title="Description",
description="Detailed text description for this Quota.",
)
QuotaOperationField = Field(
QuotaOperation.EXACT,
title="Operation",
description=(
"Quotas can have one of three `operations`:"
"- `=` : The quota is exactly the amount specified"
"- `+` : The amount specified will be added to the amounts of the user's other associated quota definitions"
"- `-` : The amount specified will be subtracted from the amounts of the user's other associated quota definitions"
)
)
class DefaultQuota(BaseModel): # TODO: should this replace lib.galaxy.model.DefaultQuotaAssociation at some point?
model_class: str = ModelClassField(DEFAULT_QUOTA_ASSOCIATION_MODEL_CLASS_NAME)
type: DefaultQuotaTypes = Field(
...,
title="Type",
description=(
"The type of the default quota. Either one of:\n"
" - `registered`: the associated quota will affect registered users.\n"
" - `unregistered`: the associated quota will affect unregistered users.\n"
)
)
class UserQuota(BaseModel):
model_class: str = ModelClassField(USER_QUOTA_ASSOCIATION_MODEL_CLASS_NAME)
user: UserModel = Field(
...,
title="User",
description="Information about a user associated with a quota.",
)
class GroupQuota(BaseModel):
model_class: str = ModelClassField(GROUP_QUOTA_ASSOCIATION_MODEL_CLASS_NAME)
group: GroupModel = Field(
...,
title="Group",
description="Information about a user group associated with a quota.",
)
class QuotaBase(BaseModel):
"""Base model containing common fields for Quotas."""
model_class: str = ModelClassField(QUOTA_MODEL_CLASS_NAME)
id: EncodedDatabaseIdField = Field(
...,
title="ID",
description="The `encoded identifier` of the quota.",
)
name: str = QuotaNameField
class QuotaSummary(QuotaBase):
"""Contains basic information about a Quota"""
url: str = Field(
...,
title="URL",
description="The relative URL to get this particular Quota details from the rest API.",
deprecated=True,
)
class QuotaSummaryList(BaseModel):
__root__: List[QuotaSummary] = Field(
default=[],
title='List with summary information of Quotas.',
)
class QuotaDetails(QuotaBase):
description: str = QuotaDescriptionField
bytes: str = Field(
...,
title="Bytes",
description="The amount, expressed in bytes, of this Quota.",
)
operation: QuotaOperation = QuotaOperationField
display_amount: str = Field(
...,
title="Display Amount",
description="Human-readable representation of the `amount` field.",
)
default: List[DefaultQuota] = Field(
[],
title="Default",
description="A list indicating which types of default user quotas, if any, are associated with this quota.",
)
users: List[UserQuota] = Field(
[],
title="Users",
description="A list of specific users associated with this quota.",
)
groups: List[GroupQuota] = Field(
[],
title="Groups",
description="A list of specific groups of users associated with this quota.",
)
class CreateQuotaResult(QuotaSummary):
message: str = Field(
...,
title="Message",
description="Text message describing the result of the operation.",
)
class CreateQuotaParams(BaseModel):
name: str = QuotaNameField
description: str = QuotaDescriptionField
amount: str = Field(
...,
title="Amount",
description="Quota size (E.g. ``10000MB``, ``99 gb``, ``0.2T``, ``unlimited``)",
)
operation: QuotaOperation = QuotaOperationField
default: DefaultQuotaValues = Field(
default=DefaultQuotaValues.NO,
title="Default",
description=(
"Whether or not this is a default quota. Valid values"
" are ``no``, ``unregistered``, ``registered``. None is"
" equivalent to ``no``."
),
)
in_users: Optional[List[str]] = Field(
default=[],
title="Users",
description="A list of user IDs or user emails to associate with this quota.",
)
in_groups: Optional[List[str]] = Field(
default=[],
title="Groups",
description="A list of group IDs or names to associate with this quota.",
)
class UpdateQuotaParams(BaseModel):
name: Optional[str] = Field(
default=None,
title="Name",
description="The new name of the quota. This must be unique within a Galaxy instance.",
)
description: Optional[str] = Field(
None,
title="Description",
description="Detailed text description for this Quota.",
)
amount: Optional[str] = Field(
None,
title="Amount",
description="Quota size (E.g. ``10000MB``, ``99 gb``, ``0.2T``, ``unlimited``)",
)
operation: QuotaOperation = Field(
QuotaOperation.EXACT,
title="Operation",
description=(
"One of (``+``, ``-``, ``=``). If you wish to change this value,"
" you must also provide the ``amount``, otherwise it will not take effect."
)
)
default: Optional[DefaultQuotaValues] = Field(
default=None,
title="Default",
description=(
"Whether or not this is a default quota. Valid values"
" are ``no``, ``unregistered``, ``registered``."
' Calling this method with ``default="no"`` on a'
" non-default quota will throw an error. Not"
" passing this parameter is equivalent to passing ``no``."
),
)
in_users: Optional[List[str]] = Field(
default=None,
title="Users",
description="A list of user IDs or user emails to associate with this quota.",
)
in_groups: Optional[List[str]] = Field(
default=None,
title="Groups",
description="A list of group IDs or names to associate with this quota.",
)
class DeleteQuotaPayload(BaseModel):
purge: bool = Field(
False,
title="Purge",
description="Whether to also purge the Quota after deleting it.",
)
lib/galaxy/schema/schema.py
+16
View File
@@ -36,6 +36,7 @@ from galaxy.schema.fields import (
from galaxy.schema.types import RelativeUrl
USER_MODEL_CLASS_NAME = "User"
GROUP_MODEL_CLASS_NAME = "Group"
HDA_MODEL_CLASS_NAME = "HistoryDatasetAssociation"
DC_MODEL_CLASS_NAME = "DatasetCollection"
DCE_MODEL_CLASS_NAME = "DatasetCollectionElement"
@@ -180,6 +181,21 @@ class UserModel(Model):
model_class: str = ModelClassField(USER_MODEL_CLASS_NAME)
class GroupModel(BaseModel):
"""User group model"""
model_class: str = ModelClassField(GROUP_MODEL_CLASS_NAME)
id: EncodedDatabaseIdField = Field(
..., # Required
title='ID',
description='Encoded group ID',
)
name: str = Field(
..., # Required
title="Name",
description="The name of the group.",
)
class JobSourceType(str, Enum):
"""Available types of job sources (model classes) that produce dataset collections."""
Job = "Job"
lib/galaxy/web/params.py
-31
View File
@@ -1,31 +0,0 @@
"""
Mixins for parsing web form and API parameters
"""
from galaxy import util
class BaseParamParser:
def get_params(self, kwargs):
params = util.Params(kwargs)
# set defaults if unset
updates = dict(webapp=params.get('webapp', 'galaxy'),
message=util.restore_text(params.get('message', '')),
status=util.restore_text(params.get('status', 'done')))
params.update(updates)
return params
class QuotaParamParser(BaseParamParser):
def get_quota_params(self, kwargs):
params = self.get_params(kwargs)
updates = dict(name=util.restore_text(params.get('name', '')),
description=util.restore_text(params.get('description', '')),
amount=util.restore_text(params.get('amount', '').strip()),
operation=params.get('operation', ''),
default=params.get('default', ''),
in_users=util.listify(params.get('in_users', [])),
out_users=util.listify(params.get('out_users', [])),
in_groups=util.listify(params.get('in_groups', [])),
out_groups=util.listify(params.get('out_groups', [])))
params.update(updates)
return params
lib/galaxy/webapps/base/controller.py
-37
View File
@@ -191,37 +191,6 @@ class BaseAPIController(BaseController):
log.exception("Exception in get_object check for %s %s.", class_name, str(id))
raise HTTPInternalServerError(comment=util.unicodify(e))
def validate_in_users_and_groups(self, trans, payload):
"""
For convenience, in_users and in_groups can be encoded IDs or emails/group names in the API.
"""
def get_id(item, model_class, column):
try:
return trans.security.decode_id(item)
except Exception:
pass # maybe an email/group name
# this will raise if the item is invalid
return trans.sa_session.query(model_class).filter(column == item).first().id
new_in_users = []
new_in_groups = []
invalid = []
for item in util.listify(payload.get('in_users', [])):
try:
new_in_users.append(get_id(item, trans.app.model.User, trans.app.model.User.table.c.email))
except Exception:
invalid.append(item)
for item in util.listify(payload.get('in_groups', [])):
try:
new_in_groups.append(get_id(item, trans.app.model.Group, trans.app.model.Group.name))
except Exception:
invalid.append(item)
if invalid:
msg = f"The following value(s) for associated users and/or groups could not be parsed: {', '.join(invalid)}."
msg += " Valid values are email addresses of users, names of groups, or IDs of both."
raise Exception(msg)
payload['in_users'] = list(map(str, new_in_users))
payload['in_groups'] = list(map(str, new_in_groups))
def not_implemented(self, trans, **kwd):
raise HTTPNotImplemented()
@@ -1346,12 +1315,6 @@ class SharableMixin:
raise NotImplementedError()
class UsesQuotaMixin:
def get_quota(self, trans, id, check_ownership=False, check_accessible=False, deleted=None):
return self.get_object(trans, id, 'Quota', check_ownership=False, check_accessible=False, deleted=deleted)
class UsesTagsMixin(SharableItemSecurityMixin):
def get_tag_handler(self, trans) -> tags.GalaxyTagHandler:
lib/galaxy/webapps/galaxy/api/quotas.py
+157 -97
View File
@@ -3,155 +3,215 @@ API operations on Quota objects.
"""
import logging
from paste.httpexceptions import HTTPBadRequest
from sqlalchemy import (
false,
true
)
from fastapi import Path
from fastapi.param_functions import Body
from galaxy import (
util,
web
web,
)
from galaxy.actions.admin import AdminActions
from galaxy.exceptions import ActionInputError
from galaxy.web.params import QuotaParamParser
from galaxy.webapps.base.controller import (
url_for,
UsesQuotaMixin
from galaxy.managers.context import ProvidesUserContext
from galaxy.managers.quotas import QuotasService
from galaxy.quota._schema import (
CreateQuotaParams,
CreateQuotaResult,
DeleteQuotaPayload,
QuotaDetails,
QuotaSummaryList,
UpdateQuotaParams,
)
from galaxy.schema.fields import EncodedDatabaseIdField
from . import (
BaseGalaxyAPIController,
depends,
DependsOnTrans,
Router,
)
from . import BaseGalaxyAPIController
log = logging.getLogger(__name__)
class QuotaAPIController(BaseGalaxyAPIController, AdminActions, UsesQuotaMixin, QuotaParamParser):
router = Router(tags=['quotas'])
QuotaIdPathParam: EncodedDatabaseIdField = Path(
..., # Required
title="Quota ID",
description="The encoded identifier of the Quota."
)
@router.cbv
class FastAPIQuota:
service: QuotasService = depends(QuotasService)
@router.get(
'/api/quotas',
summary="Displays a list with information of quotas that are currently active.",
require_admin=True,
)
def index(
self,
trans: ProvidesUserContext = DependsOnTrans,
) -> QuotaSummaryList:
"""Displays a list with information of quotas that are currently active."""
return self.service.index(trans)
@router.get(
'/api/quotas/deleted',
summary="Displays a list with information of quotas that have been deleted.",
require_admin=True,
)
def index_deleted(
self,
trans: ProvidesUserContext = DependsOnTrans,
) -> QuotaSummaryList:
"""Displays a list with information of quotas that have been deleted."""
return self.service.index(trans, deleted=True)
@router.get(
'/api/quotas/{id}',
summary="Displays details on a particular active quota.",
require_admin=True,
)
def show(
self,
trans: ProvidesUserContext = DependsOnTrans,
id: EncodedDatabaseIdField = QuotaIdPathParam
) -> QuotaDetails:
"""Displays details on a particular active quota."""
return self.service.show(trans, id)
@router.get(
'/api/quotas/deleted/{id}',
summary="Displays details on a particular quota that has been deleted.",
require_admin=True,
)
def show_deleted(
self,
trans: ProvidesUserContext = DependsOnTrans,
id: EncodedDatabaseIdField = QuotaIdPathParam,
) -> QuotaDetails:
"""Displays details on a particular quota that has been deleted."""
return self.service.show(trans, id, deleted=True)
@router.post(
'/api/quotas',
summary="Creates a new quota.",
require_admin=True,
)
def create(
self,
payload: CreateQuotaParams,
trans: ProvidesUserContext = DependsOnTrans,
) -> CreateQuotaResult:
"""Creates a new quota."""
return self.service.create(trans, payload)
@router.put(
'/api/quotas/{id}',
summary="Updates an existing quota.",
require_admin=True,
)
def update(
self,
payload: UpdateQuotaParams,
id: EncodedDatabaseIdField = QuotaIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> str:
"""Updates an existing quota."""
return self.service.update(trans, id, payload)
@router.delete(
'/api/quotas/{id}',
summary="Deletes an existing quota.",
require_admin=True,
)
def delete(
self,
id: EncodedDatabaseIdField = QuotaIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
payload: DeleteQuotaPayload = Body(None), # Optional
) -> str:
"""Deletes an existing quota."""
return self.service.delete(trans, id, payload)
@router.post(
'/api/quotas/deleted/{id}/undelete',
summary="Restores a previously deleted quota.",
require_admin=True,
)
def undelete(
self,
id: EncodedDatabaseIdField = QuotaIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> str:
"""Restores a previously deleted quota."""
return self.service.undelete(trans, id)
class QuotaAPIController(BaseGalaxyAPIController):
service: QuotasService = depends(QuotasService)
@web.require_admin
@web.legacy_expose_api
@web.expose_api
def index(self, trans, deleted='False', **kwd):
"""
GET /api/quotas
GET /api/quotas/deleted
Displays a collection (list) of quotas.
"""
rval = []
deleted = util.string_as_bool(deleted)
query = trans.sa_session.query(trans.app.model.Quota)
if deleted:
route = 'deleted_quota'
query = query.filter(trans.app.model.Quota.deleted == true())
else:
route = 'quota'
query = query.filter(trans.app.model.Quota.deleted == false())
for quota in query:
item = quota.to_dict(value_mapper={'id': trans.security.encode_id})
encoded_id = trans.security.encode_id(quota.id)
item['url'] = url_for(route, id=encoded_id)
rval.append(item)
return rval
return self.service.index(trans, deleted)
@web.require_admin
@web.legacy_expose_api
@web.expose_api
def show(self, trans, id, deleted='False', **kwd):
"""
GET /api/quotas/{encoded_quota_id}
GET /api/quotas/deleted/{encoded_quota_id}
Displays information about a quota.
"""
quota = self.get_quota(trans, id, deleted=util.string_as_bool(deleted))
return quota.to_dict(view='element', value_mapper={'id': trans.security.encode_id, 'total_disk_usage': float})
deleted = util.string_as_bool(deleted)
return self.service.show(trans, id, deleted)
@web.require_admin
@web.legacy_expose_api
@web.expose_api
def create(self, trans, payload, **kwd):
"""
POST /api/quotas
Creates a new quota.
"""
try:
self.validate_in_users_and_groups(trans, payload)
except Exception as e:
raise HTTPBadRequest(detail=util.unicodify(e))
params = self.get_quota_params(payload)
try:
quota, message = self._create_quota(params)
except ActionInputError as e:
raise HTTPBadRequest(detail=util.unicodify(e))
item = quota.to_dict(value_mapper={'id': trans.security.encode_id})
item['url'] = url_for('quota', id=trans.security.encode_id(quota.id))
item['message'] = message
return item
params = CreateQuotaParams(**payload)
return self.service.create(trans, params)
@web.require_admin
@web.legacy_expose_api
@web.expose_api
def update(self, trans, id, payload, **kwd):
"""
PUT /api/quotas/{encoded_quota_id}
Modifies a quota.
"""
try:
self.validate_in_users_and_groups(trans, payload)
except Exception as e:
raise HTTPBadRequest(detail=util.unicodify(e))
quota = self.get_quota(trans, id, deleted=False)
# FIXME: Doing it this way makes the update non-atomic if a method fails after an earlier one has succeeded.
payload['id'] = id
params = self.get_quota_params(payload)
methods = []
if payload.get('name', None) or payload.get('description', None):
methods.append(self._rename_quota)
if payload.get('amount', None):
methods.append(self._edit_quota)
if payload.get('default', None) == 'no':
methods.append(self._unset_quota_default)
elif payload.get('default', None):
methods.append(self._set_quota_default)
if payload.get('in_users', None) or payload.get('in_groups', None):
methods.append(self._manage_users_and_groups_for_quota)
messages = []
for method in methods:
try:
message = method(quota, params)
except ActionInputError as e:
raise HTTPBadRequest(detail=util.unicodify(e))
messages.append(message)
return '; '.join(messages)
params = UpdateQuotaParams(**payload)
return self.service.update(trans, id, params)
@web.require_admin
@web.legacy_expose_api
@web.expose_api
def delete(self, trans, id, **kwd):
"""
DELETE /api/quotas/{encoded_quota_id}
Deletes a quota
"""
quota = self.get_quota(trans, id, deleted=False) # deleted quotas are not technically members of this collection
# a request body is optional here
payload = kwd.get('payload', {})
payload['id'] = id
params = self.get_quota_params(payload)
try:
message = self._delete_quota(quota, params)
if util.string_as_bool(payload.get('purge', False)):
message += self._purge_quota(quota, params)
except ActionInputError as e:
raise HTTPBadRequest(detail=util.unicodify(e))
return message
payload = DeleteQuotaPayload(**kwd.get('payload', {}))
return self.service.delete(trans, id, payload)
@web.require_admin
@web.legacy_expose_api
@web.expose_api
def undelete(self, trans, id, **kwd):
"""
POST /api/quotas/deleted/{encoded_quota_id}/undelete
Undeletes a quota
"""
quota = self.get_quota(trans, id, deleted=True)
try:
return self._undelete_quota(quota)
except ActionInputError as e:
raise HTTPBadRequest(detail=util.unicodify(e))
return self.service.undelete(trans, id)
lib/galaxy/webapps/galaxy/controllers/admin.py
+15 -13
View File
@@ -9,8 +9,8 @@ from galaxy import (
util,
web
)
from galaxy.actions.admin import AdminActions
from galaxy.exceptions import ActionInputError, MessageException
from galaxy.managers.quotas import QuotaManager
from galaxy.model import tool_shed_install as install_model
from galaxy.security.validate_user_input import validate_password
from galaxy.tool_shed.util.repository_util import get_ids_of_tool_shed_repositories_being_installed
@@ -22,9 +22,7 @@ from galaxy.util import (
from galaxy.util.tool_shed import common_util, encoding_util
from galaxy.web import url_for
from galaxy.web.framework.helpers import grids, time_ago
from galaxy.web.params import QuotaParamParser
from galaxy.webapps.base import controller
from galaxy.webapps.base.controller import UsesQuotaMixin
from tool_shed.util.web_util import escape
@@ -503,7 +501,7 @@ class ToolVersionListGrid(grids.Grid):
return trans.install_model.context.query(self.model_class)
class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaParamParser):
class AdminGalaxy(controller.JSAppLauncher):
user_list_grid = UserListGrid()
role_list_grid = RoleListGrid()
@@ -522,6 +520,10 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
activate_operation = grids.GridOperation("Activate User", condition=(lambda item: not item.active), allow_multiple=False)
resend_activation_email = grids.GridOperation("Resend Activation Email", condition=(lambda item: not item.active), allow_multiple=False)
def __init__(self, app):
super().__init__(app)
self.quota_manager: QuotaManager = QuotaManager(app)
@web.expose
@web.require_admin
def index(self, trans, **kwd):
@@ -646,13 +648,13 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
operation = kwargs.pop('operation').lower()
try:
if operation == 'delete':
message = self._delete_quota(quotas)
message = self.quota_manager.delete_quota(quotas)
elif operation == 'undelete':
message = self._undelete_quota(quotas)
message = self.quota_manager.undelete_quota(quotas)
elif operation == 'purge':
message = self._purge_quota(quotas)
message = self.quota_manager.purge_quota(quotas)
elif operation == 'unset as default':
message = self._unset_quota_default(quotas[0])
message = self.quota_manager.unset_quota_default(quotas[0])
except ActionInputError as e:
message, status = (e.err_msg, 'error')
if message:
@@ -703,7 +705,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
build_select_input('in_users', 'Users', all_users, [])]}
else:
try:
quota, message = self._create_quota(util.Params(payload), decode_id=trans.security.decode_id)
quota, message = self.quota_manager.create_quota(payload, decode_id=trans.security.decode_id)
return {'message': message}
except ActionInputError as e:
return self.message_exception(trans, e.err_msg)
@@ -730,7 +732,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
}
else:
try:
return {'message': self._rename_quota(quota, util.Params(payload))}
return {'message': self.quota_manager.rename_quota(quota, util.Params(payload))}
except ActionInputError as e:
return self.message_exception(trans, e.err_msg)
@@ -766,7 +768,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
build_select_input('in_users', 'Users', all_users, in_users)]}
else:
try:
return {'message': self._manage_users_and_groups_for_quota(quota, util.Params(payload), decode_id=trans.security.decode_id)}
return {'message': self.quota_manager.manage_users_and_groups_for_quota(quota, util.Params(payload), decode_id=trans.security.decode_id)}
except ActionInputError as e:
return self.message_exception(trans, e.err_msg)
@@ -794,7 +796,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
}
else:
try:
return {'message': self._edit_quota(quota, util.Params(payload))}
return {'message': self.quota_manager.edit_quota(quota, util.Params(payload))}
except ActionInputError as e:
return self.message_exception(trans, e.err_msg)
@@ -822,7 +824,7 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
}
else:
try:
return {'message': self._set_quota_default(quota, util.Params(payload))}
return {'message': self.quota_manager.set_quota_default(quota, util.Params(payload))}
except ActionInputError as e:
return self.message_exception(trans, e.err_msg)
packages/data/galaxy/schema
Symlink
+1
View File
@@ -0,0 +1 @@
../../../lib/galaxy/schema
test/integration/test_quota.py
+170 -10
View File
@@ -1,3 +1,8 @@
import json
from requests import delete
from galaxy_test.base.populators import (
DatasetPopulator,
)
@@ -15,18 +20,173 @@ class QuotaIntegrationTestCase(integration_util.IntegrationTestCase):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
def test_quota_crud(self):
quotas = self.dataset_populator.get_quotas()
assert len(quotas) == 0
def test_create(self):
self._create_quota_with_name("test-create-quota")
payload = {
'name': 'defaultquota1',
'description': 'first default quota',
def test_index(self):
self._create_quota_with_name("test-index-quota")
index_response = self._get("quotas")
index_response.raise_for_status()
json_response = index_response.json()
assert len(json_response) > 0
def test_index_deleted(self):
quota = self._create_quota_with_name("test-index-deleted-quota")
quota_id = quota["id"]
delete_response = self._delete(f"quotas/{quota_id}")
delete_response.raise_for_status()
index_response = self._get("quotas/deleted")
index_response.raise_for_status()
json_response = index_response.json()
assert len(json_response) > 0
def test_show(self):
quota_name = "test-show-quota"
quota = self._create_quota_with_name(quota_name)
quota_id = quota["id"]
show_response = self._get(f"quotas/{quota_id}")
show_response.raise_for_status()
json_response = show_response.json()
assert json_response["name"] == quota["name"]
def test_show_deleted(self):
quota_name = "test-show-deleted-quota"
quota = self._create_quota_with_name(quota_name)
quota_id = quota["id"]
delete_response = self._delete(f"quotas/{quota_id}")
delete_response.raise_for_status()
show_response = self._get(f"quotas/deleted/{quota_id}")
show_response.raise_for_status()
json_response = show_response.json()
assert json_response["name"] == quota["name"]
def test_update(self):
quota_name = "test-update-quota"
quota = self._create_quota_with_name(quota_name)
quota_id = quota["id"]
new_quota_name = "updated-quota-name"
update_payload = json.dumps({
'name': new_quota_name,
})
put_response = self._put(f"quotas/{quota_id}", data=update_payload)
put_response.raise_for_status()
assert "has been renamed to" in put_response.text
show_response = self._get(f"quotas/{quota_id}")
show_response.raise_for_status()
json_response = show_response.json()
assert json_response["name"] == new_quota_name
def test_delete(self):
quota_name = "test-delete-quota"
quota = self._create_quota_with_name(quota_name)
quota_id = quota["id"]
delete_response = self._delete(f"quotas/{quota_id}")
delete_response.raise_for_status()
self._assert_quota_is_deleted(quota_id)
def test_delete_and_purge(self):
quota_name = "test-delete-purge-quota"
quota = self._create_quota_with_name(quota_name)
quota_id = quota["id"]
delete_response = self._delete_and_purge(f"quotas/{quota_id}")
delete_response.raise_for_status()
self._assert_quota_is_deleted(quota_id)
def test_delete_and_purge_with_user(self):
user_email = "test@galaxy.test"
self.galaxy_interactor.ensure_user_with_email(user_email)
quota_name = "test-delete-purge-quota-user"
payload = self._build_quota_payload_with_name(quota_name)
payload["in_users"].append(user_email)
create_response = self._post("quotas", data=payload, json=True)
create_response.raise_for_status()
quota = create_response.json()
quota_id = quota["id"]
show_response = self._get(f"quotas/{quota_id}")
show_response.raise_for_status()
json_response = show_response.json()
assert user_email in str(json_response["users"])
delete_response = self._delete_and_purge(f"quotas/{quota_id}")
delete_response.raise_for_status()
show_response = self._get(f"quotas/deleted/{quota_id}")
show_response.raise_for_status()
json_response = show_response.json()
assert user_email not in str(json_response["users"])
def test_undelete(self):
quota_name = "test-undelete-quota"
quota = self._create_quota_with_name(quota_name)
quota_id = quota["id"]
delete_response = self._delete(f"quotas/{quota_id}")
delete_response.raise_for_status()
self._assert_quota_is_deleted(quota_id)
undelete_response = self._post(f"quotas/deleted/{quota_id}/undelete")
undelete_response.raise_for_status()
show_response = self._get(f"quotas/{quota_id}")
show_response.raise_for_status()
show_response = self._get(f"quotas/deleted/{quota_id}")
self._assert_status_code_is(show_response, 400)
def test_400_when_delete_default(self):
quota_name = "test-delete-default-quota"
quota = self._create_quota_with_name(quota_name, is_default=True)
quota_id = quota["id"]
delete_response = self._delete(f"quotas/{quota_id}")
self._assert_status_code_is(delete_response, 400)
def test_400_when_quota_name_already_exists(self):
quota_name = "test-duplicated-quota"
self._create_quota_with_name(quota_name)
payload = self._build_quota_payload_with_name(quota_name)
create_response = self._post("quotas", data=payload)
self._assert_status_code_is(create_response, 400)
def test_400_when_show_unknown_quota(self):
quota_id = "unknown-id"
show_response = self._get(f"quotas/{quota_id}")
self._assert_status_code_is(show_response, 400)
def _create_quota_with_name(self, quota_name: str, is_default: bool = False):
payload = self._build_quota_payload_with_name(quota_name, is_default)
create_response = self._post("quotas", data=payload, json=True)
create_response.raise_for_status()
return create_response.json()
def _build_quota_payload_with_name(self, quota_name: str, is_default: bool = False):
default = "registered" if is_default else "no"
return {
'name': quota_name,
'description': f'Quota {quota_name} description',
'amount': '100MB',
'operation': '=',
'default': 'registered',
'default': default,
'in_users': [],
'in_groups': []
}
self.dataset_populator.create_quota(payload)
quotas = self.dataset_populator.get_quotas()
assert len(quotas) == 1
def _delete_and_purge(self, url):
headers = self.galaxy_interactor.api_key_header(
key=self.galaxy_interactor.api_key,
admin=True,
anon=False,
headers=None,
)
json_data = json.dumps({
'purge': 'true'
})
api_url = self._api_url(url)
return delete(api_url, data=json_data, headers=headers)
def _assert_quota_is_deleted(self, quota_id: str):
show_response = self._get(f"quotas/deleted/{quota_id}")
show_response.raise_for_status()
json_response = show_response.json()
assert json_response["id"] == quota_id