Produits Entr'ouvert » w.c.s.

from wcs.blocks import BlockDef

from wcs.qommon.substitution import CompatibilityNamesDict

    formdata.just_created()

    formdata.just_created()

def test_workflow_form_structured_data(pub):

    FormDef.wipe()

    Workflow.wipe()

    BlockDef.wipe()

    user = create_user(pub)

    block = BlockDef()

    block.name = 'foobar'

    block.fields = [

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

    ]

    block.store()

    wf = Workflow(name='test')

    status = wf.add_status('New', 'st1')

    status.items = []

    display_form = FormWorkflowStatusItem()

    display_form.id = '_display_form'

    display_form.by = ['_submitter']

    display_form.varname = 'blah'

    display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)

    display_form.formdef.fields = [

        fields.BlockField(id='1', label='test', type='block:foobar', varname='fooblock'),

    ]

    status.items.append(display_form)

    display_form.parent = status

    wf.store()

    formdef = create_formdef()

    formdef.workflow_id = wf.id

    formdef.fields = []

    formdef.store()

    formdef.data_class().wipe()

    formdata = formdef.data_class()()

    formdata.user_id = user.id

    formdata.just_created()

    formdata.store()

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

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

    resp.form['fblah_1$element0$f123'] = 'ABC'

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

    resp.form['fblah_1$element0$f123'] = 'XYZ'

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

    formdata.refresh_from_storage()

    assert formdata.workflow_data == {

        'blah_var_fooblock_raw': {'data': [{'123': 'XYZ'}], 'schema': {'123': 'string'}},

        'blah_var_fooblock': 'foobar',

    }

    substvars = CompatibilityNamesDict()

    substvars.update(formdata.get_substitution_variables())

    keys = substvars.get_flat_keys()

    for key in keys:

        # noqa pylint: disable=unused-variable

        var = substvars[key]  # check it doesn't raise, ignore the value

    assert substvars['form_workflow_form_blah_var_fooblock_var_test'] == 'XYZ'

    assert substvars['form_workflow_form_blah_0_var_fooblock_var_test'] == 'ABC'

    assert substvars['form_workflow_form_blah_1_var_fooblock_var_test'] == 'XYZ'

    # disable dumping in workflow_data

    pub.load_site_options()

    if not pub.site_options.has_section('options'):

        pub.site_options.add_section('options')

    pub.site_options.set('options', 'disable-workflow-form-to-workflow-data', 'true')

    with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:

        pub.site_options.write(fd)

    formdata = formdef.data_class()()

    formdata.user_id = user.id

    formdata.just_created()

    formdata.store()

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

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

    resp.form['fblah_1$element0$f123'] = 'ABC'

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

    resp.form['fblah_1$element0$f123'] = 'XYZ'

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

    formdata.refresh_from_storage()

    assert not formdata.workflow_data

from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormEvolutionPart, WorkflowFormFieldsFormDef

        # file from workflow form

        formdata = formdef.data_class()()

        formdata.just_created()

        formdata.data = {}

        formdata.store()

        display_form = FormWorkflowStatusItem()

        display_form.varname = 'blah'

        display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)

        display_form.formdef.fields = []

        data = {'blah_1': PicklableUpload('test.txt', 'text/plain')}

        data['blah_1'].receive([b'hello world wf form'])

        formdata.evolution[-1].parts = [

            WorkflowFormEvolutionPart(display_form, data),

        ]

        count = len(os.listdir(os.path.join(pub.app_dir, 'uploads')))

        formdata.store()

        assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == count + 1

        clean_unused_files(pub)

        assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == count + 1

        formdata.evolution[-1].parts = []

        formdata.store()

        clean_unused_files(pub)

        assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == count

            # 4 files ("hello world" + "hello world 2" + "hello world block" + "hello world wf form")

            assert len(os.listdir(os.path.join(pub.app_dir, 'unused-files/uploads/'))) == 4

        from wcs.wf.form import WorkflowFormEvolutionPart

                    elif isinstance(part, WorkflowFormEvolutionPart):

                        for field_data in (part.data or {}).values():

                            if is_upload(field_data):

                                yield field_data.get_fs_filename()

    @property

    def workflow_form(self):

        # form_ workflow_form_ <slug (action varname)> _ <index> _var_ etc.

        # ex: form_workflow_form_xxx_2_var_file

        # (index can be "latest")

        from .wf.form import LazyFormDataWorkflowForms

        return LazyFormDataWorkflowForms(self._formdata)

from wcs.variables import LazyFormDataVar

from wcs.workflows import EvolutionPart, RedisplayFormException, WorkflowStatusItem, register_item_class

class WorkflowFormEvolutionPart(EvolutionPart):

    data = None

    formdef = None

    varname = None

    def __init__(self, action, data, live=False):

        self.varname = action.varname

        self.formdef = action.formdef

        self.data = data

        self.live = live

                fields_directory.field_var_prefix = 'form_workflow_form_%s_var_' % self.varname

    def evaluate_live_form(self, form, formdata, user, submit=False):

        formdef_data = self.formdef.get_data(form)

        for k, v in get_dict_with_varnames(self.formdef.fields, formdef_data, varnames_only=True).items():

        if not get_publisher().has_site_option('disable-workflow-form-to-workflow-data'):

            formdata.update_workflow_data(workflow_data)

        if self.varname and submit:

            # do not keep parts used for live evaluation (evaluate_live_form will

            # be called when creating form and it shouldn't record twice the values)

            formdata.evolution[-1].parts = [

                x

                for x in formdata.evolution[-1].parts or []

                if not (isinstance(x, WorkflowFormEvolutionPart) and x.live)

            ]

            formdata.evolution[-1].add_part(

                WorkflowFormEvolutionPart(self, formdef_data, live=bool(not submit))

            )

            self.evaluate_live_form(form, formdata, user, submit=True)

class LazyFormDataWorkflowForms:

    def __init__(self, formdata):

        self._formdata = formdata

    def __getattr__(self, varname):

        wfform_formdatas = []

        for part in self._formdata.iter_evolution_parts():

            if not isinstance(part, WorkflowFormEvolutionPart):

                continue

            if part.varname == varname and part.data:

                wfform_formdatas.append(LazyFormDataWorkflowFormsItem(part))

        if wfform_formdatas:

            return LazyFormDataWorkflowFormsItems(wfform_formdatas)

        raise AttributeError(varname)

    def inspect_keys(self):

        for part in self._formdata.iter_evolution_parts():

            if isinstance(part, WorkflowFormEvolutionPart) and part.varname and part.data:

                yield part.varname

class LazyFormDataWorkflowFormsItems:

    def __init__(self, wfform_formdatas):

        self._wfform_formdatas = wfform_formdatas

    def inspect_keys(self):

        return [str(x) for x in range(len(self._wfform_formdatas))] + ['var']

    @property

    def var(self):

        # alias to latest values

        return self._wfform_formdatas[-1].var

    def __getitem__(self, key):

        try:

            key = int(key)

        except ValueError:

            try:

                return getattr(self, key)

            except AttributeError:

                return self._wfform_formdatas[0][key]

        return self._wfform_formdatas[key]

    def __len__(self):

        return len(self._wfform_formdatas)

    def __iter__(self):

        yield from self._wfform_formdatas

class LazyFormDataWorkflowFormsItem:

    def __init__(self, part):

        self._part = part

        self.data = part.data

    def inspect_keys(self):

        return ['var']

    @property

    def var(self):

        # pass self as formdata, it will be used to access self.data in LazyFieldVarBlock

        return LazyFormDataVar(self._part.formdef.get_all_fields(), self._part.data, formdata=self)