Produits Entr'ouvert » w.c.s.

from wcs.wf.notification import SendNotificationWorkflowStatusItem

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

    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

    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'])

    def count(self):

        return len(self.requests)

# 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_('Notification')

    key = 'notification'

    category = 'interaction'

    support_substitution_variables = True

    # parameters

    to = []

    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 ('to', 'title', 'body', 'origin', 'condition')

    def add_parameters_widgets(self, form, parameters, prefix='', formdef=None):

        if 'to' in 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)

    import wf.notification