1237 lines
38 KiB
Python
1237 lines
38 KiB
Python
import datetime
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
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.wf.jump import JumpWorkflowStatusItem, _apply_timeouts
|
|
from wcs.workflow_tests import WorkflowTestError
|
|
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
|
|
|
|
|
|
@pytest.fixture
|
|
def pub():
|
|
pub = create_temporary_pub()
|
|
|
|
req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
|
|
pub.set_app_dir(req)
|
|
pub.cfg['identification'] = {'methods': ['password']}
|
|
pub.write_cfg()
|
|
|
|
pub.user_class.wipe()
|
|
pub.role_class.wipe()
|
|
FormDef.wipe()
|
|
Workflow.wipe()
|
|
return pub
|
|
|
|
|
|
def test_workflow_tests_ignore_unsupported_items(pub, monkeypatch):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='End status')
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = end_status.id
|
|
|
|
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.AssertStatus(status_name='End status'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
monkeypatch.setattr(
|
|
JumpWorkflowStatusItem, 'get_workflow_test_action', WorkflowStatusItem.get_workflow_test_action
|
|
)
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "End status" but is in status "New status".'
|
|
|
|
|
|
def test_workflow_tests_no_actions(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
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 = []
|
|
|
|
with mock.patch('wcs.workflow_tests.WorkflowTests.run') as mocked_run:
|
|
testdef.run(formdef)
|
|
mocked_run.assert_not_called()
|
|
|
|
|
|
def test_workflow_tests_action_not_configured(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
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.ButtonClick(),
|
|
]
|
|
|
|
with mock.patch('wcs.workflow_tests.ButtonClick.perform') as mocked_perform:
|
|
testdef.run(formdef)
|
|
mocked_perform.assert_not_called()
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='xxx'),
|
|
]
|
|
|
|
with mock.patch('wcs.workflow_tests.ButtonClick.perform') as mocked_perform:
|
|
testdef.run(formdef)
|
|
mocked_perform.assert_called_once()
|
|
|
|
|
|
def test_workflow_tests_new_action_id(pub):
|
|
wf_tests = workflow_tests.WorkflowTests()
|
|
|
|
for i in range(15):
|
|
wf_tests.add_action(workflow_tests.ButtonClick)
|
|
|
|
assert [x.id for x in wf_tests.actions] == [str(i) for i in range(1, 16)]
|
|
|
|
|
|
def test_workflow_tests_button_click(pub):
|
|
role = pub.role_class(name='test role')
|
|
role.store()
|
|
user = pub.user_class(name='test user')
|
|
user.roles = [role.id]
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='End status')
|
|
|
|
jump = new_status.add_action('choice')
|
|
jump.label = 'Go to end status'
|
|
jump.status = end_status.id
|
|
jump.by = [role.id]
|
|
|
|
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.ButtonClick(button_name='Go to end status'),
|
|
workflow_tests.AssertStatus(status_name='End status'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
# change jump target status
|
|
other_status = workflow.add_status(name='Other status')
|
|
jump.status = other_status.id
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "End status" but is in status "Other status".'
|
|
|
|
# hide button from test user
|
|
other_role = pub.role_class(name='test role 2')
|
|
other_role.store()
|
|
jump.by = [other_role.id]
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Button "Go to end status" is not displayed.'
|
|
|
|
# change button label
|
|
jump.by = [role.id]
|
|
jump.label = 'Go to xxx'
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Button "Go to end status" is not displayed.'
|
|
|
|
|
|
def test_workflow_tests_button_click_global_action(pub):
|
|
role = pub.role_class(name='test role')
|
|
role.store()
|
|
user = pub.user_class(name='test user')
|
|
user.roles = [role.id]
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='End status')
|
|
|
|
global_action = workflow.add_global_action('Go to end status')
|
|
global_action.triggers[0].roles = [role.id]
|
|
|
|
sendmail = global_action.add_action('sendmail')
|
|
sendmail.to = ['test@example.org']
|
|
sendmail.subject = 'In new status'
|
|
sendmail.body = 'xxx'
|
|
|
|
jump = global_action.add_action('jump')
|
|
jump.status = end_status.id
|
|
|
|
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.AssertStatus(status_name='End status'),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "End status" but is in status "New status".'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to end status'),
|
|
workflow_tests.AssertEmail(),
|
|
workflow_tests.AssertStatus(status_name='End status'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
# hide button from test user
|
|
user.roles = []
|
|
user.store()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Button "Go to end status" is not displayed.'
|
|
|
|
|
|
def test_workflow_tests_button_click_who(pub):
|
|
role = pub.role_class(name='test role')
|
|
role.store()
|
|
agent_user = pub.user_class(name='agent user')
|
|
agent_user.roles = [role.id]
|
|
agent_user.store()
|
|
other_role = pub.role_class(name='other test role')
|
|
other_role.store()
|
|
other_user = pub.user_class(name='other user')
|
|
other_user.roles = [other_role.id]
|
|
other_user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
jump_by_unknown = workflow.add_status(name='Jump by unknown')
|
|
jump_by_receiver = workflow.add_status(name='Jump by receiver')
|
|
jump_by_submitter = workflow.add_status(name='Jump by submitter')
|
|
jump_by_other_user = workflow.add_status(name='Jump by other user')
|
|
|
|
jump = new_status.add_action('choice')
|
|
jump.label = 'Go to next status'
|
|
jump.status = jump_by_unknown.id
|
|
jump.by = ['unknown']
|
|
|
|
receiver_jump = new_status.add_action('choice')
|
|
receiver_jump.label = 'Go to next status'
|
|
receiver_jump.status = jump_by_receiver.id
|
|
receiver_jump.by = ['_receiver']
|
|
|
|
submitter_jump = new_status.add_action('choice')
|
|
submitter_jump.label = 'Go to next status'
|
|
submitter_jump.status = jump_by_submitter.id
|
|
submitter_jump.by = ['_submitter']
|
|
|
|
other_user_jump = new_status.add_action('choice')
|
|
other_user_jump.label = 'Go to next status'
|
|
other_user_jump.status = jump_by_other_user.id
|
|
other_user_jump.by = [other_role.id]
|
|
|
|
workflow.store()
|
|
|
|
formdef = FormDef()
|
|
formdef.name = 'test title'
|
|
formdef.workflow_id = workflow.id
|
|
formdef.workflow_roles = {'_receiver': role.id}
|
|
formdef.store()
|
|
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
|
|
testdef = TestDef.create_from_formdata(formdef, formdata)
|
|
testdef.agent_id = agent_user.id
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to next status', who='receiver'),
|
|
workflow_tests.AssertStatus(status_name='Jump by receiver'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to next status', who='submitter'),
|
|
workflow_tests.AssertStatus(status_name='Jump by submitter'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to next status', who='other', who_id=other_user.id),
|
|
workflow_tests.AssertStatus(status_name='Jump by other user'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
other_user.remove_self()
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Broken, missing user'
|
|
|
|
# submitter is anonymous
|
|
submitter_jump.by = ['logged-users']
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to next status', who='submitter'),
|
|
workflow_tests.AssertStatus(status_name='Jump by submitter'),
|
|
]
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Button "Go to next status" is not displayed.'
|
|
|
|
# not anonymous submitter
|
|
submitter_user = pub.user_class(name='submitter user')
|
|
submitter_user.email = 'test@example.com'
|
|
submitter_user.store()
|
|
|
|
formdata.user = submitter_user
|
|
|
|
testdef = TestDef.create_from_formdata(formdef, formdata)
|
|
testdef.agent_id = agent_user.id
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to next status', who='submitter'),
|
|
workflow_tests.AssertStatus(status_name='Jump by submitter'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
|
|
def test_workflow_tests_automatic_jump(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='End status')
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = end_status.id
|
|
|
|
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.AssertStatus(status_name='End status'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
new_end_status = workflow.add_status(name='New end status')
|
|
|
|
jump = end_status.add_action('jump')
|
|
jump.status = new_end_status.id
|
|
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "End status" but is in status "New end status".'
|
|
|
|
|
|
def test_workflow_tests_automatic_jump_condition(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
frog_status = workflow.add_status(name='Frog status')
|
|
bear_status = workflow.add_status(name='Bear status')
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = frog_status.id
|
|
jump.condition = {'type': 'django', 'value': 'form_var_animal == "frog"'}
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = bear_status.id
|
|
jump.condition = {'type': 'django', 'value': 'form_var_animal == "bear"'}
|
|
|
|
workflow.store()
|
|
|
|
formdef = FormDef()
|
|
formdef.name = 'test title'
|
|
formdef.workflow_id = workflow.id
|
|
formdef.fields = [
|
|
fields.StringField(id='1', label='Text', varname='animal'),
|
|
]
|
|
formdef.store()
|
|
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.data['1'] = 'frog'
|
|
|
|
testdef = TestDef.create_from_formdata(formdef, formdata)
|
|
testdef.agent_id = user.id
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertStatus(status_name='Frog status'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
testdef.data['fields']['1'] = 'bear'
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "Frog status" but is in status "Bear status".'
|
|
|
|
|
|
@pytest.mark.freeze_time('2024-02-19 12:00')
|
|
def test_workflow_tests_automatic_jump_timeout(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
stalled_status = workflow.add_status(name='Stalled')
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = stalled_status.id
|
|
jump.timeout = 120 * 60 # 2 hours
|
|
jump.condition = {'type': 'django', 'value': 'form_receipt_datetime|age_in_days >= 1'}
|
|
|
|
sendmail = new_status.add_action('sendmail')
|
|
sendmail.to = ['test@example.org']
|
|
sendmail.subject = 'In new status'
|
|
sendmail.body = 'xxx'
|
|
|
|
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.AssertStatus(status_name='New status'),
|
|
workflow_tests.SkipTime(seconds=119 * 60),
|
|
workflow_tests.AssertStatus(status_name='New status'),
|
|
workflow_tests.SkipTime(seconds=60),
|
|
workflow_tests.AssertStatus(status_name='New status'),
|
|
workflow_tests.SkipTime(seconds=24 * 60 * 60),
|
|
workflow_tests.AssertStatus(status_name='Stalled'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
jump.condition = {'type': 'django', 'value': 'form_receipt_datetime|age_in_hours >= 1'}
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "New status" but is in status "Stalled".'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertStatus(status_name='New status'),
|
|
workflow_tests.SkipTime(seconds=119 * 60),
|
|
workflow_tests.AssertStatus(status_name='New status'),
|
|
workflow_tests.SkipTime(seconds=60),
|
|
workflow_tests.AssertStatus(status_name='Stalled'),
|
|
]
|
|
testdef.run(formdef)
|
|
|
|
|
|
@mock.patch('wcs.qommon.emails.send_email')
|
|
def test_workflow_tests_sendmail(mocked_send_email, pub):
|
|
role = pub.role_class(name='test role')
|
|
role.store()
|
|
user = pub.user_class(name='test user')
|
|
user.roles = [role.id]
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='End status')
|
|
|
|
sendmail = new_status.add_action('sendmail')
|
|
sendmail.to = ['test@example.org']
|
|
sendmail.subject = 'In new status'
|
|
sendmail.body = 'xxx'
|
|
|
|
jump = new_status.add_action('choice')
|
|
jump.label = 'Go to end status'
|
|
jump.status = end_status.id
|
|
jump.by = [role.id]
|
|
|
|
sendmail = end_status.add_action('sendmail')
|
|
sendmail.to = ['test@example.org']
|
|
sendmail.subject = 'In end status'
|
|
sendmail.body = 'yyy'
|
|
|
|
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.AssertEmail(
|
|
addresses=['test@example.org'], subject_strings=['In new status'], body_strings=['xxx']
|
|
),
|
|
workflow_tests.ButtonClick(button_name='Go to end status'),
|
|
workflow_tests.AssertStatus(status_name='End status'),
|
|
workflow_tests.AssertEmail(subject_strings=['end status'], body_strings=['yyy']),
|
|
]
|
|
|
|
testdef.run(formdef)
|
|
mocked_send_email.assert_not_called()
|
|
|
|
testdef.workflow_tests.actions.append(workflow_tests.AssertEmail())
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'No email was sent.'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertEmail(subject_strings=['bla'], body_strings=['xxx']),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Email subject does not contain "bla".'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertEmail(body_strings=['xxx', 'bli']),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Email body does not contain "bli".'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertEmail(addresses=['test@example.org', 'other@example.org']),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Email was not sent to address "other@example.org".'
|
|
assert 'Email addresses: test@example.org' in excinfo.value.details
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.ButtonClick(button_name='Go to end status'),
|
|
workflow_tests.AssertEmail(subject_strings=['In new status'], body_strings=['xxx']),
|
|
workflow_tests.AssertEmail(subject_strings=['end status'], body_strings=['yyy']),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Email subject does not contain "In new status".'
|
|
|
|
|
|
def test_workflow_tests_sms(pub):
|
|
pub.cfg['sms'] = {'sender': 'xxx', 'passerelle_url': 'http://passerelle.invalid/'}
|
|
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
|
|
sendsms = new_status.add_action('sendsms')
|
|
sendsms.to = ['0123456789']
|
|
sendsms.body = 'Hello'
|
|
|
|
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.AssertSMS(phone_numbers=['0123456789'], body='Hello'),
|
|
]
|
|
|
|
testdef.run(formdef)
|
|
|
|
testdef.workflow_tests.actions.append(workflow_tests.AssertSMS())
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'No SMS was sent.'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertSMS(phone_numbers=['0612345678']),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'SMS was not sent to 0612345678.'
|
|
assert 'SMS phone numbers: 0123456789' in excinfo.value.details
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertSMS(body='Goodbye'),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'SMS body mismatch.'
|
|
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()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
|
|
workflow.backoffice_fields_formdef.fields = [
|
|
fields.StringField(id='bo1', label='Text'),
|
|
fields.StringField(id='bo2', label='Text 2'),
|
|
]
|
|
|
|
new_status = workflow.add_status(name='New status')
|
|
set_backoffice_fields = new_status.add_action('set-backoffice-fields')
|
|
set_backoffice_fields.fields = [{'field_id': 'bo2', 'value': '{{ form_var_text }}'}]
|
|
|
|
workflow.store()
|
|
|
|
formdef = FormDef()
|
|
formdef.name = 'test title'
|
|
formdef.workflow_id = workflow.id
|
|
formdef.fields = [
|
|
fields.StringField(id='1', label='Text', varname='text'),
|
|
]
|
|
formdef.store()
|
|
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.data['1'] = 'abc'
|
|
|
|
testdef = TestDef.create_from_formdata(formdef, formdata)
|
|
testdef.agent_id = user.id
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertBackofficeFieldValues(id='1', fields=[{'field_id': 'bo2', 'value': 'abc'}]),
|
|
]
|
|
|
|
testdef.run(formdef)
|
|
|
|
testdef.data['fields']['1'] = 'def'
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Wrong value for backoffice field "Text 2" (expected "abc", got "def").'
|
|
|
|
workflow.backoffice_fields_formdef.fields = []
|
|
workflow.store()
|
|
formdef.refresh_from_storage()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Field bo2 not found (expected value "abc").'
|
|
|
|
|
|
def test_workflow_tests_webservice(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='End status')
|
|
|
|
wscall = new_status.add_action('webservice_call')
|
|
wscall.url = 'http://example.com/json'
|
|
wscall.varname = 'test_webservice'
|
|
wscall.qs_data = {'a': 'b'}
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = end_status.id
|
|
jump.condition = {'type': 'django', 'value': 'form_workflow_data_test_webservice_response_foo == "bar"'}
|
|
|
|
wscall = end_status.add_action('webservice_call')
|
|
wscall.url = 'http://example.com/json'
|
|
wscall.varname = 'test_webservice_2'
|
|
wscall.method = 'POST'
|
|
wscall.post_data = {'a': 'b'}
|
|
|
|
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.store()
|
|
|
|
response = WebserviceResponse()
|
|
response.testdef_id = testdef.id
|
|
response.name = 'Fake response'
|
|
response.url = 'http://example.com/json'
|
|
response.payload = '{"foo": "foo"}'
|
|
response.store()
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertStatus(status_name='End status'),
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=1),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Form should be in status "End status" but is in status "New status".'
|
|
|
|
response.payload = '{"foo": "bar"}'
|
|
response.store()
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Webservice response Fake response was used 2 times (instead of 1).'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=2),
|
|
]
|
|
|
|
testdef.run(formdef)
|
|
|
|
response.qs_data = {'a': 'b'}
|
|
response.store()
|
|
|
|
response2 = WebserviceResponse()
|
|
response2.testdef_id = testdef.id
|
|
response2.name = 'Fake response 2'
|
|
response2.url = 'http://example.com/json'
|
|
response2.payload = '{}'
|
|
response2.method = 'POST'
|
|
response2.store()
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=1),
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response2.id, call_count=1),
|
|
]
|
|
|
|
testdef.run(formdef)
|
|
|
|
testdef.workflow_tests.actions = reversed(testdef.workflow_tests.actions)
|
|
|
|
testdef.run(formdef)
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=1),
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=1),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Webservice response Fake response was used 0 times (instead of 1).'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=0),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Webservice response Fake response was used 1 times (instead of 0).'
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id='xxx', call_count=1),
|
|
]
|
|
|
|
with pytest.raises(WorkflowTestError) as excinfo:
|
|
testdef.run(formdef)
|
|
assert str(excinfo.value) == 'Broken, missing webservice response'
|
|
|
|
|
|
def test_workflow_tests_webservice_status_jump(pub):
|
|
user = pub.user_class(name='test user')
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
new_status = workflow.add_status(name='New status')
|
|
end_status = workflow.add_status(name='Error status')
|
|
|
|
wscall = new_status.add_action('webservice_call')
|
|
wscall.url = 'http://example.com/json'
|
|
wscall.varname = 'test_webservice'
|
|
wscall.action_on_4xx = end_status.id
|
|
|
|
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.store()
|
|
|
|
response = WebserviceResponse()
|
|
response.testdef_id = testdef.id
|
|
response.name = 'Fake response'
|
|
response.url = 'http://example.com/json'
|
|
response.payload = '{"foo": "foo"}'
|
|
response.store()
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertStatus(status_name='New status'),
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=1),
|
|
]
|
|
|
|
testdef.run(formdef)
|
|
|
|
response.status_code = 400
|
|
response.store()
|
|
|
|
testdef.workflow_tests.actions = [
|
|
workflow_tests.AssertStatus(status_name='End status'),
|
|
workflow_tests.AssertWebserviceCall(webservice_response_id=response.id, call_count=1),
|
|
]
|
|
|
|
|
|
def test_workflow_tests_create_from_formdata(pub, http_requests, freezer):
|
|
role = pub.role_class(name='test role')
|
|
role.store()
|
|
user = create_user(pub, is_admin=True)
|
|
user.roles = [role.id]
|
|
user.store()
|
|
|
|
workflow = Workflow(name='Workflow One')
|
|
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
|
|
workflow.backoffice_fields_formdef.fields = [fields.StringField(id='bo1', label='Text')]
|
|
|
|
new_status = workflow.add_status('New status', 'new-status')
|
|
status_with_timeout_jump = workflow.add_status('Status with timeout jump', 'status-with-timeout-jump')
|
|
status_with_button = workflow.add_status('Status with button', 'status-with-button')
|
|
transition_status = workflow.add_status('Transition status', 'transition-status')
|
|
transition_status2 = workflow.add_status('Transition status 2', 'transition-status-2')
|
|
end_status = workflow.add_status('End status', 'end-status')
|
|
|
|
jump = new_status.add_action('jump')
|
|
jump.status = status_with_timeout_jump.id
|
|
|
|
jump = status_with_timeout_jump.add_action('jump')
|
|
jump.status = status_with_button.id
|
|
jump.timeout = '{{ 1 }} day'
|
|
|
|
choice = status_with_button.add_action('choice')
|
|
choice.label = 'Accept'
|
|
choice.status = transition_status.id
|
|
choice.by = [role.id]
|
|
|
|
wscall = transition_status.add_action('webservice_call')
|
|
wscall.url = 'http://remote.example.net/json'
|
|
wscall.varname = 'test_webservice'
|
|
|
|
sendmail = transition_status.add_action('sendmail')
|
|
sendmail.to = ['test@example.org']
|
|
sendmail.subject = 'In new status'
|
|
sendmail.body = 'xxx'
|
|
|
|
set_backoffice_fields = transition_status.add_action('set-backoffice-fields')
|
|
set_backoffice_fields.fields = [{'field_id': 'bo1', 'value': 'xxx'}]
|
|
|
|
sendsms = transition_status.add_action('sendsms')
|
|
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')
|
|
|
|
global_action = workflow.add_global_action('Action 1')
|
|
global_action.triggers[0].roles = [role.id]
|
|
|
|
jump = global_action.add_action('jump')
|
|
jump.status = transition_status2.id
|
|
|
|
jump = transition_status2.add_action('jump')
|
|
jump.status = end_status.id
|
|
|
|
workflow.store()
|
|
|
|
formdef = FormDef()
|
|
formdef.name = 'test title'
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdata = formdef.data_class()()
|
|
formdata.user_id = user.id
|
|
formdata.just_created()
|
|
formdata.store()
|
|
|
|
formdata.record_workflow_event('frontoffice-created')
|
|
formdata.perform_workflow()
|
|
formdata.store()
|
|
|
|
freezer.tick(datetime.timedelta(days=2))
|
|
_apply_timeouts(pub)
|
|
|
|
app = login(get_app(pub))
|
|
resp = app.get(formdata.get_url())
|
|
resp.form.submit('button1').follow()
|
|
resp.form.submit('button-action-1').follow()
|
|
formdata.refresh_from_storage()
|
|
assert formdata.status == 'wf-end-status'
|
|
|
|
testdef = TestDef.create_from_formdata(formdef, formdata, add_workflow_tests=True)
|
|
testdef.run(formdef)
|
|
|
|
actions = testdef.workflow_tests.actions
|
|
assert len(actions) == 15
|
|
|
|
assert actions[0].key == 'assert-status'
|
|
assert actions[0].status_name == 'Status with timeout jump'
|
|
|
|
assert actions[1].key == 'skip-time'
|
|
assert actions[1].seconds == 172800
|
|
|
|
assert actions[2].key == 'assert-status'
|
|
assert actions[2].status_name == 'Status with button'
|
|
|
|
assert actions[3].key == 'button-click'
|
|
assert actions[3].button_name == 'Accept'
|
|
|
|
assert actions[4].key == 'assert-webservice-call'
|
|
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[12].key == 'assert-status'
|
|
assert actions[12].status_name == 'Transition status'
|
|
|
|
assert actions[13].key == 'button-click'
|
|
assert actions[13].button_name == 'Action 1'
|
|
|
|
assert actions[-1].key == 'assert-status'
|
|
assert actions[-1].status_name == 'End status'
|