Improvements on cloning

docs/src/user-guide/resource_info.md

@@ -92,6 +92,27 @@ From the lower right toolbar on the thumbnail part of the properties panel, it i
 - Copy the resource URL
 - Copy the OGC resource web services URL (in the case of a `Dataset`)
 
+## Cloning a resource
+
+Cloning creates a new, fully independent resource. It gets its own UUID, its own permissions record, and, for a `Dataset`, its own copy of the data on the GIS backend. Ownership of the clone is transferred to whoever triggers it, regardless of who owned the source.
+
+What is carried over from the source:
+
+- Metadata: title, abstract, category, license, and every other descriptive field
+- Keywords, regions, and thesaurus keywords
+- Contacts and their roles (point of contact, metadata author, and so on)
+- Geographic access limits (per-user and per-group)
+- Permissions: the clone starts with the same permission spec as the source, not the default permissions a newly created resource would get
+- Linked resources (e.g. a `Map`'s linked `Datasets`)
+- Type-specific data: a `Dataset`'s attribute table, a `Map`'s layers, and the underlying files/assets
+
+What does not carry over:
+
+- The owner, which becomes the user who triggered the clone
+- The `featured` flag, always reset to off on the clone
+
+Because the clone owns its own copy of everything above rather than sharing rows with the source, deleting the source resource afterward does not affect the clone.
+
 You can access the resource details page by clicking the button on the right (`View dataset` in the case of a `dataset`) in the overview panel.
 That page looks like the one shown in the picture below.
 

geonode/base/api/tests.py

@@ -2985,7 +2985,7 @@ def test_is_copyable_remote_dataset(self):
 
     @patch.dict(os.environ, {"ASYNC_SIGNALS": "False"})
     @override_settings(ASYNC_SIGNALS=False)
-    def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
+    def test_resource_service_copy_with_perms_dataset_keep_original_perms(self):
         with self.settings(ASYNC_SIGNALS=False):
             files = os.path.join(gisdata.GOOD_DATA, "vector/single_point.shp")
             files_as_dict, _ = get_files(files)
@@ -3026,7 +3026,7 @@ def test_resource_service_copy_with_perms_dataset_set_default_perms(self):
         self.assertEqual("finished", self.client.get(response.json().get("status_url")).json().get("status"))
         _resource = Dataset.objects.filter(title__icontains="test_copy_with_perms").last()
         self.assertIsNotNone(_resource)
-        self.assertNotIn(
+        self.assertIn(
             "bobby",
             [x.username for x in permissions_registry.get_perms(instance=_resource).get("users", [])],
         )

geonode/geoapps/tests.py

@@ -21,8 +21,10 @@
 from django.contrib.auth import get_user_model
 
 from geonode.geoapps.models import GeoApp
-from geonode.base.models import TopicCategory
+from geonode.base.models import TopicCategory, Region, Thesaurus, ThesaurusKeyword
+from geonode.groups.models import GroupProfile
 from geonode.resource.registry import resource_manager_registry, geoapp_manager
+from geonode.security.registry import permissions_registry
 from geonode.tests.base import GeoNodeBaseTestSupport
 from geonode.metadata.manager import metadata_manager
 from geonode.base.populate_test_data import all_public, create_models, remove_models
@@ -82,11 +84,67 @@ def test_geoapp_copy(self):
         self.client.login(username="admin", password="admin")
         geoapp_copy = None
         try:
+            # owner must be whoever triggers the clone (self.bobby), not self.geoapp's own owner (self.user)
             geoapp_copy = resource_manager_registry.get_for_instance(self.geoapp).copy(
-                self.geoapp, defaults=dict(title="Testing GeoApp 2")
+                self.geoapp, owner=self.bobby, defaults=dict(title="Testing GeoApp 2")
             )
             self.assertIsNotNone(geoapp_copy)
             self.assertEqual(geoapp_copy.title, "Testing GeoApp 2")
+            self.assertEqual(geoapp_copy.owner, self.bobby)
+        finally:
+            if geoapp_copy:
+                geoapp_copy.delete()
+            self.assertIsNotNone(self.geoapp)
+
+    def test_geoapp_copy_carries_over_metadata_and_permissions(self):
+        """M2M metadata and perm_spec must survive a GeoApp copy too, same as Dataset/Map."""
+        self.client.login(username="admin", password="admin")
+        # update() drops keywords/regions kwargs, set M2M fields directly instead
+        region = Region.objects.first()
+        self.geoapp = geoapp_manager.update(
+            self.geoapp.uuid, instance=self.geoapp, vals={"abstract": "test abstract", "purpose": "test purpose"}
+        )
+        self.geoapp.keywords.add("foo", "bar")
+        self.geoapp.regions.add(region)
+        thesaurus = Thesaurus.objects.create(identifier="test_thesaurus_geoapp_copy", title="Test Thesaurus")
+        tkeyword = ThesaurusKeyword.objects.create(thesaurus=thesaurus, alt_label="test_tkeyword")
+        self.geoapp.tkeywords.add(tkeyword)
+
+        custom_group, _ = GroupProfile.objects.get_or_create(
+            slug="geoapp_copy_group", title="geoapp_copy_group", access="private"
+        )
+        custom_perms = {
+            "users": {self.bobby.username: ["view_resourcebase", "change_resourcebase"]},
+            "groups": {custom_group.slug: ["view_resourcebase"]},
+        }
+        geoapp_manager.set_permissions(self.geoapp.uuid, instance=self.geoapp, permissions=custom_perms)
+
+        geoapp_copy = None
+        try:
+            geoapp_copy = resource_manager_registry.get_for_instance(self.geoapp).copy(
+                self.geoapp, owner=self.bobby, defaults=dict(title="Testing GeoApp Metadata Copy")
+            )
+            self.assertIsNotNone(geoapp_copy)
+            self.assertEqual(geoapp_copy.owner, self.bobby)
+            self.assertEqual(self.geoapp.abstract, geoapp_copy.abstract)
+            self.assertEqual(self.geoapp.purpose, geoapp_copy.purpose)
+            self.assertCountEqual(
+                [k.name for k in self.geoapp.keywords.all()], [k.name for k in geoapp_copy.keywords.all()]
+            )
+            self.assertCountEqual(list(self.geoapp.regions.all()), list(geoapp_copy.regions.all()))
+            self.assertCountEqual(list(self.geoapp.tkeywords.all()), list(geoapp_copy.tkeywords.all()))
+
+            source_perms = permissions_registry.get_perms(instance=self.geoapp, include_virtual=False)
+            copy_perms = permissions_registry.get_perms(instance=geoapp_copy, include_virtual=False)
+            for perm_key in ("users", "groups"):
+                source_entries = {profile.pk: set(perms) for profile, perms in source_perms.get(perm_key, {}).items()}
+                copy_entries = {profile.pk: set(perms) for profile, perms in copy_perms.get(perm_key, {}).items()}
+                # bobby is also the clone's new owner, who always gets full owner perms on top
+                # of whatever perm_spec is passed, regardless of what the source spec granted him
+                source_entries.pop(self.bobby.pk, None)
+                copy_entries.pop(self.bobby.pk, None)
+                self.assertEqual(source_entries, copy_entries)
+            self.assertIn("change_resourcebase_permissions", copy_perms["users"][self.bobby])
         finally:
             if geoapp_copy:
                 geoapp_copy.delete()

geonode/resource/manager.py

@@ -41,7 +41,7 @@
 from django.core.exceptions import ValidationError, FieldDoesNotExist
 
 from geonode.assets.utils import create_asset_and_link_dict, rollback_asset_and_link, copy_assets_and_links, create_link
-from geonode.base.models import ResourceBase, LinkedResource
+from geonode.base.models import ResourceBase, LinkedResource, ContactRole, UserGeoLimit, GroupGeoLimit
 from geonode.documents.tasks import create_document_thumbnail
 from geonode.metadata.manager import metadata_manager
 from geonode.thumbs.thumbnails import _generate_thumbnail_name
@@ -496,26 +496,60 @@ def update(
     def copy(
         self, instance: ResourceBase, /, uuid: str = None, owner: settings.AUTH_USER_MODEL = None, defaults: dict = {}
     ) -> ResourceBase:
+        """Clone a resource under a new pk/uuid, owned by `owner`.
+
+        Scalar/FK fields (title, abstract, license, ...) come along via copy.copy().
+        Everything else lives in a join table keyed by the old pk and needs re-linking by
+        hand: keywords/regions/tkeywords, contacts (ContactRole), geolimits (own new rows,
+        not shared with the source), linked resources, dataset attributes/maplayers, assets,
+        and the source's permission spec (re-applied after update(), which resets to defaults).
+        """
+        if owner is None:
+            raise ValueError("copy() requires an explicit 'owner': the user who triggered the clone")
         _resource = None
         if instance:
             try:
                 instance.set_processing_state(enumerations.STATE_RUNNING)
                 with transaction.atomic():
                     _resource = copy.copy(instance.get_real_instance())
-                    _resource.owner = owner or instance.get_real_instance().owner
+                    _resource.owner = owner
                     _resource.pk = _resource.id = None
                     _resource.uuid = uuid or str(uuid4())
                     # Ensure that the featured flag is set to False
                     _resource.featured = False
                     try:
-                        # Avoid Integrity errors...
                         _resource.get_real_instance()._meta.get_field("name")
-                        _name = defaults.get("name", _resource.get_real_instance().name)
-                        _resource.get_real_instance().name = defaults["name"] = f"{_name}_{uuid1().hex[:8]}"
+                        if "name" in defaults:
+                            _resource.get_real_instance().name = defaults["name"]
+                        else:
+                            _name = _resource.get_real_instance().name
+                            _resource.get_real_instance().name = defaults["name"] = f"{_name}_{uuid1().hex[:8]}"
                     except FieldDoesNotExist:
                         if "name" in defaults:
                             defaults.pop("name")
                     _resource.save()
+                    _src = instance.get_real_instance()
+                    _dst = _resource.get_real_instance()
+                    # M2M fields live in join tables keyed by pk, not carried by copy.copy()
+                    for _field in ("keywords", "regions", "tkeywords"):
+                        getattr(_dst, _field).set(getattr(_src, _field).all())
+                    # contacts goes through ContactRole (has a "role" field), plain .set() doesn't work
+                    for _contact_role in ContactRole.objects.filter(resource=instance):
+                        ContactRole.objects.get_or_create(
+                            resource=_dst, contact=_contact_role.contact, role=_contact_role.role
+                        )
+                    # UserGeoLimit/GroupGeoLimit own a `resource` FK (CASCADE), reusing the source's
+                    # rows via .set() would tie the clone's limits to the source resource's lifetime
+                    for _geolimit in UserGeoLimit.objects.filter(resource=instance):
+                        _new_geolimit = UserGeoLimit.objects.create(
+                            user=_geolimit.user, resource=_dst, wkt=_geolimit.wkt
+                        )
+                        _dst.users_geolimits.add(_new_geolimit)
+                    for _geolimit in GroupGeoLimit.objects.filter(resource=instance):
+                        _new_geolimit = GroupGeoLimit.objects.create(
+                            group=_geolimit.group, resource=_dst, wkt=_geolimit.wkt
+                        )
+                        _dst.groups_geolimits.add(_new_geolimit)
                     for lr in LinkedResource.get_linked_resources(source=instance.pk, is_internal=False):
                         LinkedResource.objects.get_or_create(
                             source_id=_resource.pk, target_id=lr.target.pk, internal=False
@@ -558,7 +592,10 @@ def copy(
                     to_update.update(defaults)
                     # Refresh from DB
                     _resource.refresh_from_db()
-                    return self.update(_resource.uuid, _resource, vals=to_update)
+                    _resource = self.update(_resource.uuid, _resource, vals=to_update)
+                    _source_perms = permissions_registry.get_perms(instance=instance, include_virtual=False)
+                    self.set_permissions(_resource.uuid, instance=_resource, permissions=_source_perms)
+                    return _resource
                 except Exception as e:
                     logger.exception(e)
                 finally:

geonode/resource/tests.py

@@ -31,7 +31,7 @@
 from geonode.base.populate_test_data import create_models
 from geonode.tests.base import GeoNodeBaseTestSupport
 from geonode.resource.manager import BaseResourceManager
-from geonode.base.models import LinkedResource, ResourceBase
+from geonode.base.models import LinkedResource, ResourceBase, Region, Thesaurus, ThesaurusKeyword
 from geonode.layers.models import Dataset
 from geonode.services.models import Service
 from geonode.documents.models import Document
@@ -295,12 +295,15 @@ def test_ingest(self):
         res.delete()
 
     def test_dataset_copy(self):
+        trigger_user = self.User.objects.create_user(username="test_dataset_copy_trigger", email="trigger@test.com")
+
         def _copy_assert_resource(res, title):
             dataset_copy = None
             try:
-                dataset_copy = self.rm.copy(res, defaults=dict(title=title))
+                dataset_copy = self.rm.copy(res, owner=trigger_user, defaults=dict(title=title))
                 self.assertIsNotNone(dataset_copy)
                 self.assertEqual(dataset_copy.title, title)
+                self.assertEqual(dataset_copy.owner, trigger_user)
             finally:
                 if dataset_copy:
                     dataset_copy.delete()
@@ -352,7 +355,7 @@ def test_resource_copy_with_linked_resources(self):
         def _copy_assert_resource(res, title):
             dataset_copy = None
             try:
-                dataset_copy = self.rm.copy(res, defaults=dict(title=title))
+                dataset_copy = self.rm.copy(res, owner=self.user, defaults=dict(title=title))
                 self.assertIsNotNone(dataset_copy)
                 self.assertEqual(dataset_copy.title, title)
             finally:
@@ -368,6 +371,74 @@ def _copy_assert_resource(res, title):
         self.assertTrue(isinstance(res, Map))
         _copy_assert_resource(res, "A Test Map 2")
 
+    def test_resource_copy_carries_over_metadata_and_permissions(self):
+        res = create_single_map("A Test Map With Metadata")
+        try:
+            region = Region.objects.first()
+            vals = {
+                "abstract": "test abstract",
+                "purpose": "test purpose",
+                "supplemental_information": "test supplemental information",
+                "data_quality_statement": "test data quality statement",
+            }
+            res = self.rm.update(res.uuid, instance=res, vals=vals, keywords=["foo", "bar"], regions=[region.name])
+
+            thesaurus = Thesaurus.objects.create(identifier="test_thesaurus_copy", title="Test Thesaurus")
+            tkeyword = ThesaurusKeyword.objects.create(thesaurus=thesaurus, alt_label="test_tkeyword")
+            res.tkeywords.add(tkeyword)
+
+            custom_group, _ = GroupProfile.objects.get_or_create(
+                slug="test_copy_group", title="test_copy_group", access="private"
+            )
+            custom_perms = {
+                "users": {self.user.username: ["view_resourcebase", "change_resourcebase"]},
+                "groups": {custom_group.slug: ["view_resourcebase"]},
+            }
+            self.rm.set_permissions(res.uuid, instance=res, permissions=custom_perms)
+
+            # owner must be whoever triggers the clone (self.user), not res's own owner
+            self.assertNotEqual(res.owner, self.user)
+            dataset_copy = self.rm.copy(res, owner=self.user, defaults=dict(title="A Test Map With Metadata Copy"))
+            try:
+                self.assertIsNotNone(dataset_copy)
+                self.assertEqual(dataset_copy.owner, self.user)
+                self.assertEqual(res.abstract, dataset_copy.abstract)
+                self.assertEqual(res.purpose, dataset_copy.purpose)
+                self.assertEqual(res.supplemental_information, dataset_copy.supplemental_information)
+                self.assertEqual(res.data_quality_statement, dataset_copy.data_quality_statement)
+                self.assertCountEqual(
+                    [k.name for k in res.keywords.all()], [k.name for k in dataset_copy.keywords.all()]
+                )
+                self.assertCountEqual(list(res.regions.all()), list(dataset_copy.regions.all()))
+                self.assertCountEqual(list(res.tkeywords.all()), list(dataset_copy.tkeywords.all()))
+
+                source_perms = permissions_registry.get_perms(instance=res, include_virtual=False)
+                copy_perms = permissions_registry.get_perms(instance=dataset_copy, include_virtual=False)
+                # self.user is the copy's new owner, so the permission engine promotes it to full
+                # owner perms on top of whatever custom perms it already had (AdvancedSecurityWorkflowManager
+                # always grants the owner admin+view perms) - it's the one entry expected to differ.
+                for perm_key in ("users", "groups"):
+                    source_entries = {
+                        profile.pk: set(perms) for profile, perms in source_perms.get(perm_key, {}).items()
+                    }
+                    copy_entries = {profile.pk: set(perms) for profile, perms in copy_perms.get(perm_key, {}).items()}
+                    if perm_key == "users":
+                        source_entries.pop(self.user.pk, None)
+                        copy_entries.pop(self.user.pk, None)
+                    self.assertEqual(source_entries, copy_entries)
+                # sanity: our own custom entries actually landed (owner promotion is a superset, not a replacement)
+                copy_user_perms = {profile.username: set(perms) for profile, perms in copy_perms["users"].items()}
+                self.assertTrue(
+                    set(custom_perms["users"][self.user.username]).issubset(copy_user_perms[self.user.username])
+                )
+                copy_group_perms = {group.name: perms for group, perms in copy_perms["groups"].items()}
+                self.assertCountEqual(copy_group_perms[custom_group.slug], custom_perms["groups"][custom_group.slug])
+            finally:
+                if dataset_copy:
+                    dataset_copy.delete()
+        finally:
+            res.delete()
+
     def test_exec(self):
         map = create_single_map("test_exec_map")
         self.assertIsNone(self.rm.exec("set_style", None, instance=None))