0003-formdata-look-at-files-in-history-when-deleting-unus.patch

Lauréline Guérin, 20 octobre 2022 11:01

Voir les différences: en ligne côte à côte

Subject: [PATCH 3/3] formdata: look at files in history when deleting unused
 files (#62800)

 tests/test_formdef.py | 93 +++++++++++++++++++++++++++++++++++++++----
 wcs/formdata.py       | 12 +++++-
 wcs/formdef.py        |  2 +-
 3 files changed, 96 insertions(+), 11 deletions(-)

     from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormEvolutionPart, WorkflowFormFieldsFormDef
     from wcs.workflows import (
         AttachmentEvolutionPart,
         ContentSnapshotPart,
         Workflow,
         WorkflowBackofficeFieldsFormDef,
         WorkflowVariablesFieldsFormDef,
-...
                 pub.site_options.set('options', 'unused-files-behaviour', behaviour)
             formdata = formdef.data_class()()
             formdata.just_created()
             formdata.data = {
                 '5': PicklableUpload('test.txt', 'text/plain'),
+            }
             formdata.data['5'].receive([b'hello world'])
             formdata.just_created()
             formdata.store()
             assert isinstance(formdata.evolution[0].parts[0], ContentSnapshotPart)
             assert formdata.evolution[0].parts[0].new_data['5'].qfilename == formdata.data['5'].qfilename
             assert formdata.data['5'].qfilename in os.listdir(os.path.join(pub.app_dir, 'uploads'))
             clean_unused_files(pub)
-...
             for _ in range(5):
                 formdata = formdef.data_class()()
                 formdata.just_created()
                 formdata.data = {
                     '5': PicklableUpload('test.txt', 'text/plain'),
+                }
                 formdata.data['5'].receive([b'hello world'])
                 formdata.just_created()
                 formdata.store()
                 assert isinstance(formdata.evolution[0].parts[0], ContentSnapshotPart)
                 assert formdata.evolution[0].parts[0].new_data['5'].qfilename == formdata.data['5'].qfilename
             # same file, deduplicated
             assert os.listdir(os.path.join(pub.app_dir, 'uploads')) == [formdata.data['5'].qfilename]
-...
             clean_unused_files(pub)
             assert os.listdir(os.path.join(pub.app_dir, 'uploads')) == []
             # file referenced in formdata history, but not in formdata's data
             formdata = formdef.data_class()()
             formdata.data = {
                 '5': PicklableUpload('test.txt', 'text/plain'),
+            }
             formdata.data['5'].receive([b'hello world'])
             formdata.just_created()
             formdata.store()
             assert isinstance(formdata.evolution[0].parts[0], ContentSnapshotPart)
             assert formdata.evolution[0].parts[0].new_data['5'].qfilename == formdata.data['5'].qfilename
             qfilename = formdata.data['5'].qfilename
             formdata.data['5'] = None
             formdata.store()
             assert formdata.evolution[0].parts[0].new_data['5'].qfilename == qfilename
             assert os.listdir(os.path.join(pub.app_dir, 'uploads')) == [qfilename]
             clean_unused_files(pub)
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 1
             formdata.anonymise()
             assert len(formdata.evolution) == 1
             assert formdata.evolution[0].parts is None
             clean_unused_files(pub)
             assert os.listdir(os.path.join(pub.app_dir, 'uploads')) == []
             # file referenced in formdef option
             workflow = Workflow(name='variables')
-...
             formdef.store()
             formdata = formdef.data_class()()
             formdata.just_created()
             formdata.data = {
                 '5': PicklableUpload('test.txt', 'text/plain'),
+            }
             formdata.data['5'].receive([b'hello world'])
             formdata.just_created()
             formdata.store()
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 1
-...
             # file in block field
             formdata = formdef.data_class()()
             formdata.just_created()
             formdata.data = {
                 '6': {
                     'data': [
-...
             formdata.data['6']['data'][0]['234'].receive([b'hello world'])
             formdata.data['6']['data'][1]['234'].receive([b'hello world block'])
             formdata.workflow_data = {'wscall': {'data': ['not', 'a', 'block'], 'err': 0}}
             formdata.just_created()
             formdata.store()
             assert isinstance(formdata.evolution[0].parts[0], ContentSnapshotPart)
             assert (
                 formdata.evolution[0].parts[0].new_data['6']['data'][0]['234'].qfilename
                 == formdata.data['6']['data'][0]['234'].qfilename
+            )
             assert (
                 formdata.evolution[0].parts[0].new_data['6']['data'][1]['234'].qfilename
                 == formdata.data['6']['data'][1]['234'].qfilename
+            )
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 2
             clean_unused_files(pub)
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 2
-...
             clean_unused_files(pub)
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 0
             # non local storage: nothing happens
             # block field: file referenced in formdata history, but not in formdata's data
             formdata = formdef.data_class()()
             formdata.data = {
                 '6': {
                     'data': [
                         {'234': PicklableUpload('test.txt', 'text/plain')},
                         {'234': PicklableUpload('test2.txt', 'text/plain')},
                     ],
                     'schema': {'234': 'file'},
                 },
+            }
             formdata.data['6']['data'][0]['234'].receive([b'hello world'])
             formdata.data['6']['data'][1]['234'].receive([b'hello world block'])
             formdata.just_created()
             formdata.store()
             assert isinstance(formdata.evolution[0].parts[0], ContentSnapshotPart)
             assert (
                 formdata.evolution[0].parts[0].new_data['6']['data'][0]['234'].qfilename
                 == formdata.data['6']['data'][0]['234'].qfilename
+            )
             assert (
                 formdata.evolution[0].parts[0].new_data['6']['data'][1]['234'].qfilename
                 == formdata.data['6']['data'][1]['234'].qfilename
+            )
             qfilename0 = formdata.data['6']['data'][0]['234'].qfilename
             qfilename1 = formdata.data['6']['data'][1]['234'].qfilename
             formdata.data['6'] = {}
             formdata.store()
             assert formdata.evolution[0].parts[0].new_data['6']['data'][0]['234'].qfilename == qfilename0
             assert formdata.evolution[0].parts[0].new_data['6']['data'][1]['234'].qfilename == qfilename1
             assert os.listdir(os.path.join(pub.app_dir, 'uploads')) == [qfilename0, qfilename1]
             clean_unused_files(pub)
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 2
             assert os.listdir(os.path.join(pub.app_dir, 'uploads')) == [qfilename0, qfilename1]
             formdata.anonymise()
             assert len(formdata.evolution) == 1
             assert formdata.evolution[0].parts is None
             clean_unused_files(pub)
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 0
             # non local storage: nothing happens
             formdata = formdef.data_class()()
             formdata.data = {
                 '5': PicklableUpload('test.txt', 'text/plain'),
+            }
             formdata.data['5'].receive([b'hello world'])
             formdata.data['5'].storage = 'remote'
             formdata.data['5'].storage_attrs = {'redirect_url': 'https://crypto.example.net/1234'}
             formdata.just_created()
             formdata.store()
             assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 0
             clean_unused_files(pub)
-...
             # workflow attachment
             formdata = formdef.data_class()()
             formdata.just_created()
             formdata.data = {}
             formdata.just_created()
             formdata.store()
             formdata.evolution[-1].parts = [
-...
             # file from workflow form
             formdata = formdef.data_class()()
             formdata.just_created()
             formdata.data = {}
             formdata.just_created()
             formdata.store()
             display_form = FormWorkflowStatusItem()
-...
         # unknown unused-files-behaviour: do nothing
         pub.site_options.set('options', 'unused-files-behaviour', 'foo')
         formdata = formdef.data_class()()
         formdata.just_created()
         formdata.data = {
             '5': PicklableUpload('test-no-remove.txt', 'text/plain'),
+        }
         formdata.data['5'].receive([b'hello world'])
         formdata.just_created()
         formdata.store()
         assert formdata.data['5'].qfilename in os.listdir(os.path.join(pub.app_dir, 'uploads'))

                 empty &= self.data.get(key) is None
             return empty
         def get_all_file_data(self):
         def get_all_file_data(self, with_history=False):
             from wcs.wf.form import WorkflowFormEvolutionPart
             from wcs.workflows import ContentSnapshotPart
             for field_data in itertools.chain((self.data or {}).values(), (self.workflow_data or {}).values()):
             def check_field_data(field_data):
                 if misc.is_upload(field_data):
                     yield field_data
                 elif isinstance(field_data, dict) and isinstance(field_data.get('data'), list):
-...
                             for block_field_data in subfield_rowdata.values():
                                 if misc.is_upload(block_field_data):
                                     yield block_field_data
             for field_data in itertools.chain((self.data or {}).values(), (self.workflow_data or {}).values()):
                 yield from check_field_data(field_data)
             for part in self.iter_evolution_parts():
                 if misc.is_attachment(part):
                     yield part
-...
                     for field_data in (part.data or {}).values():
                         if misc.is_upload(field_data):
                             yield field_data
                 elif isinstance(part, ContentSnapshotPart):
                     # look into old and new values (belt and suspenders)
                     for field_data in list((part.old_data or {}).values()) + list((part.new_data or {}).values()):
                         yield from check_field_data(field_data)
         @classmethod
         def get_actionable_count(cls, user_roles):

                     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 formdata.get_all_file_data():
                     for field_data in formdata.get_all_file_data(with_history=True):
                         if is_upload(field_data):
                             yield field_data.get_fs_filename()
                         elif is_attachment(field_data):
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0003-formdata-look-at-files-in-history-when-deleting-unus.patch