External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-15 03:40:42 -03:30
Code Issues Packages Projects Releases Wiki Activity

add period to active job conflict error

Browse Source 
Rename StateConflict to ActiveJobConflict and used shared
message inside of that exception class.
This commit is contained in:
AlanCoding 2017-10-06 13:23:25 -04:00 committed by Matthew Jones
parent f93506fe2c
commit 5380d57ce8
2 changed files with 17 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
awx/main/access.py
View File

@ -26,7 +26,7 @@ from awx.conf.license import LicenseForbids, feature_enabled
__all__ = ['get_user_queryset', 'check_user_access', 'check_user_access_with_errors',
'user_accessible_objects', 'consumer_access',
'user_admin_role', 'StateConflict',]
'user_admin_role', 'ActiveJobConflict',]
logger = logging.getLogger('awx.main.access')
@ -36,9 +36,15 @@ access_registry = {
}
class StateConflict(ValidationError):
class ActiveJobConflict(ValidationError):
status_code = 409
def __init__(self, active_jobs):
super(ActiveJobConflict, self).__init__({
"conflict": _("Resource is being used by running jobs."),
"active_jobs": active_jobs
})
def register_access(model_class, access_class):
access_registry[model_class] = access_class
@ -550,8 +556,7 @@ class OrganizationAccess(BaseAccess):
active_jobs.extend([dict(type="inventory_update", id=o.id)
for o in InventoryUpdate.objects.filter(inventory_source__inventory__organization=obj, status__in=ACTIVE_STATES)])
if len(active_jobs) > 0:
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": active_jobs})
raise ActiveJobConflict(active_jobs)
return True
def can_attach(self, obj, sub_obj, relationship, *args, **kwargs):
@ -644,8 +649,7 @@ class InventoryAccess(BaseAccess):
active_jobs.extend([dict(type="ad_hoc_command", id=o.id)
for o in AdHocCommand.objects.filter(inventory=obj, status__in=ACTIVE_STATES)])
if len(active_jobs) > 0:
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": active_jobs})
raise ActiveJobConflict(active_jobs)
return True
def can_run_ad_hoc_commands(self, obj):
@ -773,8 +777,7 @@ class GroupAccess(BaseAccess):
active_jobs.extend([dict(type="inventory_update", id=o.id)
for o in InventoryUpdate.objects.filter(inventory_source__in=obj.inventory_sources.all(), status__in=ACTIVE_STATES)])
if len(active_jobs) > 0:
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": active_jobs})
raise ActiveJobConflict(active_jobs)
return True
def can_start(self, obj, validate_license=True):
@ -826,8 +829,7 @@ class InventorySourceAccess(BaseAccess):
return False
active_jobs_qs = InventoryUpdate.objects.filter(inventory_source=obj, status__in=ACTIVE_STATES)
if active_jobs_qs.exists():
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": [dict(type="inventory_update", id=o.id) for o in active_jobs_qs.all()]})
raise ActiveJobConflict([dict(type="inventory_update", id=o.id) for o in active_jobs_qs.all()])
return True
@check_superuser
@ -1092,8 +1094,7 @@ class ProjectAccess(BaseAccess):
active_jobs.extend([dict(type="project_update", id=o.id)
for o in ProjectUpdate.objects.filter(project=obj, status__in=ACTIVE_STATES)])
if len(active_jobs) > 0:
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": active_jobs})
raise ActiveJobConflict(active_jobs)
return True
@check_superuser
@ -1298,8 +1299,7 @@ class JobTemplateAccess(BaseAccess):
active_jobs = [dict(type="job", id=o.id)
for o in obj.jobs.filter(status__in=ACTIVE_STATES)]
if len(active_jobs) > 0:
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": active_jobs})
raise ActiveJobConflict(active_jobs)
return True
@check_superuser
@ -1764,8 +1764,7 @@ class WorkflowJobTemplateAccess(BaseAccess):
active_jobs = [dict(type="workflow_job", id=o.id)
for o in obj.workflow_jobs.filter(status__in=ACTIVE_STATES)]
if len(active_jobs) > 0:
raise StateConflict({"conflict": _("Resource is being used by running jobs"),
"active_jobs": active_jobs})
raise ActiveJobConflict(active_jobs)
return True

4
awx/main/tests/functional/test_rbac_inventory.py
View File

@ -13,7 +13,7 @@ from awx.main.access import (
InventoryUpdateAccess,
CustomInventoryScriptAccess,
ScheduleAccess,
StateConflict
ActiveJobConflict
)
@ -21,7 +21,7 @@ from awx.main.access import (
def test_running_job_protection(inventory, admin_user):
AdHocCommand.objects.create(inventory=inventory, status='running')
access = InventoryAccess(admin_user)
with pytest.raises(StateConflict):
with pytest.raises(ActiveJobConflict):
access.can_delete(inventory)