Produits Entr'ouvert » w.c.s.

def test_workflows_edit_dispatch_action(pub):

    create_superuser(pub)

    role = create_role()

    Workflow.wipe()

    workflow = Workflow(name='foo')

    workflow.add_status(name='baz')

    workflow.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/workflows/1/')

    resp = resp.click('baz')

    resp.forms[0]['type'] = 'Assign a Function'

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

    resp = resp.follow()

    resp = resp.click('Assign a Function')

    resp.form['rules$element0$value'].value = 'FOOBAR'

    resp.form['rules$element0$role_id'].value = str(role.id)

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

    resp = resp.follow()

    resp = resp.follow()

    resp = resp.click('Assign a Function')

    assert resp.form['rules$element0$value'].value == 'FOOBAR'

    resp.form['rules$element1$value'].value = 'BARFOO'

    resp.form['rules$element1$role_id'].value = str(role.id)

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

    workflow = Workflow.get(workflow.id)

    assert workflow.possible_status[0].items[0].rules == [

            {'value': 'FOOBAR', 'role_id': '1'}, {'value': 'BARFOO', 'role_id': '1'}]

from wcs.wf.dispatch import DispatchWorkflowStatusItem

def test_complex_dispatch_action():

    wf = Workflow(name='status')

    st1 = wf.add_status('Status1', 'st1')

    Role.wipe()

    role1 = Role()

    role1.name = 'Test Role 1'

    role1.store()

    role2 = Role()

    role2.name = 'Test Role 2'

    role2.store()

    dispatch = DispatchWorkflowStatusItem()

    dispatch.id = '_dispatch'

    dispatch.role_key = '_receiver'

    dispatch.dispatch_type = 'automatic'

    dispatch.variable = 'plop'

    dispatch.rules = [{'value': 'a', 'role_id': role1.id},

                      {'value': 'b', 'role_id': role2.id}]

    st1.items.append(dispatch)

    dispatch.parent = st1

    wf2 = assert_import_export_works(wf)

    assert wf2.possible_status[0].items[0].variable == dispatch.variable

    assert wf2.possible_status[0].items[0].rules == dispatch.rules

    assert wf2.possible_status[0].items[0].dispatch_type == 'automatic'

    Role.wipe()

    role3 = Role()

    role3.name = 'Test Role 1'

    role3.store()

    role4 = Role()

    role4.name = 'Test Role 2'

    role4.store()

    role1.remove_self()

    role2.remove_self()

    xml_export_orig = export_to_indented_xml(wf, include_id=True)

    wf2 = Workflow.import_from_xml_tree(xml_export_orig)

    assert wf2.possible_status[0].items[0].variable == dispatch.variable

    assert wf2.possible_status[0].items[0].rules == [

            {'value': 'a', 'role_id': role3.id}, {'value': 'b', 'role_id': role4.id}]

    assert wf2.possible_status[0].items[0].dispatch_type == 'automatic'

def test_dispatch_auto(pub):

    formdef = FormDef()

    formdef.name = 'baz'

    formdef.fields = [

        StringField(id='1', label='Test', type='string', varname='foo'),

    ]

    formdef.store()

    item = DispatchWorkflowStatusItem()

    item.role_key = '_receiver'

    item.dispatch_type = 'automatic'

    formdata = formdef.data_class()()

    pub.substitutions.feed(formdata)

    item.perform(formdata)

    assert not formdata.workflow_roles

    item.variable = 'form_var_foo'

    item.rules = [

        {'role_id': '1', 'value': 'foo'},

        {'role_id': '2', 'value': 'bar'},

    ]

    item.perform(formdata)

    assert not formdata.workflow_roles

    # no match

    formdata.data = {'1': 'XXX'}

    item.perform(formdata)

    assert not formdata.workflow_roles

    # match

    formdata.data = {'1': 'foo'}

    item.perform(formdata)

    assert formdata.workflow_roles == {'_receiver': '1'}

    # other match

    formdata.data = {'1': 'bar'}

    item.perform(formdata)

    assert formdata.workflow_roles == {'_receiver': '2'}

import collections

import xml.etree.ElementTree as ET

from qommon import get_logger

from wcs.workflows import XmlSerialisable, WorkflowStatusItem, register_item_class

class AutomaticDispatchRowWidget(CompositeWidget):

    def __init__(self, name, value=None, **kwargs):

        CompositeWidget.__init__(self, name, value, **kwargs)

        if not value:

            value = {}

        self.add(StringWidget, name='value', title=_('Value'),

                value=value.get('value'), **kwargs)

        self.add(SingleSelectWidget, name='role_id', title=_('Role'),

                value=value.get('role_id'),

                options=[(None, '----', None)] + get_user_roles())

    def _parse(self, request):

        if self.get('value') or self.get('role_id'):

            self.value = {

                'value': self.get('value'),

                'role_id': self.get('role_id')

            }

        else:

            self.value = None

class AutomaticDispatchTableWidget(WidgetListAsTable):

    readonly = False

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

        super(AutomaticDispatchTableWidget, self).__init__(name,

                element_type=AutomaticDispatchRowWidget, **kwargs)

class RuleNode(XmlSerialisable):

    node_name = 'rule'

    def __init__(self, rule={}):

        self.role_id = rule.get('role_id')

        self.value = rule.get('value')

    def as_dict(self):

        return {

            'role_id': self.role_id,

            'value': self.value

        }

    def get_parameters(self):

        return ('role_id', 'value')

    def role_id_export_to_xml(self, item, charset, include_id=False):

        self._role_export_to_xml('role_id', item, charset,

                include_id=include_id)

    def role_id_init_with_xml(self, elem, charset, include_id=False):

        self._role_init_with_xml('role_id', elem, charset,

                include_id=include_id)

    dispatch_type = 'manual'

    variable = None

    rules = None

        return ('role_key', 'role_id', 'dispatch_type', 'variable', 'rules')

    def rules_export_to_xml(self, item, charset, include_id=False):

        if not self.rules:

            return

        rules_node = ET.SubElement(item, 'rules')

        for rule in self.rules:

            rules_node.append(RuleNode(rule).export_to_xml(charset=charset,

                include_id=include_id))

        return rules_node

    def rules_init_with_xml(self, elem, charset, include_id=False):

        rules = []

        if elem is None:

            return

        for rule_xml_node in elem.findall('rule'):

            rule_node = RuleNode()

            rule_node.init_with_xml(rule_xml_node, charset,

                    include_id=include_id)

            rules.append(rule_node.as_dict())

        if rules:

            self.rules = rules

        dispatch_types = collections.OrderedDict(

                [('manual', _('Manual')), ('automatic', _('Automatic'))])

        if 'dispatch_type' in parameters:

            form.add(RadiobuttonsWidget, '%sdispatch_type' % prefix,

                    title=_('Type'),

                    options=dispatch_types.items(),

                    value=self.dispatch_type,

                    required=True,

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

                     options=[(None, '----', None)] + get_user_roles(),

                     attrs={

                         'data-dynamic-display-child-of': '%sdispatch_type' % prefix,

                         'data-dynamic-display-value': dispatch_types.get('manual'),

                         }

                     )

        if 'variable' in parameters:

            form.add(StringWidget, '%svariable' % prefix,

                    title=_('Variable'),

                    value=self.variable,

                    attrs={

                        'data-dynamic-display-child-of': '%sdispatch_type' % prefix,

                        'data-dynamic-display-value': dispatch_types.get('automatic'),

                        }

                    )

        if 'rules' in parameters:

            form.add(AutomaticDispatchTableWidget, '%srules' % prefix,

                    title=_('Rules'),

                    value=self.rules,

                    attrs={

                        'data-dynamic-display-child-of': '%sdispatch_type' % prefix,

                        'data-dynamic-display-value': dispatch_types.get('automatic'),

                        }

                    )

        new_role_id = None

        if self.dispatch_type == 'manual' or not self.dispatch_type:

            if not (self.role_id and self.role_key):

                return

            new_role_id = self.role_id

        elif self.dispatch_type == 'automatic':

            if not (self.role_key and self.variable and self.rules):

                return

            variables = get_publisher().substitutions.get_context_variables()

            # convert the given value to a few different types, to allow more

            # diversity in matching.

            variable_values = [variables.get(self.variable)]

            if not variable_values[0]:

                variable_values.append(None)

            try:

                variable_values.append(int(variable_values[0]))

            except (ValueError, TypeError):

                pass

            for rule in self.rules:

                if rule.get('value') in variable_values:

                    new_role_id = rule.get('role_id')

                    break

        if new_role_id:

            if not Role.has_key(new_role_id):

                get_logger().error('error in dispatch, missing role %s' % new_role_id)

            else:

                formdata.workflow_roles[self.role_key] = str(new_role_id)

                formdata.store()

register_item_class(DispatchWorkflowStatusItem)

    key = None

        if self.key:

            node.attrib['type'] = self.key