Produits Entr'ouvert » w.c.s.

def test_attachment():

    hello_word_b64 = base64.encodestring('hello world')

    d = evalutils.attachment(varname='var1')

    assert d == {'varname'     : 'var1'

                ,'filename'    : ''

                ,'content_type': 'application/octet-stream'

                ,'b64_content' : ''}

    d = evalutils.attachment(varname ='var1'

                            ,filename='hello.txt')

    assert d == {'varname'     : 'var1'

                ,'filename'    : 'hello.txt'

                ,'content_type': 'application/octet-stream'

                ,'b64_content' : ''}

    d = evalutils.attachment(varname     ='var1'

                            ,content_type='text/plain')

    assert d == {'varname'     : 'var1'

                ,'filename'    : ''

                ,'content_type': 'text/plain'

                ,'b64_content' : ''}

    d = evalutils.attachment(varname='var1'

                            ,content='hello world')

    assert d == {'varname'     : 'var1'

                ,'filename'    : ''

                ,'content_type': 'application/octet-stream'

                ,'b64_content' : hello_word_b64}

    d = evalutils.attachment(varname    ='var1'

                            ,b64_content=hello_word_b64)

    assert d == {'varname'     : 'var1'

                ,'filename'    : ''

                ,'content_type': 'application/octet-stream'

                ,'b64_content' : hello_word_b64}

    d = evalutils.attachment(varname     ='var1'

                            ,filename    ='hello.txt'

                            ,content_type='text/plain'

                            ,content     ='hello world')

    assert d == {'varname'     : 'var1'

                ,'filename'    : 'hello.txt'

                ,'content_type': 'text/plain'

                ,'b64_content' : hello_word_b64}

    d = evalutils.attachment(varname     ='var1'

                            ,filename    ='hello.txt'

                            ,content_type='text/plain'

                            ,b64_content =hello_word_b64)

    assert d == {'varname'     : 'var1'

                ,'filename'    : 'hello.txt'

                ,'content_type': 'text/plain'

                ,'b64_content' : hello_word_b64}

    d = evalutils.attachment(varname     ='var1'

                            ,filename    ='hello.txt'

                            ,content_type='text/plain'

                            ,content     ='hello world'

                            ,b64_content =hello_word_b64)

    assert d == {'varname'     : 'var1'

                ,'filename'    : 'hello.txt'

                ,'content_type': 'text/plain'

                ,'b64_content' : hello_word_b64}

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

    item = RegisterCommenterWorkflowStatusItem()

    item.attachments = ['form_var_backoffice_file1_raw']

    item.comment = '<div>{{ attachments.%s.url }}</div>' % 'backoffice_file1'

    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].varname == 'backoffice_file1'

    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 len(comment_view) > len('<div></div>')

    assert re.search('<div>[^<]+</div>', comment_view)

    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.attachment(varname='{0}'"

                                        ",filename={1}_filename"

                                        ",b64_content ={1}_b64_content"

                                        ",content_type={1}_content_type)"

                        "".format('afile', ws_response_varname)]

    item.comment = '{{ attachments.%s.url }}' % 'afile'

    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].varname == 'afile'

    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 len(comment_view) > len('<div></div>')

    assert re.search('<div>[^<]+</div>', comment_view)

def attachment(content='', filename='', content_type='application/octet-stream', b64_content='', varname=None):

    d = {

        'filename'    : filename if filename else '',

        'content_type': content_type if content_type else 'application/octet-stream',

        'b64_content' : b64_content if b64_content else base64.encodestring(content)

    if varname:

        d['varname'] = varname

    return d

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

        # 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 get_attachment_varname(self, attachment, upload=None):

        varname = str(attachment)

        if isinstance(attachment, str) and re.match(r'form_var_', attachment):

            varname = attachment

            m = re.search(r'^form_var_(\w+)$', attachment)

            if m:

                varname = re.sub(r'_raw$', '', m.group(1)) if re.search(r'_raw$', m.group(1)) else m.group(1)

        elif upload:

            if isinstance(upload, PUpload):

                varname = str(upload)

            elif isinstance(upload, dict) and 'varname' in upload:

                varname = upload['varname']

            elif hasattr(upload, varname):

                varname = upload.varname

        return varname

    def convert_attachments_to_uploads(self):

        # forget previous attachment converted to uploads

        # (file with same content and same varname, will not be duplicated

        #  because the path of the file is made of the hash of the content)

        uploads = OrderedDict()

        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

                varname = self.get_attachment_varname(attachment, picklableupload)

                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

                # exsiting attachments upload will be replaced by the new one (last one wins)

                uploads[varname] = picklableupload

        return uploads

    def formdata_has_attachment(self, formdata, varname, attachment):

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

            return False

        if not varname and hasattr(attachment, 'varname'):

            varname = attachment.varname

        for part in formdata.evolution[-1].parts:

            if isinstance(part, AttachmentEvolutionPart):

                if attachment:

                    # not testing content here, only metadata

                    if  part.orig_filename == attachment.orig_filename \

                    and part.base_filename == attachment.base_filename \

                    and  part.content_type == attachment.content_type  \

                    and       part.charset == attachment.charset       \

                    and (not varname or part.varname == varname):

                        return True

                elif varname and part.varname == varname:

                    return True

        return False

    def attach_uploads_to_formdata(self, formdata, uploads):

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

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

        for varname, upload in uploads.items():

            # attach to form only if new (to prevent double attachments)

            if not self.formdata_has_attachment(formdata, varname=varname, attachment=upload):

                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, varname=varname))

                except:

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

                                                        context='[workflow/attachments]')

                continue

        attachments = [x for _,x in self.convert_attachments_to_uploads().items()]