fix: Fixing the issue of deactivating user while seat allocation

enterprise_access/apps/api_client/lms_client.py

@@ -607,6 +607,49 @@ 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
+        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 the account lookup returns no usable user ID.
+
+        Raises ``requests.exceptions.HTTPError`` on LMS API failures (e.g. 5xx) so that
+        calling Celery tasks retry rather than silently proceeding with a stale state.
+
+        Arguments:
+            enterprise_customer_uuid (UUID): UUID of the enterprise customer.
+            learner_email (str): Email address of the learner to check.
+        """
+        # get_lms_user_account returns None on 404 and raises HTTPError on 5xx.
+        user_accounts = self.get_lms_user_account(email=learner_email)
+        if not user_accounts:
+            return None
+
+        # get_lms_user_account(email=...) always returns a list from the accounts endpoint.
+        lms_user_id = user_accounts[0].get('id') if isinstance(user_accounts, list) else user_accounts.get('id')
+        if lms_user_id is None:
+            return None
+
+        ec_uuid = str(enterprise_customer_uuid)
+        response = self.client.get(
+            self.enterprise_learner_endpoint,
+            params={'enterprise_customer_uuid': ec_uuid, 'user_ids': lms_user_id},
+            timeout=settings.LMS_CLIENT_TIMEOUT,
+        )
+        response.raise_for_status()  # raises HTTPError on 4xx/5xx — callers (tasks) should retry
+        for result in response.json().get('results', []):
+            returned_customer = result.get('enterprise_customer', {})
+            returned_user = result.get('user', {})
+            if returned_customer.get('uuid') == ec_uuid and returned_user.get('id') == lms_user_id:
+                return result
+        return None
+
     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,107 @@ 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, user linked → returns enterprise user record
+        {
+            'mock_user_accounts': [{'id': TEST_USER_ID}],
+            'enterprise_results': [TEST_USER_RECORD],
+            'expected_result': TEST_USER_RECORD,
+            'expect_enterprise_called': True,
+        },
+        # empty list → None, no enterprise call
+        {
+            'mock_user_accounts': [],
+            'enterprise_results': None,
+            'expected_result': None,
+            'expect_enterprise_called': False,
+        },
+        # None → None, no enterprise call
+        {
+            'mock_user_accounts': None,
+            'enterprise_results': None,
+            'expected_result': None,
+            'expect_enterprise_called': False,
+        },
+        # user has account but is not linked to this enterprise → None
+        {
+            'mock_user_accounts': [{'id': TEST_USER_ID}],
+            'enterprise_results': [],
+            'expected_result': None,
+            'expect_enterprise_called': True,
+        },
+        # list entry missing 'id' → None, no enterprise call
+        {
+            'mock_user_accounts': [{'email': 'someone@example.com'}],
+            'enterprise_results': None,
+            'expected_result': None,
+            'expect_enterprise_called': False,
+        },
+    )
+    @ddt.unpack
+    @mock.patch('enterprise_access.apps.api_client.base_oauth.OAuthAPIClient')
+    def test_get_enterprise_learner_by_email(
+            self,
+            mock_oauth_client,
+            mock_user_accounts,
+            enterprise_results,
+            expected_result,
+            expect_enterprise_called,
+    ):
+        """
+        Verify get_enterprise_learner_by_email resolves an email to an enterprise user record,
+        returning None when the learner has no account, no matching ID, or is not linked.
+        HTTP errors from either LMS call propagate to the caller.
+        """
+        learner_email = 'test@example.com'
+
+        if enterprise_results is not None:
+            mock_oauth_client.return_value.get.return_value = MockResponse(
+                {'results': enterprise_results}, 200
+            )
+
+        client = LmsApiClient()
+        with mock.patch.object(client, 'get_lms_user_account', return_value=mock_user_accounts) as mock_get_account:
+            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_enterprise_called:
+            mock_oauth_client.return_value.get.assert_called_once_with(
+                client.enterprise_learner_endpoint,
+                params={'enterprise_customer_uuid': str(TEST_ENTERPRISE_UUID), 'user_ids': TEST_USER_ID},
+                timeout=settings.LMS_CLIENT_TIMEOUT,
+            )
+        else:
+            mock_oauth_client.return_value.get.assert_not_called()
+
+    @mock.patch('enterprise_access.apps.api_client.base_oauth.OAuthAPIClient')
+    def test_get_enterprise_learner_by_email_account_http_error(self, mock_oauth_client):
+        """
+        HTTPError from get_lms_user_account propagates to the caller instead of being swallowed.
+        This ensures calling Celery tasks retry rather than silently proceeding.
+        """
+        client = LmsApiClient()
+        with mock.patch.object(
+            client, 'get_lms_user_account', side_effect=requests.exceptions.HTTPError('500')
+        ):
+            with self.assertRaises(requests.exceptions.HTTPError):
+                client.get_enterprise_learner_by_email(str(TEST_ENTERPRISE_UUID), 'test@example.com')
+        mock_oauth_client.return_value.get.assert_not_called()
+
+    @mock.patch('enterprise_access.apps.api_client.base_oauth.OAuthAPIClient')
+    def test_get_enterprise_learner_by_email_enterprise_http_error(self, mock_oauth_client):
+        """
+        HTTPError from the enterprise-learner endpoint propagates to the caller.
+        This ensures calling Celery tasks retry on LMS 5xx rather than treating
+        the failure as 'not linked' and proceeding to create_pending_enterprise_users.
+        """
+        mock_oauth_client.return_value.get.return_value = MockResponse(None, 503)
+        client = LmsApiClient()
+        with mock.patch.object(client, 'get_lms_user_account', return_value=[{'id': TEST_USER_ID}]):
+            with self.assertRaises(requests.exceptions.HTTPError):
+                client.get_enterprise_learner_by_email(str(TEST_ENTERPRISE_UUID), 'test@example.com')
+
 
 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,100 @@ 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
+
+    @mock.patch('enterprise_access.apps.api_client.base_oauth.OAuthAPIClient')
+    def test_max_retries_on_link_check_error(self, mock_oauth_client):
+        """
+        When get_enterprise_learner_by_email raises HTTPError (e.g. LMS 5xx), the task
+        retries until max retries, then sets the assignment to ERRORED.  This verifies
+        that the deactivation guard does not silently swallow LMS errors and proceed to
+        call create_pending_enterprise_users with a potentially wrong state.
+        """
+        error_response = MockResponse({'detail': 'Service Unavailable'}, status.HTTP_503_SERVICE_UNAVAILABLE)
+        self.mock_get_enterprise_learner_by_email.side_effect = HTTPError(response=error_response)
+
+        task_result = create_pending_enterprise_learner_for_assignment_task.delay(self.assignment.uuid)
+
+        assert task_result.state == celery_states.FAILURE
+        assert isinstance(task_result.result, HTTPError)
+
+        # Called once per attempt: 1 initial + max retries
+        assert self.mock_get_enterprise_learner_by_email.call_count == 1 + settings.TASK_MAX_RETRIES
+        # create_pending_enterprise_users must never be called when the link check errors
+        mock_oauth_client.return_value.post.assert_not_called()
+
+        self.assignment.refresh_from_db()
+        assert self.assignment.state == LearnerContentAssignmentStateChoices.ERRORED
+        action = self.assignment.actions.filter(action_type=AssignmentActions.LEARNER_LINKED).first()
+        self.assertIsNotNone(action)
+        self.assertEqual(action.error_reason, AssignmentActionErrors.INTERNAL_API_ERROR)
+
 
 @ddt.ddt
 class TestBrazeEmailTasks(APITestWithMocks):