Produits Entr'ouvert » w.c.s.

def test_backoffice_multi_actions_interactive(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 'id="multi-actions"' in resp.text  # always there

    workflow = Workflow.get_default_workflow()

    workflow.id = '2'

    action = workflow.add_global_action('FOOBAR')

    form_action = action.add_action('form')

    form_action.varname = 'blah'

    form_action.formdef = WorkflowFormFieldsFormDef(item=form_action)

    form_action.formdef.fields.append(

        fields.StringField(id='1', label='Test', varname='test', type='string', required=True)

    )

    register_comment = action.add_action('register-comment')

    register_comment.comment = 'HELLO {{ form_workflow_form_blah_var_test }}'

    trigger = action.triggers[0]

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

    workflow.store()

    formdef.workflow_id = workflow.id

    formdef.store()

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

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

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

        checkbox.checked = True

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

    assert '/actions/' in resp.location

    resp = resp.follow()

    resp = resp.follow()  # back to form listing

    assert 'Error: empty action' in resp.text

    form_action.by = trigger.roles

    workflow.store()

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

    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 '/actions/' in resp.location

    resp = resp.follow()

    assert '5 selected forms' in resp.text

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

    assert resp.pyquery('#form_error_fblah_1').text() == 'required field'

    resp.form['fblah_1'] = 'GLOBAL INTERACTIVE ACTION'

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

    assert '?job=' in resp.location

    resp = resp.follow()

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

    assert '>completed<' in resp.text

    assert (

        resp.pyquery.find('[data-redirect-auto]').attr['href']

        == '/backoffice/management/form-title/?limit=20'

    )

    for id in ids:

        pub.substitutions.reset()

        pub.substitutions.feed(formdef.data_class().get(id))

        context = pub.substitutions.get_context_variables(mode='lazy')

        assert context['form_number_raw'] == id

        assert context['form_workflow_form_blah_var_test'].get_value() == 'GLOBAL INTERACTIVE ACTION'

def test_backoffice_global_interactive_action(pub):

    create_user(pub)

    formdef = FormDef()

    formdef.name = 'test global action'

    formdef.fields = []

    workflow = Workflow.get_default_workflow()

    workflow.id = '2'

    action = workflow.add_global_action('FOOBAR')

    display = action.add_action('displaymsg')

    display.message = 'This is a message'

    display.to = []

    form_action = action.add_action('form')

    form_action.varname = 'blah'

    form_action.formdef = WorkflowFormFieldsFormDef(item=form_action)

    form_action.formdef.fields.append(

        fields.StringField(id='1', label='Test', varname='test', type='string', required=True)

    )

    register_comment = action.add_action('register-comment')

    register_comment.comment = 'HELLO {{ form_workflow_form_blah_var_test }}'

    trigger = action.triggers[0]

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

    workflow.store()

    formdef.workflow_id = workflow.id

    formdef.workflow_roles = {'_receiver': 1}

    formdef.store()

    formdata = formdef.data_class()()

    formdata.just_created()

    formdata.store()

    app = login(get_app(pub))

    resp = app.get(formdata.get_url(backoffice=True))

    assert 'button-action-1' in resp.form.fields

    resp = resp.form.submit('button-action-1')

    resp = resp.follow()  # -> error, empty action

    resp = resp.follow()  # -> back to form

    assert 'Error: empty action' in resp.text

    form_action.by = trigger.roles

    workflow.store()

    resp = app.get(formdata.get_url(backoffice=True))

    resp = resp.form.submit('button-action-1')

    resp = resp.follow()

    assert 'This is a message' in resp.text

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

    assert resp.pyquery('#form_error_fblah_1').text() == 'required field'

    resp.form['fblah_1'] = 'GLOBAL INTERACTIVE ACTION'

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

    assert resp.location == formdata.get_url(backoffice=True)

    resp = resp.follow()

    assert 'HELLO GLOBAL INTERACTIVE ACTION' in resp.text

from wcs.wf.form import WorkflowFormFieldsFormDef

def test_backoffice_card_global_interactive_action(pub):

    user = create_user(pub)

    workflow = CardDef.get_default_workflow()

    workflow.id = None

    action = workflow.add_global_action('FOOBAR')

    display = action.add_action('displaymsg')

    display.message = 'This is a message'

    display.to = []

    form_action = action.add_action('form')

    form_action.varname = 'blah'

    form_action.formdef = WorkflowFormFieldsFormDef(item=form_action)

    form_action.formdef.fields.append(

        fields.StringField(id='1', label='Test', varname='test', type='string', required=True)

    )

    register_comment = action.add_action('register-comment')

    register_comment.comment = 'HELLO {{ form_workflow_form_blah_var_test }}'

    trigger = action.triggers[0]

    trigger.roles = [user.roles[0]]

    workflow.store()

    CardDef.wipe()

    carddef = CardDef()

    carddef.name = 'foo'

    carddef.fields = []

    carddef.workflow_id = workflow.id

    carddef.workflow_roles = {'_editor': user.roles[0]}

    carddef.store()

    carddef.data_class().wipe()

    carddata = carddef.data_class()()

    carddata.data = {}

    carddata.just_created()

    carddata.store()

    app = login(get_app(pub))

    resp = app.get(carddata.get_url(backoffice=True))

    assert 'button-action-1' in resp.form.fields

    resp = resp.form.submit('button-action-1')

    resp = resp.follow()  # -> error, empty action

    resp = resp.follow()  # -> back to form

    assert 'Error: empty action' in resp.text

    form_action.by = trigger.roles

    workflow.store()

    resp = app.get(carddata.get_url(backoffice=True))

    resp = resp.form.submit('button-action-1')

    resp = resp.follow()

    assert 'This is a message' in resp.text

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

    assert resp.pyquery('#form_error_fblah_1').text() == 'required field'

    resp.form['fblah_1'] = 'GLOBAL INTERACTIVE ACTION'

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

    assert resp.location == carddata.get_url(backoffice=True)

    resp = resp.follow()

    assert 'HELLO GLOBAL INTERACTIVE ACTION' in resp.text

def test_global_interactive_action(pub):

    user = create_user(pub)

    formdef = FormDef()

    formdef.name = 'test global action'

    formdef.fields = []

    workflow = Workflow.get_default_workflow()

    workflow.id = '2'

    action = workflow.add_global_action('FOOBAR')

    display = action.add_action('displaymsg')

    display.message = 'This is a message'

    display.to = []

    form_action = action.add_action('form')

    form_action.varname = 'blah'

    form_action.formdef = WorkflowFormFieldsFormDef(item=form_action)

    form_action.formdef.fields.append(

        fields.StringField(id='1', label='Test', varname='test', type='string', required=True)

    )

    register_comment = action.add_action('register-comment')

    register_comment.comment = 'HELLO {{ form_workflow_form_blah_var_test }}'

    trigger = action.triggers[0]

    trigger.roles = ['_submitter']

    workflow.store()

    formdef.workflow_id = workflow.id

    formdef.workflow_roles = {'_receiver': 1}

    formdef.store()

    formdata = formdef.data_class()()

    formdata.user_id = user.id

    formdata.just_created()

    formdata.perform_workflow()

    formdata.store()

    app = login(get_app(pub), username='foo', password='foo')

    resp = app.get(formdata.get_url(backoffice=False))

    assert 'button-action-1' in resp.form.fields

    resp = resp.form.submit('button-action-1')

    resp = resp.follow()  # -> error, empty action

    resp = resp.follow()  # -> back to form

    assert 'Error: empty action' in resp.text

    form_action.by = trigger.roles

    workflow.store()

    resp = app.get(formdata.get_url(backoffice=False))

    resp = resp.form.submit('button-action-1')

    resp = resp.follow()

    assert 'This is a message' in resp.text

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

    assert resp.pyquery('#form_error_fblah_1').text() == 'required field'

    resp.form['fblah_1'] = 'GLOBAL INTERACTIVE ACTION'

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

    assert resp.location == formdata.get_url(backoffice=False)

    resp = resp.follow()

    assert 'HELLO GLOBAL INTERACTIVE ACTION' in resp.text

    assert display_message.get_message(formdata) == '<p>test</p>'

    assert display_message.get_message(formdata) == '<p>%s</p>' % formdata.id

    assert display_message.get_message(formdata) == '<p>%s</p>' % formdata.id

    assert display_message.get_message(formdata) == '<p>Foobar</p>'

    assert display_message.get_message(formdata) == '<p>Foobar</p>'

    assert display_message.get_message(formdata) == '<p>1 &lt; 3</p>'

    assert display_message.get_message(formdata) == '<p>1 &lt; 3</p>'

    assert display_message.get_message(formdata) == '<p>all</p>'

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>all</p>']

    assert display_message.get_message(formdata) == '<p>to-role</p>'

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>to-role</p>']

    assert display_message.get_message(formdata) == '<p>to-submitter</p>'

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>to-submitter</p>']

    assert display_message.get_message(formdata) == '<p>to-role-or-submitter</p>'

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>to-role-or-submitter</p>']

    assert display_message.get_message(formdata) == '<p>to-role-or-submitter</p>'

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>to-role-or-submitter</p>']

    assert display_message.get_message(formdata) == '<p>to-role-or-submitter</p>'

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>to-role-or-submitter</p>']

    assert formdata.get_workflow_messages(user=pub._request._user) == ['<p>d2</p>']

    assert '<p>d1</p>' in formdata.get_workflow_messages(user=pub._request._user)

    assert '<p>d2</p>' in formdata.get_workflow_messages(user=pub._request._user)

            options = [

                (x.key, x(parent=self.status).description) for x in available_items if x.category == category

            ]

        'actions',

        'actions',

        from wcs.forms.actions import ActionsDirectory

        self.actions = ActionsDirectory()

        if action['action'].is_interactive():

            return redirect(

                action['action'].get_global_interactive_form_url(formdef=self.formdef, ids=item_ids)

            )

from .qommon import _, force_text, misc, pgettext_lazy

    item_name = pgettext_lazy('item', 'card')

    item_name_plural = pgettext_lazy('item', 'cards')

        return wf_status.get_messages(formdata=self, position=position)

from .qommon import PICKLE_KWARGS, _, force_str, get_cfg, pgettext_lazy

    item_name = pgettext_lazy('item', 'form')

    item_name_plural = pgettext_lazy('item', 'forms')

from quixote import get_publisher, get_request, get_response, get_session, redirect

from wcs.qommon import _, errors, misc, template

from wcs.qommon.afterjobs import AfterJob

from wcs.qommon.backoffice.menu import html_top as backoffice_html_top

from wcs.qommon.form import Form

from wcs.qommon.http_request import HTTPRequest

from wcs.workflows import perform_items

        if token.type == 'action':

            return ActionDirectory(token)

        if token.type == 'global-interactive-action':

            return GlobalInteractiveActionDirectory(token)

        raise errors.TraversalError()

class GlobalInteractiveActionDirectory(Directory, FormTemplateMixin):

    _q_exports = ['']

    def __init__(self, token):

        self.token = token

        formdef_type = self.token.context.get('form_type', 'formdef')

        if formdef_type == 'carddef':

            formdef_class = CardDef

        elif formdef_type == 'formdef':

            formdef_class = FormDef

        else:

            raise errors.TraversalError()

        self.formdef = formdef_class.get_by_urlname(self.token.context['form_slug'])

        try:

            self.formdata = self.formdef.data_class().get(self.token.context['form_ids'][0])

        except KeyError:

            raise MissingFormdata()

        self.action = None

        for action in self.formdef.workflow.global_actions or []:

            if action.id == self.token.context['action_id']:

                self.action = action

                break

        else:

            raise MissingOrExpiredToken()

    def _q_index(self):

        if get_request().is_in_backoffice():

            if isinstance(self.formdef, CardDef):

                section = 'data_management'

            else:

                section = 'management'

            get_response().breadcrumb.append(('', self.action.name))

            backoffice_html_top(section, title=self.formdef.name)

            template_name = 'wcs/backoffice/global-interactive-action.html'

        else:

            template.html_top(title=self.formdef.name)

            template_name = 'wcs/global-interactive-action.html'

        form = self.action.get_action_form(self.formdata, user=get_request().user)

        if not form:

            # empty form, nothing to do

            get_session().message = ('error', _('Error: empty action'))

            return redirect(self.token.context['return_url'])

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

            messages = self.action.get_messages()

            context = {

                'html_form': form,

                'action': self.action,

                'ids': self.token.context['form_ids'],

                'formdata': self.formdata,

                'workflow_messages': messages,

            }

            return template.QommonTemplateResponse(templates=[template_name], context=context)

        if len(self.token.context['form_ids']) > 1:

            # mass action

            job = get_response().add_after_job(

                GlobalInteractiveMassActionAfterJob(

                    label=_('Executing task "%s" on forms') % self.action.name,

                    formdef=self.formdef,

                    request_form=get_request().form,

                    user_id=get_request().user.id,

                    action_id=self.action.id,

                    item_ids=self.token.context['form_ids'],

                    return_url=self.token.context['return_url'],

                )

            )

            job.store()

            self.token.remove_self()

            return redirect(job.get_processing_url())

        get_publisher().substitutions.reset()

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

        get_publisher().substitutions.feed(self.formdata)

        status = self.formdata.status

        url = self.action.handle_form(form, self.formdata, user=get_request().user)

        if self.formdata.status == status:

            # if there's no status change run non-interactive items from global ction

            url = (

                perform_items(

                    self.action.items, self.formdata, event=('global-interactive-action', self.action.id)

                )

                or url

            )

        if not url:

            url = self.formdata.get_url(backoffice=bool(get_request().is_in_backoffice()))

        self.token.remove_self()

        return redirect(url)

class GlobalInteractiveMassActionAfterJob(AfterJob):

    def __init__(self, formdef, **kwargs):

        super().__init__(formdef_class=formdef.__class__, formdef_id=formdef.id, **kwargs)

    def execute(self):

        self.total_count = len(self.kwargs['item_ids'])

        # restore request form

        publisher = get_publisher()

        req = HTTPRequest(None, {'SERVER_NAME': publisher.tenant.hostname, 'SCRIPT_NAME': ''})

        req.form = self.kwargs['request_form']

        formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])

        data_class = formdef.data_class()

        for action in formdef.workflow.global_actions or []:

            if action.id == self.kwargs['action_id']:

                break

        else:

            # maybe action got removed from workflow?

            return

        user = publisher.user_class.get(self.kwargs['user_id'])

        for item_id in self.kwargs['item_ids']:

            publisher._set_request(req)

            publisher.substitutions.reset()

            publisher.substitutions.feed(get_publisher())

            formdata = data_class.get(item_id)

            publisher.substitutions.feed(formdata)

            status = formdata.status

            form = action.get_action_form(formdata, user=user)

            form.method = 'get'

            action.handle_form(form, formdata, user=user)

            # reset request to avoid emails being created as afterjobs

            publisher._set_request(None)

            if formdata.status == status:

                # if there's no status change run non-interactive items from global ction

                perform_items(action.items, formdata, event=('global-interactive-action', action.id))

            self.increment_count()

    def done_action_url(self):

        return self.kwargs['return_url']

    def done_action_label(self):

        return _('Back to Listing')

    def done_button_attributes(self):

        return {'data-redirect-auto': 'true'}

                    r += htmltext(workflow_message)

{% extends "wcs/backoffice/base.html" %}

{% load i18n %}

{% block appbar-title %}{{ action.name }} -

{% if ids|length == 1 %}

{{ formdata.formdef.item_name }} - {{ formdata.id_display }}

{% else %}

{% blocktrans with len=ids|length item_name=formdata.formdef.item_name_plural %}{{ len }} selected {{ item_name }}{% endblocktrans %}

{% endif %}

{% endblock %}

{% block content %}

{% include "wcs/includes/global-interactive-action.html" %}

{% endblock %}

{% extends template_base %}

{% block body %}

{% include "wcs/includes/global-interactive-action.html" %}

{% endblock %}

{% if workflow_messages %}

  <div class="workflow-messages">

  {% for message in workflow_messages %}{{ message|safe }}{% endfor %}

  </div>

{% endif %}

{{ html_form.render|safe }}

    ok_in_global_action = True

    def is_interactive(self):

        return True

from wcs.workflows import WorkflowGlobalAction, WorkflowStatus, WorkflowStatusJumpItem, register_item_class

    ok_in_global_action = True

    @property

    def description(self):

        if isinstance(self.parent, WorkflowGlobalAction):

            return _('Manual Jump (interactive)')

        else:

            return _('Manual Jump')

    def is_interactive(self):

        return True

    ok_in_global_action = True

    def is_interactive(self):

        return True

from wcs.workflows import WorkflowGlobalAction, WorkflowStatusItem, register_item_class

    ok_in_global_action = True

        in_global_action = isinstance(self.parent, WorkflowGlobalAction)

        if in_global_action:

            pass

        elif self.position == 'top':

    def get_message(self, formdata, position='top'):

        if not self.message:

        if formdata and not formdata.is_for_current_user(self.to):

            return ''

        in_global_action = isinstance(self.parent, WorkflowGlobalAction)

        if self.position != position and not in_global_action:

            return

        if formdata:

            ctx = copy.copy(get_publisher().substitutions.get_context_variables('lazy'))

            ctx['date'] = misc.localstrftime(formdata.receipt_time)

            ctx['number'] = formdata.id

            handling_role = formdata.get_handling_role()

            if handling_role and handling_role.details:

                ctx['receiver'] = handling_role.details.replace('\n', '<br />')

        else:

            ctx = get_publisher().substitutions.get_context_variables('lazy')

        elif not message.startswith('<'):

            message = '<p>%s</p>' % message

            return Template(message, ezt_format=ezt.FORMAT_HTML).render(ctx)

        in_global_action = isinstance(self.parent, WorkflowGlobalAction)

        if 'position' in parameters and not in_global_action:

    def is_interactive(self):

        return bool(self.method == 'interactive')

    ok_in_global_action = True

    def is_interactive(self):

        return True

            'global-interactive-action': _('Global action (interactive)'),

    def get_action_form(self, filled, user, displayed_fields=None):

        form = Form(enctype='multipart/form-data', use_tokens=False)

        form.attrs['id'] = 'wf-actions'

        for item in self.items:

            if not item.check_auth(filled, user):

                continue

            if not item.check_condition(filled):

                continue

            item.fill_form(form, filled, user, displayed_fields=displayed_fields)

        if form.widgets or form.submit_widgets:

            return form

        else:

            return None

    def get_active_items(self, form, filled, user):

        for item in self.items:

            if hasattr(item, 'by'):

                for role in item.by or []:

                    if role == logged_users_role().id:

                        break

                    if role == '_submitter':

                        if filled.is_submitter(user):

                            break

                        continue

                    if user is None:

                        continue

                    if filled.get_function_roles(role).intersection(user.get_roles()):

                        break

                else:

                    continue

            if not item.check_condition(filled):

                continue

            yield item

    def get_messages(self, formdata=None, position='top'):

        messages = []

        for item in self.items or []:

            if not hasattr(item, 'get_message'):

                continue

            if not item.check_condition(formdata):

                continue

            message = item.get_message(formdata, position=position)

            if message:

                messages.append(message)

        return messages

    def handle_form(self, form, filled, user, evo):

        evo.time = time.localtime()

        if user:

            if filled.is_submitter(user):

                evo.who = '_submitter'

            else:

                evo.who = user.id

        if not filled.evolution:

            filled.evolution = []

        for item in self.get_active_items(form, filled, user):

            next_url = item.submit_form(form, filled, user, evo)

            if next_url is True:

                break

            if next_url:

                if not form.has_errors():

                    if evo.parts or evo.status or evo.comment or evo.status:

                        # add evolution entry only if there's some content

                        # within, i.e. do not register anything in the case of

                        # a single edit action (where the evolution should be

                        # appended only after successful edit).

                        filled.evolution.append(evo)

                        if evo.status:

                            filled.status = evo.status

                        filled.store()

                return next_url

        return next_url

    def is_interactive(self):

        for item in self.items or []:

            if item.is_interactive():

                return True

        return False

    def get_global_interactive_form_url(self, formdef=None, ids=None):

        token = get_publisher().token_class(size=32)

        token.type = 'global-interactive-action'

        token.context = {

            'action_id': self.id,

            'form_slug': formdef.slug,

            'form_type': formdef.xml_root_node,

            'form_ids': ids,

            'return_url': get_request().get_path_query(),

        }

        token.store()

        if get_request().is_in_backoffice():

            return formdef.get_url(backoffice=True) + 'actions/%s/#' % token.id

        else:

            return '/actions/%s/#' % token.id

    def handle_form(self, form, filled, user):

        evo = Evolution()

        url = super().handle_form(form, filled, user, evo)

        if isinstance(url, str):

            return url

        filled.evolution.append(evo)

        if evo.status:

            filled.status = evo.status

        filled.store()

        form = super().get_action_form(filled, user, displayed_fields=displayed_fields)

        if form is None:

            form = Form(enctype='multipart/form-data', use_tokens=False)

            form.attrs['id'] = 'wf-actions'

                if action.is_interactive():

                    return action.get_global_interactive_form_url(formdef=filled.formdef, ids=[filled.id])

        url = super().handle_form(form, filled, user, evo)

        if isinstance(url, str):

            return url

            def is_interactive(self):

                return False

    def __init__(self, parent=None):

        self.parent = parent

    def is_interactive(self):

        return False