From 1e343fdb72fa9bf7dcbd094cbbbc90ef839f55e0 Mon Sep 17 00:00:00 2001 From: RegisSinjari Date: Tue, 19 Mar 2024 08:36:13 +0100 Subject: [PATCH 1/4] [Fixes #11995] Implement endpoint to transfer ownership --- geonode/people/tests.py | 93 +++++++++++++++++++++++++++++++++++++++++ geonode/people/views.py | 22 ++++++++++ 2 files changed, 115 insertions(+) diff --git a/geonode/people/tests.py b/geonode/people/tests.py index dc89fd3c3e1..94c6dc3e486 100644 --- a/geonode/people/tests.py +++ b/geonode/people/tests.py @@ -911,3 +911,96 @@ def test_delete_a_user_with_resource(self): # bobby cant be deleted self.assertNotEqual(get_user_model().objects.filter(username="bobby").first(), None) self.assertTrue("user_has_resources" in response.json()["errors"][0]) + + def test_transfer_resources_all(self): + """ + user wants to transfer resources to target + """ + bobby = get_user_model().objects.get(username="bobby") + norman = get_user_model().objects.get(username="norman") + # group = GroupProfile.objects.get(slug="bar") + + self.assertTrue(self.client.login(username="bobby", password="bob")) + self.assertTrue(bobby.is_authenticated) + # check bobbys resources + prior_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(len(prior_bobby_resources)) + # call api + response = self.client.post( + path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": norman.id} + ) + # check that bobby owns the resources no more + later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertFalse(len(later_bobby_resources)) + self.assertEqual(response.status_code, 200) + # check that the resources have been transvered to norman + norman_resources = ResourceBase.objects.filter(owner=norman).all() + self.assertTrue(set(prior_bobby_resources).issubset(set(norman_resources))) + + def test_transfer_resources_invalid_user(self): + """ + user wants to transfer resources to target + """ + bobby = get_user_model().objects.get(username="bobby") + invalid_user_id = get_user_model().objects.last().id + 1 + + self.assertTrue(self.client.login(username="bobby", password="bob")) + self.assertTrue(bobby.is_authenticated) + # check bobbys resources + prior_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(len(prior_bobby_resources)) + # call api + response = self.client.post( + path=f"{reverse('users-list')}/{bobby}/transfer_resources", data={"owner": invalid_user_id} + ) + # response should be 404 + self.assertEqual(response.status_code, 404) + # check that bobby still owns the resources + later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(len(later_bobby_resources)) + # and no change has happened to them + self.assertTrue(set(prior_bobby_resources) == set(later_bobby_resources)) + self.assertTrue(len(later_bobby_resources)) + + def test_transfer_resources_default(self): + """ + user wants to transfer resources to target + """ + bobby = get_user_model().objects.get(username="bobby") + admin = get_user_model().objects.get(username="admin") + + self.assertTrue(self.client.login(username="bobby", password="bob")) + self.assertTrue(bobby.is_authenticated) + + # check bobbys resources + prior_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(len(prior_bobby_resources)) + # call api + response = self.client.post( + path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": "DEFAULT"} + ) + self.assertTrue(response.status_code == 200) + # check that bobby owns the resources no more + later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertFalse(len(later_bobby_resources)) + # check that the resources have been transfered to admin + admin_resources = ResourceBase.objects.filter(owner=admin).all() + self.assertTrue(set(prior_bobby_resources).issubset(set(admin_resources))) + + def test_transfer_resources_nopayload(self): + """ + user wants to transfer resources to target + """ + bobby = get_user_model().objects.get(username="bobby") + self.assertTrue(self.client.login(username="bobby", password="bob")) + self.assertTrue(bobby.is_authenticated) + # check bobbys resources + bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(len(bobby_resources)) + # call api + response = self.client.post(path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={}) + # response should be 404 + self.assertEqual(response.status_code, 404) + # check that bobby still owns the resources + bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(len(bobby_resources)) diff --git a/geonode/people/views.py b/geonode/people/views.py index 184f6a28b60..5acd9923f1d 100644 --- a/geonode/people/views.py +++ b/geonode/people/views.py @@ -251,6 +251,28 @@ def groups(self, request, pk=None): groups = GroupProfile.objects.filter(id__in=qs_ids) return Response(GroupProfileSerializer(embed=True, many=True).to_representation(groups)) + @action(detail=True, methods=["post"]) + def transfer_resources(self, request, pk=None): + user = self.get_object() + admin = ( + get_user_model().objects.filter(is_superuser=True, is_staff=True).first() + ) # admin=get_object_or_404(get_user_model(),username=admin) + target_user = request.data.get("owner") + # initalize as self + target = user + if target_user == "DEFAULT": + if not admin: + return Response("Principal User not found", status=500) + target = admin + else: + target = get_object_or_404(get_user_model(), id=target_user) + # transfer to target + user_resources = ResourceBase.objects.filter(owner=user).all() + for resource in user_resources: + resource.owner = target + resource.save() + return Response("Resources transfered successfully", status=200) + class ProfileAutocomplete(autocomplete.Select2QuerySetView): def get_queryset(self): From 58cf50f44812613cc64630f82c14e5e27c360271 Mon Sep 17 00:00:00 2001 From: RegisSinjari Date: Tue, 19 Mar 2024 15:02:12 +0100 Subject: [PATCH 2/4] [Fixes #11995] Implement endpoint to transfer ownership --- geonode/people/tests.py | 95 +++++++++++++++++++++++++++++++++-------- geonode/people/views.py | 14 +++--- 2 files changed, 86 insertions(+), 23 deletions(-) diff --git a/geonode/people/tests.py b/geonode/people/tests.py index 94c6dc3e486..7bc783864fc 100644 --- a/geonode/people/tests.py +++ b/geonode/people/tests.py @@ -36,6 +36,7 @@ from geonode.people import profileextractors from geonode.base.populate_test_data import all_public, create_models, remove_models +from django.db.models import Q class PeopleAndProfileTests(GeoNodeBaseTestSupport): @@ -923,15 +924,15 @@ def test_transfer_resources_all(self): self.assertTrue(self.client.login(username="bobby", password="bob")) self.assertTrue(bobby.is_authenticated) # check bobbys resources - prior_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertTrue(len(prior_bobby_resources)) + bobby_resources = ResourceBase.objects.filter(owner=bobby) + prior_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) # call api response = self.client.post( path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": norman.id} ) # check that bobby owns the resources no more - later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertFalse(len(later_bobby_resources)) + self.assertFalse(bobby_resources.exists()) self.assertEqual(response.status_code, 200) # check that the resources have been transvered to norman norman_resources = ResourceBase.objects.filter(owner=norman).all() @@ -947,8 +948,9 @@ def test_transfer_resources_invalid_user(self): self.assertTrue(self.client.login(username="bobby", password="bob")) self.assertTrue(bobby.is_authenticated) # check bobbys resources - prior_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertTrue(len(prior_bobby_resources)) + bobby_resources = ResourceBase.objects.filter(owner=bobby) + prior_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) # call api response = self.client.post( path=f"{reverse('users-list')}/{bobby}/transfer_resources", data={"owner": invalid_user_id} @@ -956,11 +958,10 @@ def test_transfer_resources_invalid_user(self): # response should be 404 self.assertEqual(response.status_code, 404) # check that bobby still owns the resources - later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertTrue(len(later_bobby_resources)) + later_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) # and no change has happened to them self.assertTrue(set(prior_bobby_resources) == set(later_bobby_resources)) - self.assertTrue(len(later_bobby_resources)) def test_transfer_resources_default(self): """ @@ -973,20 +974,77 @@ def test_transfer_resources_default(self): self.assertTrue(bobby.is_authenticated) # check bobbys resources - prior_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertTrue(len(prior_bobby_resources)) + bobby_resources = ResourceBase.objects.filter(owner=bobby) + prior_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) # call api response = self.client.post( path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": "DEFAULT"} ) self.assertTrue(response.status_code == 200) # check that bobby owns the resources no more - later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertFalse(len(later_bobby_resources)) + self.assertFalse(bobby_resources.exists()) # check that the resources have been transfered to admin admin_resources = ResourceBase.objects.filter(owner=admin).all() self.assertTrue(set(prior_bobby_resources).issubset(set(admin_resources))) + def test_transfer_resources_to_missing_default(self): + """ + user wants to transfer resources to principal, + but a principal account is missing + """ + bobby = get_user_model().objects.get(username="bobby") + admin = get_user_model().objects.get(username="admin") + + self.assertTrue(self.client.login(username="bobby", password="bob")) + self.client.force_login(bobby) + self.assertTrue(bobby.is_authenticated) + # removal of admin accounts + admin.is_superuser = False + admin.is_staff = False + admin.save() + self.assertFalse(get_user_model().objects.filter(Q(is_superuser=True) | Q(is_staff=True)).exists()) + + # check bobbys resources + bobby_resources = ResourceBase.objects.filter(owner=bobby) + prior_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) + # call api + response = self.client.post( + path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": "DEFAULT"} + ) + self.assertTrue(response.status_code == 500) + self.assertEqual(response.data, "Principal User not found") + # check that bobby still owns the resources + later_bobby_resources = bobby_resources.all() + # check that the resources havent changed + self.assertTrue(set(prior_bobby_resources) == set(later_bobby_resources)) + + def test_transfer_resources_to_self(self): + """ + user wants to transfer resources to self but should be unable to + """ + bobby = get_user_model().objects.get(username="bobby") + + self.assertTrue(self.client.login(username="bobby", password="bob")) + self.assertTrue(bobby.is_authenticated) + + # check bobbys resources + bobby_resources = ResourceBase.objects.filter(owner=bobby) + prior_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) + + # call api + response = self.client.post( + path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={"owner": bobby.pk} + ) + self.assertTrue(response.status_code == 400) + self.assertEqual(response.data, "Cannot reassign to self") + # check that bobby still owns the resources + later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + # check that the resources havent changed + self.assertTrue(set(prior_bobby_resources) == set(later_bobby_resources)) + def test_transfer_resources_nopayload(self): """ user wants to transfer resources to target @@ -995,12 +1053,15 @@ def test_transfer_resources_nopayload(self): self.assertTrue(self.client.login(username="bobby", password="bob")) self.assertTrue(bobby.is_authenticated) # check bobbys resources - bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertTrue(len(bobby_resources)) + bobby_resources = ResourceBase.objects.filter(owner=bobby) + prior_bobby_resources = bobby_resources.all() + self.assertTrue(bobby_resources.exists()) + # call api response = self.client.post(path=f"{reverse('users-list')}/{bobby.pk}/transfer_resources", data={}) # response should be 404 self.assertEqual(response.status_code, 404) # check that bobby still owns the resources - bobby_resources = ResourceBase.objects.filter(owner=bobby).all() - self.assertTrue(len(bobby_resources)) + self.assertTrue(bobby_resources.exists()) + later_bobby_resources = ResourceBase.objects.filter(owner=bobby).all() + self.assertTrue(set(prior_bobby_resources) == set(later_bobby_resources)) diff --git a/geonode/people/views.py b/geonode/people/views.py index 5acd9923f1d..e396728294f 100644 --- a/geonode/people/views.py +++ b/geonode/people/views.py @@ -258,19 +258,21 @@ def transfer_resources(self, request, pk=None): get_user_model().objects.filter(is_superuser=True, is_staff=True).first() ) # admin=get_object_or_404(get_user_model(),username=admin) target_user = request.data.get("owner") - # initalize as self - target = user + + target = None if target_user == "DEFAULT": if not admin: return Response("Principal User not found", status=500) target = admin else: target = get_object_or_404(get_user_model(), id=target_user) + + if target == user: + return Response("Cannot reassign to self", status=400) + # transfer to target - user_resources = ResourceBase.objects.filter(owner=user).all() - for resource in user_resources: - resource.owner = target - resource.save() + ResourceBase.objects.filter(owner=user).update(owner=target or user) + return Response("Resources transfered successfully", status=200) From 6615d755f058b6ea5bc2981654f056a85aeb2cb2 Mon Sep 17 00:00:00 2001 From: RegisSinjari Date: Wed, 20 Mar 2024 07:02:16 +0100 Subject: [PATCH 3/4] [Fixes #11995] Implement endpoint to transfer ownership --- geonode/people/views.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/geonode/people/views.py b/geonode/people/views.py index e396728294f..4c973322281 100644 --- a/geonode/people/views.py +++ b/geonode/people/views.py @@ -254,9 +254,7 @@ def groups(self, request, pk=None): @action(detail=True, methods=["post"]) def transfer_resources(self, request, pk=None): user = self.get_object() - admin = ( - get_user_model().objects.filter(is_superuser=True, is_staff=True).first() - ) # admin=get_object_or_404(get_user_model(),username=admin) + admin = get_user_model().objects.filter(is_superuser=True, is_staff=True).first() target_user = request.data.get("owner") target = None @@ -271,7 +269,7 @@ def transfer_resources(self, request, pk=None): return Response("Cannot reassign to self", status=400) # transfer to target - ResourceBase.objects.filter(owner=user).update(owner=target or user) + ResourceBase.objects.filter(owner=user).update(owner=target) return Response("Resources transfered successfully", status=200) From 544bb60eb34543f2d3c932c961e23e6363bde25e Mon Sep 17 00:00:00 2001 From: RegisSinjari Date: Wed, 20 Mar 2024 09:57:16 +0100 Subject: [PATCH 4/4] [Fixes #11995] Implement endpoint to transfer ownership --- geonode/people/views.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/geonode/people/views.py b/geonode/people/views.py index 4c973322281..7da510cf711 100644 --- a/geonode/people/views.py +++ b/geonode/people/views.py @@ -269,7 +269,7 @@ def transfer_resources(self, request, pk=None): return Response("Cannot reassign to self", status=400) # transfer to target - ResourceBase.objects.filter(owner=user).update(owner=target) + ResourceBase.objects.filter(owner=user).update(owner=target or user) return Response("Resources transfered successfully", status=200)