Produits Entr'ouvert » w.c.s.

def test_backoffice_multi_actions(pub):

    create_superuser(pub)

    create_environment(pub)

    formdef = FormDef.get_by_urlname('form-title')

    app = login(get_app(pub))

    resp = app.get('/backoffice/management/form-title/')

    assert not 'id="multi-actions"' in resp.body

    workflow = Workflow.get_default_workflow()

    workflow.id = '2'

    action = workflow.add_global_action('FOOBAR')

    jump = action.append_item('jump')

    jump.status = 'finished'

    trigger = action.triggers[0]

    trigger.roles = ['whatever']

    workflow.store()

    formdef.workflow_id = workflow.id

    formdef.store()

    resp = app.get('/backoffice/management/form-title/')

    assert not 'id="multi-actions"' in resp.body

    trigger.roles = [x.id for x in Role.select() if x.name == 'foobar']

    workflow.store()

    resp = app.get('/backoffice/management/form-title/')

    assert 'id="multi-actions"' in resp.body

    ids = []

    for checkbox in resp.forms[0].fields['select[]'][1:6]:

        ids.append(checkbox._value)

        checkbox.checked = True

    resp = resp.forms[0].submit('button-action-1')

    assert '?job=' in resp.location

    resp = resp.follow()

    assert 'Executing task "FOOBAR" on forms' in resp.body

    assert '>completed<' in resp.body

    for id in ids:

        assert formdef.data_class().get(id).status == 'wf-finished'

    draft_ids = [x.id for x in formdef.data_class().select() if x.status == 'draft']

    resp = app.get('/backoffice/management/form-title/')

    assert resp.forms[0].fields['select[]'][0]._value == '_all'

    resp.forms[0].fields['select[]'][0].checked = True

    resp = resp.forms[0].submit('button-action-1')

    for formdata in formdef.data_class().select():

        if formdata.id in draft_ids:

            assert formdata.status == 'draft'

        else:

            assert formdata.status == 'wf-finished'

    for formdata in formdef.data_class().select():

        if formdata.status != 'draft':

            formdata.jump_status('new')

            formdata.store()

    # action for other role

    action2 = workflow.add_global_action('OTHER ACTION')

    jump = action2.append_item('jump')

    jump.status = 'accepted'

    trigger = action2.triggers[0]

    trigger.roles = ['whatever']

    workflow.store()

    resp = app.get('/backoffice/management/form-title/')

    assert 'id="multi-actions"' in resp.body

    assert not 'OTHER ACTION' in resp.body

    # action for function

    trigger.roles = ['_foobar']

    workflow.store()

    resp = app.get('/backoffice/management/form-title/')

    assert 'id="multi-actions"' in resp.body

    assert 'OTHER ACTION' not in resp.body

    workflow.roles['_foobar'] = 'Foobar'

    workflow.store()

    resp = app.get('/backoffice/management/form-title/')

    assert 'id="multi-actions"' in resp.body

    assert 'OTHER ACTION' in resp.body

    # alter some formdata to simulate dispatch action

    stable_ids = []

    for checkbox in resp.forms[0].fields['select[]'][1:6]:

        formdata = formdef.data_class().get(checkbox._value)

        formdata.workflow_roles = {'_foobar': formdef.workflow_roles['_receiver']}

        formdata.store()

        stable_ids.append(formdata.id)

    resp = app.get('/backoffice/management/form-title/')

    assert 'OTHER ACTION' in resp.body

    resp.forms[0].fields['select[]'][0].checked = True  # _all

    resp = resp.forms[0].submit('button-action-2')

    assert '?job=' in resp.location

    resp = resp.follow()

    assert 'Executing task "OTHER ACTION" on forms' in resp.body

    # check only dispatched formdata have been moved by global action executed

    # on all formdatas

    for formdata in formdef.data_class().select():

        if formdata.id in draft_ids:

            assert formdata.status == 'draft'

        elif formdata.id in stable_ids:

            assert formdata.status == 'wf-accepted'

        else:

            assert formdata.status != 'wf-accepted'

    def get_multi_actions(self, user):

        global_actions = self.formdef.workflow.get_global_manual_actions()

        workflow_roles = self.formdef.workflow.roles or {}

        mass_actions = []

        for action_dict in global_actions:

            action_dict['roles'] = [x for x in user.get_roles() if x in action_dict.get('roles') or []]

            if action_dict['functions'] or action_dict['roles']:

                mass_actions.append(action_dict)

        return mass_actions

        if 'job' in get_request().form:

            return self.job_multi()

        multi_actions = self.get_multi_actions(get_request().user)

        if not get_request().form.get('ajax') == 'true':

            multi_form = Form(id='multi-actions')

            for action in multi_actions:

                attrs = {}

                if action.get('functions'):

                    for function in action.get('functions'):

                        attrs['data-visible_for_%s' % function] = 'true'

                else:

                    attrs['data-visible_for_all'] = 'true'

                multi_form.add_submit('button-action-%s' % action['action'].id, action['action'].name, attrs=attrs)

            if multi_form.is_submitted() and get_request().form.get('select[]'):

                for action in multi_actions:

                    if multi_form.get_submit() == 'button-action-%s' % action['action'].id:

                        return self.submit_multi(

                                action,

                                selected_filter=selected_filter,

                                query=query,

                                criterias=criterias)

                        order_by=order_by, criterias=criterias,

                        include_checkboxes=bool(multi_actions))

        if multi_actions:

            multi_form.widgets.append(HtmlWidget(table))

            r += multi_form.render()

        else:

            r += table

    def submit_multi(self, action, selected_filter, query, criterias):

        class ActionJob(object):

            def __init__(self, formdef, query_string, action, item_ids):

                self.formdef = formdef

                self.query_string = query_string

                self.action = action

                self.item_ids = item_ids

                self.user = get_request().user

            def execute(self, job=None):

                formdatas = self.formdef.data_class().get_ids(self.item_ids)

                publisher = get_publisher()

                for formdata in formdatas:

                    publisher.substitutions.reset()

                    publisher.substitutions.feed(publisher)

                    publisher.substitutions.feed(self.formdef)

                    publisher.substitutions.feed(formdata)

                    formdata.perform_global_action(self.action['action'].id, self.user)

        item_ids = get_request().form['select[]']

        if '_all' in item_ids:

            item_ids = FormDefUI(self.formdef).get_listing_item_ids(

                        selected_filter, user=get_request().user, query=query,

                        criterias=criterias)

        action_job = ActionJob(self.formdef, get_request().get_query(), action, item_ids)

        job = get_response().add_after_job(

                N_('Executing task "%s" on forms') % action['action'].name,

                action_job.execute)

        job.query_string = get_request().get_query()

        job.store()

        return redirect('./?job=%s' % job.id)

    def job_multi(self):

        try:

            job = AfterJob.get(get_request().form.get('job'))

        except KeyError:

            return redirect('.')

        html_top('management', title=_('Executing Task'))

        r = TemplateIO(html=True)

        r += get_session().display_message()

        get_response().add_javascript(['jquery.js', 'afterjob.js'])

        r += htmltext('<dl class="job-status">')

        r += htmltext('<dt>')

        r += _(job.label)

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

        r += htmltext('<dd>')

        r += htmltext('<span class="afterjob" id="%s">') % job.id

        r += _(job.status)

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

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

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

        r += htmltext('<div class="done">')

        r += htmltext('<a data-redirect-auto="true" href="./?%s">%s</a>') % (job.query_string, _('Back to Listing'))

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

        return r.getvalue()

        for k in self.django_request.POST:

            if k.endswith('[]'):

                v = [x.encode(site_charset) for x in self.django_request.POST.getlist(k)]

            else:

                v = self.django_request.POST[k]

                if isinstance(v, unicode):

                    v = v.encode(site_charset)

    def perform_global_action(self, action_id, user):

        from wcs.workflows import perform_items

        for action in self.formdef.workflow.get_global_actions_for_user(self, user):

            if action.id != action_id:

                continue

            perform_items(action.items, self)

            break

                    query=None, order_by=None, criterias=None,

                    include_checkboxes=False):

        if include_checkboxes:

            r += htmltext('<col/>')  # checkbox

        if include_checkboxes:

            r += htmltext('<th class="select"><input type="checkbox" name="select[]" value="_all"/></th>')

        r += htmltext(self.tbody(fields, items, url_action,

            include_checkboxes=include_checkboxes))

    def get_listing_item_ids(self, selected_filter='all', query=None, order_by=None, user=None, criterias=None, anonymise=False):

        return item_ids

    def get_listing_items(self, selected_filter='all', offset=None,

            limit=None, query=None, order_by=None, user=None, criterias=None, anonymise=False):

        user = user or get_request().user

        formdata_class = self.formdef.data_class()

        item_ids = self.get_listing_item_ids(

                selected_filter=selected_filter,

                query=query,

                user=user,

                criterias=criterias,

                anonymise=anonymise)

    def tbody(self, fields=None, items=None, url_action=None, include_checkboxes=False):

        user = get_request().user

        visited_objects = get_publisher().get_visited_objects(exclude_user=user.id)

            if include_checkboxes:

                r += htmltext('<td class="select"><input type="checkbox" name="select[]" ')

                r += htmltext('value="%s"') % filled.id

                workflow_roles = {}

                workflow_roles.update(self.formdef.workflow_roles)

                if filled.workflow_roles:

                    workflow_roles.update(filled.workflow_roles)

                for function_key, function_value in workflow_roles.items():

                    if function_value in user.get_roles():

                        r += htmltext(' data-is-%s="true" ' % function_key)

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

#listing tbody tr td {

	cursor: pointer;

}

#listing tbody tr td.select {

	cursor: default;

}

table#listing .select {

	width: 1rem;

	text-align: center;

}

table#listing .select input {

	margin: 2px 0;

}

#main-content form#multi-actions {

	padding: 0;

}

#main-content form#multi-actions div.buttons {

	padding: 0 0.5rem;

}

table#listing tr.checked td {

	background: #ddf;

}

                        if ($('div.done a[data-redirect-auto]').length) {

                            window.location = $('div.done a[data-redirect-auto]').attr('href');

                        }

    var $target = $(event.target);

    if ($target.is('input[type=checkbox]')) {

      return false;

    }

    if ($target.is('td.select')) {

      $target.find('input').click();

      return false;

    }

  if ($('#listing tbody input[type=checkbox]:checked').length == 0) {

    $('form#multi-actions div.buttons').hide();

  } else {

    $('form#multi-actions div.buttons').show();

  }

  $('#listing tbody input[type=checkbox]').each(function() {

    if ($(this).is(':checked')) {

      $(this).parents('tr').addClass('checked');

    } else {

      $(this).parents('tr').removeClass('checked');

    }

  });

  $('#listing input[type=checkbox]').on('change', function() {

    if ($(this).is(':checked')) {

      if ($(this).is('[value=_all]')) {

        $(this).parents('table').find('tbody td.select input').prop('checked', true);

        $(this).parents('table').find('tbody tr').addClass('checked');

      } else {

        $(this).parents('tr').addClass('checked');

      }

    } else {

      if ($(this).is('[value=_all]')) {

        $(this).parents('table').find('tbody td.select input').prop('checked', false);

        $(this).parents('table').find('tbody tr').removeClass('checked');

      } else {

        $(this).parents('tr').removeClass('checked');

        $('#listing input[type=checkbox][value=_all]').prop('checked', false);

      }

    }

    if ($('#listing tbody input[type=checkbox]:checked').length == 0) {

      $('form#multi-actions div.buttons').hide();

      return;

    } else {

      $('form#multi-actions div.buttons button').each(function(idx, elem) {

        var visible = false;

        for (var key in $(elem).first().data()) {

          if (key == 'visible_for_all') {

            visible = true;

            break;

          }

          if ($('input[type=checkbox][data-is-' + key.substr(12) + ']:checked').length) {

            visible = true;

            break;

          }

        }

        if (visible) {

          $(elem).parents('div.widget').show();

        } else {

          $(elem).parents('div.widget').hide();

        }

      });

      $('form#multi-actions div.buttons').show();

    }

  });

  if ($('#multi-actions input:checked').length) {

    // disable autorefresh when multiselection is enabled

    return;

  }

    def get_global_manual_actions(self):

        actions = []

        for action in self.global_actions or []:

            roles = []

            for trigger in action.triggers or []:

                if not isinstance(trigger, WorkflowGlobalActionManualTrigger):

                    continue

                roles.extend(trigger.roles or [])

            functions = [x for x in roles if x in self.roles]

            roles = [x for x in roles if x not in self.roles]

            if functions or roles:

                actions.append({'action': action, 'roles': roles, 'functions': functions})

        return actions