workflows: add option to limit global actions to some statuses (#65898) #126
|
@ -709,55 +709,150 @@ def test_backoffice_multi_actions(pub):
|
|||
else:
|
||||
assert formdata.status != 'wf-accepted'
|
||||
|
||||
|
||||
def test_backoffice_multi_actions_jump(pub):
|
||||
create_superuser(pub)
|
||||
create_environment(pub)
|
||||
fpeters marked this conversation as resolved
Outdated
|
||||
formdef = FormDef.get_by_urlname('form-title')
|
||||
|
||||
app = login(get_app(pub))
|
||||
# check webservice (external) triggers are not displayed
|
||||
action3 = workflow.add_global_action('THIRD ACTION')
|
||||
action3.triggers = []
|
||||
trigger = action3.append_trigger('webservice')
|
||||
trigger.roles = [x.id for x in pub.role_class.select() if x.name == 'foobar']
|
||||
workflow.store()
|
||||
resp = app.get('/backoffice/management/form-title/')
|
||||
assert 'id="multi-actions"' in resp.text
|
||||
assert 'THIRD ACTION' not in resp.text
|
||||
|
||||
|
||||
def test_backoffice_multi_actions_some_status(pub):
|
||||
create_superuser(pub)
|
||||
Workflow.wipe()
|
||||
FormDef.wipe()
|
||||
|
||||
workflow = Workflow.get_default_workflow()
|
||||
workflow.id = '2'
|
||||
action = workflow.add_global_action('FOOBAR')
|
||||
jump = action.add_action('jump')
|
||||
jump.status = 'finished'
|
||||
trigger = action.triggers[0]
|
||||
trigger.statuses = ['new']
|
||||
trigger.roles = ['_receiver']
|
||||
workflow.store()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'form title'
|
||||
formdef.workflow_roles = {'_receiver': 1}
|
||||
formdef.workflow_id = workflow.id
|
||||
formdef.store()
|
||||
|
||||
resp = app.get('/backoffice/management/form-title/')
|
||||
assert 'select[]' not in resp.forms['multi-actions'].fields
|
||||
initial_statuses = {}
|
||||
for i in range(15):
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
if i < 5:
|
||||
formdata.jump_status('accepted')
|
||||
elif i % 3 == 0:
|
||||
formdata.jump_status('new')
|
||||
else:
|
||||
formdata.jump_status('finished')
|
||||
initial_statuses[str(formdata.id)] = formdata.status
|
||||
|
||||
resp.forms['listing-settings']['filter'] = 'new'
|
||||
resp = resp.forms['listing-settings'].submit()
|
||||
assert 'select[]' not in resp.forms['multi-actions'].fields
|
||||
app = login(get_app(pub))
|
||||
resp = app.get('/backoffice/management/form-title/?filter=all')
|
||||
ids = []
|
||||
for checkbox in resp.forms[0].fields['select[]']:
|
||||
if checkbox._value == '_all':
|
||||
continue
|
||||
# check them all
|
||||
ids.append(checkbox._value)
|
||||
checkbox.checked = True
|
||||
|
||||
assert len(resp.pyquery('[data-status_new]')) == 3
|
||||
assert len(resp.pyquery('[data-status_finished]')) == 7
|
||||
assert len(resp.pyquery('[data-status_accepted]')) == 5
|
||||
|
||||
resp = resp.forms[0].submit('button-action-1')
|
||||
assert '?job=' in resp.location
|
||||
resp = resp.follow()
|
||||
assert 'Executing task "FOOBAR" on forms' in resp.text
|
||||
assert '>completed<' in resp.text
|
||||
new_statuses = {str(x.id): x.status for x in formdef.data_class().select()}
|
||||
for id in ids:
|
||||
# check action was only executed on "new"
|
||||
if initial_statuses[id] == 'wf-new':
|
||||
assert new_statuses[id] == 'wf-finished'
|
||||
else:
|
||||
assert new_statuses[id] == initial_statuses[id]
|
||||
lguerin
commented
il manque un Workflow.wipe() non ? il manque un Workflow.wipe() non ?
fpeters
commented
Ça passe sans mais oui autant nettoyer. Ça passe sans mais oui autant nettoyer.
|
||||
|
||||
resp = app.get('/backoffice/management/form-title/?filter=all')
|
||||
assert len(resp.pyquery('[data-status_new]')) == 0
|
||||
assert len(resp.pyquery('[data-status_finished]')) == 10
|
||||
assert len(resp.pyquery('[data-status_accepted]')) == 5
|
||||
|
||||
|
||||
def test_backoffice_multi_actions_jump(pub):
|
||||
create_superuser(pub)
|
||||
FormDef.wipe()
|
||||
Workflow.wipe()
|
||||
|
||||
# add identifier to jumps
|
||||
workflow = Workflow.get_default_workflow()
|
||||
workflow.id = '2'
|
||||
workflow.get_status('new').items[1].identifier = 'accept'
|
||||
workflow.get_status('new').items[2].identifier = 'reject'
|
||||
workflow.get_status('new').items[2].require_confirmation = True
|
||||
workflow.store()
|
||||
|
||||
resp.forms['listing-settings']['filter-operator'] = 'ne'
|
||||
resp = resp.forms['listing-settings'].submit()
|
||||
assert 'select[]' not in resp.forms['multi-actions'].fields
|
||||
formdef = FormDef()
|
||||
formdef.name = 'form title'
|
||||
formdef.workflow_roles = {'_receiver': 1}
|
||||
formdef.workflow_id = workflow.id
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
resp.forms['listing-settings']['filter-operator'] = 'eq'
|
||||
resp = resp.forms['listing-settings'].submit()
|
||||
initial_statuses = {}
|
||||
for i in range(15):
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
if i < 5:
|
||||
formdata.jump_status('accepted')
|
||||
elif i % 3 == 0:
|
||||
formdata.jump_status('new')
|
||||
else:
|
||||
formdata.jump_status('finished')
|
||||
initial_statuses[str(formdata.id)] = formdata.status
|
||||
|
||||
app = login(get_app(pub))
|
||||
resp = app.get('/backoffice/management/form-title/?filter=all')
|
||||
assert 'select[]' in resp.forms['multi-actions'].fields
|
||||
assert len(resp.pyquery('[data-status_new]')) == 3
|
||||
assert len(resp.pyquery('[data-status_finished]')) == 7
|
||||
assert len(resp.pyquery('[data-status_accepted]')) == 5
|
||||
assert len(resp.pyquery.find('#multi-actions div.buttons button')) == 2
|
||||
assert len(resp.pyquery.find('#multi-actions div.buttons button[data-ask-for-confirmation]')) == 1
|
||||
|
||||
ids = []
|
||||
for checkbox in resp.forms[0].fields['select[]'][1:6]:
|
||||
for checkbox in resp.forms[0].fields['select[]']:
|
||||
if checkbox._value == '_all':
|
||||
continue
|
||||
# check them all
|
||||
ids.append(checkbox._value)
|
||||
checkbox.checked = True
|
||||
resp = resp.forms['multi-actions'].submit('button-action-st-accept')
|
||||
|
||||
resp = resp.forms['multi-actions'].submit('button-action-st-new-accept')
|
||||
assert '?job=' in resp.location
|
||||
resp = resp.follow()
|
||||
assert 'Executing task "Accept" on forms' in resp.text
|
||||
assert '>completed<' in resp.text
|
||||
|
||||
new_statuses = {str(x.id): x.status for x in formdef.data_class().select()}
|
||||
for id in ids:
|
||||
assert formdef.data_class().get(id).status == 'wf-accepted'
|
||||
# check action was only executed on "new"
|
||||
if initial_statuses[id] == 'wf-new':
|
||||
assert new_statuses[id] == 'wf-accepted'
|
||||
else:
|
||||
assert new_statuses[id] == initial_statuses[id]
|
||||
|
||||
resp = app.get('/backoffice/management/form-title/?filter=all')
|
||||
assert len(resp.pyquery('[data-status_new]')) == 0
|
||||
assert len(resp.pyquery('[data-status_finished]')) == 7
|
||||
assert len(resp.pyquery('[data-status_accepted]')) == 8
|
||||
|
||||
|
||||
def test_backoffice_multi_actions_jump_condition(pub):
|
||||
|
@ -810,7 +905,7 @@ def test_backoffice_multi_actions_jump_condition(pub):
|
|||
}
|
||||
workflow.store()
|
||||
|
||||
resp = resp.forms['multi-actions'].submit('button-action-st-accept')
|
||||
resp = resp.forms['multi-actions'].submit('button-action-st-new-accept')
|
||||
assert '?job=' in resp.location
|
||||
resp = resp.follow()
|
||||
assert 'Executing task "Accept" on forms' in resp.text
|
||||
|
|
|
@ -7623,6 +7623,52 @@ def test_user_global_action_along_form(pub):
|
|||
assert formdef.data_class().get(formdata.id).status == 'wf-finished'
|
||||
|
||||
|
||||
def test_user_global_action_specific_statuses(pub):
|
||||
user = create_user(pub)
|
||||
|
||||
workflow = Workflow.get_default_workflow()
|
||||
workflow.id = '2'
|
||||
action = workflow.add_global_action('FOOBAR')
|
||||
register_comment = action.add_action('register-comment')
|
||||
register_comment.comment = 'HELLO WORLD GLOBAL ACTION'
|
||||
trigger = action.triggers[0]
|
||||
trigger.roles = ['_submitter']
|
||||
workflow.store()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test global action'
|
||||
formdef.fields = []
|
||||
formdef.workflow_id = workflow.id
|
||||
formdef.workflow_roles = {}
|
||||
formdef.store()
|
||||
|
||||
formdef.data_class().wipe()
|
||||
formdata = formdef.data_class()()
|
||||
formdata.user_id = user.id
|
||||
formdata.just_created()
|
||||
formdata.store()
|
||||
|
||||
app = login(get_app(pub), username='foo', password='foo')
|
||||
resp = app.get(formdata.get_url())
|
||||
assert 'button-action-1' in resp.text
|
||||
|
||||
trigger.statuses = ['accepted']
|
||||
workflow.store()
|
||||
|
||||
resp = app.get(formdata.get_url())
|
||||
assert 'button-action-1' not in resp.text
|
||||
|
||||
formdata.jump_status('accepted')
|
||||
formdata.store()
|
||||
|
||||
resp = app.get(formdata.get_url())
|
||||
assert 'button-action-1' in resp.form.fields
|
||||
resp = resp.form.submit('button-action-1')
|
||||
|
||||
resp = app.get(formdata.get_url())
|
||||
assert 'HELLO WORLD GLOBAL ACTION' in resp.text
|
||||
|
||||
|
||||
def test_email_actions(pub, emails):
|
||||
create_user(pub)
|
||||
|
||||
|
|
|
@ -2095,18 +2095,11 @@ class FormPage(FormdefDirectoryBase):
|
|||
return ''
|
||||
|
||||
@classmethod
|
||||
def get_multi_actions(cls, formdef, user, status_filter, status_filter_operator):
|
||||
def get_multi_actions(cls, formdef, user):
|
||||
global_actions = formdef.workflow.get_global_manual_actions()
|
||||
|
||||
if status_filter not in ('open', 'waiting', 'done', 'all') and status_filter_operator == 'eq':
|
||||
# when the listing is filtered on a specific status, include
|
||||
# manual jumps with identifiers
|
||||
try:
|
||||
status = formdef.workflow.get_status(status_filter)
|
||||
except KeyError:
|
||||
status = None
|
||||
else:
|
||||
global_actions.extend(status.get_status_manual_actions())
|
||||
# include manual jumps with identifiers
|
||||
global_actions.extend(formdef.workflow.get_status_manual_actions())
|
||||
|
||||
mass_actions = []
|
||||
for action_dict in global_actions:
|
||||
|
@ -2151,12 +2144,7 @@ class FormPage(FormdefDirectoryBase):
|
|||
if get_request().get_query():
|
||||
qs = '?' + get_request().get_query()
|
||||
|
||||
multi_actions = self.get_multi_actions(
|
||||
self.formdef,
|
||||
get_request().user,
|
||||
status_filter=selected_filter,
|
||||
status_filter_operator=selected_filter_operator,
|
||||
)
|
||||
multi_actions = self.get_multi_actions(self.formdef, get_request().user)
|
||||
multi_form = Form(id='multi-actions')
|
||||
for action in multi_actions:
|
||||
attrs = {}
|
||||
|
@ -2167,6 +2155,11 @@ class FormPage(FormdefDirectoryBase):
|
|||
attrs['data-visible_for_%s' % function.replace('-', '_')] = 'true'
|
||||
else:
|
||||
attrs['data-visible_for_all'] = 'true'
|
||||
if action.get('statuses'):
|
||||
for status in action.get('statuses'):
|
||||
attrs['data-visible_status_%s' % status] = 'true'
|
||||
else:
|
||||
attrs['data-visible_all_status'] = 'true'
|
||||
if getattr(action['action'], 'require_confirmation', False):
|
||||
attrs['data-ask-for-confirmation'] = 'true'
|
||||
multi_form.add_submit(
|
||||
|
@ -4187,11 +4180,9 @@ class MassActionAfterJob(AfterJob):
|
|||
formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])
|
||||
item_ids = self.kwargs['item_ids']
|
||||
action_id = self.kwargs['action_id']
|
||||
status_filter = self.kwargs['status_filter']
|
||||
status_filter_operator = self.kwargs['status_filter_operator']
|
||||
user = get_publisher().user_class.get(self.kwargs['user_id'])
|
||||
|
||||
multi_actions = FormPage.get_multi_actions(formdef, user, status_filter, status_filter_operator)
|
||||
multi_actions = FormPage.get_multi_actions(formdef, user)
|
||||
for action in multi_actions:
|
||||
if action['action'].id == action_id:
|
||||
break
|
||||
|
@ -4224,15 +4215,18 @@ class MassActionAfterJob(AfterJob):
|
|||
)
|
||||
if getattr(action['action'], 'status_action', False):
|
||||
# manual jump action
|
||||
if action['action'].action.check_condition(formdata):
|
||||
if formdata.status.removeprefix('wf-') == action['statuses'][0] and action[
|
||||
'action'
|
||||
].action.check_condition(formdata):
|
||||
from wcs.wf.jump import jump_and_perform
|
||||
|
||||
formdata.record_workflow_event('mass-jump', action_item_id=action['action'].action.id)
|
||||
jump_and_perform(formdata, action['action'].action)
|
||||
else:
|
||||
# global action
|
||||
formdata.record_workflow_event('global-action-mass', global_action_id=action['action'].id)
|
||||
formdata.perform_global_action(action['action'].id, user)
|
||||
if action['action'].check_executable(formdata, user):
|
||||
formdata.record_workflow_event('global-action-mass', global_action_id=action['action'].id)
|
||||
formdata.perform_global_action(action['action'].id, user)
|
||||
self.increment_count()
|
||||
|
||||
def done_action_url(self):
|
||||
|
|
|
@ -315,6 +315,7 @@ class FormDefUI:
|
|||
# dashes are replaced by underscores to prevent HTML5
|
||||
# normalization to CamelCase.
|
||||
r += htmltext(' data-is_%s="true" ' % function_key.replace('-', '_'))
|
||||
r += htmltext(' data-status_%s="true" ' % filled.status.removeprefix('wf-'))
|
||||
r += htmltext('/></td>')
|
||||
for i, f in enumerate(fields):
|
||||
field_value = filled.get_field_view_value(f, max_length=30)
|
||||
|
|
|
@ -72,18 +72,25 @@ function prepare_row_links() {
|
|||
return;
|
||||
} else {
|
||||
$('form#multi-actions div.buttons button').each(function(idx, elem) {
|
||||
var visible = false;
|
||||
var role_visible = false;
|
||||
var status_visible = false;
|
||||
for (var key in $(elem).first().data()) {
|
||||
if (key == 'visible_for_all') {
|
||||
visible = true;
|
||||
break;
|
||||
}
|
||||
if ($('input[type=checkbox][data-is_' + key.substr(12) + ']:checked').length) {
|
||||
visible = true;
|
||||
break;
|
||||
role_visible = true;
|
||||
} else if (key == 'visible_all_status') {
|
||||
status_visible = true;
|
||||
} else if (key.startsWith('visible_status')) {
|
||||
lguerin
commented
pourquoi continue ici seulement ? pourquoi continue ici seulement ?
fpeters
commented
En effet inutile vu le reste de la boucle. En effet inutile vu le reste de la boucle.
|
||||
if ($('input[type=checkbox][data-status_' + key.substr(15) + ']:checked').length) {
|
||||
status_visible = true;
|
||||
}
|
||||
} else if (key.startsWith('visible_for')) {
|
||||
if ($('input[type=checkbox][data-is_' + key.substr(12) + ']:checked').length) {
|
||||
role_visible = true;
|
||||
}
|
||||
}
|
||||
if (role_visible && status_visible) break;
|
||||
}
|
||||
if (visible) {
|
||||
if (role_visible && status_visible) {
|
||||
$(elem).parents('div.widget').show();
|
||||
} else {
|
||||
$(elem).parents('div.widget').hide();
|
||||
|
|
127
wcs/workflows.py
127
wcs/workflows.py
|
@ -863,34 +863,62 @@ class Workflow(StorableObject):
|
|||
actions = []
|
||||
for action in self.global_actions or []:
|
||||
roles = []
|
||||
statuses = []
|
||||
for trigger in action.triggers or []:
|
||||
if not isinstance(trigger, WorkflowGlobalActionManualTrigger):
|
||||
if not trigger.key == 'manual':
|
||||
continue
|
||||
roles.extend(trigger.roles or [])
|
||||
statuses.extend(trigger.statuses or [])
|
||||
functions = [x for x in roles if x in self.roles]
|
||||
roles = [x for x in roles if x not in self.roles]
|
||||
if functions or roles:
|
||||
actions.append({'action': action, 'roles': roles, 'functions': functions})
|
||||
actions.append(
|
||||
{'action': action, 'roles': roles, 'functions': functions, 'statuses': statuses}
|
||||
)
|
||||
return actions
|
||||
|
||||
def get_status_manual_actions(self):
|
||||
class StatusAction:
|
||||
def __init__(self, action):
|
||||
self.status_id = action.parent.id
|
||||
self.id = 'st-%s-%s' % (self.status_id, action.identifier)
|
||||
self.action_id = action.identifier
|
||||
self.name = action.get_label()
|
||||
self.status_action = True
|
||||
self.require_confirmation = action.require_confirmation
|
||||
self.action = action
|
||||
|
||||
def is_interactive(self):
|
||||
return False
|
||||
|
||||
def get_actions(workflow):
|
||||
for status in workflow.possible_status or []:
|
||||
yield from status.items or []
|
||||
|
||||
actions = []
|
||||
choices = [x for x in get_actions(self) if x.key == 'choice' and x.identifier]
|
||||
|
||||
for action in choices:
|
||||
roles = action.by or []
|
||||
functions = [x for x in roles if x in (self.roles or [])]
|
||||
roles = [x for x in roles if x not in (self.roles or [])]
|
||||
if functions or roles:
|
||||
actions.append(
|
||||
{
|
||||
'action': StatusAction(action),
|
||||
'roles': roles,
|
||||
'functions': functions,
|
||||
'statuses': [action.parent.id],
|
||||
}
|
||||
)
|
||||
|
||||
return actions
|
||||
|
||||
def get_global_actions_for_user(self, formdata, user):
|
||||
actions = []
|
||||
for action in self.global_actions or []:
|
||||
for trigger in action.triggers or []:
|
||||
if trigger.key == 'manual':
|
||||
if '_submitter' in (trigger.roles or []) and formdata.is_submitter(user):
|
||||
actions.append(action)
|
||||
break
|
||||
if not user:
|
||||
continue
|
||||
roles = set()
|
||||
for role_id in trigger.roles or []:
|
||||
if role_id == '_submitter':
|
||||
continue
|
||||
roles |= formdata.get_function_roles(role_id)
|
||||
if roles.intersection(user.get_roles()):
|
||||
actions.append(action)
|
||||
break
|
||||
if action.check_executable(formdata, user):
|
||||
actions.append(action)
|
||||
return actions
|
||||
|
||||
def get_subdirectories(self, formdata):
|
||||
|
@ -1452,9 +1480,10 @@ class WorkflowGlobalActionTrigger(XmlSerialisable):
|
|||
class WorkflowGlobalActionManualTrigger(WorkflowGlobalActionTrigger):
|
||||
key = 'manual'
|
||||
roles = None
|
||||
statuses = None
|
||||
|
||||
def get_parameters(self):
|
||||
return ('roles',)
|
||||
return ('roles', 'statuses')
|
||||
|
||||
def render_as_line(self):
|
||||
if self.roles:
|
||||
|
@ -1475,6 +1504,17 @@ class WorkflowGlobalActionManualTrigger(WorkflowGlobalActionTrigger):
|
|||
add_element_label=workflow.get_add_role_label(),
|
||||
element_kwargs={'render_br': False, 'options': options},
|
||||
)
|
||||
status_options = [(None, '---', None)]
|
||||
status_options += [(str(x.id), x.name, str(x.id)) for x in self.parent.parent.possible_status]
|
||||
form.add(
|
||||
WidgetList,
|
||||
'statuses',
|
||||
title=_('Only display to following statuses'),
|
||||
element_type=SingleSelectWidget,
|
||||
value=self.statuses,
|
||||
add_element_label=_('Add a status'),
|
||||
element_kwargs={'render_br': False, 'options': status_options},
|
||||
)
|
||||
return form
|
||||
|
||||
def roles_export_to_xml(self, item, charset, include_id=False):
|
||||
|
@ -2088,6 +2128,26 @@ class WorkflowGlobalAction(SerieOfActionsMixin):
|
|||
for action in self.items or []:
|
||||
yield from action.i18n_scan(location)
|
||||
|
||||
def check_executable(self, formdata, user):
|
||||
# check action is executable for given formdata and user (appropriate status and roles)
|
||||
current_status_id = (formdata.status or '').removeprefix('wf-')
|
||||
for trigger in self.triggers or []:
|
||||
if trigger.key == 'manual':
|
||||
if trigger.statuses and current_status_id not in trigger.statuses:
|
||||
continue
|
||||
if '_submitter' in (trigger.roles or []) and formdata.is_submitter(user):
|
||||
return True
|
||||
if not user:
|
||||
continue
|
||||
roles = set()
|
||||
for role_id in trigger.roles or []:
|
||||
if role_id == '_submitter':
|
||||
continue
|
||||
roles |= formdata.get_function_roles(role_id)
|
||||
if roles.intersection(user.get_roles()):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class WorkflowCriticalityLevel:
|
||||
id = None
|
||||
|
@ -2266,37 +2326,6 @@ class WorkflowStatus(SerieOfActionsMixin):
|
|||
waitpoint = item.waitpoint or waitpoint
|
||||
return bool(endpoint or waitpoint)
|
||||
|
||||
def get_status_manual_actions(self):
|
||||
actions = []
|
||||
status_id = self.id
|
||||
|
||||
from .wf.choice import ChoiceWorkflowStatusItem
|
||||
|
||||
class StatusAction:
|
||||
def __init__(self, action):
|
||||
self.id = 'st-%s' % action.identifier
|
||||
self.status_id = status_id
|
||||
self.action_id = action.identifier
|
||||
self.name = action.get_label()
|
||||
self.status_action = True
|
||||
self.require_confirmation = action.require_confirmation
|
||||
self.action = action
|
||||
|
||||
def is_interactive(self):
|
||||
return False
|
||||
|
||||
for action in self.items or []:
|
||||
if not isinstance(action, ChoiceWorkflowStatusItem):
|
||||
continue
|
||||
if not action.identifier:
|
||||
continue
|
||||
roles = action.by or []
|
||||
functions = [x for x in roles if x in (self.parent.roles or [])]
|
||||
roles = [x for x in roles if x not in (self.parent.roles or [])]
|
||||
if functions or roles:
|
||||
actions.append({'action': StatusAction(action), 'roles': roles, 'functions': functions})
|
||||
return actions
|
||||
|
||||
def get_contrast_color(self):
|
||||
colour = self.colour or 'ffffff'
|
||||
return misc.get_foreground_colour(colour)
|
||||
|
|
Loading…
Reference in New Issue
yeah :)