diff --git a/docs/src/user-guide/resource_info.md b/docs/src/user-guide/resource_info.md index 3f13119e833..c02ff1cc746 100644 --- a/docs/src/user-guide/resource_info.md +++ b/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. diff --git a/geonode/base/api/tests.py b/geonode/base/api/tests.py index 92ea47db3bf..b1ab3c2073b 100644 --- a/geonode/base/api/tests.py +++ b/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", [])], ) diff --git a/geonode/geoapps/tests.py b/geonode/geoapps/tests.py index 903055016d6..685d5b69db3 100644 --- a/geonode/geoapps/tests.py +++ b/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() diff --git a/geonode/resource/manager.py b/geonode/resource/manager.py index 5d57aa22878..9380cf7dfcc 100644 --- a/geonode/resource/manager.py +++ b/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: diff --git a/geonode/resource/tests.py b/geonode/resource/tests.py index dc2d9b1e51c..7b427ebe282 100644 --- a/geonode/resource/tests.py +++ b/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)) diff --git a/geonode/upload/handlers/common/tests_vector.py b/geonode/upload/handlers/common/tests_vector.py index a5cb2a006f3..c5ecfaabad9 100644 --- a/geonode/upload/handlers/common/tests_vector.py +++ b/geonode/upload/handlers/common/tests_vector.py @@ -43,6 +43,11 @@ from geonode.upload.tests.utils import TransactionImporterBaseTestSupport from geonode.utils import OGC_Servers_Handler from geonode.upload.utils import create_vrt_file, has_incompatible_field_names +from geonode.resource.enumerator import ExecutionRequestAction as exa +from geonode.resource.registry import dataset_manager +from geonode.security.registry import permissions_registry +from geonode.groups.models import GroupProfile +from geonode.base.models import ResourceBase, Region, Thesaurus, ThesaurusKeyword class TestBaseVectorFileHandler(TestCase): @@ -65,6 +70,80 @@ def setUp(self) -> None: shutil.copy(self.valid_gpkg, "/tmp") super().setUp() + def test_copy_geonode_resource_should_copy_metadata_and_permissions(self): + dataset = create_single_dataset(name="dataset_to_clone_metadata", owner=self.owner) + region, _ = Region.objects.get_or_create(code="test_vector_copy_region", defaults={"name": "Test Region"}) + dataset = dataset_manager.update( + dataset.uuid, + instance=dataset, + vals={"abstract": "test abstract", "purpose": "test purpose"}, + keywords=["foo", "bar"], + regions=[region.name], + ) + thesaurus = Thesaurus.objects.create(identifier="test_thesaurus_vector_copy", title="Test Thesaurus") + tkeyword = ThesaurusKeyword.objects.create(thesaurus=thesaurus, alt_label="test_tkeyword") + dataset.tkeywords.add(tkeyword) + + other_user, _ = get_user_model().objects.get_or_create(username="vector_copy_other_user") + custom_group, _ = GroupProfile.objects.get_or_create( + slug="vector_copy_group", title="vector_copy_group", access="private" + ) + custom_perms = { + "users": {other_user.username: ["view_resourcebase", "change_resourcebase"]}, + "groups": {custom_group.slug: ["view_resourcebase"]}, + } + dataset_manager.set_permissions(dataset.uuid, instance=dataset, permissions=custom_perms) + + resource = ResourceBase.objects.get(pk=dataset.pk) + # distinct from self.owner (the dataset's owner) so the owner assertion below actually + # proves the clone follows the trigger, not just falls back to the source's owner + trigger_user, _ = get_user_model().objects.get_or_create(username="vector_copy_trigger") + exec_id = orchestrator.create_execution_request( + user=trigger_user, + func_name="start_import", + step="start_import", + action=exa.COPY.value, + input_params={"title": "cloned dataset metadata"}, + ) + _exec = orchestrator.get_execution_object(str(exec_id)) + + new_resource = self.handler.copy_geonode_resource( + alternate=dataset.alternate, + resource=resource, + _exec=_exec, + data_to_update={"title": "cloned dataset metadata"}, + new_alternate="dataset_to_clone_metadata_copy", + kwargs={"original_dataset_alternate": dataset.alternate}, + ) + + try: + self.assertNotEqual(dataset.pk, new_resource.pk) + self.assertEqual(trigger_user, new_resource.owner) + self.assertEqual(dataset.abstract, new_resource.abstract) + self.assertEqual(dataset.purpose, new_resource.purpose) + self.assertCountEqual( + [k.name for k in dataset.keywords.all()], [k.name for k in new_resource.keywords.all()] + ) + self.assertCountEqual(list(dataset.regions.all()), list(new_resource.regions.all())) + self.assertCountEqual(list(dataset.tkeywords.all()), list(new_resource.tkeywords.all())) + + # read via the plain ResourceBase, like production does (resource, not the downcast + # dataset) - get_all_level_info() only merges in the dataset-specific perms when + # called on a ResourceBase with a '.dataset' accessor, not on an already-downcast Dataset + source_perms = permissions_registry.get_perms(instance=resource, include_virtual=False) + copy_perms = permissions_registry.get_perms(instance=new_resource, 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()} + # trigger_user is the clone's new owner, always granted full owner perms on top + # of the reapplied source perm_spec, on top of whatever the source spec granted + copy_entries.pop(trigger_user.pk, None) + self.assertEqual(source_entries, copy_entries) + self.assertIn("change_resourcebase_permissions", copy_perms["users"][trigger_user]) + finally: + new_resource.delete() + dataset.delete() + def test_create_error_log(self): """ Should return the formatted way for the log of the handler diff --git a/geonode/upload/handlers/common/vector.py b/geonode/upload/handlers/common/vector.py index 43d06804cbd..c8249fcdfdb 100644 --- a/geonode/upload/handlers/common/vector.py +++ b/geonode/upload/handlers/common/vector.py @@ -61,8 +61,9 @@ from osgeo import ogr, gdal from geonode.upload.api.exceptions import ImportException, UpsertException from geonode.upload.celery_app import importer_app -from geonode.assets.utils import copy_assets_and_links, get_default_asset +from geonode.assets.utils import get_default_asset +from geonode.resource.registry import dataset_manager from geonode.upload.handlers.utils import create_alternate, should_be_imported from geonode.upload.models import ResourceHandlerInfo from geonode.upload.orchestrator import orchestrator @@ -188,8 +189,10 @@ def extract_params_from_data(_data, action=None): all the other are returned """ if action == exa.COPY.value: - title = json.loads(_data.get("defaults")) - return {"title": title.pop("title"), "store_spatial_file": True}, _data + data = _data.get("defaults") + if isinstance(data, str): + data = json.loads(data) + return {"title": data.pop("title"), "store_spatial_file": True}, _data return { "skip_existing_layers": _data.pop("skip_existing_layers", "False"), @@ -1030,14 +1033,22 @@ def copy_geonode_resource( ).first() if previous_resource: subtype = previous_resource.subtype - new_resource = self.create_geonode_resource( - layer_name=data_to_update.get("title"), - alternate=new_alternate, - execution_id=str(_exec.exec_id), - subtype=subtype, - ) - copy_assets_and_links(resource, target=new_resource) + workspace = getattr(settings, "DEFAULT_WORKSPACE", getattr(settings, "CASCADE_WORKSPACE", "geonode")) + defaults = { + "name": new_alternate, + "alternate": f"{workspace}:{new_alternate}", + "workspace": workspace, + "store": os.environ.get("GEONODE_GEODATABASE", "geonode_data"), + } + if data_to_update.get("title"): + defaults["title"] = data_to_update["title"] + if subtype: + defaults["subtype"] = subtype + + new_resource = dataset_manager.copy(resource, owner=_exec.user, defaults=defaults) + # copy() only relinks GeoNode-side data, the GIS backend still needs its own style + self.handle_sld_file(new_resource, _exec) if resource.dataset.has_time is True: diff --git a/geonode/upload/handlers/document/tests.py b/geonode/upload/handlers/document/tests.py index 9f0129fb195..bd1d4a68cae 100644 --- a/geonode/upload/handlers/document/tests.py +++ b/geonode/upload/handlers/document/tests.py @@ -30,10 +30,13 @@ from geonode.assets.models import Asset from geonode.assets.utils import get_default_asset from geonode.base.enumerations import SOURCE_TYPE_LOCAL, SOURCE_TYPE_REMOTE -from geonode.base.models import Link +from geonode.base.models import Link, Region, Thesaurus, ThesaurusKeyword from geonode.base.populate_test_data import create_single_doc from geonode.documents.models import Document +from geonode.groups.models import GroupProfile from geonode.resource.models import ExecutionRequest +from geonode.resource.registry import document_manager +from geonode.security.registry import permissions_registry from geonode.upload.api.exceptions import InvalidInputFileException from geonode.upload.celery_tasks import create_geonode_resource, import_resource from geonode.upload.handlers.document.exceptions import InvalidDocumentException @@ -390,6 +393,55 @@ def test_copy_geonode_resource_should_clone_the_document(self): self.assertEqual("cloned document", cloned.title) self.assertTrue(cloned.files) + def test_copy_geonode_resource_should_copy_metadata_and_permissions(self): + other_owner, _ = get_user_model().objects.get_or_create(username="document_copy_source_owner") + document = create_single_doc("document with metadata to clone", owner=other_owner) + region = Region.objects.first() + document = document_manager.update( + document.uuid, + instance=document, + vals={"abstract": "test abstract", "purpose": "test purpose"}, + keywords=["foo", "bar"], + regions=[region.name], + ) + thesaurus = Thesaurus.objects.create(identifier="test_thesaurus_document_copy", title="Test Thesaurus") + tkeyword = ThesaurusKeyword.objects.create(thesaurus=thesaurus, alt_label="test_tkeyword") + document.tkeywords.add(tkeyword) + + other_user, _ = get_user_model().objects.get_or_create(username="document_copy_other_user") + custom_group, _ = GroupProfile.objects.get_or_create( + slug="document_copy_group", title="document_copy_group", access="private" + ) + custom_perms = { + "users": {other_user.username: ["view_resourcebase", "change_resourcebase"]}, + "groups": {custom_group.slug: ["view_resourcebase"]}, + } + document_manager.set_permissions(document.uuid, instance=document, permissions=custom_perms) + + exec_id = self._create_execution_request(action=ira.DOCUMENT_COPY.value, title="cloned document metadata") + _exec = orchestrator.get_execution_object(exec_id) + + cloned = self.handler.copy_geonode_resource(document, _exec) + + self.assertNotEqual(document.pk, cloned.pk) + self.assertEqual(self.user, cloned.owner) + self.assertEqual(document.abstract, cloned.abstract) + self.assertEqual(document.purpose, cloned.purpose) + self.assertCountEqual([k.name for k in document.keywords.all()], [k.name for k in cloned.keywords.all()]) + self.assertCountEqual(list(document.regions.all()), list(cloned.regions.all())) + self.assertCountEqual(list(document.tkeywords.all()), list(cloned.tkeywords.all())) + + source_perms = permissions_registry.get_perms(instance=document, include_virtual=False) + copy_perms = permissions_registry.get_perms(instance=cloned, 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()} + # self.user is the clone's new owner, always granted full owner perms on top of + # the reapplied source perm_spec, on top of whatever the source spec granted it + copy_entries.pop(self.user.pk, None) + self.assertEqual(source_entries, copy_entries) + self.assertIn("change_resourcebase_permissions", copy_perms["users"][self.user]) + def test_copy_geonode_resource_should_clone_a_remote_document(self): document = create_single_doc("remote document to clone") document.doc_url = self.remote_url