Fix form permissions not updated when project is transferred (#2775)

kelvin-muchiri · web-flow · commit 92d914b01cde · 2025-02-13T08:21:25.000+03:00
* fix forms not transferred when project is transferred

* refactor code

* enhance test

* enhance test

* fix failing tests

* transfer project to individual users using management command

users are only disallowed to transfer project to individual users when using the RESTFUL API

* revert file changes

revert changes made to onadata/apps/logger/tests/test_transfer_project_command.py

* add test

* update comment

* update test

* update test

* address failing tests
diff --git a/onadata/apps/api/tests/viewsets/test_project_viewset.py b/onadata/apps/api/tests/viewsets/test_project_viewset.py
@@ -1296,11 +1296,16 @@ def test_project_transfer_upgrades_permissions(self):
         self.assertEqual(self.project.created_by, alice_profile.user)
         alice_project = self.project
 
-        # create org owned by bob then make alice admin
+        # Publish a form to Alice's project
+        self._publish_xls_form_to_project()
+        alice_xform = self.xform
+
+        # Create organization owned by Bob
         self._login_user_and_profile({"username": bob.username, "email": bob.email})
         self._org_create()
         self.assertEqual(self.organization.created_by, bob)
-        org_url = f"http://testserver/api/v1/users/{self.organization.user.username}"
+
+        # Add Alice as admin to Bob's organization
         view = OrganizationProfileViewSet.as_view({"post": "members"})
         data = {"username": alice_profile.user.username, "role": OwnerRole.name}
         request = self.factory.post(
@@ -1312,6 +1317,20 @@ def test_project_transfer_upgrades_permissions(self):
         owners_team = get_or_create_organization_owners_team(self.organization)
         self.assertIn(alice_profile.user, owners_team.user_set.all())
 
+        # Add Jane to Bob's organization with dataentry role
+        jane_data = {"username": "jane", "email": "janedoe@example.com"}
+        jane_profile = self._create_user_profile(jane_data)
+        data = {"username": jane_profile.user.username, "role": DataEntryRole.name}
+        request = self.factory.post("/", data=data, **self.extra)
+        request = self.factory.post(
+            "/", data=json.dumps(data), content_type="application/json", **self.extra
+        )
+        response = view(request, user=self.organization.user.username)
+        self.assertEqual(response.status_code, 201)
+        self.assertTrue(
+            DataEntryRole.user_has_role(jane_profile.user, self.organization)
+        )
+
         # Share project to bob as editor
         data = {"username": bob.username, "role": EditorRole.name}
         view = ProjectViewSet.as_view({"post": "share"})
@@ -1322,28 +1341,24 @@ def test_project_transfer_upgrades_permissions(self):
         self.assertEqual(response.status_code, 204)
 
         # Transfer project to Bobs Organization
+        org_url = f"http://testserver/api/v1/users/{self.organization.user.username}"
         data = {"owner": org_url, "name": alice_project.name}
         view = ProjectViewSet.as_view({"patch": "partial_update"})
         request = self.factory.patch("/", data=data, **auth_credentials)
         response = view(request, pk=alice_project.pk)
         self.assertEqual(response.status_code, 200)
 
-        # Ensure all Admins have admin privileges to the project
-        # once transferred
-        view = ProjectViewSet.as_view({"get": "retrieve"})
-        request = self.factory.get("/", **self.extra)
-        response = view(request, pk=alice_project.pk)
-        self.assertEqual(response.status_code, 200)
-        project_users = response.data["users"]
-
-        org_owners = get_or_create_organization_owners_team(
-            self.organization
-        ).user_set.all()
-
-        for user in project_users:
-            owner = org_owners.filter(username=user["user"]).first()
-            if owner:
-                self.assertEqual(user["role"], OwnerRole.name)
+        # Admins have owner privileges to the transferred project
+        # and forms
+        self.assertTrue(OwnerRole.user_has_role(bob, alice_project))
+        self.assertTrue(OwnerRole.user_has_role(bob, alice_xform))
+        self.assertTrue(OwnerRole.user_has_role(alice_profile.user, alice_project))
+        self.assertTrue(OwnerRole.user_has_role(alice_profile.user, alice_xform))
+
+        # Non-admins have readonly privileges to the transferred project
+        # and forms
+        self.assertTrue(ReadOnlyRole.user_has_role(jane_profile.user, alice_project))
+        self.assertTrue(ReadOnlyRole.user_has_role(jane_profile.user, alice_xform))
 
     # pylint: disable=invalid-name
     @override_settings(ALLOW_PUBLIC_DATASETS=False)
diff --git a/onadata/apps/logger/management/commands/transferproject.py b/onadata/apps/logger/management/commands/transferproject.py
@@ -5,10 +5,13 @@
 from django.core.management.base import BaseCommand
 from django.db import transaction
 
-from onadata.apps.logger.models import DataView, MergedXForm, Project, XForm
+from onadata.apps.api.models import Team
+from onadata.apps.api.models.organization_profile import get_organization_members_team
+from onadata.apps.logger.models import MergedXForm, Project, XForm
 from onadata.apps.logger.models.project import (
     set_object_permissions as set_project_permissions,
 )
+from onadata.libs.permissions import ReadOnlyRole
 from onadata.libs.utils.project_utils import set_project_perms_to_xform
 
 
@@ -64,55 +67,72 @@ def get_user(self, username):
             self.errors.append(f"User {username} does not exist")
         return user
 
-    def update_xform_with_new_user(self, project, user):
-        """
-        Update XForm user update the DataViews and also set the permissions
-        for the xForm and the project.
-        """
+    def _transfer_xform(self, project, to_user):
+        """Transfer XForm to the new owner."""
         xforms = XForm.objects.filter(
             project=project, deleted_at__isnull=True, downloadable=True
         )
         for form in xforms:
-            form.user = user
-            form.created_by = user
+            form.user = to_user
+            form.created_by = to_user
             form.save()
-            self.update_data_views(form)
             set_project_perms_to_xform(form, project)
 
-    @staticmethod
-    def update_data_views(form):
-        """Update DataView project for the XForm given."""
-        dataviews = DataView.objects.filter(
-            xform=form, project=form.project, deleted_at__isnull=True
-        )
-        for data_view in dataviews:
-            data_view.project = form.project
-            data_view.save()
-
-    @staticmethod
-    def update_merged_xform(project, user):
-        """Update ownership of MergedXforms."""
+    def _transfer_merged_xform(self, project, to_user):
+        """Transfer MergedXForm to the new owner."""
         merged_xforms = MergedXForm.objects.filter(
             project=project, deleted_at__isnull=True
         )
         for form in merged_xforms:
-            form.user = user
-            form.created_by = user
+            form.user = to_user
+            form.created_by = to_user
             form.save()
             set_project_perms_to_xform(form, project)
 
+    def transfer_project(self, project, to_user):
+        """Transfer Project to the new owner."""
+        project.organization = to_user
+        project.created_by = to_user
+        project.save()
+
+        set_project_permissions(Project, project, created=True)
+
+        if hasattr(to_user, "profile") and hasattr(
+            to_user.profile, "organizationprofile"
+        ):
+            # Give readonly permission to members
+            organization = to_user.profile.organizationprofile
+            owners_team = Team.objects.get(
+                name=f"{to_user.username}#{Team.OWNER_TEAM_NAME}", organization=to_user
+            )
+            members_team = get_organization_members_team(organization)
+
+            ReadOnlyRole.add(members_team, project)
+
+            # Owners are also members so we exclude them
+            members = members_team.user_set.exclude(
+                username__in=[user.username for user in owners_team.user_set.all()]
+            )
+
+            for member in members:
+                ReadOnlyRole.add(member, project)
+
+        self._transfer_xform(project, to_user)
+        self._transfer_merged_xform(project, to_user)
+
     @transaction.atomic()
     def handle(self, *args, **options):
         """Transfer projects from one user to another."""
         from_user = self.get_user(options["current_owner"])
         to_user = self.get_user(options["new_owner"])
-        project_id = options.get("project_id")
-        transfer_all_projects = options.get("all_projects")
 
         if self.errors:
             self.stdout.write("".join(self.errors))
             return
 
+        project_id = options.get("project_id")
+        transfer_all_projects = options.get("all_projects")
+
         # No need to validate project ownership as they filtered
         # against current_owner
         projects = []
@@ -124,12 +144,6 @@ def handle(self, *args, **options):
             projects = Project.objects.filter(id=project_id, organization=from_user)
 
         for project in projects:
-            project.organization = to_user
-            project.created_by = to_user
-            project.save()
-
-            set_project_permissions(Project, project, created=True)
-            self.update_xform_with_new_user(project, to_user)
-            self.update_merged_xform(project, to_user)
+            self.transfer_project(project, to_user)
 
         self.stdout.write("Projects transferred successfully")
diff --git a/onadata/apps/logger/tests/test_transfer_project_command.py b/onadata/apps/logger/tests/test_transfer_project_command.py
@@ -1,13 +1,17 @@
 """Tests for project transfer command"""
+
 import os
 import sys
 
 from django.contrib.auth import get_user_model
 from django.core.management import call_command
+
 from six import StringIO
 
+from onadata.apps.api.models import Team
 from onadata.apps.logger.models import Project, XForm
 from onadata.apps.main.tests.test_base import TestBase
+from onadata.libs.permissions import OwnerRole, ReadOnlyRole
 
 
 class TestMoveProjectToAnewOwner(TestBase):  # pylint: disable=C0111
@@ -141,6 +145,58 @@ def test_xforms_are_transferred_as_well(self):  # pylint: disable=C0103
         self.assertEqual(0, bobs_forms.count())
         self.assertEqual(1, new_owner_forms.count())
 
+    def test_transfer_to_org(self):
+        """Transferring to an organization works"""
+        # Create users
+        bob = self._create_user("bob", "test_pass")
+        alice = self._create_user("alice", "test_pass")
+        jane = self._create_user("jane", "test_pass")
+        # Create Project owned by Bob and publish form
+        project = Project.objects.create(
+            name="Test_project",
+            organization=bob,
+            created_by=bob,
+        )
+        self.project = project
+        self._publish_transportation_form()
+        # Create organization owned by Alice
+        alice_org = self._create_organization("alice_inc", "Alice Inc", alice)
+        owners_team, _ = Team.objects.get_or_create(
+            name=f"{alice_org.user.username}#Owners",
+            organization=alice_org.user,
+        )
+        members_team, _ = Team.objects.get_or_create(
+            name=f"{alice_org.user.username}#members",
+            organization=alice_org.user,
+        )
+        # Add Bob as admin to Alice's organization
+        bob.groups.add(owners_team)
+        bob.groups.add(members_team)
+        # Add Jane as member to Alice's organization
+        jane.groups.add(members_team)
+        # Transfer Bob's project to Alice's organization
+        call_command(
+            "transferproject",
+            current_owner=bob.username,
+            new_owner=alice_org.user.username,
+            project_id=project.id,
+        )
+        self.xform.refresh_from_db()
+        self.project.refresh_from_db()
+        self.assertEqual(self.project.organization, alice_org.user)
+        self.assertEqual(self.project.organization, alice_org.user)
+        self.assertEqual(self.xform.user, alice_org.user)
+        self.assertEqual(self.xform.created_by, alice_org.user)
+        # Admins have owner privileges
+        self.assertTrue(OwnerRole.user_has_role(bob, project))
+        self.assertTrue(OwnerRole.user_has_role(bob, self.xform))
+        self.assertTrue(OwnerRole.user_has_role(alice, project))
+        self.assertTrue(OwnerRole.user_has_role(alice, self.xform))
+        # Non-admins have readonly privileges
+        self.assertFalse(OwnerRole.user_has_role(jane, project))
+        self.assertTrue(ReadOnlyRole.user_has_role(jane, project))
+        self.assertTrue(ReadOnlyRole.user_has_role(jane, self.xform))
+
 
 class TestUserValidation(TestBase):
     """Created this class to specifically test for the user validation.
diff --git a/onadata/apps/main/tests/test_base.py b/onadata/apps/main/tests/test_base.py
@@ -2,6 +2,7 @@
 """
 TestBase - a TestCase base class.
 """
+
 from __future__ import unicode_literals
 
 import base64
@@ -13,8 +14,6 @@
 from io import StringIO
 from tempfile import NamedTemporaryFile
 
-from pyxform.builder import create_survey_element_from_dict
-
 from django.conf import settings
 from django.contrib.auth import authenticate, get_user_model
 from django.core.files.uploadedfile import InMemoryUploadedFile
@@ -24,10 +23,12 @@
 
 from django_digest.test import Client as DigestClient
 from django_digest.test import DigestAuth
+from pyxform.builder import create_survey_element_from_dict
 from rest_framework.test import APIRequestFactory
 from six.moves.urllib.error import URLError
 from six.moves.urllib.request import urlopen
 
+from onadata.apps.api.models import OrganizationProfile
 from onadata.apps.api.viewsets.xform_viewset import XFormViewSet
 from onadata.apps.logger.models import Instance, MergedXForm, XForm, XFormVersion
 from onadata.apps.logger.views import submission
@@ -86,6 +87,13 @@ def _create_user(self, username, password, create_profile=False):
 
         return user
 
+    def _create_organization(self, username, name, created_by):
+        user = self._create_user(username, "", False)
+        organization, _ = OrganizationProfile.objects.get_or_create(
+            user=user, defaults={"name": name, "creator": created_by}
+        )
+        return organization
+
     # pylint: disable=no-self-use
     def _login(self, username, password):
         client = Client()
diff --git a/onadata/libs/serializers/project_serializer.py b/onadata/libs/serializers/project_serializer.py
@@ -6,27 +6,18 @@
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.core.cache import cache
+from django.core.management import call_command
 from django.db.utils import IntegrityError
 from django.utils.translation import gettext as _
 
 from rest_framework import serializers
 from six import itervalues
 
-from onadata.apps.api.models.organization_profile import (
-    OrganizationProfile,
-    get_or_create_organization_owners_team,
-    get_organization_members_team,
-)
+from onadata.apps.api.models.organization_profile import OrganizationProfile
 from onadata.apps.logger.models.project import Project
 from onadata.apps.logger.models.xform import XForm
 from onadata.apps.main.models.user_profile import UserProfile
-from onadata.libs.permissions import (
-    ManagerRole,
-    OwnerRole,
-    ReadOnlyRole,
-    get_role,
-    is_organization,
-)
+from onadata.libs.permissions import ManagerRole, OwnerRole, get_role, is_organization
 from onadata.libs.serializers.fields.json_field import JsonField
 from onadata.libs.serializers.tag_list_serializer import TagListSerializer
 from onadata.libs.utils.analytics import TrackObjectEvent
@@ -542,23 +533,12 @@ def update(self, instance, validated_data):
             set_owners_permission(owner, instance)
 
             if is_organization(owner.profile):
-                owners_team = get_or_create_organization_owners_team(owner.profile)
-                members_team = get_organization_members_team(owner.profile)
-                OwnerRole.add(owners_team, instance)
-                ReadOnlyRole.add(members_team, instance)
-                owners = owners_team.user_set.all()
-                # Owners are also members
-                members = members_team.user_set.exclude(
-                    username__in=[user.username for user in owners]
+                call_command(
+                    "transferproject",
+                    current_owner=instance.organization,
+                    new_owner=owner,
+                    project_id=instance.pk,
                 )
-                # Exclude new owner if in members
-                members = members.exclude(username=owner.username)
-
-                # Add permissions to all users in Owners and Members team
-                for owner in owners:
-                    OwnerRole.add(owner, instance)
-                for member in members:
-                    ReadOnlyRole.add(member, instance)
 
             # clear cache
             safe_delete(f"{PROJ_PERM_CACHE}{instance.pk}")
