diff --git a/chris_backend/core/models.py b/chris_backend/core/models.py index f396676e..658dee0e 100755 --- a/chris_backend/core/models.py +++ b/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') diff --git a/chris_backend/core/tests/test_models.py b/chris_backend/core/tests/test_models.py new file mode 100755 index 00000000..bd064d90 --- /dev/null +++ b/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) diff --git a/chris_backend/userfiles/serializers.py b/chris_backend/userfiles/serializers.py index ed60a092..fd7c5a8a 100755 --- a/chris_backend/userfiles/serializers.py +++ b/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/ # where 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): diff --git a/chris_backend/userfiles/tests/test_serializers.py b/chris_backend/userfiles/tests/test_serializers.py index 12d69689..3dfc9339 100755 --- a/chris_backend/userfiles/tests/test_serializers.py +++ b/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): """