0001-forms-add-direct-download-of-files-from-workflow-for.patch

         assert not formdata.workflow_data
     def test_workflow_form_file_access(pub):
         FormDef.wipe()
         Workflow.wipe()
         BlockDef.wipe()
         user = create_user(pub)
         block = BlockDef()
         block.name = 'foobar'
         block.fields = [
             fields.FileField(id='123', required=True, label='Test', type='file', varname='test'),
+        ]
         block.store()
         wf = Workflow(name='test')
         status = wf.add_status('New', 'st1')
         next_status = wf.add_status('Next', 'st2')
         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', max_items=3),
             fields.FileField(id='2', label='test2', type='file', varname='file'),
+        ]
         status.items.append(display_form)
         display_form.parent = status
         jump = JumpOnSubmitWorkflowStatusItem()
         jump.id = '_jump'
         jump.status = next_status.id
         status.items.append(jump)
         jump.parent = status
         register_comment = RegisterCommenterWorkflowStatusItem()
         register_comment.id = '_register'
         register_comment.comment = '''<p>
             <a href="{{ form_workflow_form_blah_var_fooblock_0_test_url }}" id="t1">1st file in block</a>
             <a href="{{ form_workflow_form_blah_var_fooblock_1_test_url }}" id="t2">2nd file in block</a>
             <a href="{{ form_workflow_form_blah_0_var_fooblock_0_test_url }}" id="t3">again 1st file in block</a>
             <a href="{{ form_workflow_form_blah_0_var_file_url }}" id="t4">file field</a>
         </p>'''
         next_status.items.append(register_comment)
         register_comment.parent = next_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$file'] = Upload('test1.txt', b'foobar1', 'text/plain')
         resp = resp.form.submit('fblah_1$add_element')
         resp.form['fblah_1$element1$f123$file'] = Upload('test2.txt', b'foobar2', 'text/plain')
         resp.form['fblah_2$file'] = Upload('test3.txt', b'foobar3', 'text/plain')
         resp = resp.form.submit('submit').follow()
         assert app.get(resp.pyquery('#t1').attr.href).body == b'foobar1'
         assert app.get(resp.pyquery('#t2').attr.href).body == b'foobar2'
         assert app.get(resp.pyquery('#t3').attr.href).body == b'foobar1'
         assert app.get(resp.pyquery('#t4').attr.href).body == b'foobar3'
         app.get(resp.pyquery('#t4').attr.href + 'X', status=404)  # wrong URL, unknown file
         # unlogged user
         assert '/login' in get_app(pub).get(resp.pyquery('#t1').attr.href).location
         # other user
         user = pub.user_class()
         user.name = 'Second user'
         user.store()
         account = PasswordAccount(id='foo2')
         account.set_password('foo2')
         account.user_id = user.id
         account.store()
         login(get_app(pub), username='foo2', password='foo2').get(resp.pyquery('#t1').attr.href, status=403)
     def test_rich_commentable_action(pub):
         create_user(pub)

             return self.get_view_value(value, include_image_thumbnail=False, max_len=max_len, **kwargs)
         def get_download_query_string(self, **kwargs):
             if kwargs.get('file_value'):
                 return 'hash=%s' % kwargs.get('file_value').file_digest()
             if kwargs.get('parent_field'):
                 return 'f=%s$%s$%s' % (kwargs['parent_field'].id, kwargs['parent_field_index'], self.id)
             return 'f=%s' % self.id

     import collections
     import copy
     import datetime
     import itertools
     import json
     import re
     import sys
-...
                 empty &= self.data.get(key) is None
             return empty
         def get_all_file_data(self):
             from wcs.wf.form import WorkflowFormEvolutionPart
             for field_data in itertools.chain((self.data or {}).values(), (self.workflow_data or {}).values()):
                 if misc.is_upload(field_data):
                     yield field_data
                 elif isinstance(field_data, dict) and isinstance(field_data.get('data'), list):
                     for subfield_rowdata in field_data.get('data'):
                         if isinstance(subfield_rowdata, dict):
                             for block_field_data in subfield_rowdata.values():
                                 if misc.is_upload(block_field_data):
                                     yield block_field_data
             for part in self.iter_evolution_parts():
                 if misc.is_attachment(part):
                     yield part
                 elif isinstance(part, WorkflowFormEvolutionPart):
                     for field_data in (part.data or {}).values():
                         if misc.is_upload(field_data):
                             yield field_data
         @classmethod
         def get_actionable_count(cls, user_roles):
             if get_publisher().is_using_postgresql():

     from .qommon.admin.emails import EmailsDirectory
     from .qommon.cron import CronJob
     from .qommon.form import Form, HtmlWidget, UploadedFile
     from .qommon.misc import JSONEncoder, get_as_datetime, simplify, xml_node_text
     from .qommon.misc import JSONEncoder, get_as_datetime, is_attachment, is_upload, simplify, xml_node_text
     from .qommon.publisher import get_publisher_class
     from .qommon.storage import Equal, StorableObject, StrictNotEqual, fix_key
     from .qommon.substitution import Substitutions
-...
         known_filenames.update([x for x in glob.glob(os.path.join(publisher.app_dir, 'uploads/*'))])
         known_filenames.update([x for x in glob.glob(os.path.join(publisher.app_dir, 'attachments/*/*'))])
         def is_upload(obj):
             # we can't use isinstance() because obj can be a
             # wcs.qommon.form.PicklableUpload or a qommon.form.PicklableUpload
             return obj.__class__.__name__ == 'PicklableUpload'
         def is_attachment(obj):
             return obj.__class__.__name__ == 'AttachmentEvolutionPart'
         def accumulate_filenames():
             from wcs.carddef import CardDef
             from wcs.wf.form import WorkflowFormEvolutionPart
             for formdef in FormDef.select(ignore_migration=True) + CardDef.select(ignore_migration=True):
                 for option_data in (formdef.workflow_options or {}).values():
                     if is_upload(option_data):
                         yield option_data.get_fs_filename()
                 for formdata in formdef.data_class().select_iterator(ignore_errors=True, itersize=200):
                     for field_data in itertools.chain(
                         (formdata.data or {}).values(), (formdata.workflow_data or {}).values()
                     ):
                     for field_data in formdata.get_all_file_data():
                         if is_upload(field_data):
                             yield field_data.get_fs_filename()
                         elif isinstance(field_data, dict) and isinstance(field_data.get('data'), list):
                             for subfield_rowdata in field_data.get('data'):
                                 if isinstance(subfield_rowdata, dict):
                                     for block_field_data in subfield_rowdata.values():
                                         if is_upload(block_field_data):
                                             yield block_field_data.get_fs_filename()
                     for part in formdata.iter_evolution_parts():
                         if is_attachment(part):
                             yield part.filename
                         elif isinstance(part, WorkflowFormEvolutionPart):
                             for field_data in (part.data or {}).values():
                                 if is_upload(field_data):
                                     yield field_data.get_fs_filename()
                         elif is_attachment(field_data):
                             yield field_data.filename
             for user in publisher.user_class.select():
                 for field_data in (user.form_data or {}).values():
                     if is_upload(field_data):

                 redirect_url = sign_url_auto_orig(redirect_url)
                 return redirect(redirect_url)
             return self.serve_file(file, thumbnail=self.thumbnails)
         @classmethod
         def serve_file(cls, file, thumbnail=False):
             response = get_response()
             if self.thumbnails:
             if thumbnail:
                 if file.can_thumbnail():
                     try:
                         thumbnail = misc.get_thumbnail(file.get_fs_filename(), content_type=file.content_type)
                         content = misc.get_thumbnail(file.get_fs_filename(), content_type=file.content_type)
                         response.set_content_type('image/png')
                         return thumbnail
                         return content
                     except misc.ThumbnailError:
                         raise errors.TraversalError()
                 else:
-...
         def download(self):
             if not is_url_signed():
                 self.check_receiver()
             try:
                 fn = get_request().form['f']
                 if '$' in fn:
                     # path to block field contents
                     fn2, idx, sub = fn.split('$', 2)
                     file = self.filled.data[fn2]['data'][int(idx)][sub]
                 else:
                     file = self.filled.data[fn]
             except (KeyError, ValueError):
                 raise errors.TraversalError()
             file = None
             if get_request().form and get_request().form.get('hash'):
                 # look in all known formdata files for file with given hash
                 file_digest = get_request().form.get('hash')
                 for field_data in self.filled.get_all_file_data():
                     if not hasattr(field_data, 'file_digest'):
                         continue
                     if field_data.file_digest() == file_digest:
                         return FileDirectory.serve_file(field_data)
             elif get_request().form and get_request().form.get('f'):
                 try:
                     fn = get_request().form['f']
                     if '$' in fn:
                         # path to block field contents
                         fn2, idx, sub = fn.split('$', 2)
                         file = self.filled.data[fn2]['data'][int(idx)][sub]
                     else:
                         file = self.filled.data[fn]
                 except (KeyError, ValueError):
                     pass
             if not hasattr(file, 'content_type'):
                 raise errors.TraversalError()

         return value
     def is_upload(obj):
         # we can't use isinstance() because obj can be a
         # wcs.qommon.form.PicklableUpload or a qommon.form.PicklableUpload
         return obj.__class__.__name__ == 'PicklableUpload'
     def is_attachment(obj):
         # ditto
         return obj.__class__.__name__ == 'AttachmentEvolutionPart'
     class QLookupRedirect:
         """
         Class to use to interrupt a _q_lookup method and redirect.

                 self.fp = io.BytesIO(self.data)
                 del self.data
         def file_digest(self):
             if getattr(self, 'qfilename', None):
                 # last file part is created using misc.file_digest()
                 return self.qfilename.split('/')[-1]
             return None
         def get_file(self):
             # quack like UploadedFile
             return self.get_file_pointer()
-...
             self.frontoffice_redirect = bool(frontoffice_redirect == 'true')
             self.backoffice_redirect = bool(backoffice_redirect == 'true')
         def file_digest(self):
             return None
         def save_tempfile(self, upload):
             if getattr(upload, 'storage_attrs', None):
                 # upload is already a remote PicklableUpload, it does not

     class LazyFormDataVar:
         def __init__(self, fields, data, formdata=None):
         def __init__(self, fields, data, formdata=None, base_formdata=None):
             self._fields = fields
             self._data = data or {}
             self._formdata = formdata
             self._base_formdata = base_formdata
         def inspect_keys(self):
             return self.varnames.keys()
-...
             return self._varnames
         def get_field_kwargs(self, field):
             return {'data': self._data, 'field': field, 'formdata': self._formdata}
             return {
                 'data': self._data,
                 'field': field,
                 'formdata': self._formdata,
                 'base_formdata': self._base_formdata,
+            }
         def __getitem__(self, key):
             try:
-...
     class LazyFieldVar:
         def __init__(self, data, field, formdata=None, **kwargs):
         def __init__(self, data, field, formdata=None, base_formdata=None, **kwargs):
             self._data = data
             self._field = field
             self._formdata = formdata
             self._base_formdata = base_formdata
             self._field_kwargs = kwargs
         @property
-...
     class LazyFieldVarFile(LazyFieldVar):
         def inspect_keys(self):
             keys = ['raw']
             if hasattr(self._formdata, 'get_file_base_url'):
             if hasattr(self._formdata, 'get_file_base_url') or self._base_formdata:
                 keys.append('url')
             return keys
         @property
         def url(self):
             if not hasattr(self._formdata, 'get_file_base_url'):
             if 'url' not in self.inspect_keys():
                 return None
             if self._base_formdata:
                 return self._field.get_download_url(formdata=self._base_formdata, file_value=self.raw)
             return self._field.get_download_url(formdata=self._formdata, **self._field_kwargs)
     class LazyBlockDataVar(LazyFormDataVar):
         def __init__(self, fields, data, formdata=None, parent_field=None, parent_field_index=0):
         def __init__(
             self, fields, data, formdata=None, parent_field=None, parent_field_index=0, base_formdata=None
         ):
             super().__init__(fields, data, formdata=formdata)
             self.parent_field = parent_field
             self.parent_field_index = parent_field_index
             self.base_formdata = base_formdata
         def get_field_kwargs(self, field):
             kwargs = super().get_field_kwargs(field)
             kwargs['parent_field'] = self.parent_field
             kwargs['parent_field_index'] = self.parent_field_index
             kwargs['base_formdata'] = self.base_formdata
             return kwargs
-...
                 formdata=self._formdata,
                 parent_field=self._field,
                 parent_field_index=int(key),
                 base_formdata=self._base_formdata,
+            )
         def __len__(self):

                 if not isinstance(part, WorkflowFormEvolutionPart):
                     continue
                 if part.varname == varname and part.data:
                     wfform_formdatas.append(LazyFormDataWorkflowFormsItem(part))
                     wfform_formdatas.append(LazyFormDataWorkflowFormsItem(part, base_formdata=self._formdata))
             if wfform_formdatas:
                 return LazyFormDataWorkflowFormsItems(wfform_formdatas)
             raise AttributeError(varname)
-...
     class LazyFormDataWorkflowFormsItem:
         def __init__(self, part):
         def __init__(self, part, base_formdata):
             self._part = part
             self.data = part.data
             self.base_formdata = base_formdata
         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)
             return LazyFormDataVar(
                 self._part.formdef.get_all_fields(),
                 self._part.data,
                 formdata=self,
                 base_formdata=self.base_formdata,
+            )
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.