[Fixes #12124] GNIP 100: Assets - add tests and (un)managed files

geonode/assets/handlers.py

@@ -17,7 +17,7 @@ def handled_asset_class(self):
     def create(self, title, description, type, owner, *args, **kwargs):
         raise NotImplementedError()
 
-    def remove_data(self, asset: Asset):
+    def remove_data(self, asset: Asset, **kwargs):
         raise NotImplementedError()
 
     def replace_data(self, asset: Asset, files: list):

geonode/assets/local.py

@@ -47,17 +47,27 @@ def create(self, title, description, type, owner, files=None, clone_files=False,
         asset.save()
         return asset
 
-    def remove_data(self, asset: LocalAsset):
+    def remove_data(self, asset: LocalAsset, force=False):
+        """
+        Removes the files related to an Asset.
+        By default, only files within the Assets directory are removed, unless `force` is set.
+        """
         removed_dir = set()
         for file in asset.location:
-            if file.startswith(settings.ASSETS_ROOT):
-                logger.info(f"Removing asset file {file}")
+            is_managed = self._is_file_managed(file)
+            if is_managed or force:
+                logger.info(f"Removing asset file {file} {'FORCED' if force and not is_managed else ''}")
                 storage_manager.delete(file)
+                # TODO: in case of forcing deletion of unmanaged files, reconsider the deletion of directories
                 removed_dir.add(os.path.dirname(file))
             else:
                 logger.info(f"Not removing asset file outside asset directory {file}")
 
+        # TODO: in case of subdirs, make sure that all the tree is removed in the proper order
         for dir in removed_dir:
+            if not os.path.exists(dir):
+                logger.warning(f"Trying to remove not existing asset directory {dir}")
+                continue
             if not os.listdir(dir):
                 logger.info(f"Removing empty asset directory {dir}")
                 os.rmdir(dir)
@@ -67,11 +77,20 @@ def replace_data(self, asset: LocalAsset, files: list):
         asset.location = files
         asset.save()
 
-    def clone(self, asset: LocalAsset) -> LocalAsset:
-        prefix = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
-        asset.location = storage_manager.copy_files_list(asset.location, dir=settings.ASSETS_ROOT, dir_prefix=prefix)
+    def clone(self, source: LocalAsset) -> LocalAsset:
+        # get a new asset instance to be edited and stored back
+        asset = LocalAsset.objects.get(pk=source.pk)
+        # only copy files if they are managed
+        if self._are_files_managed(asset.location):
+            asset.location = storage_manager.copy_files_list(
+                asset.location, dir=settings.ASSETS_ROOT, dir_prefix=datetime.datetime.now().strftime("%Y%m%d%H%M%S")
+            )
+        # it's a polymorphic object, we need to null both IDs
+        # https://django-polymorphic.readthedocs.io/en/stable/advanced.html#copying-polymorphic-objects
         asset.pk = None
+        asset.id = None
         asset.save()
+        asset.refresh_from_db()
         return asset
 
     def create_download_url(self, asset) -> str:
@@ -80,6 +99,28 @@ def create_download_url(self, asset) -> str:
     def create_link_url(self, asset) -> str:
         return build_absolute_uri(reverse("assets-link", args=(asset.pk,)))
 
+    def _is_file_managed(self, file) -> bool:
+        assets_root = os.path.normpath(settings.ASSETS_ROOT)
+        return file.startswith(assets_root)
+
+    def _are_files_managed(self, files: list) -> bool:
+        """
+        :param files: files to be checked
+        :return: True if all files are managed, False is no file is managed
+        :raise: ValueError if both managed and unmanaged files are in the list
+        """
+        managed = unmanaged = None
+        for file in files:
+            if self._is_file_managed(file):
+                managed = True
+            else:
+                unmanaged = True
+            if managed and unmanaged:
+                logger.error(f"Both managed and unmanaged files are present: {files}")
+                raise ValueError("Both managed and unmanaged files are present")
+
+        return bool(managed)
+
 
 class LocalAssetDownloadHandler(AssetDownloadHandlerInterface):
 

geonode/assets/tests.py

@@ -0,0 +1,193 @@
+#########################################################################
+#
+# Copyright (C) 2024 OSGeo
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+#########################################################################
+
+import os
+import logging
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+
+from rest_framework.test import APITestCase
+
+from geonode.assets.handlers import asset_handler_registry
+from geonode.assets.local import LocalAssetHandler
+from geonode.assets.models import Asset, LocalAsset
+
+logger = logging.getLogger(__name__)
+
+TEST_GIF = os.path.join(os.path.dirname(os.path.dirname(__file__)), "base/tests/data/img.gif")
+
+
+class AssetsTests(APITestCase):
+
+    def test_default_handler(self):
+        asset_handler = asset_handler_registry.get_default_handler()
+        self.assertIsNotNone(asset_handler)
+        # in the default config, the default handler is the Local one
+        self.assertIsInstance(asset_handler, LocalAssetHandler, "Bad default Asset handler found")
+
+    def test_creation_and_delete_data_cloned(self):
+        u, _ = get_user_model().objects.get_or_create(username="admin")
+        assets_root = os.path.normpath(settings.ASSETS_ROOT)
+
+        asset_handler = asset_handler_registry.get_default_handler()
+        asset = asset_handler.create(
+            title="Test Asset",
+            description="Description of test asset",
+            type="NeverMind",
+            owner=u,
+            files=[TEST_GIF],
+            clone_files=True,
+        )
+        asset.save()
+        self.assertIsInstance(asset, LocalAsset)
+
+        reloaded = Asset.objects.get(pk=asset.pk)
+        self.assertIsNotNone(reloaded)
+        self.assertIsInstance(reloaded, LocalAsset)
+        file = reloaded.location[0]
+        self.assertTrue(os.path.exists(file), "Asset file does not exist")
+        self.assertTrue(file.startswith(assets_root), f"Asset file is not inside the assets root: {file}")
+
+        cloned_file = file
+        reloaded.delete()
+        self.assertFalse(Asset.objects.filter(pk=asset.pk).exists())
+        self.assertFalse(os.path.exists(cloned_file))
+        self.assertFalse(os.path.exists(os.path.dirname(cloned_file)))
+        self.assertTrue(os.path.exists(TEST_GIF))
+
+    def test_creation_and_delete_data_external(self):
+        u, _ = get_user_model().objects.get_or_create(username="admin")
+
+        asset_handler = asset_handler_registry.get_default_handler()
+        asset = asset_handler.create(
+            title="Test Asset",
+            description="Description of test asset",
+            type="NeverMind",
+            owner=u,
+            files=[TEST_GIF],
+            clone_files=False,
+        )
+        asset.save()
+        self.assertIsInstance(asset, LocalAsset)
+
+        reloaded = Asset.objects.get(pk=asset.pk)
+        self.assertIsNotNone(reloaded)
+        self.assertIsInstance(reloaded, LocalAsset)
+        file = reloaded.location[0]
+        self.assertEqual(TEST_GIF, file)
+
+        reloaded.delete()
+        self.assertFalse(Asset.objects.filter(pk=asset.pk).exists())
+        self.assertTrue(os.path.exists(TEST_GIF))
+
+    def test_clone_and_delete_data_managed(self):
+        u, _ = get_user_model().objects.get_or_create(username="admin")
+
+        asset_handler = asset_handler_registry.get_default_handler()
+        asset = asset_handler.create(
+            title="Test Asset",
+            description="Description of test asset",
+            type="NeverMind",
+            owner=u,
+            files=[TEST_GIF],
+            clone_files=True,
+        )
+        asset.save()
+        self.assertIsInstance(asset, LocalAsset)
+
+        reloaded = Asset.objects.get(pk=asset.pk)
+        cloned = asset_handler.clone(reloaded)
+        self.assertNotEqual(reloaded.pk, cloned.pk)
+
+        reloaded_file = reloaded.location[0]
+        cloned_file = cloned.location[0]
+
+        self.assertNotEqual(reloaded_file, cloned_file)
+        self.assertTrue(os.path.exists(reloaded_file))
+        self.assertTrue(os.path.exists(cloned_file))
+
+        reloaded.delete()
+        self.assertFalse(os.path.exists(reloaded_file))
+        self.assertTrue(os.path.exists(cloned_file))
+
+        cloned.delete()
+        self.assertFalse(os.path.exists(cloned_file))
+
+    def test_clone_and_delete_data_unmanaged(self):
+        u, _ = get_user_model().objects.get_or_create(username="admin")
+
+        asset_handler = asset_handler_registry.get_default_handler()
+        asset = asset_handler.create(
+            title="Test Asset",
+            description="Description of test asset",
+            type="NeverMind",
+            owner=u,
+            files=[TEST_GIF],
+            clone_files=False,
+        )
+        asset.save()
+        self.assertIsInstance(asset, LocalAsset)
+
+        reloaded = Asset.objects.get(pk=asset.pk)
+        cloned = asset_handler.clone(reloaded)
+
+        self.assertEqual(reloaded.location[0], cloned.location[0])
+        self.assertTrue(os.path.exists(reloaded.location[0]))
+
+        reloaded.delete()
+        self.assertTrue(os.path.exists(reloaded.location[0]))
+
+        cloned.delete()
+        self.assertTrue(os.path.exists(reloaded.location[0]))
+
+    def test_clone_mixed_data(self):
+        u, _ = get_user_model().objects.get_or_create(username="admin")
+
+        asset_handler = asset_handler_registry.get_default_handler()
+        managed_asset = asset_handler.create(
+            title="Test Asset",
+            description="Description of test asset",
+            type="NeverMind",
+            owner=u,
+            files=[TEST_GIF],
+            clone_files=True,
+        )
+        managed_asset.save()
+
+        # TODO: dunno if mixed files should be allowed at all
+        mixed_asset = asset_handler.create(
+            title="Mixed Asset",
+            description="Description of test asset",
+            type="NeverMind",
+            owner=u,
+            files=[TEST_GIF, managed_asset.location[0]],
+            clone_files=False,  # let's keep both managed and unmanaged together
+        )
+
+        reloaded = Asset.objects.get(pk=mixed_asset.pk)
+
+        try:
+            asset_handler.clone(reloaded)
+            self.fail("A mixed LocalAsset has been cloned")
+        except ValueError:
+            pass
+
+        mixed_asset.delete()
+        managed_asset.delete()

geonode/resource/manager.py

@@ -40,8 +40,7 @@
 from django.core.exceptions import ObjectDoesNotExist, ValidationError, FieldDoesNotExist
 
 
-from geonode.base.models import ResourceBase, LinkedResource, Link
-from geonode.geoapps.models import GeoApp
+from geonode.base.models import ResourceBase, LinkedResource
 from geonode.thumbs.thumbnails import _generate_thumbnail_name
 from geonode.documents.tasks import create_document_thumbnail
 from geonode.security.permissions import PermSpecCompact, DATA_STYLABLE_RESOURCES_SUBTYPES
@@ -50,7 +49,6 @@
 from . import settings as rm_settings
 from .utils import update_resource, resourcebase_post_save
 from geonode.assets.utils import create_asset_and_link_dict, rollback_asset_and_link, copy_assets_and_links
-from geonode.assets.handlers import asset_handler_registry
 
 from ..base import enumerations
 from ..security.utils import AdvancedSecurityWorkflowManager