tests/admin_pages/test_workflow_tests.py

@@ -9,7 +9,7 @@ from wcs import workflow_tests
 from wcs.formdef import FormDef, fields
 from wcs.qommon.http_request import HTTPRequest
 from wcs.testdef import TestDef, WebserviceResponse
-from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
+from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef, WorkflowCriticalityLevel
 
 from ..utilities import create_temporary_pub, get_app, login
 from .test_all import create_superuser
@@ -418,6 +418,156 @@ def test_workflow_tests_action_assert_sms(pub):
     assert escape('SMS to 0123456789 (+2)') in resp.text
 
 
+def test_workflow_tests_action_assert_anonymise(pub):
+    create_superuser(pub)
+
+    formdef = FormDef()
+    formdef.name = 'test title'
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertAnonymise(id='1'),
+    ]
+    testdef.store()
+
+    app = login(get_app(pub))
+
+    resp = app.get('/backoffice/forms/1/tests/%s/workflow/' % testdef.id)
+    assert 'Edit' not in resp.text
+
+
+def test_workflow_tests_action_assert_redirect(pub):
+    create_superuser(pub)
+
+    formdef = FormDef()
+    formdef.name = 'test title'
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertRedirect(id='1'),
+    ]
+    testdef.store()
+
+    app = login(get_app(pub))
+
+    resp = app.get('/backoffice/forms/1/tests/%s/workflow/' % testdef.id)
+    assert 'not configured' in resp.text
+
+    resp = resp.click('Edit')
+    resp.form['url'] = 'http://example.com'
+    resp = resp.form.submit().follow()
+
+    assert 'not configured' not in resp.text
+    assert 'http://example.com' in resp.text
+
+
+def test_workflow_tests_action_assert_history_message(pub):
+    create_superuser(pub)
+
+    formdef = FormDef()
+    formdef.name = 'test title'
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertHistoryMessage(id='1'),
+    ]
+    testdef.store()
+
+    app = login(get_app(pub))
+
+    resp = app.get('/backoffice/forms/1/tests/%s/workflow/' % testdef.id)
+    assert 'not configured' in resp.text
+
+    resp = resp.click('Edit')
+    resp.form['message'] = 'Hello'
+    resp = resp.form.submit().follow()
+
+    assert 'not configured' not in resp.text
+
+
+def test_workflow_tests_action_assert_alert(pub):
+    create_superuser(pub)
+
+    formdef = FormDef()
+    formdef.name = 'test title'
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertAlert(id='1'),
+    ]
+    testdef.store()
+
+    app = login(get_app(pub))
+
+    resp = app.get('/backoffice/forms/1/tests/%s/workflow/' % testdef.id)
+    assert 'not configured' in resp.text
+
+    resp = resp.click('Edit')
+    resp.form['message'] = 'Hello'
+    resp = resp.form.submit().follow()
+
+    assert 'not configured' not in resp.text
+
+
+def test_workflow_tests_action_assert_criticality(pub):
+    create_superuser(pub)
+
+    workflow = Workflow(name='Workflow One')
+    workflow.add_status(name='New status')
+    workflow.store()
+
+    formdef = FormDef()
+    formdef.workflow_id = workflow.id
+    formdef.name = 'test title'
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+
+    testdef = TestDef.create_from_formdata(formdef, formdata)
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertCriticality(id='1'),
+    ]
+    testdef.store()
+
+    app = login(get_app(pub))
+
+    resp = app.get('/backoffice/forms/1/tests/%s/workflow/' % testdef.id)
+    assert 'not configured' in resp.text
+
+    resp = resp.click('Edit')
+    assert 'Workflow has no criticality levels.' in resp.text
+
+    workflow.criticality_levels = [
+        WorkflowCriticalityLevel(name='green'),
+        WorkflowCriticalityLevel(name='red'),
+    ]
+    workflow.store()
+
+    resp = app.get('/backoffice/forms/1/tests/%s/workflow/1/' % testdef.id)
+    resp.form['level_id'].select(text='green')
+    resp = resp.form.submit().follow()
+
+    assert 'not configured' not in resp.text
+    assert escape('Criticality is "green"') in resp.text
+
+
 def test_workflow_tests_action_assert_backoffice_field(pub):
     create_superuser(pub)
 

tests/test_workflow_tests.py

@@ -9,7 +9,12 @@ from wcs.qommon.http_request import HTTPRequest
 from wcs.testdef import TestDef, WebserviceResponse
 from wcs.wf.jump import JumpWorkflowStatusItem, _apply_timeouts
 from wcs.workflow_tests import WorkflowTestError
-from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef, WorkflowStatusItem
+from wcs.workflows import (
+    Workflow,
+    WorkflowBackofficeFieldsFormDef,
+    WorkflowCriticalityLevel,
+    WorkflowStatusItem,
+)
 
 from .backoffice_pages.test_all import create_user
 from .utilities import create_temporary_pub, get_app, login
@@ -603,6 +608,230 @@ def test_workflow_tests_sms(pub):
     assert 'SMS body: "Hello"' in excinfo.value.details
 
 
+def test_workflow_tests_anonymise(pub):
+    user = pub.user_class(name='test user')
+    user.store()
+
+    workflow = Workflow(name='Workflow One')
+    new_status = workflow.add_status(name='New status')
+    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.agent_id = user.id
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertAnonymise(),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'Form was not anonymised.'
+
+    anonymise_action = new_status.add_action('anonymise')
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+    anonymise_action.mode = 'intermediate'
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+    anonymise_action.mode = 'unlink_user'
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+
+def test_workflow_tests_redirect(pub):
+    user = pub.user_class(name='test user')
+    user.store()
+
+    workflow = Workflow(name='Workflow One')
+    new_status = workflow.add_status(name='New status')
+    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.agent_id = user.id
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertRedirect(url='https://example.com/'),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'No redirection occured.'
+
+    redirect_action = new_status.add_action('redirect_to_url')
+    redirect_action.url = 'https://test.com/'
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert (
+        str(excinfo.value)
+        == 'Expected redirection to https://example.com/ but was redirected to https://test.com/.'
+    )
+
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertRedirect(url='https://test.com/'),
+    ]
+
+    testdef.run(formdef)
+
+
+def test_workflow_tests_history_message(pub):
+    user = pub.user_class(name='test user')
+    user.store()
+
+    workflow = Workflow(name='Workflow One')
+    new_status = workflow.add_status(name='New status')
+    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.agent_id = user.id
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertHistoryMessage(message='Hello 42'),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'No history message.'
+
+    register_comment = new_status.add_action('register-comment')
+    register_comment.comment = 'Hello {{ 41|add:1 }}'
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertHistoryMessage(message='Hello 43'),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'Wrong history message content.'
+    assert 'Displayed history message: <div>Hello 42</div>' in excinfo.value.details
+    assert 'Expected history message: Hello 43' in excinfo.value.details
+
+
+def test_workflow_tests_alert(pub):
+    user = pub.user_class(name='test user')
+    user.store()
+
+    workflow = Workflow(name='Workflow One')
+    new_status = workflow.add_status(name='New status')
+    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.agent_id = user.id
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertAlert(message='Hello 42'),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'No alert matching message.'
+    assert 'Displayed alerts: None' in excinfo.value.details
+    assert 'Expected alert: Hello 42' in excinfo.value.details
+
+    alert = new_status.add_action('displaymsg')
+    alert.message = 'Hello {{ 41|add:1 }}'
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertAlert(message='Hello 43'),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'No alert matching message.'
+    assert 'Displayed alerts: <p>Hello 42</p>' in excinfo.value.details
+    assert 'Expected alert: Hello 43' in excinfo.value.details
+
+
+def test_workflow_tests_criticality(pub):
+    user = pub.user_class(name='test user')
+    user.store()
+
+    workflow = Workflow(name='Workflow One')
+    new_status = workflow.add_status(name='New status')
+    green_level = WorkflowCriticalityLevel(name='green')
+    red_level = WorkflowCriticalityLevel(name='red')
+    workflow.criticality_levels = [green_level, red_level]
+    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.agent_id = user.id
+    testdef.workflow_tests.actions = [
+        workflow_tests.AssertCriticality(level_id=red_level.id),
+    ]
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'Form should have criticality level "red" but has level "green".'
+
+    new_status.add_action('modify_criticality')
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    testdef.run(formdef)
+
+    workflow.criticality_levels = []
+    workflow.store()
+    formdef.refresh_from_storage()
+
+    with pytest.raises(WorkflowTestError) as excinfo:
+        testdef.run(formdef)
+    assert str(excinfo.value) == 'Broken, missing criticality level'
+
+
 def test_workflow_tests_backoffice_fields(pub):
     user = pub.user_class(name='test user')
     user.store()
@@ -864,6 +1093,17 @@ def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
     sendsms.to = ['0123456789']
     sendsms.body = 'Hello'
 
+    anonymise_action = transition_status.add_action('anonymise')
+    anonymise_action.mode = 'intermediate'
+
+    redirect_action = transition_status.add_action('redirect_to_url')
+    redirect_action.url = 'https://test.com/'
+
+    register_comment = transition_status.add_action('register-comment')
+    register_comment.comment = 'Hello'
+
+    transition_status.add_action('modify_criticality')
+
     jump = transition_status.add_action('jump')
     jump.status = end_status.id
 
@@ -896,7 +1136,7 @@ def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
     testdef.run(formdef)
 
     actions = testdef.workflow_tests.actions
-    assert len(actions) == 9
+    assert len(actions) == 13
 
     assert actions[0].key == 'assert-status'
     assert actions[0].status_name == 'Status with timeout jump'
@@ -914,6 +1154,10 @@ def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
     assert actions[5].key == 'assert-email'
     assert actions[6].key == 'assert-backoffice-field'
     assert actions[7].key == 'assert-sms'
+    assert actions[8].key == 'assert-anonymise'
+    assert actions[9].key == 'assert-redirect'
+    assert actions[10].key == 'assert-history-message'
+    assert actions[11].key == 'assert-criticality'
 
     assert actions[-1].key == 'assert-status'
     assert actions[-1].status_name == 'End status'

wcs/admin/tests.py

@@ -155,6 +155,7 @@ class TestPage(FormBackOfficeStatusPage):
             self.testdef = TestDef.get(component)
         except KeyError:
             raise TraversalError()
+        self.testdef.formdef = objectdef
 
         filled = self.testdef.build_formdata(objectdef, include_fields=True)
         super().__init__(objectdef, filled)

wcs/templates/wcs/backoffice/workflow-tests.html

@@ -31,9 +31,11 @@
           </span>
         </span>
         <p class="commands">
-          <span class="edit">
-            <a href="{{ action.id }}/" rel="popup" title="{% trans "Edit" %}">{% trans "Edit" %}</a>
-          </span>
+          {% if action.editable %}
+            <span class="edit">
+              <a href="{{ action.id }}/" rel="popup" title="{% trans "Edit" %}">{% trans "Edit" %}</a>
+            </span>
+          {% endif %}
           <span class="duplicate">
             <a href="{{ action.id }}/duplicate" title="{% trans "Duplicate" %}">{% trans "Duplicate" %}</a>
           </span>

wcs/wf/anonymise.py

@@ -82,5 +82,15 @@ class AnonymiseWorkflowStatusItem(WorkflowStatusItem):
                 default_value=self.__class__.mode,
             )
 
+    def get_workflow_test_action(self, formdata, *args, **kwargs):
+        original_perform = self.perform
+
+        def perform(formdata):
+            original_perform(formdata)
+            formdata.anonymisation_performed = True
+
+        setattr(self, 'perform', perform)
+        return self
+
 
 register_item_class(AnonymiseWorkflowStatusItem)

wcs/wf/criticality.py

@@ -86,5 +86,8 @@ class ModifyCriticalityWorkflowStatusItem(WorkflowStatusItem):
         elif self.mode == MODE_SET:
             formdata.set_criticality_level(int(self.absolute_value))
 
+    def get_workflow_test_action(self, *args, **kwargs):
+        return self
+
 
 register_item_class(ModifyCriticalityWorkflowStatusItem)

wcs/wf/display_message.py

@@ -169,5 +169,8 @@ class DisplayMessageWorkflowStatusItem(WorkflowStatusItem):
         location = '%sitems/%s/' % (base_location, self.id)
         yield location, None, self.message
 
+    def get_workflow_test_action(self, *args, **kwargs):
+        return self
+
 
 register_item_class(DisplayMessageWorkflowStatusItem)

wcs/wf/redirect_to_url.py

@@ -63,5 +63,16 @@ class RedirectToUrlWorkflowStatusItem(WorkflowStatusItem):
             return  # don't redirect
         return url
 
+    def get_workflow_test_action(self, formdata, *args, **kwargs):
+        original_perform = self.perform
+
+        def perform(formdata):
+            url = original_perform(formdata)
+            formdata.redirect_to_url = url
+            return url
+
+        setattr(self, 'perform', perform)
+        return self
+
 
 register_item_class(RedirectToUrlWorkflowStatusItem)

wcs/wf/register_comment.py

@@ -228,20 +228,34 @@ class RegisterCommenterWorkflowStatusItem(WorkflowStatusItem):
 
         # the comment can use attachments done above
         if comment:
-            try:
-                formdata.evolution[-1].add_part(
-                    JournalEvolutionPart(formdata, get_publisher().translate(comment), self.to, self.level)
-                )
+            part = self.get_journal_evolution_part(formdata, comment)
+            if part:
+                formdata.evolution[-1].add_part(part)
                 formdata.store()
-            except TemplateError as e:
-                get_publisher().record_error(
-                    _('Error in template, comment could not be generated'), formdata=formdata, exception=e
-                )
+
+    def get_journal_evolution_part(self, formdata, comment):
+        try:
+            return JournalEvolutionPart(formdata, get_publisher().translate(comment), self.to, self.level)
+        except TemplateError as e:
+            get_publisher().record_error(
+                _('Error in template, comment could not be generated'), formdata=formdata, exception=e
+            )
 
     def i18n_scan(self, base_location):
         location = '%sitems/%s/' % (base_location, self.id)
         if not self.comment_template:
             yield location, None, self.comment
 
+    def get_workflow_test_action(self, formdata, *args, **kwargs):
+        original_get_journal_evolution_part = self.get_journal_evolution_part
+
+        def get_journal_evolution_part(formdata, comment):
+            part = original_get_journal_evolution_part(formdata, comment)
+            formdata.history_messages.append(part.content)
+            return part
+
+        setattr(self, 'get_journal_evolution_part', get_journal_evolution_part)
+        return self
+
 
 register_item_class(RegisterCommenterWorkflowStatusItem)

wcs/workflow_tests.py

@@ -28,6 +28,7 @@ from wcs.qommon.form import (
     RadiobuttonsWidget,
     SingleSelectWidget,
     StringWidget,
+    TextWidget,
     WidgetList,
 )
 from wcs.qommon.humantime import humanduration2seconds, seconds2humanduration, timewords
@@ -90,9 +91,7 @@ class WorkflowTests(XmlStorableObject):
         formdata.workflow_test = True
 
         formdata.frozen_receipt_time = formdata.receipt_time
-        formdata.sent_sms = []
-        formdata.sent_emails = []
-        formdata.used_webservice_responses = self.testdef.used_webservice_responses = []
+        self.reset_formdata_test_attributes(formdata)
 
         formdata.perform_workflow()
         for action in self.actions:
@@ -102,9 +101,7 @@ class WorkflowTests(XmlStorableObject):
                 continue
 
             if not action.is_assertion:
-                formdata.sent_sms.clear()
-                formdata.sent_emails.clear()
-                formdata.used_webservice_responses.clear()
+                self.reset_formdata_test_attributes(formdata)
 
             try:
                 action.perform(formdata)
@@ -127,6 +124,14 @@ class WorkflowTests(XmlStorableObject):
 
         formdata.store = lambda *args, **kwargs: None
 
+    def reset_formdata_test_attributes(self, formdata):
+        formdata.sent_sms = []
+        formdata.sent_emails = []
+        formdata.used_webservice_responses = self.testdef.used_webservice_responses = []
+        formdata.anonymisation_performed = False
+        formdata.redirect_to_url = None
+        formdata.history_messages = []
+
     def get_new_action_id(self):
         if not self.actions:
             return '1'
@@ -146,6 +151,10 @@ class WorkflowTests(XmlStorableObject):
             'set-backoffice-fields': AssertBackofficeFieldValues,
             'button': ButtonClick,
             'timeout-jump': SkipTime,
+            'anonymise': AssertAnonymise,
+            'redirect_to_url': AssertRedirect,
+            'register-comment': AssertHistoryMessage,
+            'modify_criticality': AssertCriticality,
         }
 
         previous_trace = None
@@ -196,6 +205,7 @@ class WorkflowTestAction(XmlStorableObject):
 
     optional_fields = []
     is_assertion = True
+    editable = True
 
     XML_NODES = [
         ('id', 'str'),
@@ -760,3 +770,163 @@ class AssertSMS(WorkflowTestAction):
             title=_('Body'),
             value=self.body,
         )
+
+
+class AssertAnonymise(WorkflowTestAction):
+    label = _('Assert anonymisation is performed')
+
+    key = 'assert-anonymise'
+
+    editable = False
+    details_label = ''
+
+    def perform(self, formdata):
+        if not formdata.anonymisation_performed:
+            raise WorkflowTestError(_('Form was not anonymised.'))
+
+
+class AssertRedirect(WorkflowTestAction):
+    label = _('Assert redirect is performed')
+
+    key = 'assert-redirect'
+    url = None
+
+    XML_NODES = WorkflowTestAction.XML_NODES + [
+        ('url', 'str'),
+    ]
+
+    @property
+    def details_label(self):
+        return self.url
+
+    def perform(self, formdata):
+        if not formdata.redirect_to_url:
+            raise WorkflowTestError(_('No redirection occured.'))
+
+        if formdata.redirect_to_url != self.url:
+            raise WorkflowTestError(
+                _('Expected redirection to %(expected_url)s but was redirected to %(url)s.')
+                % {'expected_url': self.url, 'url': formdata.redirect_to_url}
+            )
+
+    def fill_admin_form(self, form, formdef):
+        form.add(
+            StringWidget,
+            'url',
+            title=_('URL'),
+            value=self.url,
+        )
+
+
+class AssertHistoryMessage(WorkflowTestAction):
+    label = _('Assert history message is displayed')
+    details_label = ''
+
+    key = 'assert-history-message'
+    message = None
+
+    XML_NODES = WorkflowTestAction.XML_NODES + [
+        ('message', 'str'),
+    ]
+
+    def perform(self, formdata):
+        try:
+            message = formdata.history_messages.pop(0)
+        except IndexError:
+            raise WorkflowTestError(_('No history message.'))
+
+        if self.message not in message:
+            details = [
+                _('Displayed history message: %s') % message,
+                _('Expected history message: %s') % self.message,
+            ]
+            raise WorkflowTestError(_('Wrong history message content.'), details=details)
+
+    def fill_admin_form(self, form, formdef):
+        form.add(
+            TextWidget,
+            'message',
+            title=_('Message'),
+            value=self.message,
+            hint=_('Assertion will pass if the text is contained in history message.'),
+        )
+
+
+class AssertAlert(WorkflowTestAction):
+    label = _('Assert alert is displayed')
+    details_label = ''
+
+    key = 'assert-alert'
+    message = None
+
+    XML_NODES = WorkflowTestAction.XML_NODES + [
+        ('message', 'str'),
+    ]
+
+    def perform(self, formdata):
+        messages = formdata.get_workflow_messages()
+
+        for message in messages:
+            if self.message in message:
+                break
+        else:
+            details = [
+                _('Displayed alerts: %s') % (', '.join(messages) if messages else _('None')),
+                _('Expected alert: %s') % self.message,
+            ]
+            raise WorkflowTestError(_('No alert matching message.'), details=details)
+
+    def fill_admin_form(self, form, formdef):
+        form.add(
+            TextWidget,
+            'message',
+            title=_('Message'),
+            value=self.message,
+            hint=_('Assertion will pass if the text is contained in alert message.'),
+        )
+
+
+class AssertCriticality(WorkflowTestAction):
+    label = _('Assert criticality level')
+    empty_form_error = _('Workflow has no criticality levels.')
+
+    key = 'assert-criticality'
+    level_id = None
+
+    XML_NODES = WorkflowTestAction.XML_NODES + [
+        ('level_id', 'str'),
+    ]
+
+    @property
+    def details_label(self):
+        levels = [
+            x for x in self.parent.testdef.formdef.workflow.criticality_levels or [] if x.id == self.level_id
+        ]
+        if not levels:
+            return _('Broken, missing criticality level')
+
+        return _('Criticality is "%s"') % levels[0].name
+
+    def perform(self, formdata):
+        levels = [x for x in formdata.formdef.workflow.criticality_levels or [] if x.id == self.level_id]
+        if not levels:
+            raise WorkflowTestError(_('Broken, missing criticality level'))
+
+        current_level = formdata.get_criticality_level_object()
+        if current_level.id != self.level_id:
+            raise WorkflowTestError(
+                _('Form should have criticality level "%(expected_level)s" but has level "%(level)s".')
+                % {'expected_level': levels[0].name, 'level': current_level.name}
+            )
+
+    def fill_admin_form(self, form, formdef):
+        if not formdef.workflow.criticality_levels:
+            return
+
+        form.add(
+            SingleSelectWidget,
+            'level_id',
+            title=_('Name'),
+            value=self.level_id,
+            options=[(x.id, x.name, x.id) for x in formdef.workflow.criticality_levels],
+        )