wcs/tests/test_workflow_tests.py

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_uuid=response.uuid, 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_uuid=response.uuid, 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_uuid=response.uuid, call_count=1),
        workflow_tests.AssertWebserviceCall(webservice_response_uuid=response2.uuid, 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_uuid=response.uuid, call_count=1),
        workflow_tests.AssertWebserviceCall(webservice_response_uuid=response.uuid, 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_uuid=response.uuid, 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_uuid='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'