Add user to organization asynchronously (#2574)

* add user to organization asynchronously

add user to organization asynchronously and refactor to remove redundant code

* address failing tests

* refactor code

* add tests

* add tests

* suppress lint warning

suppress lint warning Function name "add_org_user_and_share_projects_async" doesn't conform to '(([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$' pattern

* fix pylint error

fix pylint: unrecognized-inline-option / Unrecognized file option 'invalid-name'

* add tests

* share project asynchronously

* add test case

* add test case

* retry async task incase of exception

* add tests

* add test case

* address typos

* send email after adding user to org successfully

* address lint warning

address pylint: no-value-for-parameter / No value for argument 'from_email'

onadata/apps/api/tasks.py

@@ -11,21 +11,27 @@
 from django.conf import settings
 from django.core.files.uploadedfile import TemporaryUploadedFile
 from django.core.files.storage import default_storage
+from django.core.mail import send_mail
 from django.contrib.auth import get_user_model
+from django.db import DatabaseError
 from django.utils import timezone
 from django.utils.datastructures import MultiValueDict
 
 from onadata.apps.api import tools
+from onadata.apps.api.models.organization_profile import OrganizationProfile
+from onadata.apps.logger.models import Instance, ProjectInvitation, XForm, Project
 from onadata.libs.utils.email import send_generic_email
 from onadata.libs.utils.model_tools import queryset_iterator
 from onadata.libs.utils.cache_tools import (
     safe_delete,
     XFORM_REGENERATE_INSTANCE_JSON_TASK,
 )
-from onadata.apps.logger.models import Instance, ProjectInvitation, XForm
+from onadata.libs.models.share_project import ShareProject
 from onadata.libs.utils.email import ProjectInvitationEmail
 from onadata.celeryapp import app
 
+logger = logging.getLogger(__name__)
+
 
 User = get_user_model()
 
@@ -145,7 +151,7 @@ def send_project_invitation_email_async(
         invitation = ProjectInvitation.objects.get(id=invitation_id)
 
     except ProjectInvitation.DoesNotExist as err:
-        logging.exception(err)
+        logger.exception(err)
 
     else:
         email = ProjectInvitationEmail(invitation, url)
@@ -161,7 +167,7 @@ def regenerate_form_instance_json(xform_id: int):
     try:
         xform: XForm = XForm.objects.get(pk=xform_id)
     except XForm.DoesNotExist as err:
-        logging.exception(err)
+        logger.exception(err)
 
     else:
         if not xform.is_instance_json_regenerated:
@@ -182,3 +188,73 @@ def regenerate_form_instance_json(xform_id: int):
             # Clear cache used to store the task id from the AsyncResult
             cache_key = f"{XFORM_REGENERATE_INSTANCE_JSON_TASK}{xform_id}"
             safe_delete(cache_key)
+
+
+class ShareProjectBaseTask(app.Task):
+    autoretry_for = (
+        DatabaseError,
+        ConnectionError,
+    )
+    retry_backoff = 3
+
+
+@app.task(base=ShareProjectBaseTask)
+def add_org_user_and_share_projects_async(
+    org_id: int,
+    user_id: int,
+    role: str = None,
+    email_subject: str = None,
+    email_msg: str = None,
+):  # pylint: disable=invalid-name
+    """Add user to organization and share projects asynchronously"""
+    try:
+        organization = OrganizationProfile.objects.get(pk=org_id)
+        user = User.objects.get(pk=user_id)
+
+    except OrganizationProfile.DoesNotExist as err:
+        logger.exception(err)
+
+    except User.DoesNotExist as err:
+        logger.exception(err)
+
+    else:
+        tools.add_org_user_and_share_projects(organization, user, role)
+
+        if email_msg and email_subject and user.email:
+            send_mail(
+                email_subject,
+                email_msg,
+                settings.DEFAULT_FROM_EMAIL,
+                (user.email,),
+            )
+
+
+@app.task(base=ShareProjectBaseTask)
+def remove_org_user_async(org_id, user_id):
+    """Remove user from organization asynchronously"""
+    try:
+        organization = OrganizationProfile.objects.get(pk=org_id)
+        user = User.objects.get(pk=user_id)
+
+    except OrganizationProfile.DoesNotExist as err:
+        logger.exception(err)
+
+    except User.DoesNotExist as err:
+        logger.exception(err)
+
+    else:
+        tools.remove_user_from_organization(organization, user)
+
+
+@app.task(base=ShareProjectBaseTask)
+def share_project_async(project_id, username, role, remove=False):
+    """Share project asynchronously"""
+    try:
+        project = Project.objects.get(pk=project_id)
+
+    except Project.DoesNotExist as err:
+        logger.exception(err)
+
+    else:
+        share = ShareProject(project, username, role, remove)
+        share.save()

onadata/apps/api/tests/test_tasks.py

@@ -1,19 +1,31 @@
 """Tests for module onadata.apps.api.tasks"""
+
 import sys
 
 from unittest.mock import patch
 
 from django.core.cache import cache
+from django.contrib.auth import get_user_model
+from django.db import DatabaseError, OperationalError
+from django.test import override_settings
 
-from onadata.apps.main.tests.test_base import TestBase
 from onadata.apps.api.tasks import (
     send_project_invitation_email_async,
     regenerate_form_instance_json,
+    add_org_user_and_share_projects_async,
+    remove_org_user_async,
+    share_project_async,
+    ShareProject,
 )
+from onadata.apps.api.models.organization_profile import OrganizationProfile
 from onadata.apps.logger.models import ProjectInvitation, Instance
+from onadata.apps.main.tests.test_base import TestBase
+from onadata.libs.permissions import ManagerRole
 from onadata.libs.utils.user_auth import get_user_default_project
 from onadata.libs.utils.email import ProjectInvitationEmail
 
+User = get_user_model()
+
 
 class SendProjectInivtationEmailAsyncTestCase(TestBase):
     """Tests for send_project_invitation_email_async"""
@@ -80,7 +92,7 @@ def mock_get_full_dict(
         instance.refresh_from_db()
         self.assertFalse("foo" in instance.json)
 
-    @patch("logging.exception")
+    @patch("onadata.apps.api.tasks.logger.exception")
     def test_form_id_invalid(self, mock_log_exception):
         """An invalid xform_id is handled"""
 
@@ -107,3 +119,219 @@ def mock_get_full_dict(
         regenerate_form_instance_json.delay(self.xform.pk)
         instance.refresh_from_db()
         self.assertFalse(instance.json)
+
+
+@patch("onadata.apps.api.tasks.tools.add_org_user_and_share_projects")
+class AddOrgUserAndShareProjectsAsyncTestCase(TestBase):
+    """Tests for add_org_user_and_share_projects_async"""
+
+    def setUp(self):
+        super().setUp()
+
+        self.org_user = User.objects.create(username="onaorg")
+        alice = self._create_user("alice", "1234&&")
+        self.org = OrganizationProfile.objects.create(
+            user=self.org_user, name="Ona Org", creator=alice
+        )
+
+    def test_user_added_to_org(self, mock_add):
+        """User is added to organization"""
+        add_org_user_and_share_projects_async.delay(
+            self.org.pk, self.user.pk, "manager"
+        )
+        mock_add.assert_called_once_with(self.org, self.user, "manager")
+
+    def test_role_optional(self, mock_add):
+        """role param is optional"""
+        add_org_user_and_share_projects_async.delay(self.org.pk, self.user.pk)
+        mock_add.assert_called_once_with(self.org, self.user, None)
+
+    @patch("onadata.apps.api.tasks.logger.exception")
+    def test_invalid_org_id(self, mock_log, mock_add):
+        """Invalid org_id is handled"""
+        add_org_user_and_share_projects_async.delay(sys.maxsize, self.user.pk)
+        mock_add.assert_not_called()
+        mock_log.assert_called_once()
+
+    @patch("onadata.apps.api.tasks.logger.exception")
+    def test_invalid_user_id(self, mock_log, mock_add):
+        """Invalid org_id is handled"""
+        add_org_user_and_share_projects_async.delay(self.org.pk, sys.maxsize)
+        mock_add.assert_not_called()
+        mock_log.assert_called_once()
+
+    @patch("onadata.apps.api.tasks.add_org_user_and_share_projects_async.retry")
+    def test_database_error(self, mock_retry, mock_add):
+        """We retry calls if DatabaseError is raised"""
+        mock_add.side_effect = DatabaseError()
+        add_org_user_and_share_projects_async.delay(self.org.pk, self.user.pk)
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], DatabaseError))
+
+    @patch("onadata.apps.api.tasks.add_org_user_and_share_projects_async.retry")
+    def test_connection_error(self, mock_retry, mock_add):
+        """We retry calls if ConnectionError is raised"""
+        mock_add.side_effect = ConnectionError()
+        add_org_user_and_share_projects_async.delay(self.org.pk, self.user.pk)
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], ConnectionError))
+
+    @patch("onadata.apps.api.tasks.add_org_user_and_share_projects_async.retry")
+    def test_operation_error(self, mock_retry, mock_add):
+        """We retry calls if OperationError is raised"""
+        mock_add.side_effect = OperationalError()
+        add_org_user_and_share_projects_async.delay(self.org.pk, self.user.pk)
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], OperationalError))
+
+    @override_settings(DEFAULT_FROM_EMAIL="[email protected]")
+    @patch("onadata.apps.api.tasks.send_mail")
+    def test_send_mail(self, mock_email, mock_add):
+        """Send mail works"""
+        self.user.email = "[email protected]"
+        self.user.save()
+        add_org_user_and_share_projects_async.delay(
+            self.org.pk, self.user.pk, "manager", "Subject", "Body"
+        )
+        mock_email.assert_called_with(
+            "Subject",
+            "Body",
+            "[email protected]",
+            ("[email protected]",),
+        )
+        mock_add.assert_called_once_with(self.org, self.user, "manager")
+
+    @override_settings(DEFAULT_FROM_EMAIL="[email protected]")
+    @patch("onadata.apps.api.tasks.send_mail")
+    def test_user_email_none(self, mock_email, mock_add):
+        """Email not sent if user email is None"""
+        add_org_user_and_share_projects_async.delay(
+            self.org.pk, self.user.pk, "manager", "Subject", "Body"
+        )
+        mock_email.assert_not_called()
+        mock_add.assert_called_once_with(self.org, self.user, "manager")
+
+
+@patch("onadata.apps.api.tasks.tools.remove_user_from_organization")
+class RemoveOrgUserAsyncTestCase(TestBase):
+    """Tests for remove_org_user_async"""
+
+    def setUp(self):
+        super().setUp()
+
+        self.org_user = User.objects.create(username="onaorg")
+        alice = self._create_user("alice", "1234&&")
+        self.org = OrganizationProfile.objects.create(
+            user=self.org_user, name="Ona Org", creator=alice
+        )
+
+    def test_user_removed_from_org(self, mock_remove):
+        """User is removed from organization"""
+        remove_org_user_async.delay(self.org.pk, self.user.pk)
+        mock_remove.assert_called_once_with(self.org, self.user)
+
+    @patch("onadata.apps.api.tasks.logger.exception")
+    def test_invalid_org_id(self, mock_log, mock_remove):
+        """Invalid org_id is handled"""
+        remove_org_user_async.delay(sys.maxsize, self.user.pk)
+        mock_remove.assert_not_called()
+        mock_log.assert_called_once()
+
+    @patch("onadata.apps.api.tasks.logger.exception")
+    def test_invalid_user_id(self, mock_log, mock_remove):
+        """Invalid user_id is handled"""
+        remove_org_user_async.delay(self.org.pk, sys.maxsize)
+        mock_remove.assert_not_called()
+        mock_log.assert_called_once()
+
+    @patch("onadata.apps.api.tasks.remove_org_user_async.retry")
+    def test_database_error(self, mock_retry, mock_remove):
+        """We retry calls if DatabaseError is raised"""
+        mock_remove.side_effect = DatabaseError()
+        remove_org_user_async.delay(self.org.pk, self.user.pk)
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], DatabaseError))
+
+    @patch("onadata.apps.api.tasks.remove_org_user_async.retry")
+    def test_connection_error(self, mock_retry, mock_remove):
+        """We retry calls if ConnectionError is raised"""
+        mock_remove.side_effect = ConnectionError()
+        remove_org_user_async.delay(self.org.pk, self.user.pk)
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], ConnectionError))
+
+    @patch("onadata.apps.api.tasks.remove_org_user_async.retry")
+    def test_operation_error(self, mock_retry, mock_remove):
+        """We retry calls if OperationError is raised"""
+        mock_remove.side_effect = OperationalError()
+        remove_org_user_async.delay(self.org.pk, self.user.pk)
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], OperationalError))
+
+
+class ShareProjectAsyncTestCase(TestBase):
+    """Tests for share_project_async"""
+
+    def setUp(self):
+        super().setUp()
+
+        self._publish_transportation_form()
+        self.alice = self._create_user("alice", "Yuao8(-)")
+
+    def test_share(self):
+        """Project is shared with user"""
+        share_project_async.delay(self.project.id, "alice", "manager")
+
+        self.assertTrue(ManagerRole.user_has_role(self.alice, self.project))
+
+    def test_remove(self):
+        """User is removed from project"""
+        # Add user to project
+        ManagerRole.add(self.alice, self.project)
+        # Remove user
+        share_project_async.delay(self.project.id, "alice", "manager", True)
+
+        self.assertFalse(ManagerRole.user_has_role(self.alice, self.project))
+
+    @patch("onadata.apps.api.tasks.logger.exception")
+    def test_invalid_project_id(self, mock_log):
+        """Invalid projecct_id is handled"""
+        share_project_async.delay(sys.maxsize, "alice", "manager")
+        self.assertFalse(ManagerRole.user_has_role(self.alice, self.project))
+        mock_log.assert_called_once()
+
+    @patch.object(ShareProject, "save")
+    @patch("onadata.apps.api.tasks.share_project_async.retry")
+    def test_database_error(self, mock_retry, mock_share):
+        """We retry calls if DatabaseError is raised"""
+        mock_share.side_effect = DatabaseError()
+        share_project_async.delay(self.project.id, self.user.pk, "manager")
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], DatabaseError))
+
+    @patch.object(ShareProject, "save")
+    @patch("onadata.apps.api.tasks.share_project_async.retry")
+    def test_connection_error(self, mock_retry, mock_share):
+        """We retry calls if ConnectionError is raised"""
+        mock_share.side_effect = ConnectionError()
+        share_project_async.delay(self.project.pk, self.user.pk, "manager")
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], ConnectionError))
+
+    @patch.object(ShareProject, "save")
+    @patch("onadata.apps.api.tasks.share_project_async.retry")
+    def test_operation_error(self, mock_retry, mock_share):
+        """We retry calls if OperationError is raised"""
+        mock_share.side_effect = OperationalError()
+        share_project_async.delay(self.project.pk, self.user.pk, "manager")
+        self.assertTrue(mock_retry.called)
+        _, kwargs = mock_retry.call_args_list[0]
+        self.assertTrue(isinstance(kwargs["exc"], OperationalError))