improve tests and mocks

osf/external/messages/celery_publishers.py

@@ -3,44 +3,43 @@
 from framework.celery_tasks import app as celery_app
 from website import settings
 from osf import features
+from osf.utils.requests import get_current_request
 
 
 def publish_deactivated_user(user):
-    if settings.USE_CELERY and waffle.switch_is_active(features.ENABLE_GV):
-        _publish_user_status_change(
-            body={
-                'action': 'deactivate',
-                'user_uri': user.get_semantic_iri(),
-            }
-        )
+    _publish_user_status_change(
+        body={
+            'action': 'deactivate',
+            'user_uri': user.get_semantic_iri(),
+        }
+    )
 
 
 def publish_reactivate_user(user):
-    if settings.USE_CELERY and waffle.switch_is_active(features.ENABLE_GV):
-        _publish_user_status_change(
-            body={
-                'action': 'reactivate',
-                'user_uri': user.get_semantic_iri(),
-            },
-        )
+    _publish_user_status_change(
+        body={
+            'action': 'reactivate',
+            'user_uri': user.get_semantic_iri(),
+        },
+    )
 
 
 def publish_merged_user(user):
     assert user.merged_by, 'User received merge signal, but has no `merged_by` reference.'
-    if settings.USE_CELERY and waffle.switch_is_active(features.ENABLE_GV):
-        _publish_user_status_change(
-            body={
-                'action': 'merge',
-                'into_user_uri': user.merged_by.get_semantic_iri(),
-                'from_user_uri': user.get_semantic_iri(),
-            },
-        )
+    _publish_user_status_change(
+        body={
+            'action': 'merge',
+            'into_user_uri': user.merged_by.get_semantic_iri(),
+            'from_user_uri': user.get_semantic_iri(),
+        },
+    )
 
 
 def _publish_user_status_change(body: dict):
-    with celery_app.producer_pool.acquire() as producer:
-        producer.publish(
-            body=body,
-            exchange=Exchange(celery_app.conf.task_account_status_changes_queue),
-            serializer='json'
-        )
+    if settings.USE_CELERY and waffle.flag_is_active(get_current_request(), features.ENABLE_GV):
+        with celery_app.producer_pool.acquire() as producer:
+            producer.publish(
+                body=body,
+                exchange=Exchange(celery_app.conf.task_account_status_changes_queue),
+                serializer='json'
+            )

tests/test_utils.py

@@ -6,6 +6,8 @@
 import time
 import unittest
 from django.utils import timezone
+from django.dispatch import receiver
+
 
 from flask import Flask
 from nose.tools import *  # noqa (PEP8 asserts)
@@ -24,13 +26,18 @@
 from framework.utils import secure_filename, throttle_period_expired
 from api.base.utils import waterbutler_api_url_for
 from osf.utils.functional import rapply
+from waffle.testutils import override_flag
 from website.routes import process_rules, OsfWebRenderer
 from website import settings
 from website.util import paths
 from website.util import web_url_for, api_url_for, is_json_request, conjunct, api_v2_url
 from website.project import utils as project_utils
 from website.profile import utils as profile_utils
 
+from osf import features
+
+from kombu import Exchange
+
 try:
     import magic  # noqa
     LIBMAGIC_AVAILABLE = True
@@ -460,10 +467,6 @@ def test_build_create_user_time_conflict(self):
         assert user_one_create.username != user_two_create.username
 
 
-# Add this import at the top of your file
-from django.dispatch import receiver
-
-
 @pytest.mark.django_db
 class TestUserSignals:
 
@@ -481,6 +484,10 @@ def deactivated_user(self, db):
         user.deactivate_account()
         return user
 
+    @pytest.fixture
+    def account_status_changes_exchange(self):
+        return Exchange('account_status_changes')
+
     @mock.patch('osf.external.messages.celery_publishers.publish_deactivated_user')
     def test_user_account_deactivated_signal(self, mock_publish_deactivated_user, user):
         # Connect a mock receiver to the signal for testing
@@ -519,3 +526,57 @@ def mock_receiver(user, **kwargs):
 
         # Verify that the mock receiver was called
         mock_publish_reactivate_user.assert_called_once_with(deactivated_user)
+
+    @pytest.mark.enable_account_status_messaging
+    @mock.patch('osf.external.messages.celery_publishers.celery_app.producer_pool.acquire')
+    def test_publish_body_on_deactivation(self, mock_publish_user_status_change, user, account_status_changes_exchange):
+        with mock.patch.object(settings, 'USE_CELERY', True):
+            with override_flag(features.ENABLE_GV, active=True):
+                user.deactivate_account()
+
+        mock_publish_user_status_change().__enter__().publish.assert_called_once_with(
+            body={'action': 'deactivate', 'user_uri': f'http://localhost:5000/{user._id}'},
+            exchange=account_status_changes_exchange,
+            serializer='json',
+        )
+
+    @pytest.mark.enable_account_status_messaging
+    @mock.patch('osf.external.messages.celery_publishers.celery_app.producer_pool.acquire')
+    def test_publish_body_on_reactivation(
+            self,
+            mock_publish_user_status_change,
+            deactivated_user,
+            account_status_changes_exchange
+    ):
+        with mock.patch.object(settings, 'USE_CELERY', True):
+            with override_flag(features.ENABLE_GV, active=True):
+                deactivated_user.reactivate_account()
+
+        mock_publish_user_status_change().__enter__().publish.assert_called_once_with(
+            body={'action': 'reactivate', 'user_uri': f'http://localhost:5000/{deactivated_user._id}'},
+            exchange=account_status_changes_exchange,
+            serializer='json',
+        )
+
+    @pytest.mark.enable_account_status_messaging
+    @mock.patch('osf.external.messages.celery_publishers.celery_app.producer_pool.acquire')
+    def test_publish_body_on_merger(
+            self,
+            mock_publish_user_status_change,
+            user,
+            old_user,
+            account_status_changes_exchange
+    ):
+        with mock.patch.object(settings, 'USE_CELERY', True):
+            with override_flag(features.ENABLE_GV, active=True):
+                user.merge_user(old_user)
+
+        mock_publish_user_status_change().__enter__().publish.assert_called_once_with(
+            body={
+                'action': 'merge',
+                'into_user_uri': f'http://localhost:5000/{user._id}',
+                'from_user_uri': f'http://localhost:5000/{old_user._id}'
+            },
+            exchange=account_status_changes_exchange,
+            serializer='json',
+        )