added management command (#912)

jrief · FinalAngel · commit e1e95d7c4c9b · 2019-09-26T10:26:02.000+02:00
* added management command * Flake8 complained, used isort * Use the storage.location as intended * change interface according to @yakky’s annotations * add management command ‘filer_check’ plus tests * Use verbosity flag <3 to only render filenames * fix indention * use ‘input’ from PY2/3 compatibility layer * isort complained for that extra line * fix typos * remove translations from managment commands * aborting a managment command does not raise an error * fix typo * fix flake8 complaint * fix isort complaints * fix all tests regarding filer_check * fix flake8 complaint
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -9,6 +9,8 @@ CHANGELOG
 * Removed support for Django <= 1.10
 * Removed outdated files
 * Code alignments with other addons
+* Added management command ``filer_check`` to check the integrity of the
+  database against the file system, and vice versa.
 * Add jQuery as AdminFileWidget Media dependency
 
 
diff --git a/docs/management_commands.rst b/docs/management_commands.rst
@@ -12,3 +12,33 @@ command.
 To generate them, use::
 
     ./manage.py generate_thumbnails
+
+
+Filesystem Checks
+-----------------
+
+**django-filer** offers a few commands to check the integrity of the database against the files
+stored on disk. By invoking::
+
+    ./manage.py filer_check --missing
+
+all files which are referenced by the database, but missing on disk are reported.
+
+Invoking::
+
+    ./manage.py filer_check --delete-missing
+
+deletes those file references from the database.
+
+Invoking::
+
+    ./manage.py filer_check --orphans
+
+lists all files found on disk belonging to the configured storage engine, but which
+are not referenced by the database.
+
+Invoking::
+
+    ./manage.py filer_check --delete-orphans
+
+deletes those orphaned files from disk.
diff --git a/filer/management/commands/filer_check.py b/filer/management/commands/filer_check.py
@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+import os
+
+from django.core.files.storage import DefaultStorage
+from django.core.management.base import BaseCommand
+from django.utils.module_loading import import_string
+from django.utils.six.moves import input
+
+from filer import settings as filer_settings
+
+
+class Command(BaseCommand):
+    help = "Look for orphaned files in media folders."
+    storage = DefaultStorage()
+    prefix = filer_settings.FILER_STORAGES['public']['main']['UPLOAD_TO_PREFIX']
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--orphans',
+            action='store_true',
+            dest='orphans',
+            default=False,
+            help="Walk through the media folders and look for orphaned files.",
+        )
+        parser.add_argument(
+            '--delete-orphans',
+            action='store_true',
+            dest='delete_orphans',
+            default=False,
+            help="Delete orphaned files from their media folders.",
+        )
+        parser.add_argument(
+            '--missing',
+            action='store_true',
+            dest='missing',
+            default=False,
+            help="Verify media folders and report about missing files.",
+        )
+        parser.add_argument(
+            '--delete-missing',
+            action='store_true',
+            dest='delete_missing',
+            default=False,
+            help="Delete references in database if files are missing in media folder.",
+        )
+        parser.add_argument(
+            '--noinput',
+            '--no-input',
+            action='store_false',
+            dest='interactive',
+            default=True,
+            help="Do NOT prompt the user for input of any kind."
+        )
+
+    def handle(self, *args, **options):
+        if options['missing']:
+            self.verify_references(options)
+        if options['delete_missing']:
+            if options['interactive']:
+                msg = "\nThis will delete entries from your database. Are you sure you want to do this?\n\n" \
+                      "Type 'yes' to continue, or 'no' to cancel: "
+                if input(msg) != 'yes':
+                    self.stdout.write("Aborted: Delete missing file entries from database.")
+                    return
+            self.verify_references(options)
+
+        if options['orphans']:
+            self.verify_storages(options)
+        if options['delete_orphans']:
+            if options['interactive']:
+                msg = "\nThis will delete orphaned files from your storage. Are you sure you want to do this?\n\n" \
+                      "Type 'yes' to continue, or 'no' to cancel: "
+                if input(msg) != 'yes':
+                    self.stdout.write("Aborted: Delete orphaned files from storage.")
+                    return
+            self.verify_storages(options)
+
+    def verify_references(self, options):
+        from filer.models.filemodels import File
+
+        for file in File.objects.all():
+            if not file.file.storage.exists(file.file.name):
+                if options['delete_missing']:
+                    file.delete()
+                    msg = "Delete missing file reference '{}/{}' from database."
+                else:
+                    msg = "Referenced file '{}/{}' is missing in media folder."
+                if options['verbosity'] > 2:
+                    self.stdout.write(msg.format(str(file.folder), str(file)))
+                elif options['verbosity']:
+                    self.stdout.write(os.path.join(str(file.folder), str(file)))
+
+    def verify_storages(self, options):
+        from filer.models.filemodels import File
+
+        def walk(prefix):
+            child_dirs, files = storage.listdir(prefix)
+            for filename in files:
+                relfilename = os.path.join(prefix, filename)
+                if not File.objects.filter(file=relfilename).exists():
+                    if options['delete_orphans']:
+                        storage.delete(relfilename)
+                        msg = "Deleted orphaned file '{}'"
+                    else:
+                        msg = "Found orphaned file '{}'"
+                    if options['verbosity'] > 2:
+                        self.stdout.write(msg.format(relfilename))
+                    elif options['verbosity']:
+                        self.stdout.write(relfilename)
+
+            for child in child_dirs:
+                walk(os.path.join(prefix, child))
+
+        filer_public = filer_settings.FILER_STORAGES['public']['main']
+        storage = import_string(filer_public['ENGINE'])()
+        walk(filer_public['UPLOAD_TO_PREFIX'])
diff --git a/tests/test_filer_check.py b/tests/test_filer_check.py
@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import
+
+import os
+import shutil
+
+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.utils.six import StringIO
+
+from filer import settings as filer_settings
+from filer.models.filemodels import File
+
+from tests.helpers import create_image
+
+
+class FilerCheckTestCase(TestCase):
+    def setUp(self):
+        # ensure that filer_public directory is empty from previous tests
+        storage = import_string(filer_settings.FILER_STORAGES['public']['main']['ENGINE'])()
+        upload_to_prefix = filer_settings.FILER_STORAGES['public']['main']['UPLOAD_TO_PREFIX']
+        if storage.exists(upload_to_prefix):
+            shutil.rmtree(storage.path(upload_to_prefix))
+
+        original_filename = 'testimage.jpg'
+        file_obj = SimpleUploadedFile(
+            name=original_filename,
+            content=create_image().tobytes(),
+            content_type='image/jpeg')
+        self.filer_file = File.objects.create(
+            file=file_obj,
+            original_filename=original_filename)
+
+    def tearDown(self):
+        self.filer_file.delete()
+
+    def test_delete_missing(self):
+        out = StringIO()
+        self.assertTrue(os.path.exists(self.filer_file.file.path))
+        file_pk = self.filer_file.id
+        call_command('filer_check', stdout=out, missing=True)
+        self.assertEqual('', out.getvalue())
+
+        os.remove(self.filer_file.file.path)
+        call_command('filer_check', stdout=out, missing=True)
+        self.assertEqual("None/testimage.jpg\n", out.getvalue())
+        self.assertIsInstance(File.objects.get(id=file_pk), File)
+
+        call_command('filer_check', delete_missing=True, interactive=False, verbosity=0)
+        with self.assertRaises(File.DoesNotExist):
+            File.objects.get(id=file_pk)
+
+    def test_delete_orphans(self):
+        out = StringIO()
+        self.assertTrue(os.path.exists(self.filer_file.file.path))
+        call_command('filer_check', stdout=out, orphans=True)
+        # folder must be clean, free of orphans
+        self.assertEqual('', out.getvalue())
+
+        # add an orphan file to our storage
+        storage = import_string(filer_settings.FILER_STORAGES['public']['main']['ENGINE'])()
+        filer_public = storage.path(filer_settings.FILER_STORAGES['public']['main']['UPLOAD_TO_PREFIX'])
+        orphan_file = os.path.join(filer_public, 'hello.txt')
+        with open(orphan_file, 'w') as fh:
+            fh.write("I don't belong here!")
+        call_command('filer_check', stdout=out, orphans=True)
+        self.assertEqual("filer_public/hello.txt\n", out.getvalue())
+        self.assertTrue(os.path.exists(orphan_file))
+
+        call_command('filer_check', delete_orphans=True, interactive=False, verbosity=0)
+        self.assertFalse(os.path.exists(orphan_file))
