Produits Entr'ouvert » w.c.s.

def test_dict_from_prefix():

    hello_word_b64 = base64.encodestring('hello world')

    d = evalutils.dict_from_prefix('var1', {})

    assert d == {}

    d = evalutils.dict_from_prefix('', {'k1':'v1'})

    assert d == {'k1':'v1'}

    d = evalutils.dict_from_prefix('k', {'k1':'v1', 'k2':'v2'})

    assert d == {'1':'v1', '2':'v2'}

    d = evalutils.dict_from_prefix('v', {'k1':'v1', 'k2':'v2'})

    assert d == {}

from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem, JournalEvolutionPart

def test_register_comment_attachment_substitution(pub):

def test_register_comment_attachment(pub):

    wf = Workflow(name='comment with attachments')

    wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)

    wf.backoffice_fields_formdef.fields = [

        FileField(id='bo1', label='bo field 1', type='file', varname='backoffice_file1'),

    ]

    st1 = wf.add_status('Status1')

    wf.store()

    upload = PicklableUpload('test.jpeg', 'image/jpeg')

    jpg = open(os.path.join(os.path.dirname(__file__), 'image-with-gps-data.jpeg')).read()

    upload.receive([jpg])

    formdef = FormDef()

    formdef.name = 'baz'

    formdef.fields = [

        FileField(id='1', label='File', type='file', varname='frontoffice_file'),

    ]

    formdef.workflow_id = wf.id

    formdef.store()

    formdata = formdef.data_class()()

    formdata.data = {'1': upload}

    formdata.just_created()

    formdata.store()

    pub.substitutions.feed(formdata)

    setbo = SetBackofficeFieldsWorkflowStatusItem()

    setbo.parent = st1

    setbo.fields = [{'field_id': 'bo1', 'value': '=form_var_frontoffice_file_raw'}]

    setbo.perform(formdata)

    if os.path.exists(os.path.join(get_publisher().app_dir, 'attachments')):

        shutil.rmtree(os.path.join(get_publisher().app_dir, 'attachments'))

    comment_text = 'File is attached to the form history'

    item = RegisterCommenterWorkflowStatusItem()

    item.attachments = ['form_var_backoffice_file1_raw']

    item.comment = comment_text

    item.perform(formdata)

    assert len(os.listdir(os.path.join(get_publisher().app_dir, 'attachments'))) == 1

    for subdir in os.listdir(os.path.join(get_publisher().app_dir, 'attachments')):

        assert len(subdir) == 4

        assert len(os.listdir(os.path.join(get_publisher().app_dir, 'attachments', subdir))) == 1

    assert len(formdata.evolution[-1].parts) == 2

    assert isinstance(formdata.evolution[-1].parts[0], AttachmentEvolutionPart)

    assert formdata.evolution[-1].parts[0].orig_filename == upload.orig_filename

    assert isinstance(formdata.evolution[-1].parts[1], JournalEvolutionPart)

    assert len(formdata.evolution[-1].parts[1].content) > 0

    comment_view = str(formdata.evolution[-1].parts[1].view())

    assert comment_view == '<p>%s</p>' % comment_text

    if os.path.exists(os.path.join(get_publisher().app_dir, 'attachments')):

        shutil.rmtree(os.path.join(get_publisher().app_dir, 'attachments'))

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

    formdata.store()

    ws_response_varname = 'ws_response_afile'

    wf_data = {

        '%s_filename'     % ws_response_varname : 'hello.txt',

        '%s_content_type' % ws_response_varname : 'text/plain',

        '%s_b64_content'  % ws_response_varname : base64.encodestring('hello world')

    }

    formdata.update_workflow_data(wf_data)

    formdata.store()

    assert hasattr(formdata, 'workflow_data')

    assert isinstance(formdata.workflow_data, dict)

    item = RegisterCommenterWorkflowStatusItem()

    item.attachments = ["utils.dict_from_prefix('%s_', locals())" % ws_response_varname]

    item.comment = comment_text

    item.perform(formdata)

    assert len(os.listdir(os.path.join(get_publisher().app_dir, 'attachments'))) == 1

    for subdir in os.listdir(os.path.join(get_publisher().app_dir, 'attachments')):

        assert len(subdir) == 4

        assert len(os.listdir(os.path.join(get_publisher().app_dir, 'attachments', subdir))) == 1

    assert len(formdata.evolution[-1].parts) == 2

    assert isinstance(formdata.evolution[-1].parts[0], AttachmentEvolutionPart)

    assert formdata.evolution[-1].parts[0].orig_filename == 'hello.txt'

    assert isinstance(formdata.evolution[-1].parts[1], JournalEvolutionPart)

    assert len(formdata.evolution[-1].parts[1].content) > 0

    comment_view = str(formdata.evolution[-1].parts[1].view())

    assert comment_view == '<p>%s</p>' % comment_text

import re

def dict_from_prefix(prefix, in_dict):

    '''Return a dict based on a dict filtered by a key prefix.

       The prefix is removed from the key.

       Intent: meant to help build a PicklableUpload from a set

               of key/values stored in the workflow data.

       Note: to use this function in a context of a Python

             expression, you should pass the _wf_data_ using

             the function locals()

       Example: utils.dict_from_prefix('akey_', locals())

         Where: the workflow data contains the key/values:

                akey_filename     = <filename>

                akey_content_type = <mime_type>

                akey_b64_content  = <content base64 encoded>

           And: it produces a dict like the key/values are:

                filename     = wf_data['akey_filename']

                content_type = wf_data['akey_content_type']

                b64_content  = wf_data['akey_b64_content']

    '''

    return {re.sub(r'^%s' % prefix, '', k): v \

            for k,v in in_dict.items() if k.startswith('%s' % prefix)}

from wcs.workflows import (WorkflowStatusItem, register_item_class, template_on_formdata,

                           AttachmentEvolutionPart)

import sys

        return ('comment', 'attachments', 'condition')

    def attach_uploads_to_formdata(self, formdata, uploads):

        if not formdata.evolution[-1].parts:

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

        for upload in uploads:

            try:

                # useless but required to restore upload.fp from serialized state, needed by 'AttachmentEvolutionPart.from_upload()

                fp = upload.get_file_pointer()

                formdata.evolution[-1].add_part(AttachmentEvolutionPart.from_upload(upload))

            except:

                get_publisher().notify_of_exception(sys.exc_info(),

                                                    context='[comment/attachments]')

            continue

        # process attachments first, they might be used in the comment

        # (with substitution vars)

        if self.attachments:

            uploads = self.convert_attachments_to_uploads()

            self.attach_uploads_to_formdata(formdata, uploads)

            formdata.store()  # store and invalidate cache, so references can be used in the comment message.

        # the comment can use attachments done above

from wcs.qommon.form import PicklableUpload as PUpload

from collections import OrderedDict

    support_substitution_variables = True

    attachments = None

        if 'attachments' in parameters:

            attachments_options, attachments = self.get_attachments_options()

            if len(attachments_options) > 1:

                form.add(WidgetList, '%sattachments' % prefix, title=_('Attachments'),

                         element_type=SingleSelectWidgetWithOther,

                         value=attachments,

                         add_element_label=_('Add attachment'),

                         element_kwargs={'render_br': False, 'options': attachments_options})

            else:

                form.add(WidgetList, '%sattachments' % prefix,

                         title=_('Attachments (Python expressions)'),

                         element_type=StringWidget,

                         value=attachments,

                         add_element_label=_('Add attachment'),

                         element_kwargs={'render_br': False, 'size': 50},

                         advanced=not(bool(attachments)))

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

        if elem is None:

            self.attachments = None

        else:

            self.attachments = [item.text.encode(charset) for item in elem.findall('attachment')]

    def get_attachments_options(self):

        attachments_options = [(None, '---', None)]

        varnameless = []

        for field in self.parent.parent.get_backoffice_fields():

            if field.key != 'file':

                continue

            if field.varname:

                codename = 'form_var_%s_raw' % field.varname

            else:

                codename = 'form_f%s' % field.id  # = form_fbo<n>

                varnameless.append(codename)

            attachments_options.append((codename, field.label, codename))

        # filter: do not consider removed fields without varname

        attachments = [attachment for attachment in self.attachments or []

                       if ((not attachment.startswith('form_fbo')) or

                           (attachment in varnameless))]

        return attachments_options, attachments

    def convert_attachments_to_uploads(self):

        uploads = []

        if self.attachments:

            global_eval_dict = get_publisher().get_global_eval_dict()

            local_eval_dict = get_publisher().substitutions.get_context_variables()

            for attachment in self.attachments:

                try:

                    # execute any Python expression

                    # and magically convert string like 'form_var_*_raw' to a PicklableUpload

                    picklableupload = eval(attachment, global_eval_dict, local_eval_dict)

                except:

                    get_publisher().notify_of_exception(sys.exc_info(),

                                                        context='[workflow/attachments]')

                    continue

                if not picklableupload:

                    continue

                try:

                    # magically convert any value to a PicklableUpload

                    # usualy a dict like one provided by qommon/evalutils:attachment()

                    picklableupload = FileField.convert_value_from_anything(picklableupload)

                except ValueError:

                    get_publisher().notify_of_exception(sys.exc_info(),

                                                        context='[workflow/attachments]')

                    continue

                uploads.append(picklableupload)

        return uploads

        attachments = self.convert_attachments_to_uploads()