tests/admin_pages/test_workflow.py

@@ -8,7 +8,7 @@ import responses
 from pyquery import PyQuery
 from webtest import Upload
 
-from wcs import fields
+from wcs import fields, workflow_tests
 from wcs.blocks import BlockDef
 from wcs.carddef import CardDef
 from wcs.categories import WorkflowCategory
@@ -17,6 +17,7 @@ from wcs.mail_templates import MailTemplate
 from wcs.qommon.afterjobs import AfterJob
 from wcs.qommon.errors import ConnectionError
 from wcs.qommon.http_request import HTTPRequest
+from wcs.testdef import TestDef, TestResult
 from wcs.wf.create_formdata import CreateFormdataWorkflowStatusItem, Mapping
 from wcs.wf.form import WorkflowFormFieldsFormDef
 from wcs.workflows import (
@@ -633,7 +634,10 @@ def test_workflows_delete_status_reassign(pub, name):
             resp = resp.follow()
             assert formdefs[-1].data_class().get(formdata2.id).status == 'wf-%s' % wf_bar.id
 
-        assert AfterJob.count() == 2  # status change + rebuild_security
+        if name in ('forms', 'cards'):
+            assert AfterJob.count() == 3  # status change + rebuild_security + form or card tests
+        else:
+            assert AfterJob.count() == 4  # status change + rebuild_security + card tests + form tests
         resp = resp.click('Back')
         assert resp.request.path == f'/backoffice/workflows/{workflow.id}/'
 
@@ -4434,3 +4438,58 @@ def test_workflows_function_and_role_with_same_name(pub):
         (str(role1.id), False, 'Foo'),
         (str(role2.id), False, 'Foobar [role]'),  # same name as function -> role suffix
     ]
+
+
+def test_workflow_test_results(pub):
+    create_superuser(pub)
+    TestDef.wipe()
+    TestResult.wipe()
+
+    Workflow.wipe()
+    workflow = Workflow(name='Workflow One')
+    workflow.add_status(name='New status')
+    workflow.store()
+
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.workflow_id = workflow.id
+    formdef.name = 'test title'
+    formdef.store()
+
+    app = login(get_app(pub))
+    resp = app.get('/backoffice/workflows/1/edit')
+    resp.form['name'] = 'test'
+    resp = resp.form.submit('submit').follow()
+    assert TestResult.count() == 0
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.store()
+
+    resp = app.get('/backoffice/workflows/1/edit')
+    resp.form['name'] = 'test 2'
+    resp = resp.form.submit('submit').follow()
+    assert TestResult.count() == 0
+
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertStatus(id='1'),
+    ]
+    testdef.store()
+
+    resp = app.get('/backoffice/workflows/1/edit')
+    resp.form['name'] = 'test 3'
+    resp = resp.form.submit('submit').follow()
+
+    assert TestResult.count() == 1
+    result = TestResult.select()[0]
+    assert result.reason == 'Change in workflow'
+
+    resp = resp.click('add status')
+    resp.forms[0]['name'] = 'new status'
+    resp = resp.forms[0].submit()
+
+    assert TestResult.count() == 2
+    result = TestResult.select(order_by='id')[1]
+    assert result.reason == 'Workflow: New status "new status"'

wcs/admin/tests.py

@@ -651,12 +651,13 @@ class TestResultsDirectory(Directory):
 
 
 class TestsAfterJob(AfterJob):
-    def __init__(self, objectdef, reason, snapshot=None, **kwargs):
+    def __init__(self, objectdef, reason, snapshot=None, triggered_by='', **kwargs):
         super().__init__(
             objectdef_class=objectdef.__class__,
             objectdef_id=objectdef.id,
             reason=reason,
             snapshot_id=snapshot.id if snapshot else None,
+            triggered_by=triggered_by,
             **kwargs,
         )
 
@@ -667,7 +668,7 @@ class TestsAfterJob(AfterJob):
             return
         reason = self.kwargs['reason']
 
-        result = self.run_tests(objectdef, reason)
+        result = self.run_tests(objectdef, reason, self.kwargs.get('triggered_by', ''))
 
         if result and self.kwargs['snapshot_id'] is not None:
             snapshot = get_publisher().snapshot_class.get(self.kwargs['snapshot_id'])
@@ -675,11 +676,14 @@ class TestsAfterJob(AfterJob):
             snapshot.store()
 
     @staticmethod
-    def run_tests(objectdef, reason):
+    def run_tests(objectdef, reason, triggered_by=''):
         testdefs = TestDef.select_for_objectdef(objectdef)
         if not testdefs:
             return
 
+        if triggered_by == 'workflow-change' and not any(x.workflow_tests.actions for x in testdefs):
+            return
+
         for test in testdefs:
             try:
                 test.run(objectdef)

wcs/workflows.py

@@ -942,6 +942,16 @@ class Workflow(StorableObject):
         if not migration_update:
             if get_response():
                 get_response().add_after_job(_('Reindexing cards and forms after workflow change'), update)
+                from wcs.admin.tests import TestsAfterJob
+
+                for formdef in itertools.chain(self.formdefs(), self.carddefs()):
+                    get_response().add_after_job(
+                        TestsAfterJob(
+                            formdef,
+                            reason=_('Workflow: %s') % comment if comment else _('Change in workflow'),
+                            triggered_by='workflow-change',
+                        )
+                    )
             else:
                 update()