upload_storage: handle empty qfilename, for non-local storage (#42236)

tests/test_formdef.py

@@ -275,6 +275,10 @@ def test_unused_file_removal_job(pub):
 
     if not pub.site_options.has_section('options'):
         pub.site_options.add_section('options')
+    pub.site_options.add_section('storage-remote')
+    pub.site_options.set('storage-remote', 'label', 'remote')
+    pub.site_options.set('storage-remote', 'class', 'wcs.qommon.upload_storage.RemoteOpaqueUploadStorage')
+    pub.site_options.set('storage-remote', 'ws', 'https://crypto.example.net/')
 
     for behaviour in (None, 'move', 'remove'):
         if behaviour:
@@ -346,6 +350,20 @@ def test_unused_file_removal_job(pub):
         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.just_created()
+        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.store()
+        assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 0
+        clean_unused_files(pub)
+        assert len(os.listdir(os.path.join(pub.app_dir, 'uploads'))) == 0
+
         # workflow attachment
         formdata = formdef.data_class()()
         formdata.just_created()

wcs/formdef.py

@@ -1617,6 +1617,8 @@ def clean_unused_files(publisher):
 
     used_filenames = set()
     for filename in accumulate_filenames():
+        if not filename:  # alternative storage
+            continue
         if not os.path.isabs(filename):
             filename = os.path.join(publisher.app_dir, filename)
         used_filenames.add(filename)

wcs/qommon/upload_storage.py

@@ -61,6 +61,8 @@ class PicklableUpload(Upload):
     def get_filename(self):
         if not hasattr(self, 'qfilename'):
             raise AttributeError('filename')
+        if hasattr(self, 'storage_attrs'):  # alternative storage
+            return None
         basedir = os.path.join(get_publisher().app_dir, 'uploads')
         return os.path.join(basedir, self.qfilename)