0001-workflows-store-workflow-form-data-for-structured-ac.patch

     from webtest import Hidden, Radio, Upload
     from wcs import fields
     from wcs.blocks import BlockDef
     from wcs.carddef import CardDef
     from wcs.categories import Category
     from wcs.data_sources import NamedDataSource
-...
     from wcs.qommon.emails import docutils
     from wcs.qommon.ident.password_accounts import PasswordAccount
     from wcs.qommon.misc import ConnectionError
     from wcs.qommon.substitution import CompatibilityNamesDict
     from wcs.roles import logged_users_role
     from wcs.tracking_code import TrackingCode
     from wcs.wf.backoffice_fields import SetBackofficeFieldsWorkflowStatusItem
-...
         formdata = formdef.data_class()()
         formdata.user_id = user.id
         formdata.just_created()
         formdata.status = 'wf-new'
         formdata.data = {'0': 'plop'}
         formdata.store()
-...
         formdata = formdef.data_class()()
         formdata.user_id = user.id
         formdata.just_created()
         formdata.status = 'wf-new'
         formdata.data = {'0': 'plop'}
         formdata.store()
-...
         # and persist after being saved again
         resp = resp.form.submit('submit').follow()
         assert '<span>test.txt</span>' in resp.text
     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'] == 'ABC'
         assert substvars['form_workflow_form_blah_0_var_fooblock_var_test'] == 'ABC'
         assert substvars['form_workflow_form_blah_1_var_fooblock_var_test'] == 'XYZ'
         assert substvars['form_workflow_form_blah_latest_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.formdef import FormDef, get_formdefs_of_all_kinds
     from wcs.qommon.http_request import HTTPRequest
     from wcs.qommon.upload_storage import PicklableUpload
     from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormFieldsFormDef
     from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormEvolutionPart, WorkflowFormFieldsFormDef
     from wcs.workflows import (
         AttachmentEvolutionPart,
         Workflow,
-...
             assert len(glob.glob(os.path.join(pub.app_dir, 'attachments', '*/*'))) == 0
             # files in user profile
             user_formdef = UserFieldsFormDef(pub)
             user_formdef.fields.append(fields.FileField(id='3', label='test', type='file'))
             user_formdef.store()
-...
             clean_unused_files(pub)
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 0
             # 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
             if behaviour == 'move':
                 # 3 files ("hello world" + "hello world 2" + "hello world block")
                 assert len(os.listdir(os.path.join(pub.app_dir, 'unused-files/uploads/'))) == 3
                 # 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
                 # 1 attachment
                 assert len(glob.glob(os.path.join(pub.app_dir, 'unused-files/attachments/*/*'))) == 1

         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():
-...
                     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()
             for user in publisher.user_class.select():
                 for field_data in (user.form_data or {}).values():
                     if is_upload(field_data):

             return LazyFormDataLinks(self._formdata)
         @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)
         @property
         def parent(self):
             formdata = self._formdata.get_parent()

     from wcs.formdef import FormDef, lax_int
     from wcs.forms.common import FileDirectory
     from wcs.forms.root import FormPage
     from wcs.workflows import RedisplayFormException, WorkflowStatusItem, register_item_class
     from wcs.variables import LazyFormDataVar
     from wcs.workflows import EvolutionPart, RedisplayFormException, WorkflowStatusItem, register_item_class
     from ..qommon import _
     from ..qommon.form import SingleSelectWidget, VarnameWidget, WidgetList
     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
     def lookup_wf_form_file(self, filename):
         # supports for URLs such as /$formdata/$id/files/form-$formvar-$fieldvar/test.txt
         try:
-...
                     self.formdef.readonly = True
                 fields_directory = WorkflowFormFieldsDirectory(self.formdef)
                 if self.varname:
                     fields_directory.field_var_prefix = '%s_var_' % self.varname
                     fields_directory.field_var_prefix = 'form_workflow_form_%s_var_' % self.varname
                 fields_directory.html_top = html_top
                 return fields_directory
             return None
-...
             FormPage.apply_field_prefills({}, form, fields)
         def evaluate_live_form(self, form, formdata, user):
         def evaluate_live_form(self, form, formdata, user, submit=False):
             if not self.formdef:
                 return
             workflow_data = {}
             self.prefix_form_fields()
             for k, v in get_dict_with_varnames(
                 self.formdef.fields, self.formdef.get_data(form), varnames_only=True
             ).items():
             formdef_data = self.formdef.get_data(form)
             for k, v in get_dict_with_varnames(self.formdef.fields, formdef_data, varnames_only=True).items():
                 workflow_data['%s_%s' % (self.varname, k)] = v
             formdata.update_workflow_data(workflow_data)
             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))
+                )
         def submit_form(self, form, formdata, user, evo):
             if not self.formdef:
-...
                         if add_element_widget and add_element_widget.parse():
                             raise RedisplayFormException()
             if form.get_submit() == 'submit' and not form.has_errors():
                 self.evaluate_live_form(form, formdata, user)
                 self.evaluate_live_form(form, formdata, user, submit=True)
                 formdata.store()
             get_publisher().substitutions.unfeed(lambda x: x.__class__.__name__ == 'ConditionVars')
-...
     register_item_class(FormWorkflowStatusItem)
     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', 'latest']
         @property
         def var(self):
             # alias when there's a single workflow form
             return self._wfform_formdatas[0].var
         @property
         def latest(self):
             # alias to latest form
             return {'var': 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)
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.