workflows: add notification action (#33817)
This commit is contained in:
parent
2ccf0c2418
commit
82758404c2
|
|
@ -45,6 +45,7 @@ from wcs.wf.export_to_model import ExportToModel, transform_to_pdf
|
|||
from wcs.wf.geolocate import GeolocateWorkflowStatusItem
|
||||
from wcs.wf.backoffice_fields import SetBackofficeFieldsWorkflowStatusItem
|
||||
from wcs.wf.redirect_to_url import RedirectToUrlWorkflowStatusItem
|
||||
from wcs.wf.notification import SendNotificationWorkflowStatusItem
|
||||
|
||||
from utilities import (create_temporary_pub, MockSubstitutionVariables,
|
||||
clean_temporary_pub)
|
||||
|
|
@ -3972,3 +3973,73 @@ def test_workflow_action_condition(two_pubs):
|
|||
assert logged_error.exception_message == "name 'foobar' is not defined"
|
||||
assert logged_error.expression == 'foobar == barfoo'
|
||||
assert logged_error.expression_type == 'python'
|
||||
|
||||
def test_notifications(pub, http_requests):
|
||||
formdef = FormDef()
|
||||
formdef.name = 'baz'
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
formdata.store()
|
||||
|
||||
assert not SendNotificationWorkflowStatusItem.is_available()
|
||||
|
||||
if not pub.site_options.has_section('variables'):
|
||||
pub.site_options.add_section('variables')
|
||||
pub.site_options.set('variables', 'portal_url', 'https://portal/')
|
||||
assert SendNotificationWorkflowStatusItem.is_available()
|
||||
|
||||
item = SendNotificationWorkflowStatusItem()
|
||||
assert item.to == ['_submitter']
|
||||
item.title = 'xxx'
|
||||
item.body = 'XXX'
|
||||
|
||||
# no user
|
||||
http_requests.empty()
|
||||
item.perform(formdata)
|
||||
assert http_requests.count() == 0
|
||||
|
||||
# user
|
||||
http_requests.empty()
|
||||
user = pub.user_class()
|
||||
user.name_identifiers = ['xxx']
|
||||
user.store()
|
||||
formdata.user_id = user.id
|
||||
formdata.store()
|
||||
|
||||
item.perform(formdata)
|
||||
assert http_requests.count() == 1
|
||||
assert http_requests.get_last('url') == 'https://portal/api/notification/add/?NameID=xxx'
|
||||
assert json.loads(http_requests.get_last('body')) == {
|
||||
'body': 'XXX',
|
||||
'url': formdata.get_url(),
|
||||
'id': 'formdata:%s' % formdata.get_display_id(),
|
||||
'origin': '',
|
||||
'summary': 'xxx'
|
||||
}
|
||||
|
||||
# roles (not exposed in current UI)
|
||||
http_requests.empty()
|
||||
|
||||
role = Role(name='blah')
|
||||
role.store()
|
||||
|
||||
user1 = pub.user_class()
|
||||
user1.roles = [role.id]
|
||||
user1.name_identifiers = ['xxy1']
|
||||
user1.store()
|
||||
user2 = pub.user_class()
|
||||
user2.roles = [role.id]
|
||||
user2.name_identifiers = ['xxy2']
|
||||
user2.store()
|
||||
|
||||
formdef.workflow_roles = {'_receiver': role.id}
|
||||
|
||||
item.to = ['_receiver']
|
||||
item.perform(formdata)
|
||||
assert http_requests.count() == 2
|
||||
assert set(x['url'] for x in http_requests.requests) == set([
|
||||
'https://portal/api/notification/add/?NameID=xxy1',
|
||||
'https://portal/api/notification/add/?NameID=xxy2'])
|
||||
|
|
|
|||
|
|
@ -378,6 +378,9 @@ class HttpRequestsMocking(object):
|
|||
def empty(self):
|
||||
self.requests = []
|
||||
|
||||
def count(self):
|
||||
return len(self.requests)
|
||||
|
||||
|
||||
class SMSMocking(wcs.qommon.sms.MobytSMS):
|
||||
def get_sms_class(self, mode):
|
||||
|
|
|
|||
|
|
@ -0,0 +1,149 @@
|
|||
# w.c.s. - web application for online forms
|
||||
# Copyright (C) 2005-2019 Entr'ouvert
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from quixote import get_publisher
|
||||
|
||||
from qommon import _, ezt
|
||||
from qommon.form import *
|
||||
from qommon.template import TemplateError
|
||||
from qommon import get_logger
|
||||
|
||||
from wcs.roles import Role
|
||||
from wcs.workflows import (WorkflowStatusItem, register_item_class,
|
||||
template_on_formdata, get_role_translation)
|
||||
from .wscall import WebserviceCallStatusItem
|
||||
|
||||
|
||||
class SendNotificationWorkflowStatusItem(WebserviceCallStatusItem):
|
||||
description = N_('User Notification')
|
||||
key = 'notification'
|
||||
category = 'interaction'
|
||||
support_substitution_variables = True
|
||||
|
||||
# parameters
|
||||
to = ['_submitter']
|
||||
title = None
|
||||
body = None
|
||||
origin = None
|
||||
|
||||
# webservice parameters
|
||||
varname = 'notification'
|
||||
post = False
|
||||
_method = 'POST'
|
||||
response_type = 'json'
|
||||
|
||||
action_on_app_error = ':pass'
|
||||
action_on_4xx = ':pass'
|
||||
action_on_5xx = ':pass'
|
||||
action_on_bad_data = ':pass'
|
||||
action_on_network_errors = ':pass'
|
||||
notify_on_errors = True
|
||||
record_errors = False
|
||||
|
||||
@classmethod
|
||||
def is_available(cls, workflow=None):
|
||||
return bool(cls.get_api_url() is not None)
|
||||
|
||||
@classmethod
|
||||
def get_api_url(cls):
|
||||
url = get_publisher().get_site_option('portal_url', 'variables')
|
||||
if not url:
|
||||
return None
|
||||
return url + 'api/notification/add/'
|
||||
|
||||
def get_jump_label(self, target_id):
|
||||
return self.description
|
||||
|
||||
def get_parameters(self):
|
||||
return ('title', 'body', 'origin', 'condition')
|
||||
|
||||
def add_parameters_widgets(self, form, parameters, prefix='', formdef=None):
|
||||
if 'to' in parameters:
|
||||
# never displayed in the current UI (no 'to' in get_parameters)
|
||||
form.add(WidgetList, '%sto' % prefix, title=_('To'),
|
||||
element_type=SingleSelectWidget,
|
||||
value=self.to,
|
||||
add_element_label=_('Add Role'),
|
||||
element_kwargs={'render_br': False,
|
||||
'options': [(None, '---', None)] +
|
||||
self.get_list_of_roles(include_logged_in_users=False)})
|
||||
if 'title' in parameters:
|
||||
form.add(StringWidget, '%stitle' % prefix, title=_('Title'),
|
||||
value=self.title, size=80,
|
||||
validation_function=ComputedExpressionWidget.validate_template)
|
||||
if 'body' in parameters:
|
||||
form.add(TextWidget, '%sbody' % prefix, title=_('Body'),
|
||||
value=self.body, cols=80, rows=5,
|
||||
validation_function=ComputedExpressionWidget.validate_template)
|
||||
if 'origin' in parameters:
|
||||
form.add(StringWidget, '%sorigin' % prefix, title=_('Origin'),
|
||||
value=self.origin, required=False,
|
||||
advanced=not(self.origin))
|
||||
WorkflowStatusItem.add_parameters_widgets(self, form, parameters,
|
||||
prefix=prefix, formdef=formdef)
|
||||
|
||||
def perform(self, formdata):
|
||||
if not (self.is_available() and self.to and self.title and self.body):
|
||||
return
|
||||
|
||||
try:
|
||||
title = template_on_formdata(formdata,
|
||||
self.compute(self.title, render=False),
|
||||
autoescape=False)
|
||||
except TemplateError as e:
|
||||
get_logger().error('error in template for notification title [%s], '
|
||||
'mail could not be generated: %s' % (notified_url, str(e)))
|
||||
return
|
||||
|
||||
try:
|
||||
body = template_on_formdata(formdata,
|
||||
self.compute(self.body, render=False),
|
||||
autoescape=False)
|
||||
except TemplateError as e:
|
||||
get_logger().error('error in template for notification body [%s], '
|
||||
'mail could not be generated: %s' % (notified_url, str(e)))
|
||||
return
|
||||
|
||||
self.post_data = {
|
||||
'summary': title,
|
||||
'body': body,
|
||||
'url': formdata.get_url(),
|
||||
'origin': self.origin or '',
|
||||
'id': 'formdata:%s' % formdata.get_display_id(),
|
||||
}
|
||||
self.url = self.get_api_url()
|
||||
|
||||
users = []
|
||||
for dest in self.to:
|
||||
if dest == '_submitter':
|
||||
users.append(formdata.get_user())
|
||||
continue
|
||||
|
||||
dest = get_role_translation(formdata, dest)
|
||||
try:
|
||||
role = Role.get(dest)
|
||||
except KeyError:
|
||||
continue
|
||||
users.extend(get_publisher().user_class.get_users_with_role(role.id))
|
||||
|
||||
for user in users:
|
||||
if not user:
|
||||
continue
|
||||
for name_id in (user.name_identifiers or []):
|
||||
self.qs_data = {'NameID': name_id}
|
||||
super(SendNotificationWorkflowStatusItem, self).perform(formdata)
|
||||
|
||||
register_item_class(SendNotificationWorkflowStatusItem)
|
||||
|
|
@ -2801,5 +2801,6 @@ def load_extra():
|
|||
import wf.profile
|
||||
import wf.backoffice_fields
|
||||
import wf.redirect_to_url
|
||||
import wf.notification
|
||||
|
||||
from wf.export_to_model import ExportToModel
|
||||
|
|
|
|||
Loading…
Reference in New Issue