0003-workflow-to-configuration-for-notification-action-an.patch

Lauréline Guérin, 08 février 2022 17:11

Voir les différences: en ligne côte à côte

Subject: [PATCH 3/3] workflow: 'to' configuration for notification action, and
 users_template (#60639)

 tests/admin_pages/test_workflow.py  |  45 +++++++++
 tests/utilities.py                  |   2 +
 tests/workflow/test_notification.py | 147 +++++++++++++++++++++++++++-
 wcs/variables.py                    |   4 +-
 wcs/wf/notification.py              | 122 +++++++++++++++++------
 5 files changed, 284 insertions(+), 36 deletions(-)

+        )
     def test_workflows_user_notification_action(pub):
         create_superuser(pub)
         Workflow.wipe()
         wf = Workflow(name='notif')
         st = wf.add_status('New', 'st')
         wf.store()
         if not pub.site_options.has_section('variables'):
             pub.site_options.add_section('variables')
         pub.site_options.set('variables', 'portal_url', 'https://www.example.net/')
         with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:
             pub.site_options.write(fd)
         app = login(get_app(pub))
         resp = app.get('/backoffice/workflows/%s/status/%s/' % (wf.id, st.id))
         assert 'User Notification' in [o[0] for o in resp.forms[0]['action-interaction'].options]
         resp.forms[0]['action-interaction'] = 'User Notification'
         resp = resp.forms[0].submit().follow()
         assert 'User Notification' in resp.text
         assert Workflow.get(wf.id).possible_status[0].items[0].to == ['_submitter']
         assert Workflow.get(wf.id).possible_status[0].items[0].users_template is None
         resp = app.get('/backoffice/workflows/%s/status/%s/items/1/' % (wf.id, st.id))
         assert resp.forms[0]['to'].value == '_submitter'
         resp.forms[0]['to'] = '_receiver'
         resp = resp.forms[0].submit('submit')
         assert Workflow.get(wf.id).possible_status[0].items[0].to == ['_receiver']
         assert Workflow.get(wf.id).possible_status[0].items[0].users_template is None
         resp = app.get('/backoffice/workflows/%s/status/%s/items/1/' % (wf.id, st.id))
         assert resp.forms[0]['to'].value == '_receiver'
         resp.forms[0]['to'] = '__other'
         resp.forms[0]['users_template$value_text'] = 'foobar'
         resp = resp.forms[0].submit('submit')
         assert Workflow.get(wf.id).possible_status[0].items[0].to == []
         assert Workflow.get(wf.id).possible_status[0].items[0].users_template == 'foobar'
         resp = app.get('/backoffice/workflows/%s/status/%s/items/1/' % (wf.id, st.id))
         assert resp.forms[0]['to'].value == '__other'
         assert resp.forms[0]['users_template$value_text'].value == 'foobar'
     def test_workflows_criticality_levels(pub):
         create_superuser(pub)

                 'http://remote.example.net/connection-error': (None, None, None),
                 'http://authentic.example.net/idp/saml2/metadata': (200, metadata, None),
                 'http://authentic2.example.net/idp/saml2/metadata': (200, metadata2, None),
                 'https://portal/api/notification/add/': (200, {}, {}),
                 'https://interco-portal/api/notification/add/': (200, {}, {}),
             }.get(base_url, (200, '', {}))
             if url.startswith('file://'):

         clean_temporary_pub()
     @pytest.fixture
     def pub(request):
         pub = create_temporary_pub()
     def pub_fixture(**kwargs):
         pub = create_temporary_pub(**kwargs)
         pub.cfg['language'] = {'language': 'en'}
         pub.write_cfg()
         req = HTTPRequest(None, {'SERVER_NAME': 'example.net', 'SCRIPT_NAME': ''})
-...
         return pub
     @pytest.fixture
     def pub(request):
         return pub_fixture()
     @pytest.fixture
     def sql_pub(request):
         return pub_fixture(sql_mode=True)
     def test_notifications(pub, http_requests):
         pub.user_class.wipe()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'baz'
         formdef.fields = []
         formdef.store()
         formdef.data_class().wipe()
         formdata = formdef.data_class()()
         formdata.just_created()
-...
         assert http_requests.count() == 1
         assert http_requests.get_last('url') == 'https://interco-portal/api/notification/add/'
         assert set(json.loads(http_requests.get_last('body'))['name_ids']) == {'xxy1'}
     def test_notifications_to_users_template(sql_pub, http_requests):
         sql_pub.user_class.wipe()
         FormDef.wipe()
         user1 = sql_pub.user_class(name='userA')
         user1.name_identifiers = ['xxy1']
         user1.email = 'user1@example.com'
         user1.store()
         user2 = sql_pub.user_class(name='userB')
         user2.name_identifiers = ['xxy2']
         user2.email = 'user2@example.com'
         user2.store()
         formdef = FormDef()
         formdef.name = 'foo'
         formdef.fields = []
         formdef.store()
         formdef.data_class().wipe()
         formdatas = []
         for i in range(2):
             formdatas.append(formdef.data_class()())
         formdatas[0].user_id = user1.id
         formdatas[1].user_id = user2.id
         for formdata in formdatas:
             formdata.just_created()
             formdata.store()
         formdef = FormDef()
         formdef.name = 'baz'
         formdef.fields = []
         formdef.store()
         formdef.data_class().wipe()
         formdata = formdef.data_class()()
         formdata.just_created()
         formdata.store()
         assert not SendNotificationWorkflowStatusItem.is_available()
         if not sql_pub.site_options.has_section('variables'):
             sql_pub.site_options.add_section('variables')
         sql_pub.site_options.set('variables', 'portal_url', 'https://portal/')
         assert SendNotificationWorkflowStatusItem.is_available()
         item = SendNotificationWorkflowStatusItem()
         item.to = []
         item.title = 'xxx'
         item.body = 'XXX'
         # no user template defined
         http_requests.empty()
         item.perform(formdata)
         assert http_requests.count() == 0
         for users_template in [
             'xxy1, , xxy2',
             'user1@example.com,user2@example.com',
             '{% for obj in forms|objects:"foo" %}{{ obj|get:"form_user_nameid" }},{% endfor %}',
             '{% for obj in forms|objects:"foo" %}{{ obj|get:"form_user_email" }},{% endfor %}',
             '{{ forms|objects:"foo"|getlist:"form_user_nameid" }}',
             '{{ forms|objects:"foo"|getlist:"form_user_nameid"|list }}',
             '{{ forms|objects:"foo"|getlist:"form_user_email" }}',
             '{{ forms|objects:"foo"|getlist:"form_user_email"|list }}',
             '{{ forms|objects:"foo"|getlist:"form_user" }}',
             '{{ forms|objects:"foo"|getlist:"form_user"|list }}',
         ]:
             item.users_template = users_template
             http_requests.empty()
             item.perform(formdata)
             assert http_requests.count() == 1
             assert http_requests.get_last('url') == 'https://portal/api/notification/add/'
             assert set(json.loads(http_requests.get_last('body'))['name_ids']) == {'xxy1', 'xxy2'}
         formdatas[1].user_id = user1.id
         formdatas[1].store()
         for users_template in [
             'xxy1,',
             'user1@example.com',
             '{% for obj in forms|objects:"foo" %}{{ obj|get:"form_user_nameid" }},{% endfor %}',
             '{% for obj in forms|objects:"foo" %}{{ obj|get:"form_user_email" }},{% endfor %}',
             '{{ forms|objects:"foo"|getlist:"form_user_nameid" }}',
             '{{ forms|objects:"foo"|getlist:"form_user_nameid"|list }}',
             '{{ forms|objects:"foo"|getlist:"form_user_email" }}',
             '{{ forms|objects:"foo"|getlist:"form_user_email"|list }}',
             '{{ forms|objects:"foo"|getlist:"form_user" }}',
             '{{ forms|objects:"foo"|getlist:"form_user"|list }}',
         ]:
             item.users_template = users_template
             http_requests.empty()
             item.perform(formdata)
             assert http_requests.count() == 1
             assert http_requests.get_last('url') == 'https://portal/api/notification/add/'
             assert set(json.loads(http_requests.get_last('body'))['name_ids']) == {'xxy1'}
         # unknown user
         item.users_template = 'xxy1, foobar'
         http_requests.empty()
         item.perform(formdata)
         assert http_requests.count() == 1
         assert set(json.loads(http_requests.get_last('body'))['name_ids']) == {'xxy1'}
         assert sql_pub.loggederror_class.count() == 1
         logged_error = sql_pub.loggederror_class.select()[0]
         assert logged_error.summary == 'Failed to notify user (not found: "foobar")'
         assert logged_error.formdata_id == str(formdata.id)
         # result bad format
         item.users_template = '{{ forms|objects:"foo"|get:"foobarbaz" }}'  # None
         http_requests.empty()
         item.perform(formdata)
         assert http_requests.count() == 0
         assert sql_pub.loggederror_class.count() == 2
         logged_error = sql_pub.loggederror_class.select()[1]
         assert logged_error.summary == 'Failed to notify users, bad template result (None)'
         assert logged_error.formdata_id == str(formdata.id)
         # template error
         item.users_template = '{% for obj in forms|objects:"foo" %}{{ obj|get:"form_user_nameid" }},'
         http_requests.empty()
         item.perform(formdata)
         assert http_requests.count() == 0
         assert sql_pub.loggederror_class.count() == 3
         logged_error = sql_pub.loggederror_class.select()[2]
         assert logged_error.summary == 'Failed to compute template'
         assert logged_error.formdata_id == str(formdata.id)

     from .qommon.evalutils import make_datetime
     from .qommon.storage import And, Equal, Intersects, Not, NotEqual, Null, Or
     from .qommon.substitution import CompatibilityNamesDict
     from .qommon.templatetags.qommon import parse_datetime
     from .qommon.templatetags.qommon import parse_datetime, unlazy
     class LazyFormDefObjectsManager:
-...
                     if hasattr(value, 'timetuple'):
                         value = value.timetuple()
                     else:
                         value = value.get_value()
                         value = unlazy(value)
                 self._cached_resultset.append(value)
         def __len__(self):

     from quixote import get_publisher
     from wcs.variables import LazyList, LazyUser
     from wcs.workflows import WorkflowStatusItem, register_item_class, template_on_formdata
     from ..qommon import _
     from ..qommon.form import ComputedExpressionWidget, SingleSelectWidget, StringWidget, TextWidget, WidgetList
     from ..qommon.form import ComputedExpressionWidget, SingleSelectWidget, StringWidget, TextWidget
     from ..qommon.template import TemplateError
     from .wscall import WebserviceCallStatusItem
-...
         # parameters
         to = ['_submitter']
         users_template = None
         title = None
         body = None
         origin = None
-...
             return _(self.description)
         def get_parameters(self):
             return ('title', 'body', 'origin', 'condition')
             return ('to', 'users_template', 'title', 'body', 'origin', 'condition')
         def add_parameters_widgets(self, form, parameters, prefix='', formdef=None, **kwargs):
             if 'to' in parameters:
                 # never displayed in the current UI (no 'to' in get_parameters)
                 options = [(None, '---', None)] + self.get_list_of_roles(include_logged_in_users=False)
                 options.append(('__other', _('Other (from template)'), '__other'))
                 form.add(
                     WidgetList,
                     SingleSelectWidget,
                     '%sto' % prefix,
                     title=_('To'),
                     element_type=SingleSelectWidget,
                     value=self.to,
                     add_element_label=self.get_add_role_label(),
                     element_kwargs={
                         'render_br': False,
                         'options': [(None, '---', None)] + self.get_list_of_roles(include_logged_in_users=False),
                     value=self.to[0] if self.to else '__other',
                     options=options,
                     attrs={'data-dynamic-display-parent': 'true'},
+                )
             if 'users_template' in parameters:
                 form.add(
                     ComputedExpressionWidget,
                     '%susers_template' % prefix,
                     title=_('Users template'),
                     value=self.users_template,
                     attrs={
                         'data-dynamic-display-child-of': '%sto' % prefix,
                         'data-dynamic-display-value': '__other',
                     },
                     allow_python=False,
+                )
             if 'title' in parameters:
                 form.add(
-...
                 self, form, parameters, prefix=prefix, formdef=formdef, **kwargs
+            )
         def submit_admin_form(self, form):
             super().submit_admin_form(form)
             if not form.has_errors():
                 if self.to == '__other':
                     self.to = []
                 elif self.to and isinstance(self.to, str):
                     self.to = [self.to]
                     self.users_template = None
         def perform(self, formdata):
             if not (self.is_available() and self.to and self.title and self.body):
             if not (self.is_available() and (self.to or self.users_template) and self.title and self.body):
                 return
             try:
-...
                 return
             users = []
             for dest in self.to:
                 if dest == '_submitter':
                     users.append(formdata.get_user())
                     continue
             if self.to:
                 for dest in self.to:
                     if dest == '_submitter':
                         users.append(formdata.get_user())
                         continue
                 for dest_id in formdata.get_function_roles(dest):
                     for dest_id in formdata.get_function_roles(dest):
                         try:
                             role = get_publisher().role_class.get(dest_id)
                         except KeyError:
                             continue
                         users.extend(get_publisher().user_class.get_users_with_role(role.id))
             else:
                 with get_publisher().complex_data():
                     try:
                         role = get_publisher().role_class.get(dest_id)
                     except KeyError:
                         continue
                     users.extend(get_publisher().user_class.get_users_with_role(role.id))
                         to = self.compute(self.users_template, allow_complex=True, raises=True, formdata=formdata)
                     except Exception:
                         return
                     else:
                         to = get_publisher().get_cached_complex_data(to)
                 if isinstance(to, LazyList):
                     to = list(to)
                 elif isinstance(to, str):
                     to = [x.strip() for x in to.split(',')]
                 if not isinstance(to, list):
                     get_publisher().record_error(
                         _('Failed to notify users, bad template result (%s)') % to,
                         formdata=formdata,
                         status_item=self,
+                    )
                     return
                 to = [v for v in to if v]
                 for value in to:
                     if isinstance(value, LazyUser):
                         value = value._user
                     if isinstance(value, get_publisher().user_class):
                         users.append(value)
                     else:
                         user = get_publisher().user_class.lookup_by_string(str(value))
                         if not user:
                             get_publisher().record_error(
                                 _('Failed to notify user (not found: "%s")') % value,
                                 formdata=formdata,
                                 status_item=self,
+                            )
                             continue
                         users.append(user)
             name_ids = set()
             for user in users:
-...
                 for name_id in user.name_identifiers or []:
                     name_ids.add(name_id)
             if name_ids:
                 self.post_data = {
                     'summary': title,
                     'body': body,
                     'url': formdata.get_url(),
                     'origin': self.origin or '',
                     'id': 'formdata:%s' % formdata.get_display_id(),
                     'name_ids': list(name_ids),
+                }
                 self.url = self.get_api_url()
                 super().perform(formdata)
             if not name_ids:
                 return
             self.post_data = {
                 'summary': title,
                 'body': body,
                 'url': formdata.get_url(),
                 'origin': self.origin or '',
                 'id': 'formdata:%s' % formdata.get_display_id(),
                 'name_ids': list(name_ids),
+            }
             self.url = self.get_api_url()
             super().perform(formdata)
     register_item_class(SendNotificationWorkflowStatusItem)
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0003-workflow-to-configuration-for-notification-action-an.patch