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, 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_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_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')
    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'}]

    jump = transition_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.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()
    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) == 8

    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[-1].key == 'assert-status'
    assert actions[-1].status_name == 'End status'