feat: add pipeline steps that register signup sources

eox_core/pipeline.py

@@ -3,9 +3,15 @@
 """
 import logging
 
+from crum import get_current_request
 from django.db.models.signals import post_save
 
-from eox_core.edxapp_wrapper.users import generate_password, get_user_attribute, get_user_profile
+from eox_core.edxapp_wrapper.users import (
+    generate_password,
+    get_user_attribute,
+    get_user_profile,
+    get_user_signup_source,
+)
 from eox_core.logging import logging_pipeline_step
 
 try:
@@ -14,6 +20,7 @@
     AuthFailed = object
     NotAllowedToDisconnect = object
 
+UserSignupSource = get_user_signup_source()  # pylint: disable=invalid-name
 LOG = logging.getLogger(__name__)
 
 
@@ -164,3 +171,51 @@ def assert_user_information(details, user, backend, *args, **kwargs):
                 **locals()
             )
             raise AuthFailed(backend, "Credentials not allowed.")  # pylint: disable=raising-non-exception
+
+
+def create_signup_source_for_new_association(user=None, *args, **kwargs):
+    """
+    This pipeline step registers a new signup source for users with new social auth link. The signup source
+    will have associated the current site and user.
+
+    It's recommended to place this step after the creation of the social auth link, i.e, after the step
+    'social_core.pipeline.social_auth.associate_user',
+
+    To avoid errors or unnecessary searches, these steps are skipped using 'is_new' key insterted by
+    'social_core.pipeline.user.create_user' after creating a user.
+    """
+    if not user or kwargs.get("is_new", False):
+        return
+
+    if kwargs.get("new_association", False):
+        _, created = UserSignupSource.objects.get_or_create(user=user, site=get_current_request().site)
+        if created:
+            logging_pipeline_step(
+                "info",
+                "Created new singup source for user during the third party pipeline.",
+                **locals()
+            )
+
+
+def ensure_user_has_signup_source(user=None, *args, **kwargs):
+    """
+    This pipeline step registers a new signup source for users that have a social auth link but
+    no signup source associated with the current site. The signup source will be associated to
+    the current site and user.
+
+    It's recommended to place this step at the end of the TPA pipeline, e.g., after the step
+    'eox_core.pipeline.ensure_user_has_profile',
+
+    To avoid errors or unnecessary searches, these steps are skipped using 'is_new' key insterted by
+    'social_core.pipeline.user.create_user' after creating a user.
+    """
+    if not user or kwargs.get("is_new", False):
+        return
+
+    _, created = UserSignupSource.objects.get_or_create(user=user, site=get_current_request().site)
+    if created:
+        logging_pipeline_step(
+            "info",
+            "Created new singup source for user during the third party pipeline.",
+            **locals()
+        )

eox_core/tests/test_pipeline.py

@@ -6,13 +6,18 @@
 from django.test import TestCase
 from mock import MagicMock, PropertyMock, patch
 
+from eox_core.edxapp_wrapper.users import get_user_signup_source
 from eox_core.pipeline import (
     assert_user_information,
     check_disconnect_pipeline_enabled,
+    create_signup_source_for_new_association,
     ensure_new_user_has_usable_password,
     ensure_user_has_profile,
+    ensure_user_has_signup_source,
 )
 
+UserSignupSource = get_user_signup_source()  # pylint: disable=invalid-name
+
 
 class EnsureUserPasswordUsableTest(TestCase):
     """
@@ -164,3 +169,136 @@ def test_connect_any_user(self):
         self.backend_mock.setting.return_value.get.return_value = []
 
         self.assertIsNone(assert_user_information(details, self.user_mock, self.backend_mock))
+
+
+class SignupSourceRegistration(TestCase):
+    """Test custom pipeline steps for signup source registration."""
+
+    def setUp(self):
+        self.user_mock = MagicMock()
+        self.site_mock = MagicMock()
+
+    @patch("eox_core.pipeline.UserSignupSource")
+    @patch("eox_core.pipeline.get_current_request")
+    def test_signup_source_after_assoc(self, get_request_mock, signup_source_mock):
+        """
+        This method tests creating a signup source after associating
+        a social auth link to the user.
+
+        Expected behavior:
+            The signup source is created using the current site and user.
+        """
+        signup_source_mock.objects.get_or_create.return_value = (MagicMock(), True,)
+        get_request_mock.return_value.site = self.site_mock
+        kwargs = {
+            "new_association": True,
+        }
+
+        create_signup_source_for_new_association(self.user_mock, **kwargs)
+
+        signup_source_mock.objects.get_or_create.called_once_with(
+            user=self.user_mock,
+            site=self.site_mock,
+        )
+
+    @patch("eox_core.pipeline.UserSignupSource")
+    @patch("eox_core.pipeline.get_current_request")
+    def test_ensure_signup_source(self, get_request_mock, signup_source_mock):
+        """
+        This method tests creating a signup source when the user has a social
+        link but does not have the signup source for the current site.
+
+        Expected behavior:
+            The signup source is created using the current site and user.
+        """
+        signup_source_mock.objects.get_or_create.return_value = (MagicMock(), True,)
+        get_request_mock.return_value.site = self.site_mock
+        kwargs = {}
+
+        ensure_user_has_signup_source(self.user_mock, **kwargs)
+
+        signup_source_mock.objects.get_or_create.called_once_with(
+            user=self.user_mock,
+            site=self.site_mock,
+        )
+
+    @patch("eox_core.pipeline.UserSignupSource")
+    @patch("eox_core.pipeline.get_current_request")
+    def test_existent_signup_source_after_assoc(self, get_request_mock, signup_source_mock):
+        """
+        This method tests executing the signup source steps when the user
+        already has a signup source for the current site. This is done after a new
+        social auth link is created.
+
+        Expected behavior:
+            No signup sources are created.
+        """
+        signup_source_mock.objects.get_or_create.return_value = (MagicMock(), False,)
+        get_request_mock.return_value.site = self.site_mock
+        kwargs = {
+            "new_association": True,
+        }
+
+        create_signup_source_for_new_association(self.user_mock, **kwargs)
+
+        signup_source_mock.objects.get_or_create.called_once_with(
+            user=self.user_mock,
+            site=self.site_mock,
+        )
+
+    @patch("eox_core.pipeline.UserSignupSource")
+    @patch("eox_core.pipeline.get_current_request")
+    def test_ensure_existent_signup_source(self, get_request_mock, signup_source_mock):
+        """
+        This method tests executing the signup source steps when the user
+        already has a signup source for the current site.
+
+        Expected behavior:
+            No signup sources are created.
+        """
+        signup_source_mock.objects.get_or_create.return_value = (MagicMock(), False,)
+        get_request_mock.return_value.site = self.site_mock
+        kwargs = {}
+
+        ensure_user_has_signup_source(self.user_mock, **kwargs)
+
+        signup_source_mock.objects.get_or_create.called_once_with(
+            user=self.user_mock,
+            site=self.site_mock,
+        )
+
+    @patch("eox_core.pipeline.UserSignupSource")
+    @patch("eox_core.pipeline.get_current_request")
+    def test_signup_source_during_register_assoc(self, _, signup_source_mock):
+        """
+        This method tests executing signup source step create_signup_source_for_new_association
+        during the registration process.
+
+        Expected behavior:
+            No signup sources are created.
+        """
+        kwargs = {
+            "is_new": True,
+        }
+
+        create_signup_source_for_new_association(self.user_mock, **kwargs)
+
+        signup_source_mock.objects.get_or_create.get_or_create.assert_not_called()
+
+    @patch("eox_core.pipeline.UserSignupSource")
+    @patch("eox_core.pipeline.get_current_request")
+    def test_ensure_signup_source_during_register(self, _, signup_source_mock):
+        """
+        This method tests executing signup source step ensure_user_has_signup_source
+        during the registration process.
+
+        Expected behavior:
+            No signup sources are created.
+        """
+        kwargs = {
+            "is_new": True,
+        }
+
+        ensure_user_has_signup_source(self.user_mock, **kwargs)
+
+        signup_source_mock.objects.get_or_create.assert_not_called()