Skip to content

Commit

Permalink
[Fixes #11995] Implement endpoint to transfer ownership
Browse files Browse the repository at this point in the history
  • Loading branch information
@RegisSinjari
RegisSinjari committed Mar 19, 2024
1 parent 1e343fd commit 58cf50f
Show file tree
Hide file tree
Showing 2 changed files with 86 additions and 23 deletions.
95 changes: 78 additions & 17 deletions geonode/people/tests.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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()
Expand All @@ -947,20 +948,20 @@ 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}
)
# 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):
"""
Expand All @@ -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
Expand All @@ -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))
14 changes: 8 additions & 6 deletions geonode/people/views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)


Expand Down

0 comments on commit 58cf50f

Please sign in to comment.