fix: Fixing the issue of deactivating user while seat allocation

Author: rgopalrao-sonata-png

enterprise_access/apps/api_client/lms_client.py

@@ -607,6 +607,41 @@ def get_enterprise_user(self, enterprise_customer_uuid, learner_id):
 
         return None
 
+    def get_enterprise_learner_by_email(self, enterprise_customer_uuid, learner_email):
+        """
+        Return the enterprise customer user record for ``learner_email`` if the user is
+        actively linked to ``enterprise_customer_uuid``, otherwise return ``None``.
+
+        Two LMS API calls are made:
+        1. ``/api/user/v1/accounts`` to resolve the learner's LMS user ID from their email.
+        2. ``enterprise-learner/`` to fetch the enterprise link for that user.
+
+        Returns ``None`` when the learner has no LMS account, is not linked to the given
+        enterprise, or any API call fails unexpectedly.
+
+        Arguments:
+            enterprise_customer_uuid (UUID): UUID of the enterprise customer.
+            learner_email (str): Email address of the learner to check.
+        """
+        try:
+            user_accounts = self.get_lms_user_account(email=learner_email)
+        except requests.exceptions.HTTPError:
+            logger.exception(
+                'get_enterprise_learner_by_email: failed to fetch LMS account for email %s',
+                learner_email,
+            )
+            return None
+
+        if not user_accounts:
+            return None
+
+        # get_lms_user_account returns a list; take the first match.
+        lms_user_id = user_accounts[0].get('id') if isinstance(user_accounts, list) else user_accounts.get('id')
+        if not lms_user_id:
+            return None
+
+        return self.get_enterprise_user(enterprise_customer_uuid, lms_user_id)
+
     def create_pending_enterprise_users(self, enterprise_customer_uuid, user_emails):
         """
         Creates a pending enterprise user in the given ``enterprise_customer_uuid`` for each of the

enterprise_access/apps/api_client/tests/test_lms_client.py

@@ -1282,6 +1282,89 @@ def test_get_lms_user_activation_link(
         if expected_link is None:
             self.assertTrue(mock_logger.error.called or mock_logger.exception.called)
 
+    @ddt.data(
+        # list response → returns enterprise user record
+        {
+            'mock_user_accounts': [{'id': TEST_USER_ID}],
+            'mock_enterprise_user': TEST_USER_RECORD,
+            'expected_result': TEST_USER_RECORD,
+            'expect_get_enterprise_user_called': True,
+        },
+        # dict response → also works (not a list)
+        {
+            'mock_user_accounts': {'id': TEST_USER_ID},
+            'mock_enterprise_user': TEST_USER_RECORD,
+            'expected_result': TEST_USER_RECORD,
+            'expect_get_enterprise_user_called': True,
+        },
+        # empty list → None
+        {
+            'mock_user_accounts': [],
+            'mock_enterprise_user': None,
+            'expected_result': None,
+            'expect_get_enterprise_user_called': False,
+        },
+        # None → None
+        {
+            'mock_user_accounts': None,
+            'mock_enterprise_user': None,
+            'expected_result': None,
+            'expect_get_enterprise_user_called': False,
+        },
+        # list entry missing 'id' → None
+        {
+            'mock_user_accounts': [{'email': 'someone@example.com'}],
+            'mock_enterprise_user': None,
+            'expected_result': None,
+            'expect_get_enterprise_user_called': False,
+        },
+        # dict missing 'id' → None
+        {
+            'mock_user_accounts': {'email': 'someone@example.com'},
+            'mock_enterprise_user': None,
+            'expected_result': None,
+            'expect_get_enterprise_user_called': False,
+        },
+    )
+    @ddt.unpack
+    def test_get_enterprise_learner_by_email(
+            self,
+            mock_user_accounts,
+            mock_enterprise_user,
+            expected_result,
+            expect_get_enterprise_user_called,
+    ):
+        """
+        Verify get_enterprise_learner_by_email resolves an email to an enterprise user record,
+        returning None for empty accounts, missing IDs, or unexpected responses.
+        """
+        client = LmsApiClient()
+        learner_email = 'test@example.com'
+
+        with mock.patch.object(client, 'get_lms_user_account', return_value=mock_user_accounts) as mock_get_account:
+            with mock.patch.object(client, 'get_enterprise_user', return_value=mock_enterprise_user) as mock_get_enterprise:
+                result = client.get_enterprise_learner_by_email(str(TEST_ENTERPRISE_UUID), learner_email)
+                self.assertEqual(result, expected_result)
+                mock_get_account.assert_called_once_with(email=learner_email)
+                if expect_get_enterprise_user_called:
+                    mock_get_enterprise.assert_called_once_with(str(TEST_ENTERPRISE_UUID), TEST_USER_ID)
+                else:
+                    mock_get_enterprise.assert_not_called()
+
+    @mock.patch('enterprise_access.apps.api_client.lms_client.logger')
+    def test_get_enterprise_learner_by_email_http_error(self, mock_logger):
+        """
+        Verify get_enterprise_learner_by_email returns None and logs when get_lms_user_account raises HTTPError.
+        """
+        client = LmsApiClient()
+
+        with mock.patch.object(
+            client, 'get_lms_user_account', side_effect=requests.exceptions.HTTPError('whoops')
+        ):
+            result = client.get_enterprise_learner_by_email(str(TEST_ENTERPRISE_UUID), 'test@example.com')
+            self.assertIsNone(result)
+            mock_logger.exception.assert_called_once()
+
 
 class TestLmsUserApiClient(TestCase):
     """

enterprise_access/apps/content_assignments/tasks.py

@@ -367,6 +367,24 @@ def create_pending_enterprise_learner_for_assignment_task(learner_content_assign
     enterprise_customer_uuid = assignment.assignment_configuration.enterprise_customer_uuid
 
     lms_client = LmsApiClient()
+
+    # Skip pending-user creation if the learner is already actively linked to this enterprise.
+    # Calling create_pending_enterprise_users when the user is already linked can inadvertently
+    # trigger the LMS serializer's "inactivate other customers" logic and deactivate the user's
+    # links to other enterprise customers.
+    existing_link = lms_client.get_enterprise_learner_by_email(
+        enterprise_customer_uuid, assignment.learner_email
+    )
+    if existing_link and existing_link.get('active', False):
+        assignment.add_successful_linked_action()
+        logger.info(
+            'Learner is already actively linked to enterprise %s; '
+            'skipping pending enterprise user creation for assignment %s',
+            enterprise_customer_uuid,
+            assignment.uuid,
+        )
+        return
+
     # Could raise HTTPError and trigger task retry.  Intentionally ignoring response since success should just not throw
     # an exception.  Two possible success statuses are 201 (created) and 200 (found), but there's no reason to
     # distinguish them for the purpose of this task.

enterprise_access/apps/content_assignments/tests/test_tasks.py

@@ -87,6 +87,14 @@ def setUp(self):
             assignment_configuration=self.assignment_configuration,
         )
 
+        # By default, the learner is NOT already actively linked.  Individual tests override this.
+        patcher = mock.patch(
+            'enterprise_access.apps.api_client.lms_client.LmsApiClient.get_enterprise_learner_by_email',
+            return_value=None,
+        )
+        self.mock_get_enterprise_learner_by_email = patcher.start()
+        self.addCleanup(patcher.stop)
+
     @ddt.data(
         # The LMS API did not find an existing PendingEnterpriseLearner, so it created one.
         {
@@ -210,6 +218,73 @@ def test_last_retry_success(self, mock_oauth_client):
         self.assignment.refresh_from_db()
         assert self.assignment.state == LearnerContentAssignmentStateChoices.ALLOCATED
 
+    @mock.patch('enterprise_access.apps.api_client.base_oauth.OAuthAPIClient')
+    def test_skip_if_already_active_link(self, mock_oauth_client):
+        """
+        If the learner is already actively linked to the enterprise, the task should
+        record a successful linked action and return early without calling the
+        pending-enterprise-learner LMS endpoint.
+        """
+        self.mock_get_enterprise_learner_by_email.return_value = {
+            'enterprise_customer': {'uuid': str(TEST_ENTERPRISE_UUID)},
+            'user': {'email': TEST_EMAIL},
+            'active': True,
+        }
+
+        task_result = create_pending_enterprise_learner_for_assignment_task.delay(self.assignment.uuid)
+
+        assert task_result.state == celery_states.SUCCESS
+
+        # The pending-enterprise-learner POST endpoint must NOT have been called.
+        mock_oauth_client.return_value.post.assert_not_called()
+
+        # The active-link check must have been called with the correct arguments.
+        self.mock_get_enterprise_learner_by_email.assert_called_once_with(
+            TEST_ENTERPRISE_UUID,
+            TEST_EMAIL,
+        )
+
+        # Assignment state stays allocated and a successful linked action is recorded.
+        self.assignment.refresh_from_db()
+        assert self.assignment.state == LearnerContentAssignmentStateChoices.ALLOCATED
+        assert self.assignment.actions.filter(action_type=AssignmentActions.LEARNER_LINKED).exists()
+
+    @ddt.data(
+        # Learner has a link but it is explicitly inactive.
+        {'active': False},
+        # Learner has a link record with no 'active' field at all.
+        {},
+    )
+    @mock.patch('enterprise_access.apps.api_client.base_oauth.OAuthAPIClient')
+    def test_proceeds_if_link_not_active(self, active_value, mock_oauth_client):
+        """
+        If the learner's enterprise link exists but is not active, the task should
+        proceed normally and call create_pending_enterprise_users.
+        """
+        self.mock_get_enterprise_learner_by_email.return_value = {
+            'enterprise_customer': {'uuid': str(TEST_ENTERPRISE_UUID)},
+            'user': {'email': TEST_EMAIL},
+            **active_value,
+        }
+        mock_oauth_client.return_value.post.return_value = MockResponse(
+            {'enterprise_customer': str(TEST_ENTERPRISE_UUID), 'user_email': TEST_EMAIL},
+            status.HTTP_201_CREATED,
+        )
+
+        task_result = create_pending_enterprise_learner_for_assignment_task.delay(self.assignment.uuid)
+
+        assert task_result.state == celery_states.SUCCESS
+
+        # The pending-enterprise-learner POST endpoint must still have been called.
+        assert len(mock_oauth_client.return_value.post.call_args_list) == 1
+        assert mock_oauth_client.return_value.post.call_args.kwargs['json'] == [{
+            'enterprise_customer': str(self.assignment.assignment_configuration.enterprise_customer_uuid),
+            'user_email': self.assignment.learner_email,
+        }]
+
+        self.assignment.refresh_from_db()
+        assert self.assignment.state == LearnerContentAssignmentStateChoices.ALLOCATED
+
 
 @ddt.ddt
 class TestBrazeEmailTasks(APITestWithMocks):