Produits Entr'ouvert » w.c.s.

    resp = resp.click(href='triggers/%s/' % Workflow.get(workflow.id).global_actions[0].triggers[0].id,

            index=0)

    resp = resp.click(href='triggers/%s/' % Workflow.get(workflow.id).global_actions[0].triggers[0].id,

            index=0)

    resp = resp.click(href='triggers/%s/' % Workflow.get(workflow.id).global_actions[0].triggers[0].id,

            index=0)

def test_workflows_global_actions_timeout_triggers(pub):

    create_superuser(pub)

    create_role()

    Workflow.wipe()

    workflow = Workflow(name='foo')

    workflow.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/workflows/%s/' % workflow.id)

    resp = resp.click('add global action')

    resp.forms[0]['name'] = 'Global Action'

    resp = resp.forms[0].submit('submit')

    resp = resp.follow()

    resp = resp.click('Global Action')

    # test removing the existing manual trigger

    resp = resp.click(href='triggers/%s/delete' % Workflow.get(workflow.id).global_actions[0].triggers[0].id)

    resp = resp.forms[0].submit()

    resp = resp.follow()

    assert len(Workflow.get(workflow.id).global_actions[0].triggers) == 0

    # test adding a timeout trigger

    resp.forms[1]['type'] = 'Timeout'

    resp = resp.forms[1].submit()

    resp = resp.follow()

    assert 'Timeout (not configured)' in resp.body

    resp = resp.click(href='triggers/%s/' % Workflow.get(workflow.id).global_actions[0].triggers[0].id, index=0)

    resp.form['timeout'] = '3'

    resp = resp.form.submit('submit')

    assert Workflow.get(workflow.id).global_actions[0].triggers[0].timeout == '3'

def test_global_timeout_trigger():

    wf = Workflow(name='global actions')

    ac1 = wf.add_global_action('Action', 'ac1')

    trigger = ac1.append_trigger('timeout')

    trigger.timeout = '2'

    trigger.anchor = 'creation'

    wf2 = assert_import_export_works(wf, include_id=True)

    assert wf2.global_actions[0].triggers[-1].id == trigger.id

    assert wf2.global_actions[0].triggers[-1].anchor == trigger.anchor

def test_global_timeouts(pub):

    FormDef.wipe()

    Workflow.wipe()

    workflow = Workflow(name='global-timeouts')

    workflow.possible_status = Workflow.get_default_workflow().possible_status[:]

    workflow.criticality_levels = [

        WorkflowCriticalityLevel(name='green'),

        WorkflowCriticalityLevel(name='yellow'),

        WorkflowCriticalityLevel(name='red'),

    ]

    action = workflow.add_global_action('Timeout Test')

    item = action.append_item('modify_criticality')

    trigger = action.append_trigger('timeout')

    trigger.anchor = 'creation'

    trigger.timeout = '2'

    workflow.store()

    formdef = FormDef()

    formdef.name = 'baz'

    formdef.fields = []

    formdef.workflow_id = workflow.id

    formdef.store()

    formdata1 = formdef.data_class()()

    formdata1.just_created()

    formdata1.store()

    # delay didn't expire yet, no change

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    formdata1.receipt_time = time.localtime(time.time()-3*86400)

    formdata1.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    # make sure it's not triggered a second time

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    # change id so it's triggered again

    trigger.id = 'XXX1'

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'red'

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'red'

    # reset formdata to initial state

    formdata1.store()

    trigger.anchor = '1st-arrival'

    trigger.anchor_status_first = None

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    formdata1.evolution[-1].time = time.localtime(time.time()-3*86400)

    formdata1.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    formdata1.store() # reset

    trigger.anchor = 'latest-arrival'

    trigger.anchor_status_latest = None

    workflow.store()

    formdata1.evolution[-1].time = time.localtime()

    formdata1.store()

    formdata1.jump_status('new')

    formdata1.evolution[-1].time = time.localtime(time.time()-7*86400)

    formdata1.jump_status('accepted')

    formdata1.jump_status('new')

    formdata1.evolution[-1].time = time.localtime(time.time()-1*86400)

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    formdata1.evolution[-1].time = time.localtime(time.time()-4*86400)

    formdata1.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    formdata1.store()

    # limit trigger to formdata with "accepted" status

    trigger.anchor_status_latest = 'wf-accepted'

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    formdata1.store()

    # limit trigger to formdata with "new" status

    trigger.anchor_status_latest = 'wf-new'

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    formdata1.store()

    # use python expression as anchor

    #  timestamp

    trigger.anchor = 'python'

    trigger.anchor_expression = repr(time.time())

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    trigger.anchor = 'python'

    trigger.anchor_expression = repr(time.time() - 10*86400)

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    formdata1.store()

    #  datetime object

    trigger.anchor = 'python'

    trigger.anchor_expression = 'datetime.datetime(%s, %s, %s, %s, %s)' % (

            datetime.datetime.now() - datetime.timedelta(days=10)).timetuple()[:5]

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    formdata1.store()

    # string object

    trigger.anchor = 'python'

    trigger.anchor_expression = '"%04d-%02d-%02d"' % (

            datetime.datetime.now() - datetime.timedelta(days=10)).timetuple()[:3]

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'yellow'

    formdata1.store()

    # invalid variable

    trigger.anchor = 'python'

    trigger.anchor_expression = 'Ellipsis'

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    formdata1.store()

    # invalid expression

    trigger.anchor = 'python'

    trigger.anchor_expression = 'XXX'

    workflow.store()

    Workflow.apply_global_action_timeouts()

    assert formdef.data_class().get(formdata1.id).get_criticality_level_object().name == 'green'

    formdata1.store()

            self.trigger.submit_admin_form(form)

                  'update_order', 'edit', 'newtrigger', ('triggers', 'triggers_dir'),

                  'update_triggers_order',

        r += htmltext('<ul id="items-list" class="biglist sortable" data-order-function="update_triggers_order">')

            r += htmltext('<li class="biglistitem" id="trigId_%s">') % trigger.id

            r += command_icon('triggers/%s/delete' % trigger.id, 'remove', popup=True)

            r += htmltext('<div id="new-trigger">')

            r += htmltext('<h3>%s</h3>') % _('New Trigger')

            r += self.get_new_trigger_form().render()

            r += htmltext('</div>')

    def update_triggers_order(self):

        request = get_request()

        new_order = request.form['order'].strip(';').split(';')

        self.action.triggers = [ [x for x in self.action.triggers if x.id == y][0] for y in new_order]

        self.workflow.store()

        return 'ok'

    def newtrigger(self):

        form = self.get_new_trigger_form()

        if not form.is_submitted() or form.has_errors():

            get_session().message = ('error', _('Submitted form was not filled properly.'))

            return redirect('.')

        if form.get_widget('type').parse():

            self.action.append_trigger(form.get_widget('type').parse())

        else:

            get_session().message = ('error', _('Submitted form was not filled properly.'))

            return redirect('.')

        self.workflow.store()

        return redirect('.')

    def get_new_trigger_form(self):

        form = Form(enctype='multipart/form-data', action='newtrigger')

        available_triggers = [

            ('timeout', _('Timeout')),

            ('manual', _('Manual')),

        ]

        form.add(SingleSelectWidget, 'type', title=_('Type'),

                required=True, options=available_triggers)

        form.add_submit('submit', _('Add'))

        return form

div#new-trigger h3,

function prepare_dynamic_widgets()

{

  $('[data-dynamic-display-parent]').on('change keyup', function() {

    var sel1 = '[data-dynamic-display-child-of="' + $(this).attr('name') + '"]';

    var sel2 = '[data-dynamic-display-value="' + $(this).val() + '"]';

    $(sel1).hide();

    $(sel1 + sel2).show();

  });

  $('[data-dynamic-display-child-of]').hide();

  $('select[data-dynamic-display-parent]').trigger('change');

  $('[data-dynamic-display-parent]:checked').trigger('change');

}

  prepare_dynamic_widgets();

  $(document).on('wcs:dialog-loaded', prepare_dynamic_widgets);

import collections

import datetime

import sys

import time

import uuid

from qommon.cron import CronJob

from qommon.misc import C_, get_as_datetime

from qommon.publisher import get_publisher_class

    @classmethod

    def apply_global_action_timeouts(cls):

        for workflow in cls.select():

            WorkflowGlobalActionTimeoutTrigger.apply(workflow)

        if include_id and self.id:

            node.attrib['id'] = self.id

        if include_id and elem.attrib.get('id'):

            self.id = elem.attrib.get('id')

    def submit_admin_form(self, form):

        for f in self.get_parameters():

            widget = form.get_widget(f)

            if widget:

                value = widget.parse()

                if hasattr(self, '%s_parse' % f):

                    value = getattr(self, '%s_parse' % f)(value)

                setattr(self, f, value)

class WorkflowGlobalActionTimeoutTriggerMarker(object):

    def __init__(self, timeout_id):

        self.timeout_id = timeout_id

class WorkflowGlobalActionTimeoutTrigger(WorkflowGlobalActionTrigger):

    key = 'timeout'

    anchor = None

    anchor_expression = None

    anchor_status_first = None

    anchor_status_latest = None

    timeout = None

    def get_parameters(self):

        return ('anchor', 'anchor_expression', 'anchor_status_first',

                'anchor_status_latest', 'timeout')

    def get_anchor_labels(self):

        return collections.OrderedDict([

                ('creation', _('Creation')),

                ('1st-arrival', _('First arrival in status')),

                ('latest-arrival', _('Latest arrival in status')),

                ('python', _('Python expression')),

        ])

    def render_as_line(self):

        if self.anchor and self.timeout:

            return _('Timeout, %(timeout)s, relative to: %(anchor)s') % {

                    'anchor': self.get_anchor_labels().get(self.anchor).lower(),

                    'timeout': _('%s days') % self.timeout}

        else:

            return _('Timeout (not configured)')

    def form(self, workflow):

        form = Form(enctype='multipart/form-data')

        options = self.get_anchor_labels().items()

        form.add(SingleSelectWidget, 'anchor', title=_('Anchor'),

                 options=options, value=self.anchor, required=True,

                 attrs={'data-dynamic-display-parent': 'true'})

        form.add(StringWidget, 'anchor_expression', title=_('Expression'), size=80,

                 value=self.anchor_expression,

                 attrs={'data-dynamic-display-child-of': 'anchor',

                        'data-dynamic-display-value': _('Python expression')})

        possible_status = [(None, _('Current Status'), None)]

        possible_status.extend([('wf-%s' % x.id, x.name, x.id) for x in workflow.possible_status])

        form.add(SingleSelectWidget, 'anchor_status_first', title=_('Status'),

                 options=possible_status,

                 value=self.anchor_status_first,

                 attrs={'data-dynamic-display-child-of': 'anchor',

                        'data-dynamic-display-value': _('First arrival in status')}

                 )

        form.add(SingleSelectWidget, 'anchor_status_latest', title=_('Status'),

                 options=possible_status,

                 value=self.anchor_status_latest,

                 attrs={'data-dynamic-display-child-of': 'anchor',

                        'data-dynamic-display-value': _('Latest arrival in status')}

                 )

        form.add(StringWidget, 'timeout', title=_('Timeout'),

                 value=self.timeout,

                 hint=_('Number of days, relative to anchor point.'))

        return form

    def must_trigger(self, formdata):

        anchor_date = None

        if self.anchor == 'creation':

            anchor_date = formdata.receipt_time

        elif self.anchor == '1st-arrival':

            anchor_status = self.anchor_status_first or formdata.status

            for evolution in formdata.evolution:

                if evolution.status == anchor_status:

                    anchor_date = evolution.time

                    break

        elif self.anchor == 'latest-arrival':

            anchor_status = self.anchor_status_latest or formdata.status

            for evolution in reversed(formdata.evolution):

                if evolution.status == anchor_status:

                    anchor_date = evolution.time

                    break

        elif self.anchor == 'python':

            variables = get_publisher().substitutions.get_context_variables()

            try:

                anchor_date = eval(self.anchor_expression,

                        get_publisher().get_global_eval_dict(), variables)

            except:

                # get the variables in the locals() namespace so they are

                # displayed within the trace.

                expression = self.anchor_expression

                global_variables = get_publisher().get_global_eval_dict()

                get_publisher().notify_of_exception(sys.exc_info(), context='[TIMEOUTS]')

        # convert anchor_date to datetime.datetime()

        if isinstance(anchor_date, datetime.datetime):

            pass

        elif isinstance(anchor_date, datetime.date):

            pass

        elif isinstance(anchor_date, time.struct_time):

            anchor_date = datetime.datetime.fromtimestamp(time.mktime(anchor_date))

        elif isinstance(anchor_date, basestring):

            try:

                anchor_date = get_as_datetime(anchor_date)

            except ValueError:

                get_publisher().notify_of_exception(sys.exc_info(), context='[TIMEOUTS]')

                anchor_date = None

        elif anchor_date:

            # timestamp

            try:

                anchor_date = datetime.datetime.fromtimestamp(anchor_date)

            except TypeError:

                get_publisher().notify_of_exception(sys.exc_info(), context='[TIMEOUTS]')

                anchor_date = None

        if anchor_date is None:

            return False

        anchor_date = anchor_date + datetime.timedelta(days=int(self.timeout))

        return bool(datetime.datetime.now() > anchor_date)

    @classmethod

    def apply(cls, workflow):

        triggers = []

        for action in workflow.global_actions or []:

            triggers.extend([(action, x) for x in action.triggers or [] if

                             isinstance(x, WorkflowGlobalActionTimeoutTrigger)])

        if not triggers:

            return

        formdefs = [x for x in FormDef.select() if x.workflow_id == workflow.id]

        not_endpoint_status = workflow.get_not_endpoint_status()

        not_endpoint_status_ids = ['wf-%s' % x.id for x in not_endpoint_status]

        for formdef in formdefs:

            open_formdata_ids = []

            data_class = formdef.data_class()

            for status in not_endpoint_status_ids:

                open_formdata_ids.extend(data_class.get_ids_with_indexed_value('status', status))

            for formdata in data_class.get_ids(open_formdata_ids, ignore_errors=True):

                get_publisher().substitutions.reset()

                get_publisher().substitutions.feed(get_publisher())

                get_publisher().substitutions.feed(formdef)

                get_publisher().substitutions.feed(formdata)

                seen_triggers = []

                for evolution in formdata.evolution:

                    for part in evolution.parts or []:

                        if isinstance(part, WorkflowGlobalActionTimeoutTriggerMarker):

                            seen_triggers.append(part.timeout_id)

                for action, trigger in triggers:

                    if trigger.id in seen_triggers:

                        continue # already triggered

                    if trigger.must_trigger(formdata):

                        if not formdata.evolution:

                            continue

                        formdata.evolution[-1].add_part(

                                WorkflowGlobalActionTimeoutTriggerMarker(trigger.id))

                        formdata.store()

                        perform_items(action.items, formdata)

                        break

            'manual': WorkflowGlobalActionManualTrigger,

            'timeout': WorkflowGlobalActionTimeoutTrigger

        o.id = str(uuid.uuid4())

if get_publisher_class():

    # every hour check for global action timeouts.

    get_publisher_class().register_cronjob(

            CronJob(Workflow.apply_global_action_timeouts, hours=range(24)))