0001-workflows-use-check_condition-method-to-evaluate-jum.patch

Frédéric Péters, 21 décembre 2021 16:30

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

Subject: [PATCH] workflows: use check_condition method to evaluate jump
 condition (#59914)

 tests/form_pages/test_all.py | 48 ++++++++++++++++++++++++++
 tests/workflow/test_all.py   | 65 ++++++++++++++++++++++++++----------
 wcs/formdata.py              |  4 +++
 wcs/wf/jump.py               | 49 ++++++++++-----------------
 4 files changed, 117 insertions(+), 49 deletions(-)

         assert formdata.evolution[1].comment == 'new-evolution-2'
         assert formdata.evolution[2].comment is None
         # mark user as owner so it can check the UI
         formdata.user_id = user.id
         formdata.store()
         resp = app.get(formdata.get_url())
         assert resp.text.count('Status1') == 3  # once in summary and two in journal
         assert resp.text.count('CSS-STATUS1') == 2
-...
             formdata.evolution[-1].add_part(JournalWsCallErrorPart('bla', 'bla', {}))
             formdata.store()
         # mark user as owner so it can check the UI
         formdata.user_id = user.id
         formdata.store()
         resp = app.get(formdata.get_url())
         assert resp.text.count('<li class="msg') == 1
-...
         formdata.refresh_from_storage()
         assert not formdata.workflow_data
     def test_jumps_with_by_and_no_trigger(pub):
         FormDef.wipe()
         Workflow.wipe()
         pub.role_class.wipe()
         role = pub.role_class(name='xxx')
         role.store()
         workflow = Workflow(name='test')
         st1 = workflow.add_status('Status1', 'st1')
         jump = JumpWorkflowStatusItem()
         jump.status = 'st2'
         st1.items.append(jump)
         jump.by = [role.id]
         jump.parent = st1
         jump = JumpWorkflowStatusItem()
         jump.status = 'st3'
         st1.items.append(jump)
         jump.by = []
         jump.parent = st1
         workflow.add_status('Status2', 'st2')
         workflow.add_status('Status3', 'st3')
         workflow.store()
         formdef = create_formdef()
         formdef.fields = []
         formdef.workflow = workflow
         formdef.store()
         formdef.data_class().wipe()
         resp = get_app(pub).get('/test/')
         resp = resp.form.submit('submit')
         resp = resp.form.submit('submit')
         # it jumps to st2, as jump.by is only related to triggers
         assert formdef.data_class().count() == 1
         assert formdef.data_class().select()[0].status == 'wf-st2'

         formdef.store()
         formdata = formdef.data_class()()
         item = JumpWorkflowStatusItem()
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
     def test_jump_datetime_condition(pub):
-...
             'type': 'python',
             'value': 'datetime.datetime.now() > datetime.datetime(%s, %s, %s)' % yesterday.timetuple()[:3],
+        }
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         tomorrow = datetime.datetime.now() + datetime.timedelta(days=1)
         item.condition = {
             'type': 'python',
             'value': 'datetime.datetime.now() > datetime.datetime(%s, %s, %s)' % tomorrow.timetuple()[:3],
+        }
         assert item.must_jump(formdata) is False
         assert item.check_condition(formdata) is False
     def test_jump_date_conditions(pub):
-...
             'type': 'python',
             'value': 'utils.make_date(form_var_date) == utils.make_date("2015-01-04")',
+        }
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item = JumpWorkflowStatusItem()
         item.condition = {'type': 'python', 'value': 'utils.time_delta(form_var_date, "2015-01-04").days == 0'}
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item = JumpWorkflowStatusItem()
         item.condition = {'type': 'python', 'value': 'utils.time_delta(utils.today(), "2015-01-04").days > 0'}
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item = JumpWorkflowStatusItem()
         item.condition = {
             'type': 'python',
             'value': 'utils.time_delta(datetime.datetime.now(), "2015-01-04").days > 0',
+        }
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item = JumpWorkflowStatusItem()
         item.condition = {
             'type': 'python',
             'value': 'utils.time_delta(utils.time.localtime(), "2015-01-04").days > 0',
+        }
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
     def test_jump_count_condition(pub):
-...
         formdata = formdef.data_class()()
         item = JumpWorkflowStatusItem()
         item.condition = {'type': 'python', 'value': 'form_objects.count < 2'}
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         for _ in range(10):
             formdata = formdef.data_class()()
             formdata.store()
         item.condition = {'type': 'python', 'value': 'form_objects.count < 2'}
         assert item.must_jump(formdata) is False
         assert item.check_condition(formdata) is False
     def test_jump_bad_python_condition(two_pubs):
-...
         item = JumpWorkflowStatusItem()
         item.condition = {'type': 'python', 'value': 'form_var_foobar == 0'}
         assert item.must_jump(formdata) is False
         assert item.check_condition(formdata) is False
         assert two_pubs.loggederror_class.count() == 1
         logged_error = two_pubs.loggederror_class.select()[0]
         assert logged_error.summary == 'Failed to evaluate condition'
-...
         two_pubs.loggederror_class.wipe()
         item.condition = {'type': 'python', 'value': '~ invalid ~'}
         assert item.must_jump(formdata) is False
         assert item.check_condition(formdata) is False
         assert two_pubs.loggederror_class.count() == 1
         logged_error = two_pubs.loggederror_class.select()[0]
         assert logged_error.summary == 'Failed to evaluate condition'
-...
         item = JumpWorkflowStatusItem()
         item.condition = {'type': 'django', 'value': '1 < 2'}
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item.condition = {'type': 'django', 'value': 'form_var_foo == "hello"'}
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item.condition = {'type': 'django', 'value': 'form_var_foo|first|upper == "H"'}
         assert item.must_jump(formdata) is True
         assert item.check_condition(formdata) is True
         item.condition = {'type': 'django', 'value': 'form_var_foo|first|upper == "X"'}
         assert item.must_jump(formdata) is False
         assert item.check_condition(formdata) is False
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         item.condition = {'type': 'django', 'value': '~ invalid ~'}
         assert item.must_jump(formdata) is False
         assert item.check_condition(formdata) is False
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
-...
             formdata.just_created()
             formdata.store()
             formdata_id = formdata.id
             with mock.patch('wcs.wf.jump.JumpWorkflowStatusItem.must_jump') as must_jump:
             with mock.patch('wcs.wf.jump.JumpWorkflowStatusItem.check_condition') as must_jump:
                 must_jump.return_value = False
                 _apply_timeouts(two_pubs)
                 assert must_jump.call_count == 0  # not enough time has passed
-...
         assert two_pubs.loggederror_class.count() == 1
         logged_error = two_pubs.loggederror_class.select()[0]
         assert logged_error.summary == 'Mismatch in target object: expected "Other data", got "Data"'
     def test_conditional_jump_vs_tracing(pub):
         workflow = Workflow(name='wf')
         st1 = workflow.add_status('Status1', 'st1')
         workflow.add_status('Status2', 'st2')
         comment = st1.append_item('register-comment')
         comment.comment = 'hello world'
         jump1 = st1.append_item('jump')
         jump1.parent = st1
         jump1.condition = {'type': 'django', 'value': 'False'}
         jump1.status = 'wf-st2'
         jump2 = st1.append_item('jump')
         jump2.parent = st1
         jump2.status = 'wf-st2'
         workflow.store()
         formdef = FormDef()
         formdef.name = 'baz'
         formdef.workflow = workflow
         formdef.store()
         formdata = formdef.data_class()()
         formdata.just_created()
         formdata.store()
         perform_items(st1.items, formdata)
         formdata.refresh_from_storage()
         assert formdata.evolution[0].parts[-1].actions[0][1:] == ('register-comment', str(comment.id))
         assert formdata.evolution[0].parts[-1].actions[1][1:] == ('jump', str(jump2.id))

             for item in wf_status.items or []:
                 if not hasattr(item, 'by') or not item.by:
                     continue
                 if item.key == 'jump':
                     # automatic jump has a 'by' attribute but it's only for triggers,
                     # it's not a real interactive action.
                     continue
                 with get_publisher().substitutions.temporary_feed(self):
                     if not item.check_condition(self, **(condition_kwargs or {})):
                         continue

     from quixote.html import htmltext
     from wcs.api import get_user_from_api_query_string, is_url_signed
     from wcs.conditions import Condition
     from wcs.workflows import Workflow, WorkflowGlobalAction, WorkflowStatusJumpItem, register_item_class
     from ..qommon import _, errors, force_str
-...
                         pass
                     elif not item.check_auth(self.formdata, user):
                         raise errors.AccessForbiddenError()
                     if item.must_jump(self.formdata, trigger=item.trigger):
                     if item.check_condition(self.formdata, trigger=component):
                         workflow_data = None
                         if hasattr(get_request(), '_json'):
                             workflow_data = get_request().json
-...
                 return None
         def perform(self, formdata):
             if not self.status:
                 return
             if self.must_jump(formdata):
                 wf_status = self.get_target_status(formdata)
                 if wf_status:
                     self.handle_markers_stack(formdata)
                     formdata.status = 'wf-%s' % wf_status[0].id
         def check_condition(self, formdata, *args, **kwargs):
             # ship condition check here so it is not evaluated twice.
             return True
         def must_jump(self, formdata, trigger=None):
             must_jump = True
             wf_status = self.get_target_status(formdata)
             if wf_status:
                 self.handle_markers_stack(formdata)
                 formdata.status = 'wf-%s' % wf_status[0].id
             if self.condition:
                 context = {'formdata': formdata, 'status_item': self}
                 try:
                     must_jump = Condition(self.condition, context).evaluate()
                 except RuntimeError:
                     must_jump = False
             if self.trigger:
                 triggered = trigger is not None and trigger == self.trigger
                 must_jump = must_jump and triggered
         def check_condition(self, formdata, *args, trigger=None, **kwargs):
             result = super().check_condition(formdata, *args, **kwargs)
             if not result:
                 return False
             if self.timeout:
                 timeout_str = self.compute(self.timeout)
-...
                             notify=False,
                             record=True,
+                        )
                         must_jump = False
                         return False
                 last = formdata.last_update_time
                 if last and timeout_seconds:
                     diff = time.time() - time.mktime(last)
                     must_jump = (diff > timeout_seconds) and must_jump
                     if diff < timeout_seconds:
                         return False
             return must_jump
             if self.trigger:
                 if trigger is None or trigger != self.trigger:
                     return False
             return True
     register_item_class(JumpWorkflowStatusItem)
-...
                             get_publisher().substitutions.feed(get_publisher())
                             get_publisher().substitutions.feed(formdef)
                             get_publisher().substitutions.feed(formdata)
                             if jump_action.must_jump(formdata):
                             if jump_action.check_condition(formdata):
                                 jump_and_perform(formdata, jump_action, event=('timeout-jump', jump_action.id))
                                 break
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-workflows-use-check_condition-method-to-evaluate-jum.patch