Produits Entr'ouvert » w.c.s.

from wcs.wf.external_workflow import ManyExternalCallsPart

def test_call_external_workflow_manual_queryset_targeting(two_pubs):

    if not two_pubs.is_using_postgresql():

        pytest.skip('this requires SQL')

        return

    FormDef.wipe()

    CardDef.wipe()

    two_pubs.loggederror_class.wipe()

    # carddef workflow, with global action to increment a counter in its

    # backoffice fields.

    carddef_wf = Workflow(name='Carddef Workflow')

    carddef_wf.add_status(name='New')

    carddef_wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(carddef_wf)

    carddef_wf.backoffice_fields_formdef.fields = [

        StringField(id='bo0', varname='bo', type='string', label='bo variable'),

    ]

    global_action = carddef_wf.add_global_action('Update')

    global_action.append_item('set-backoffice-fields')

    setbo = global_action.items[0]

    setbo.fields = [{'field_id': 'bo0', 'value': '{{ form_var_bo|default:"0"|add:1 }}'}]

    trigger = global_action.append_trigger('webservice')  # external call

    trigger.identifier = 'update'

    carddef_wf.store()

    # associated carddef

    carddef = CardDef()

    carddef.name = 'Data'

    carddef.fields = [

        StringField(id='0', label='string', varname='card_string'),

    ]

    carddef.workflow = carddef_wf

    carddef.store()

    carddef.data_class().wipe()

    # and sample carddatas

    for i in range(1, 5):

        carddata = carddef.data_class()()

        carddata.data = {'0': 'Text %s' % i}

        carddata.store()

        carddata.just_created()

        carddata.store()

    # formdef workflow that will trigger the global action

    wf = Workflow(name='External actions')

    wf.add_status('Blah')

    update_global_action = wf.add_global_action('Update linked object data')

    update_action = update_global_action.append_item('external_workflow_global_action')

    update_action.slug = 'carddef:%s' % carddef.url_name

    update_action.target_mode = 'manual'

    update_action.target_id = None  # not configured

    update_action.trigger_id = 'action:update'

    wf.store()

    # associated formdef

    formdef = FormDef()

    formdef.name = 'External action form'

    formdef.fields = []

    formdef.workflow = wf

    formdef.store()

    # and formdata

    formdata = formdef.data_class()()

    formdata.data = {}

    formdata.store()

    formdata.just_created()

    # target not configured

    perform_items([update_action], formdata)

    assert carddef.data_class().count() == 4

    assert carddef.data_class().get(1).data['bo0'] is None

    assert carddef.data_class().get(2).data['bo0'] is None

    assert carddef.data_class().get(3).data['bo0'] is None

    assert carddef.data_class().get(4).data['bo0'] is None

    # target all cards

    update_action.target_id = '{{cards|objects:"%s"}}' % carddef.url_name

    wf.store()

    perform_items([update_action], formdata)

    assert carddef.data_class().get(1).data['bo0'] == '1'

    assert carddef.data_class().get(2).data['bo0'] == '1'

    assert carddef.data_class().get(3).data['bo0'] == '1'

    assert carddef.data_class().get(4).data['bo0'] == '1'

    status_part = [x for x in formdata.evolution[-1].parts if isinstance(x, ManyExternalCallsPart)][0]

    assert status_part.running is False

    assert status_part.is_hidden() is True

    assert '4 processed' in str(status_part.view())

    assert set(status_part.processed_ids) == {x.get_display_id() for x in carddef.data_class().select()}

    # target some cards

    update_action.target_id = (

        '{{cards|objects:"%s"|filter_by:"card_string"|filter_value:"Text 2"}}' % carddef.url_name

    )

    wf.store()

    perform_items([update_action], formdata)

    assert carddef.data_class().get(1).data['bo0'] == '1'

    assert carddef.data_class().get(2).data['bo0'] == '2'

    assert carddef.data_class().get(3).data['bo0'] == '1'

    assert carddef.data_class().get(4).data['bo0'] == '1'

    # target a single formdata

    update_action.target_id = (

        '{{cards|objects:"%s"|filter_by:"card_string"|filter_value:"Text 2"|first}}' % carddef.url_name

    )

    wf.store()

    perform_items([update_action], formdata)

    assert carddef.data_class().get(1).data['bo0'] == '1'

    assert carddef.data_class().get(2).data['bo0'] == '3'

    assert carddef.data_class().get(3).data['bo0'] == '1'

    assert carddef.data_class().get(4).data['bo0'] == '1'

    # mismatch in target

    carddef2 = CardDef()

    carddef2.name = 'Other data'

    carddef2.fields = []

    carddef2.workflow = carddef_wf

    carddef2.store()

    update_action.slug = 'carddef:%s' % carddef2.url_name

    update_action.target_id = (

        '{{cards|objects:"%s"|filter_by:"card_string"|filter_value:"Text 2"}}' % carddef.url_name

    )

    wf.store()

    perform_items([update_action], formdata)

    assert two_pubs.loggederror_class.count() == 1

    logged_error = two_pubs.loggederror_class.select()[0]

    assert logged_error.summary == 'Mismatch in target objects: expected "Other data", got "Data"'

    # mismatch in target, with formdata

    two_pubs.loggederror_class.wipe()

    update_action.target_id = (

        '{{cards|objects:"%s"|filter_by:"card_string"|filter_value:"Text 2"|first}}' % carddef.url_name

    )

    wf.store()

    perform_items([update_action], formdata)

    assert two_pubs.loggederror_class.count() == 1

    logged_error = two_pubs.loggederror_class.select()[0]

    assert logged_error.summary == 'Mismatch in target object: expected "Other data", got "Data"'

    def __eq__(self, other):

        return bool(

            isinstance(other, FormDef) and self.xml_root_node == other.xml_root_node and self.id == other.id

        )

import uuid

from quixote import get_publisher, get_request

from quixote.html import TemplateIO, htmltext

from wcs.variables import LazyFormData, LazyFormDefObjectsManager

    EvolutionPart,

class ManyExternalCallsPart(EvolutionPart):

    processed_ids = None

    label = None

    running = True

    uuid = None

    def __init__(self, label):

        self.label = label

        self.uuid = str(uuid.uuid4())

        self.processed_ids = []

    def is_hidden(self):

        return bool(not self.running) or not (

            get_request() and get_request().get_path().startswith('/backoffice/')

        )

    def view(self):

        r = TemplateIO(html=True)

        r += htmltext('<div>')

        r += (

            htmltext('<p>%s</p>')

            % _('Running external actions on "%(label)s" (%(count)s processed)')

            % {'label': self.label, 'count': len(self.processed_ids)}

        )

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

        return r.getvalue()

        with get_publisher().complex_data():

            target_id = self.compute(self.target_id, formdata=formdata, status_item=self, allow_complex=True)

            if target_id:

                target_id = get_publisher().get_cached_complex_data(target_id)

        if isinstance(target_id, LazyFormData):

            if target_id._formdef != objectdef:

                # abort if it's not the correct formdef/carddef

                get_publisher().record_error(

                    _('Mismatch in target object: expected "%(object_name)s", got "%(object_name2)s"')

                    % {'object_name': objectdef.name, 'object_name2': target_id._formdef.name},

                    formdata=formdata,

                    status_item=self,

                )

                return

            yield target_id._formdata

            return

        if isinstance(target_id, LazyFormDefObjectsManager):

            if target_id._formdef != objectdef:

                # abort if it's not the correct formdef/carddef

                get_publisher().record_error(

                    _('Mismatch in target objects: expected "%(object_name)s", got "%(object_name2)s"')

                    % {'object_name': objectdef.name, 'object_name2': target_id._formdef.name},

                    formdata=formdata,

                    status_item=self,

                )

                return

            for lazy_formdata in target_id:

                yield lazy_formdata._formdata

            return

            yield objectdef.data_class().get(target_id)

            # return targets

            yield from self.get_manual_target(formdata)

        else:

            yield from formdata.iter_target_datas(

                objectdef=objectdef, object_type=self.slug, status_item=self

            )

        status_part = ManyExternalCallsPart(label=objectdef.name)

        for i, target_data in enumerate(self.iter_target_datas(formdata, objectdef)):

            # update local object as it may have been modified by target_data

            # workflow executions.

            formdata.refresh_from_storage()

            if i == 0:

                # if there are iterations, add tracking status to object

                formdata.evolution[-1].add_part(status_part)

            elif i:

                # get status object back

                for evolution in reversed(formdata.evolution):

                    try:

                        status_part = [

                            x

                            for x in evolution.parts

                            if isinstance(x, ManyExternalCallsPart) and x.uuid == status_part.uuid

                        ][0]

                    except IndexError:

                        # probably the status changed and the tracking object is no longer available,

                        # do without

                        continue

                    break

            status_part.processed_ids.append(target_data.get_display_id())

            # after iterating, store

            formdata.store()

        # note it's now done.

        status_part.running = False

        formdata.store()