Move send_message to non-blocking thread

onadata/apps/api/tests/viewsets/test_data_viewset.py

@@ -1752,9 +1752,11 @@ def test_delete_submission(self):
         response = view(request, pk=formid)
         self.assertEqual(len(response.data), 2)
 
-    @patch("onadata.apps.api.viewsets.data_viewset.send_message")
+    @patch("onadata.apps.api.viewsets.data_viewset.send_actstream_message_async.delay")
     @patch("onadata.apps.viewer.signals._post_process_submissions")
-    def test_post_save_signal_on_submission_deletion(self, mock, send_message_mock):
+    def test_post_save_signal_on_submission_deletion(
+        self, mock, send_actstream_message_async_mock
+    ):
         # test that post_save_submission signal is sent
         # create form
         xls_file_path = os.path.join(
@@ -1791,7 +1793,7 @@ def test_post_save_signal_on_submission_deletion(self, mock, send_message_mock):
         # check that call_count is still one after saving instance again
         instance.save()
         self.assertEqual(mock.call_count, 1)
-        self.assertTrue(send_message_mock.called)
+        self.assertTrue(send_actstream_message_async_mock.called)
 
     @patch("onadata.apps.api.viewsets.data_viewset.safe_cache_set")
     def test_deletion_of_bulk_submissions(self, mock_cache_set):
@@ -1842,8 +1844,8 @@ def test_deletion_of_bulk_submissions(self, mock_cache_set):
         )
 
     @override_settings(ENABLE_SUBMISSION_PERMANENT_DELETE=True)
-    @patch("onadata.apps.api.viewsets.data_viewset.send_message")
-    def test_submissions_permanent_deletion(self, send_message_mock):
+    @patch("onadata.apps.api.viewsets.data_viewset.send_actstream_message_async.delay")
+    def test_submissions_permanent_deletion(self, send_actstream_message_async_mock):
         """
         Test that permanent submission deletions work
         """
@@ -1876,12 +1878,12 @@ def test_submissions_permanent_deletion(self, send_message_mock):
         )
 
         # message sent upon delete
-        self.assertTrue(send_message_mock.called)
-        send_message_mock.assert_called_with(
+        self.assertTrue(send_actstream_message_async_mock.called)
+        send_actstream_message_async_mock.assert_called_with(
             instance_id=dataid,
             target_id=formid,
             target_type=XFORM,
-            user=request.user,
+            user=request.user.id,
             message_verb=SUBMISSION_DELETED,
         )
 
@@ -1936,8 +1938,10 @@ def test_permanent_deletions_bulk_submissions(self):
         self.assertEqual(self.xform.instances.count(), 2)
 
     @override_settings(ENABLE_SUBMISSION_PERMANENT_DELETE=True)
-    @patch("onadata.apps.api.viewsets.data_viewset.send_message")
-    def test_permanent_instance_delete_inactive_form(self, send_message_mock):
+    @patch("onadata.apps.api.viewsets.data_viewset.send_actstream_message_async.delay")
+    def test_permanent_instance_delete_inactive_form(
+        self, send_actstream_message_async_mock
+    ):
         """
         Test that permanent submission deletions works on inactive forms
         """
@@ -1978,7 +1982,7 @@ def test_permanent_instance_delete_inactive_form(self, send_message_mock):
         # test that xform submission count is updated
         self.xform.refresh_from_db()
         self.assertEqual(self.xform.num_of_submissions, 2)
-        self.assertTrue(send_message_mock.called)
+        self.assertTrue(send_actstream_message_async_mock.called)
 
         # check number of instances and num_of_submissions field
         self.assertEqual(self.xform.instances.count(), 2)
@@ -2002,8 +2006,8 @@ def test_failed_permanent_deletion(self):
         error_msg = "Permanent submission deletion is not enabled for this server."
         self.assertEqual(response.data, {"error": error_msg})
 
-    @patch("onadata.apps.api.viewsets.data_viewset.send_message")
-    def test_delete_submission_inactive_form(self, send_message_mock):
+    @patch("onadata.apps.api.viewsets.data_viewset.send_actstream_message_async.delay")
+    def test_delete_submission_inactive_form(self, send_actstream_message_async_mock):
         self._make_submissions()
         formid = self.xform.pk
         dataid = self.xform.instances.all().order_by("id")[0].pk
@@ -2028,7 +2032,7 @@ def test_delete_submission_inactive_form(self, send_message_mock):
         response = view(request, pk=formid, dataid=dataid)
 
         self.assertEqual(response.status_code, 400)
-        self.assertTrue(send_message_mock.called)
+        self.assertTrue(send_actstream_message_async_mock.called)
 
     def test_delete_submissions(self):
         xls_file_path = os.path.join(
@@ -2099,8 +2103,8 @@ def test_delete_submissions(self):
         self.assertEqual(delete_all_current_count, 0)
         self.assertEqual(self.xform.num_of_submissions, 0)
 
-    @patch("onadata.apps.api.viewsets.data_viewset.send_message")
-    def test_delete_submission_by_editor(self, send_message_mock):
+    @patch("onadata.apps.api.viewsets.data_viewset.send_actstream_message_async.delay")
+    def test_delete_submission_by_editor(self, send_actstream_message_async_mock):
         self._make_submissions()
         formid = self.xform.pk
         dataid = self.xform.instances.all().order_by("id")[0].pk
@@ -2126,10 +2130,10 @@ def test_delete_submission_by_editor(self, send_message_mock):
         request = self.factory.get("/", **self.extra)
         response = view(request, pk=formid)
         self.assertEqual(len(response.data), 3)
-        self.assertTrue(send_message_mock.called)
+        self.assertTrue(send_actstream_message_async_mock.called)
 
-    @patch("onadata.apps.api.viewsets.data_viewset.send_message")
-    def test_delete_submission_by_owner(self, send_message_mock):
+    @patch("onadata.apps.api.viewsets.data_viewset.send_actstream_message_async.delay")
+    def test_delete_submission_by_owner(self, send_actstream_message_async_mock):
         self._make_submissions()
         formid = self.xform.pk
         dataid = self.xform.instances.all().order_by("id")[0].pk
@@ -2155,7 +2159,7 @@ def test_delete_submission_by_owner(self, send_message_mock):
         request = self.factory.get("/", **self.extra)
         response = view(request, pk=formid)
         self.assertEqual(len(response.data), 3)
-        self.assertTrue(send_message_mock.called)
+        self.assertTrue(send_actstream_message_async_mock.called)
 
     def test_geojson_pagination(self):
         self._publish_submit_geojson()

onadata/apps/api/tests/viewsets/test_organization_profile_viewset.py

@@ -1531,18 +1531,22 @@ def test_rotation_errors_captured(self, mock_rotate_key):
         self.assertEqual(response.status_code, 400)
         self.assertIn("Key is disabled", str(response.data["id"]))
 
-    @patch("onadata.apps.api.viewsets.organization_profile_viewset.send_message")
-    def test_audit_log_capture(self, mock_send_message, mock_rotate_key):
+    @patch(
+        "onadata.apps.api.viewsets.organization_profile_viewset.send_actstream_message_async.delay"
+    )
+    def test_audit_log_capture(
+        self, mock_send_actstream_message_async, mock_rotate_key
+    ):
         """Audit log is captured."""
         request = self.factory.post("/", data=self.data, **self.extra)
         response = self.view(request, user="denoinc")
 
         self.assertEqual(response.status_code, 200)
-        mock_send_message.assert_called_once_with(
+        mock_send_actstream_message_async.assert_called_once_with(
             instance_id=self.kms_key.id,
             target_id=self.kms_key.id,
             target_type="kmskey",
-            user=self.user,
+            user=self.user.id,
             message_verb="kmskey_rotated",
         )
 

onadata/apps/api/tests/viewsets/test_submission_review_viewset.py

@@ -1,6 +1,7 @@
 """
 Submission Review ViewSet Tests Module
 """
+
 from __future__ import unicode_literals
 
 from unittest.mock import patch
@@ -55,8 +56,10 @@ def _create_submission_review(self):
 
         return response.data
 
-    @patch("onadata.apps.api.viewsets.submission_review_viewset.send_message")
-    def test_submission_review_create(self, mock_send_message):
+    @patch(
+        "onadata.apps.api.viewsets.submission_review_viewset.send_actstream_message_async.delay"
+    )
+    def test_submission_review_create(self, mock_send_actstream_message_async):
         """
         Test we can create a submission review
         """
@@ -65,17 +68,19 @@ def test_submission_review_create(self, mock_send_message):
             id=submission_review_data["id"]
         )
         # sends message upon saving the submission review
-        self.assertTrue(mock_send_message.called)
-        mock_send_message.assert_called_with(
+        self.assertTrue(mock_send_actstream_message_async.called)
+        mock_send_actstream_message_async.assert_called_with(
             instance_id=submission_review.instance_id,
             target_id=submission_review.instance.xform.id,
             target_type=XFORM,
-            user=submission_review.created_by,
+            user=submission_review.created_by.id,
             message_verb=SUBMISSION_REVIEWED,
         )
 
-    @patch("onadata.apps.api.viewsets.submission_review_viewset.send_message")
-    def test_bulk_create_submission_review(self, mock_send_message):
+    @patch(
+        "onadata.apps.api.viewsets.submission_review_viewset.send_actstream_message_async.delay"
+    )
+    def test_bulk_create_submission_review(self, mock_send_actstream_message_async):
         """
         Test that we can bulk create submission reviews
         """
@@ -100,12 +105,12 @@ def test_bulk_create_submission_review(self, mock_send_message):
         self.assertEqual(4, len(response.data))
         already_seen = []
         # sends message upon saving the submission review
-        self.assertTrue(mock_send_message.called)
-        mock_send_message.assert_called_with(
+        self.assertTrue(mock_send_actstream_message_async.called)
+        mock_send_actstream_message_async.assert_called_with(
             instance_id=[s.id for s in self.xform.instances.all()],
             target_id=self.xform.id,
             target_type=XFORM,
-            user=request.user,
+            user=request.user.id,
             message_verb=SUBMISSION_REVIEWED,
         )
         for item in response.data:

onadata/apps/api/tests/viewsets/test_xform_viewset.py

@@ -708,9 +708,11 @@ def setUp(self):
             }
         )
 
-    @patch("onadata.apps.api.viewsets.xform_viewset.send_message")
+    @patch("onadata.apps.api.viewsets.xform_viewset.send_actstream_message_async.delay")
     @flaky
-    def test_replace_form_with_external_choices(self, mock_send_message):
+    def test_replace_form_with_external_choices(
+        self, mock_send_actstream_message_async
+    ):
         with HTTMock(enketo_mock):
             xls_file_path = os.path.join(
                 settings.PROJECT_ROOT,
@@ -754,12 +756,12 @@ def test_replace_form_with_external_choices(self, mock_send_message):
                 response = view(request, pk=form_id)
                 self.assertEqual(response.status_code, 200)
             # send message upon form update
-            self.assertTrue(mock_send_message.called)
-            mock_send_message.assert_called_with(
+            self.assertTrue(mock_send_actstream_message_async.called)
+            mock_send_actstream_message_async.assert_called_with(
                 instance_id=self.xform.id,
                 target_id=self.xform.id,
                 target_type=XFORM,
-                user=request.user,
+                user=request.user.id,
                 message_verb=FORM_UPDATED,
             )
 

onadata/apps/api/viewsets/data_viewset.py

@@ -31,7 +31,7 @@
 from onadata.apps.logger.models.instance import FormInactiveError, Instance
 from onadata.apps.logger.models.xform import XForm
 from onadata.apps.messaging.constants import SUBMISSION_DELETED, XFORM
-from onadata.apps.messaging.serializers import send_message
+from onadata.apps.messaging.tasks import send_actstream_message_async
 from onadata.apps.viewer.models.parsed_instance import (
     ParsedInstance,
     _get_sort_fields,
@@ -428,11 +428,11 @@ def destroy(self, request, *args, **kwargs):
                 self.object.xform.project.save(update_fields=["date_modified"])
 
                 # send message
-                send_message(
+                send_actstream_message_async.delay(
                     instance_id=instance_id,
                     target_id=self.object.xform.id,
                     target_type=XFORM,
-                    user=request.user,
+                    user=request.user.id,
                     message_verb=SUBMISSION_DELETED,
                 )
             else:

onadata/apps/api/viewsets/organization_profile_viewset.py

@@ -20,7 +20,7 @@
 from onadata.apps.api.tools import get_baseviewset_class, get_org_profile_cache_key
 from onadata.apps.logger.models import KMSKey
 from onadata.apps.messaging.constants import KMS_KEY, KMS_KEY_ROTATED
-from onadata.apps.messaging.serializers import send_message
+from onadata.apps.messaging.tasks import send_actstream_message_async
 from onadata.libs.filters import (
     OrganizationPermissionFilter,
     OrganizationsSharedWithUserFilter,
@@ -159,11 +159,11 @@ def rotate_key(self, request, *args, **kwargs):
         old_key = KMSKey.objects.get(id=serializer.validated_data["id"])
         new_key = serializer.save()
         # Capture audit log
-        send_message(
+        send_actstream_message_async.delay(
             instance_id=old_key.id,
             target_id=old_key.id,
             target_type=KMS_KEY,
-            user=request.user,
+            user=request.user.id,
             message_verb=KMS_KEY_ROTATED,
         )
         data = KMSKeyInlineSerializer(new_key).data

onadata/apps/api/viewsets/submission_review_viewset.py

@@ -2,6 +2,7 @@
 """
 Submission Review Viewset Module
 """
+
 from __future__ import unicode_literals
 
 from django_filters.rest_framework import DjangoFilterBackend
@@ -12,31 +13,37 @@
 from onadata.apps.api.permissions import SubmissionReviewPermissions
 from onadata.apps.api.tools import get_baseviewset_class
 from onadata.apps.logger.models import SubmissionReview
-from onadata.apps.messaging.constants import XFORM, SUBMISSION_REVIEWED
-from onadata.apps.messaging.serializers import send_message
-from onadata.libs.mixins.authenticate_header_mixin import \
-    AuthenticateHeaderMixin
+from onadata.apps.messaging.constants import SUBMISSION_REVIEWED, XFORM
+from onadata.apps.messaging.tasks import send_actstream_message_async
+from onadata.libs.mixins.authenticate_header_mixin import AuthenticateHeaderMixin
 from onadata.libs.mixins.bulk_create_mixin import BulkCreateMixin
 from onadata.libs.mixins.cache_control_mixin import CacheControlMixin
 from onadata.libs.mixins.etags_mixin import ETagsMixin
-from onadata.libs.serializers.submission_review_serializer import \
-    SubmissionReviewSerializer
+from onadata.libs.serializers.submission_review_serializer import (
+    SubmissionReviewSerializer,
+)
 
 BaseViewset = get_baseviewset_class()
 
 
 # pylint: disable=too-many-ancestors
-class SubmissionReviewViewSet(AuthenticateHeaderMixin, CacheControlMixin,
-                              ETagsMixin, BulkCreateMixin, BaseViewset,
-                              ModelViewSet):
+class SubmissionReviewViewSet(
+    AuthenticateHeaderMixin,
+    CacheControlMixin,
+    ETagsMixin,
+    BulkCreateMixin,
+    BaseViewset,
+    ModelViewSet,
+):
     """
     Submission Review ViewSet class
     """
+
     queryset = SubmissionReview.objects.filter(deleted_at__isnull=True)
     serializer_class = SubmissionReviewSerializer
     permission_classes = [SubmissionReviewPermissions]
-    filter_backends = (DjangoFilterBackend, )
-    filterset_fields = ('instance', 'created_by', 'status')
+    filter_backends = (DjangoFilterBackend,)
+    filterset_fields = ("instance", "created_by", "status")
 
     def destroy(self, request, *args, **kwargs):
         """
@@ -57,24 +64,28 @@ def create(self, request, *args, **kwargs):
             serializer = self.get_serializer(data=request.data, many=True)
             serializer.is_valid(raise_exception=True)
             self.perform_create(serializer)
-            instance_ids = [sub_review['instance'] for sub_review in
-                            serializer.data]
+            instance_ids = [sub_review["instance"] for sub_review in serializer.data]
             headers = self.get_success_headers(serializer.data)
             xform = SubmissionReview.objects.get(
-                id=serializer.data[0]['id']).instance.xform
+                id=serializer.data[0]["id"]
+            ).instance.xform
 
         else:
             serializer = self.get_serializer(data=request.data)
             serializer.is_valid(raise_exception=True)
             serializer.save()
-            instance_ids = serializer.data['instance']
+            instance_ids = serializer.data["instance"]
             headers = self.get_success_headers(serializer.data)
             xform = SubmissionReview.objects.get(
-                id=serializer.data['id']).instance.xform
-        send_message(
+                id=serializer.data["id"]
+            ).instance.xform
+        send_actstream_message_async.delay(
             instance_id=instance_ids,
             target_id=xform.id,
-            target_type=XFORM, user=request.user,
-            message_verb=SUBMISSION_REVIEWED)
-        return Response(serializer.data, status=status.HTTP_201_CREATED,
-                        headers=headers)
+            target_type=XFORM,
+            user=request.user.id,
+            message_verb=SUBMISSION_REVIEWED,
+        )
+        return Response(
+            serializer.data, status=status.HTTP_201_CREATED, headers=headers
+        )