Fix issue that prevented deletion of a folder's descendant folders when the raw storage was a POSIX filesystem

Author: jbernal0019

chris_backend/core/models.py

@@ -337,6 +337,17 @@ def _update_public_access(self, public_tf):
         ChrisLinkFile.objects.bulk_update(link_files, ['public'])
 
 
+@receiver(post_delete, sender=ChrisFolder)
+def auto_delete_folder_from_storage(sender, instance, **kwargs):
+    storage_path = instance.path
+    storage_manager = connect_storage(settings)
+    try:
+        if storage_manager.path_exists(storage_path):
+            storage_manager.delete_path(storage_path)
+    except Exception as e:
+        logger.error('Storage error, detail: %s' % str(e))
+
+
 class ChrisFolderFilter(FilterSet):
     path = django_filters.CharFilter(field_name='path')
 

chris_backend/core/tests/test_models.py

@@ -0,0 +1,65 @@
+
+import logging
+import os
+from unittest import mock
+
+from django.test import TestCase
+from django.contrib.auth.models import User
+
+from core.models import ChrisFolder, ChrisFile
+
+
+class ModelTests(TestCase):
+
+    def setUp(self):
+        # avoid cluttered console output (for instance logging all the http requests)
+        logging.disable(logging.WARNING)
+
+        # superuser chris (owner of root folders)
+        self.chris_username = 'chris'
+
+        # create normal user
+        self.username = 'foo'
+        self.password = 'bar'
+        User.objects.create_user(username=self.username, password=self.password)
+
+    def tearDown(self):
+        # re-enable logging
+        logging.disable(logging.NOTSET)
+
+
+class ChrisFileModelTests(ModelTests):
+
+    def setUp(self):
+        super(ChrisFileModelTests, self).setUp()
+
+        # create a ChrisFile instance
+        owner = User.objects.get(username=self.username)
+        self.upload_path = f'home/{self.username}/uploads/file1.txt'
+        folder_path = os.path.dirname(self.upload_path)
+        (parent_folder, _) = ChrisFolder.objects.get_or_create(path=folder_path,
+                                                               owner=owner)
+        self.file = ChrisFile(parent_folder=parent_folder, owner=owner)
+        self.file.fname.name = self.upload_path
+        self.file.save()
+
+    def test_move(self):
+        """
+        Test whether custom 'move' method successfully updates a ChrisFile with
+        the correct path and parent folder.
+        """
+        new_path =  f'home/{self.username}/tests/file1.txt'
+
+        storage_manager_mock = mock.Mock()
+        storage_manager_mock.obj_exists = mock.Mock()
+        storage_manager_mock.copy_obj = mock.Mock()
+        storage_manager_mock.delete_obj = mock.Mock()
+
+        with mock.patch('core.models.connect_storage') as connect_storage_mock:
+            connect_storage_mock.return_value=storage_manager_mock
+            self.file.move(new_path)
+            storage_manager_mock.obj_exists.assert_called_with(new_path)
+            storage_manager_mock.copy_obj.assert_called_with(self.upload_path, new_path)
+            self.assertEqual(self.file.fname.name, new_path)
+            self.assertEqual(self.file.parent_folder.path, f'home/{self.username}/tests')
+            storage_manager_mock.delete_obj.assert_called_with(self.upload_path)

chris_backend/userfiles/serializers.py

@@ -1,11 +1,9 @@
 
 import os
 
-from django.conf import settings
 from rest_framework import serializers
 
 from core.models import ChrisFolder
-from core.storage import connect_storage
 from core.serializers import ChrisFileSerializer
 from .models import UserFile
 
@@ -46,38 +44,31 @@ def create(self, validated_data):
 
     def update(self, instance, validated_data):
         """
-        Overriden to set the file's saving path and parent folder and delete the old
-        path from storage.
+        Overriden to grant or remove public access to the file and/or move it to a new
+        path.
         """
-        if 'public' in validated_data:
-            instance.public = validated_data['public']
+        public = instance.public
+
+        if public and 'public' in validated_data and not validated_data['public']:
+            instance.remove_public_link()
+            instance.remove_public_access()
 
         upload_path = validated_data.pop('upload_path', None)
 
         if upload_path:
+            if public and ('public' not in validated_data or validated_data['public']):
+                instance.remove_public_link()
+
             # user file will be stored at: SWIFT_CONTAINER_NAME/<upload_path>
             # where <upload_path> must start with home/
-            old_storage_path = instance.fname.name
-
-            storage_manager = connect_storage(settings)
-            if storage_manager.obj_exists(upload_path):
-                storage_manager.delete_obj(upload_path)
-
-            storage_manager.copy_obj(old_storage_path, upload_path)
-            storage_manager.delete_obj(old_storage_path)
-
-            folder_path = os.path.dirname(upload_path)
-            owner = instance.owner
-
-            try:
-                parent_folder = ChrisFolder.objects.get(path=folder_path)
-            except ChrisFolder.DoesNotExist:
-                parent_folder = ChrisFolder.objects.create(path=folder_path, owner=owner)
+            instance.move(upload_path)
 
-            instance.parent_folder = parent_folder
-            instance.fname.name = upload_path
+            if public and ('public' not in validated_data or validated_data['public']):
+                instance.create_public_link()  # recreate public link
 
-        instance.save()
+        if not public and 'public' in validated_data and validated_data['public']:
+            instance.grant_public_access()
+            instance.create_public_link()
         return instance
 
     def validate_upload_path(self, upload_path):

chris_backend/userfiles/tests/test_serializers.py

@@ -53,11 +53,12 @@ def test_create(self):
         user_file = userfiles_serializer.create(validated_data)
         self.assertEqual(user_file.fname.name, f'home/{self.username}/uploads/file1.txt')
         self.assertEqual(user_file.parent_folder.path, f'home/{self.username}/uploads')
+        user_file.delete()
 
     def test_update(self):
         """
-        Test whether overriden 'update' method successfully updates a UserFile with
-        the correct path and parent folder
+        Test whether overriden 'update' method successfully moves a UserFile to the
+        correct path.
         """
         owner = User.objects.get(username=self.username)
         upload_path = f'home/{self.username}/uploads/file1.txt'
@@ -71,18 +72,9 @@ def test_update(self):
         userfiles_serializer = UserFileSerializer()
         validated_data = {'upload_path': f'home/{self.username}/tests/file1.txt'}
 
-        storage_manager_mock = mock.Mock()
-        storage_manager_mock.copy_obj = mock.Mock()
-        storage_manager_mock.delete_obj = mock.Mock()
-
-        with mock.patch('userfiles.serializers.connect_storage') as connect_storage_mock:
-            connect_storage_mock.return_value=storage_manager_mock
-            user_file = userfiles_serializer.update(user_file, validated_data)
-            storage_manager_mock.copy_obj.assert_called_with(upload_path, f'home/{self.username}/tests/file1.txt')
-            self.assertEqual(user_file.fname.name,f'home/{self.username}/tests/file1.txt')
-            self.assertEqual(user_file.parent_folder.path, f'home/{self.username}/tests')
-            connect_storage_mock.assert_called_with(settings)
-            storage_manager_mock.delete_obj.assert_called_with(upload_path)
+        user_file.move = mock.Mock()
+        user_file = userfiles_serializer.update(user_file, validated_data)
+        user_file.move.assert_called_with(f'home/{self.username}/tests/file1.txt')
 
     def test_validate_upload_path_failure_contains_commas(self):
         """