workflow_tests: add support for dispatch to user action (#89601)

tests/test_workflow_tests.py

@@ -1132,6 +1132,60 @@ def test_workflow_tests_dispatch(pub):
     assert str(excinfo.value) == 'Button "Go to middle status" is not displayed.'
 
 
+def test_workflow_tests_dispatch_user(pub):
+    user = pub.user_class(name='test user')
+    user.email = 'test@example.com'
+    user.test_uuid = '42'
+    user.store()
+
+    workflow = Workflow(name='Workflow One')
+    new_status = workflow.add_status(name='New status')
+    middle_status = workflow.add_status(name='Middle status')
+    end_status = workflow.add_status(name='End status')
+
+    choice = new_status.add_action('choice')
+    choice.label = 'Go to middle status'
+    choice.status = middle_status.id
+    choice.by = ['logged-users']
+
+    choice = middle_status.add_action('choice')
+    choice.label = 'Go to end status'
+    choice.status = end_status.id
+    choice.by = ['_receiver']
+
+    workflow.store()
+
+    formdef = FormDef()
+    formdef.name = 'test title'
+    formdef.workflow_id = workflow.id
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertStatus(status_name='New status'),
+        workflow_tests.ButtonClick(button_name='Go to middle status', who='other', who_id=user.test_uuid),
+        workflow_tests.AssertStatus(status_name='Middle status'),
+        workflow_tests.ButtonClick(button_name='Go to end status', who='other', who_id=user.test_uuid),
+        workflow_tests.AssertStatus(status_name='End status'),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'Button "Go to end status" is not displayed.'
+
+    dispatch = middle_status.add_action('dispatch')
+    dispatch.dispatch_type = 'manual'
+    dispatch.role_key = '_receiver'
+    dispatch.role_id = '{{ session_user_email }}'
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+
 def test_workflow_tests_webservice(pub):
     workflow = Workflow(name='Workflow One')
     new_status = workflow.add_status(name='New status')

wcs/qommon/publisher.py

@@ -150,6 +150,8 @@ class QommonPublisher(Publisher):
 
     sql_application_name = None
 
+    is_running_test = False
+
     def get_root_url(self):
         if self.get_request():
             return self.get_request().environ['SCRIPT_NAME'] + '/'

wcs/testdef.py

@@ -306,16 +306,19 @@ class TestDef(sql.TestDef):
         wsgi_request = WSGIRequest({'REQUEST_METHOD': 'POST', 'wsgi.input': io.StringIO()})
         fake_request = CompatHTTPRequest(wsgi_request)
         fake_request.is_in_backoffice_forced_value = self.is_in_backoffice
+        base_user_class = get_publisher().user_class
         try:
             get_publisher()._set_request(fake_request)
             fake_request.session = get_session_manager().new_session(None)
             get_publisher().record_error = record_error
             get_publisher()._http_adapter = MockWebserviceResponseAdapter(self)
+            get_publisher().user_class = get_publisher().test_user_class
             yield
         finally:
             get_publisher()._set_request(true_request)
             get_publisher().record_error = real_record_error
             get_publisher()._http_adapter = real_http_adapter
+            get_publisher().user_class = base_user_class
 
     def run(self, objectdef):
         self.exception = None