refactor: enhance output messages and improve storage handling in filer_check

Author: Baraff24

filer/management/commands/filer_check.py

@@ -67,7 +67,8 @@ def handle(self, *args, **options):
                     "\nThis will delete missing file references from the database.\n"
                     "Type 'yes' to continue, or 'no' to cancel: "
                 ) != 'yes':
-                    self.stdout.write("Aborted: Missing file references were not deleted.")
+                    self.stdout.write("Aborted: Missing file references were not deleted.\n")
+                    self.stdout.flush()
                     return
             self.verify_references(options)
 
@@ -77,7 +78,8 @@ def handle(self, *args, **options):
                     "\nThis will delete orphaned files from storage.\n"
                     "Type 'yes' to continue, or 'no' to cancel: "
                 ) != 'yes':
-                    self.stdout.write("Aborted: Orphaned files were not deleted.")
+                    self.stdout.write("Aborted: Orphaned files were not deleted.\n")
+                    self.stdout.flush()
                     return
             self.verify_storages(options)
 
@@ -86,7 +88,7 @@ def handle(self, *args, **options):
 
     def verify_references(self, options):
         """
-        Checks that every file reference in the database exists in the storage.
+        Checks that every file reference in the database exists in storage.
         If a file is missing, either report it or delete the reference based on the provided options.
         """
         for file in File.objects.all():
@@ -95,21 +97,24 @@ def verify_references(self, options):
                     file.delete()
                     verbose_msg = f"Deleted missing file reference '{file.folder}/{file}' from the database."
                 else:
-                    verbose_msg = f"File reference '{file.folder}/{file}' is missing in the storage."
-                # For higher verbosity, print the full verbose message.
+                    verbose_msg = f"File reference '{file.folder}/{file}' is missing in storage."
                 if options.get('verbosity', 1) > 2:
-                    self.stdout.write(verbose_msg)
-                # Otherwise, just output the relative file path.
+                    self.stdout.write(verbose_msg + "\n")
+                    self.stdout.flush()
                 elif options.get('verbosity'):
-                    self.stdout.write(os.path.join(str(file.folder), str(file)))
+                    self.stdout.write(os.path.join(str(file.folder), str(file)) + "\n")
+                    self.stdout.flush()
 
     def verify_storages(self, options):
         """
-        Scans all storages defined in FILER_STORAGES (e.g. public and private)
+        Scans all storages defined in FILER_STORAGES (e.g., public and private)
         for orphaned files, then reports or deletes them based on the options.
         """
 
         def walk(storage, prefix, label_prefix):
+            # If the directory does not exist, there is nothing to scan
+            if not storage.exists(prefix):
+                return
             child_dirs, files = storage.listdir(prefix)
             for filename in files:
                 actual_path = os.path.join(prefix, filename)
@@ -121,26 +126,25 @@ def walk(storage, prefix, label_prefix):
                     else:
                         message = f"Found orphaned file '{relfilename}'"
                     if options.get('verbosity', 1) > 2:
-                        self.stdout.write(message)
+                        self.stdout.write(message + "\n")
+                        self.stdout.flush()
                     elif options.get('verbosity'):
-                        self.stdout.write(relfilename)
+                        self.stdout.write(relfilename + "\n")
+                        self.stdout.flush()
             for child in child_dirs:
                 walk(storage, os.path.join(prefix, child), os.path.join(label_prefix, child))
 
-        # Loop through each storage configuration (e.g. public, private, etc.)
+        # Loop through each storage configuration (e.g., public, private, etc.)
         for storage_name, storage_config in filer_settings.FILER_STORAGES.items():
             storage_settings = storage_config.get('main')
             if not storage_settings:
                 continue
             storage = import_string(storage_settings['ENGINE'])()
             if storage_settings.get('OPTIONS', {}).get('location'):
                 storage.location = storage_settings['OPTIONS']['location']
-            # Use a label prefix: for public and private storages, use their names.
-            if storage_name in ['public', 'private']:
-                label_prefix = storage_name
-            else:
-                label_prefix = storage_settings['UPLOAD_TO_PREFIX']
-            walk(storage, storage_settings['UPLOAD_TO_PREFIX'], label_prefix)
+            # Set label_prefix: for public and private storages, use their names.
+            label_prefix = storage_name if storage_name in ['public', 'private'] else storage_settings.get('UPLOAD_TO_PREFIX', '')
+            walk(storage, storage_settings.get('UPLOAD_TO_PREFIX', ''), label_prefix)
 
     def image_dimensions(self, options):
         """
@@ -154,9 +158,10 @@ def image_dimensions(self, options):
 
         ImageModel = load_model(filer_settings.FILER_IMAGE_MODEL)
         images_without_dimensions = ImageModel.objects.filter(Q(_width=0) | Q(_width__isnull=True))
-        self.stdout.write(f"Setting dimensions for {images_without_dimensions.count()} images")
+        self.stdout.write(f"Setting dimensions for {images_without_dimensions.count()} images" + "\n")
+        self.stdout.flush()
         for image in images_without_dimensions:
-            file_holder = image.file_ptr if hasattr(image, 'file_ptr') and image.file_ptr else image
+            file_holder = image.file_ptr if getattr(image, 'file_ptr', None) else image
             try:
                 imgfile = file_holder.file
                 imgfile.seek(0)

tests/test_filer_check.py

@@ -1,11 +1,13 @@
 import os
 import shutil
+import tempfile
 from io import BytesIO, StringIO
 
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.core.management import call_command
 from django.test import TestCase
 from django.utils.module_loading import import_string
+from django.core.files.base import ContentFile
 
 from filer import settings as filer_settings
 from filer.models.filemodels import File
@@ -27,13 +29,18 @@ class FilerCheckTestCase(TestCase):
     </svg>"""
 
     def setUp(self):
-        # Clean up the public folder to avoid interference between tests.
-        public_settings = filer_settings.FILER_STORAGES['public']['main']
-        storage = import_string(public_settings['ENGINE'])()
-        upload_prefix = public_settings['UPLOAD_TO_PREFIX']
-        if storage.exists(upload_prefix):
-            shutil.rmtree(storage.path(upload_prefix))
+        # Clear all configured storages to ensure a clean state for each test.
+        # This prevents interference from files left in any storage.
+        for storage_alias, storage_configs in filer_settings.FILER_STORAGES.items():
+            config = storage_configs.get('main')
+            if not config:
+                continue
+            storage = import_string(config['ENGINE'])()
+            upload_prefix = config.get('UPLOAD_TO_PREFIX', '')
+            if storage.exists(upload_prefix):
+                shutil.rmtree(storage.path(upload_prefix))
 
+        # Create a sample file for testing in the public storage.
         original_filename = 'testimage.jpg'
         file_obj = SimpleUploadedFile(
             name=original_filename,
@@ -55,8 +62,10 @@ def test_delete_missing(self):
         call_command('filer_check', stdout=out, missing=True)
         self.assertEqual('', out.getvalue())
 
+        # Remove the file to simulate a missing file.
         os.remove(self.filer_file.file.path)
         call_command('filer_check', stdout=out, missing=True)
+        # When verbosity is low, a simple relative file path is output.
         self.assertEqual("None/testimage.jpg\n", out.getvalue())
         self.assertIsInstance(File.objects.get(id=file_pk), File)
 
@@ -65,26 +74,36 @@ def test_delete_missing(self):
             File.objects.get(id=file_pk)
 
     def test_delete_orphans_public(self):
+        # First check - should be empty initially
         out = StringIO()
-        self.assertTrue(os.path.exists(self.filer_file.file.path))
-        call_command('filer_check', stdout=out, orphans=True)
-        # The public folder should be free of orphaned files.
+        call_command('filer_check', stdout=out, orphans=True, verbosity=1)
         self.assertEqual('', out.getvalue())
 
-        # Add an orphan file to the public storage.
+        # Add an orphan file using the storage API directly
         public_settings = filer_settings.FILER_STORAGES['public']['main']
         storage = import_string(public_settings['ENGINE'])()
-        public_path = storage.path(public_settings['UPLOAD_TO_PREFIX'])
-        orphan_file = os.path.join(public_path, 'hello.txt')
-        os.makedirs(public_path, exist_ok=True)
-        with open(orphan_file, 'w') as fh:
-            fh.write("I don't belong here!")
-        call_command('filer_check', stdout=out, orphans=True)
+
+        # Configure storage location if specified in settings
+        if public_settings.get('OPTIONS', {}).get('location'):
+            storage.location = public_settings['OPTIONS']['location']
+
+        # Get upload prefix and create file path
+        prefix = public_settings.get('UPLOAD_TO_PREFIX', '')
+        file_path = 'hello.txt'
+        rel_path = os.path.join(prefix, file_path) if prefix else file_path
+
+        # Save file through storage API
+        storage.save(rel_path, ContentFile(b"I don't belong here!"))
+        self.assertTrue(storage.exists(rel_path))
+
+        # Check if orphan is detected
+        out = StringIO()
+        call_command('filer_check', stdout=out, orphans=True, verbosity=1)
         self.assertEqual("public/hello.txt\n", out.getvalue())
-        self.assertTrue(os.path.exists(orphan_file))
 
+        # Delete orphans
         call_command('filer_check', delete_orphans=True, interactive=False, verbosity=0)
-        self.assertFalse(os.path.exists(orphan_file))
+        self.assertFalse(storage.exists(rel_path))
 
     def test_delete_orphans_private(self):
         # Skip test if private storage is not configured.
@@ -94,15 +113,16 @@ def test_delete_orphans_private(self):
         out = StringIO()
         private_settings = filer_settings.FILER_STORAGES['private']['main']
         storage = import_string(private_settings['ENGINE'])()
+        # Set storage location if defined in OPTIONS.
         if private_settings.get('OPTIONS', {}).get('location'):
             storage.location = private_settings['OPTIONS']['location']
-        private_path = storage.path(private_settings['UPLOAD_TO_PREFIX'])
+        private_path = storage.path(private_settings.get('UPLOAD_TO_PREFIX', ''))
         os.makedirs(private_path, exist_ok=True)
 
         orphan_file = os.path.join(private_path, 'private_orphan.txt')
         with open(orphan_file, 'w') as fh:
             fh.write("I don't belong here!")
-        # Verify that the command detects the orphan file.
+        # Run the command and check that it detects the private orphan file.
         call_command('filer_check', stdout=out, orphans=True)
         self.assertIn("private_orphan.txt", out.getvalue())
         self.assertTrue(os.path.exists(orphan_file))
@@ -111,11 +131,84 @@ def test_delete_orphans_private(self):
         call_command('filer_check', delete_orphans=True, interactive=False, verbosity=0)
         self.assertFalse(os.path.exists(orphan_file))
 
+    def test_delete_orphans_multiple_storages(self):
+        """
+        Test that the filer_check command correctly handles orphaned files in multiple storages
+        without permanently modifying the settings. We use monkey-patching to assign temporary
+        directories to the storage configurations.
+        """
+        out = StringIO()
+
+        # --- Monkey-patch public storage location ---
+        public_config = filer_settings.FILER_STORAGES['public']['main']
+        temp_public_dir = tempfile.mkdtemp()
+        if 'OPTIONS' in public_config:
+            public_config['OPTIONS']['location'] = temp_public_dir
+        else:
+            public_config['OPTIONS'] = {'location': temp_public_dir}
+        # Determine the upload prefix (if any) and ensure the corresponding directory exists.
+        public_upload_prefix = public_config.get('UPLOAD_TO_PREFIX', '')
+        if public_upload_prefix:
+            public_full_dir = os.path.join(temp_public_dir, public_upload_prefix)
+        else:
+            public_full_dir = temp_public_dir
+        os.makedirs(public_full_dir, exist_ok=True)
+
+        # --- Monkey-patch private storage location ---
+        private_config = filer_settings.FILER_STORAGES.get('private', {}).get('main')
+        if private_config:
+            temp_private_dir = tempfile.mkdtemp()
+            if 'OPTIONS' in private_config:
+                private_config['OPTIONS']['location'] = temp_private_dir
+            else:
+                private_config['OPTIONS'] = {'location': temp_private_dir}
+            private_upload_prefix = private_config.get('UPLOAD_TO_PREFIX', '')
+            if private_upload_prefix:
+                private_full_dir = os.path.join(temp_private_dir, private_upload_prefix)
+            else:
+                private_full_dir = temp_private_dir
+            os.makedirs(private_full_dir, exist_ok=True)
+        else:
+            self.skipTest("Private storage not configured in FILER_STORAGES.")
+
+        # --- Initialize storages using the patched locations ---
+        from django.core.files.storage import FileSystemStorage
+        storage_public = FileSystemStorage(location=temp_public_dir)
+        storage_private = FileSystemStorage(location=private_config['OPTIONS']['location'])
+
+        # --- Save dummy orphan files in both storages ---
+        # For public storage, include the upload prefix in the filename if needed.
+        if public_upload_prefix:
+            filename_public = os.path.join(public_upload_prefix, 'orphan_public.txt')
+        else:
+            filename_public = 'orphan_public.txt'
+        if private_config.get('UPLOAD_TO_PREFIX', ''):
+            filename_private = os.path.join(private_config['UPLOAD_TO_PREFIX'], 'orphan_private.txt')
+        else:
+            filename_private = 'orphan_private.txt'
+
+        storage_public.save(filename_public, ContentFile(b"dummy content"))
+        storage_private.save(filename_private, ContentFile(b"dummy content"))
+
+        # --- Run the filer_check command ---
+        call_command('filer_check', stdout=out, orphans=True)
+        output = out.getvalue()
+
+        # Verify that the output contains indicators for both storages.
+        self.assertIn('public', output)
+        self.assertIn('private', output)
+
+        # --- Clean up ---
+        storage_public.delete(filename_public)
+        storage_private.delete(filename_private)
+        shutil.rmtree(temp_public_dir)
+        shutil.rmtree(private_config['OPTIONS']['location'])
+
     def test_image_dimensions_corrupted_file(self):
         original_filename = 'testimage.jpg'
         file_obj = SimpleUploadedFile(
             name=original_filename,
-            content=create_image().tobytes(),  # corrupted file
+            content=create_image().tobytes(),  # Simulate a corrupted file.
             content_type='image/jpeg'
         )
         self.filer_image = Image.objects.create(
@@ -189,3 +282,4 @@ def test_image_dimensions_svg(self):
         call_command('filer_check', image_dimensions=True)
         self.filer_image.refresh_from_db()
         self.assertGreater(self.filer_image._width, 0)
+