workflows: add option to limit global actions to some statuses (#65898) #126

Merged
fpeters merged 3 commits from wip/65898-not-global-global-action into main 2023-02-28 16:09:04 +01:00
6 changed files with 272 additions and 100 deletions

View File

@ -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

yeah :)

yeah :)
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 &quot;FOOBAR&quot; 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]

il manque un Workflow.wipe() non ?

il manque un Workflow.wipe() non ?

Ç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 &quot;Accept&quot; 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 &quot;Accept&quot; on forms' in resp.text

View File

@ -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)

View File

@ -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):

View File

@ -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)

View File

@ -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')) {

pourquoi continue ici seulement ?

pourquoi continue ici seulement ?

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();

View File

@ -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)