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())

    # 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: "Other data" vs "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: "Other data" vs "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.html import TemplateIO, htmltext

from wcs.variables import LazyFormData, LazyFormDefObjectsManager

    EvolutionPart,

class ManyExternalCallsPart(EvolutionPart):

    count = 0

    label = None

    running = True

    uuid = None

    def __init__(self, label):

        self.label = label

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

    def is_hidden(self):

        return bool(not self.running)

    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': self.count}

        )

        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: "%(object_name)s" vs "%(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: "%(object_name)s" vs "%(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 == 1:

                # if there are several 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

            if i:

                status_part.count = i + 1

                # after iterating, store

                formdata.store()

        # if there were many calls, note it's now done.

        if status_part.count > 1:

            status_part.running = False

            formdata.store()