workflow_tests: allow testing criticality workflow action (#88108)
gitea/wcs/pipeline/head This commit looks good
Details
gitea/wcs/pipeline/head This commit looks good
Details
This commit is contained in:
parent
d4c3e7dc4e
commit
6fd4b87ff5
|
@ -9,7 +9,7 @@ from wcs import workflow_tests
|
||||||
from wcs.formdef import FormDef, fields
|
from wcs.formdef import FormDef, fields
|
||||||
from wcs.qommon.http_request import HTTPRequest
|
from wcs.qommon.http_request import HTTPRequest
|
||||||
from wcs.testdef import TestDef, WebserviceResponse
|
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 ..utilities import create_temporary_pub, get_app, login
|
||||||
from .test_all import create_superuser
|
from .test_all import create_superuser
|
||||||
|
@ -525,6 +525,49 @@ def test_workflow_tests_action_assert_alert(pub):
|
||||||
assert 'not configured' not in resp.text
|
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):
|
def test_workflow_tests_action_assert_backoffice_field(pub):
|
||||||
create_superuser(pub)
|
create_superuser(pub)
|
||||||
|
|
||||||
|
|
|
@ -9,7 +9,12 @@ from wcs.qommon.http_request import HTTPRequest
|
||||||
from wcs.testdef import TestDef, WebserviceResponse
|
from wcs.testdef import TestDef, WebserviceResponse
|
||||||
from wcs.wf.jump import JumpWorkflowStatusItem, _apply_timeouts
|
from wcs.wf.jump import JumpWorkflowStatusItem, _apply_timeouts
|
||||||
from wcs.workflow_tests import WorkflowTestError
|
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 .backoffice_pages.test_all import create_user
|
||||||
from .utilities import create_temporary_pub, get_app, login
|
from .utilities import create_temporary_pub, get_app, login
|
||||||
|
@ -783,6 +788,50 @@ def test_workflow_tests_alert(pub):
|
||||||
assert 'Expected alert: Hello 43' 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):
|
def test_workflow_tests_backoffice_fields(pub):
|
||||||
user = pub.user_class(name='test user')
|
user = pub.user_class(name='test user')
|
||||||
user.store()
|
user.store()
|
||||||
|
@ -1053,6 +1102,8 @@ def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
|
||||||
register_comment = transition_status.add_action('register-comment')
|
register_comment = transition_status.add_action('register-comment')
|
||||||
register_comment.comment = 'Hello'
|
register_comment.comment = 'Hello'
|
||||||
|
|
||||||
|
transition_status.add_action('modify_criticality')
|
||||||
|
|
||||||
jump = transition_status.add_action('jump')
|
jump = transition_status.add_action('jump')
|
||||||
jump.status = end_status.id
|
jump.status = end_status.id
|
||||||
|
|
||||||
|
@ -1085,7 +1136,7 @@ def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
|
||||||
testdef.run(formdef)
|
testdef.run(formdef)
|
||||||
|
|
||||||
actions = testdef.workflow_tests.actions
|
actions = testdef.workflow_tests.actions
|
||||||
assert len(actions) == 12
|
assert len(actions) == 13
|
||||||
|
|
||||||
assert actions[0].key == 'assert-status'
|
assert actions[0].key == 'assert-status'
|
||||||
assert actions[0].status_name == 'Status with timeout jump'
|
assert actions[0].status_name == 'Status with timeout jump'
|
||||||
|
@ -1106,6 +1157,7 @@ def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
|
||||||
assert actions[8].key == 'assert-anonymise'
|
assert actions[8].key == 'assert-anonymise'
|
||||||
assert actions[9].key == 'assert-redirect'
|
assert actions[9].key == 'assert-redirect'
|
||||||
assert actions[10].key == 'assert-history-message'
|
assert actions[10].key == 'assert-history-message'
|
||||||
|
assert actions[11].key == 'assert-criticality'
|
||||||
|
|
||||||
assert actions[-1].key == 'assert-status'
|
assert actions[-1].key == 'assert-status'
|
||||||
assert actions[-1].status_name == 'End status'
|
assert actions[-1].status_name == 'End status'
|
||||||
|
|
|
@ -155,6 +155,7 @@ class TestPage(FormBackOfficeStatusPage):
|
||||||
self.testdef = TestDef.get(component)
|
self.testdef = TestDef.get(component)
|
||||||
except KeyError:
|
except KeyError:
|
||||||
raise TraversalError()
|
raise TraversalError()
|
||||||
|
self.testdef.formdef = objectdef
|
||||||
|
|
||||||
filled = self.testdef.build_formdata(objectdef, include_fields=True)
|
filled = self.testdef.build_formdata(objectdef, include_fields=True)
|
||||||
super().__init__(objectdef, filled)
|
super().__init__(objectdef, filled)
|
||||||
|
|
|
@ -86,5 +86,8 @@ class ModifyCriticalityWorkflowStatusItem(WorkflowStatusItem):
|
||||||
elif self.mode == MODE_SET:
|
elif self.mode == MODE_SET:
|
||||||
formdata.set_criticality_level(int(self.absolute_value))
|
formdata.set_criticality_level(int(self.absolute_value))
|
||||||
|
|
||||||
|
def get_workflow_test_action(self, *args, **kwargs):
|
||||||
|
return self
|
||||||
|
|
||||||
|
|
||||||
register_item_class(ModifyCriticalityWorkflowStatusItem)
|
register_item_class(ModifyCriticalityWorkflowStatusItem)
|
||||||
|
|
|
@ -154,6 +154,7 @@ class WorkflowTests(XmlStorableObject):
|
||||||
'anonymise': AssertAnonymise,
|
'anonymise': AssertAnonymise,
|
||||||
'redirect_to_url': AssertRedirect,
|
'redirect_to_url': AssertRedirect,
|
||||||
'register-comment': AssertHistoryMessage,
|
'register-comment': AssertHistoryMessage,
|
||||||
|
'modify_criticality': AssertCriticality,
|
||||||
}
|
}
|
||||||
|
|
||||||
previous_trace = None
|
previous_trace = None
|
||||||
|
@ -883,3 +884,49 @@ class AssertAlert(WorkflowTestAction):
|
||||||
value=self.message,
|
value=self.message,
|
||||||
hint=_('Assertion will pass if the text is contained in alert 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],
|
||||||
|
)
|
||||||
|
|
Loading…
Reference in New Issue