diff --git a/eox_core/pipeline.py b/eox_core/pipeline.py index 3a2334f88..a772abf53 100644 --- a/eox_core/pipeline.py +++ b/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() + ) diff --git a/eox_core/tests/test_pipeline.py b/eox_core/tests/test_pipeline.py index 9cebfdc19..72a6e2366 100644 --- a/eox_core/tests/test_pipeline.py +++ b/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()