Produits Entr'ouvert » w.c.s.

def test_compute_timeout(two_pubs):

    workflow = Workflow(name='timeout')

    st1 = workflow.add_status('Status1', 'st1')

    st2 = workflow.add_status('Status2', 'st2')

    jump = JumpWorkflowStatusItem()

    jump.id = '_jump'

    jump.by = ['_submitter', '_receiver']

    jump.status = 'st2'

    st1.items.append(jump)

    jump.parent = st1

    workflow.store()

    formdef = FormDef()

    formdef.name = 'baz'

    formdef.fields = [StringField(id='1', label='Test', type='string', varname='foo')]

    formdef.workflow_id = workflow.id

    assert formdef.get_workflow().id == workflow.id

    formdef.store()

    formdata = formdef.data_class()()

    formdata.data = {'1': '10'}

    formdata.just_created()

    formdata.store()

    formdata_id = formdata.id

    two_pubs.substitutions.feed(formdata)

    # Django

    jump.timeout = '{{ form_var_foo|subtract:form_var_foo }}'  # 10-10 = 0

    workflow.store()

    formdata.status = 'wf-st1'

    formdata.store()

    LoggedError.wipe()

    time.sleep(0.3)

    _apply_timeouts(two_pubs)

    assert formdef.data_class().get(formdata_id).status == 'wf-st2'

    assert LoggedError.count() == 0

    jump.timeout = '{{ bad + syntax }}'

    workflow.store()

    formdata.status = 'wf-st1'

    formdata.store()

    LoggedError.wipe()

    _apply_timeouts(two_pubs)

    assert formdef.data_class().get(formdata_id).status == 'wf-st1'

    assert LoggedError.count() == 1

    logged_error = LoggedError.select()[0]

    assert logged_error.summary == 'Failed to compute template'

    jump.timeout = '{{ form_var_foo|divide:3 }}'  # 10/3 = 3.333... not an integer

    workflow.store()

    formdata.status = 'wf-st1'

    formdata.store()

    LoggedError.wipe()

    _apply_timeouts(two_pubs)

    assert formdef.data_class().get(formdata_id).status == 'wf-st1'

    assert LoggedError.count() == 1

    logged_error = LoggedError.select()[0]

    assert "Invalid timeout '3.333" in logged_error.summary

    # Python

    jump.timeout = '=1/int(form_var_foo)'  # 1/10 = 0.1s

    workflow.store()

    formdata.status = 'wf-st1'

    formdata.store()

    time.sleep(0.3)

    LoggedError.wipe()

    _apply_timeouts(two_pubs)

    assert formdef.data_class().get(formdata_id).status == 'wf-st2'

    assert LoggedError.count() == 0

    jump.timeout = '=int(form_var_foo)/0'

    workflow.store()

    formdata.status = 'wf-st1'

    formdata.store()

    LoggedError.wipe()

    _apply_timeouts(two_pubs)

    assert formdef.data_class().get(formdata_id).status == 'wf-st1'

    assert LoggedError.count() == 1

    logged_error = LoggedError.select()[0]

    assert logged_error.summary == 'Failed to compute Python expression'

    assert logged_error.exception_class == 'ZeroDivisionError'

    jump.timeout = '="string"'

    workflow.store()

    formdata.status = 'wf-st1'

    formdata.store()

    LoggedError.wipe()

    _apply_timeouts(two_pubs)

    assert formdef.data_class().get(formdata_id).status == 'wf-st1'

    assert LoggedError.count() == 1

    logged_error = LoggedError.select()[0]

    assert logged_error.summary == "Invalid timeout 'string' (not an integer)"

    assert logged_error.exception_class == 'ValueError'

            try:

                timeout = self.compute(self.timeout, raises=True, formdata=formdata, status_item=self)

            except Exception:

                # already logged by self.compute

                return False

            try:

                timeout = int(timeout)

            except (ValueError, TypeError) as ex:

                from wcs.logged_errors import LoggedError

                LoggedError.record(_('Invalid timeout %r (not an integer)') % timeout,

                                   formdata=formdata, status_item=self, exception=ex)

                return False