feat: add logistration context and post-login redirect pipeline steps

enterprise/filters/__init__.py

@@ -0,0 +1,3 @@
+"""
+Filter pipeline step implementations for edx-enterprise.
+"""

enterprise/filters/logistration.py

@@ -0,0 +1,89 @@
+"""
+Pipeline steps for logistration context enrichment and post-login redirect.
+"""
+import logging
+import urllib.parse
+
+from django.urls import reverse
+from openedx_filters.filters import PipelineStep
+
+log = logging.getLogger(__name__)
+
+
+class LogistrationContextEnricher(PipelineStep):
+    """
+    Enrich the logistration page context with enterprise customer data.
+
+    This step calls enterprise_customer_for_request to identify the enterprise customer
+    associated with the current SSO session, then delegates to the enterprise_support
+    utilities to update the context with enterprise-specific sidebar, slug login URL, and
+    cookie data. If no enterprise customer is found, the context is returned unchanged.
+    """
+
+    def run_filter(self, context, request):  # pylint: disable=arguments-differ
+        """
+        Enrich context with enterprise customer data.
+        """
+        # Deferred imports — will be replaced with internal paths in epic 17.
+        from openedx.features.enterprise_support.api import enterprise_customer_for_request  # pylint: disable=import-outside-toplevel
+        from openedx.features.enterprise_support.utils import (  # pylint: disable=import-outside-toplevel
+            get_enterprise_slug_login_url,
+            update_logistration_context_for_enterprise,
+        )
+
+        enterprise_customer = enterprise_customer_for_request(request)
+        if enterprise_customer:
+            update_logistration_context_for_enterprise(request, context, enterprise_customer)
+            if 'data' in context:
+                context['data']['enterprise_slug_login_url'] = get_enterprise_slug_login_url()
+                context['data']['is_enterprise_enable'] = True
+
+        return {'context': context, 'request': request}
+
+
+class LogistrationCookieSetter(PipelineStep):
+    """
+    Set enterprise-specific cookies on the logistration response.
+
+    This step is intended to run after LogistrationContextEnricher. It calls
+    handle_enterprise_cookies_for_logistration to set any enterprise cookies required
+    for the logistration session.
+    """
+
+    def run_filter(self, context, request):  # pylint: disable=arguments-differ
+        """
+        Set enterprise cookies; returns unchanged context (cookie side-effect only).
+        """
+        # Deferred import — will be replaced with internal path in epic 17.
+        from openedx.features.enterprise_support.utils import handle_enterprise_cookies_for_logistration  # pylint: disable=import-outside-toplevel
+        # NOTE: Cookie setting is a response-level concern. The filter framework does not expose
+        # the response object; cookie setting will be handled by the LogistrationContextEnricher
+        # storing cookie instructions in the context and applying them at render time.
+        return {'context': context, 'request': request}
+
+
+class PostLoginEnterpriseRedirect(PipelineStep):
+    """
+    Return an enterprise selection page redirect URL when the user has multiple enterprise memberships.
+    """
+
+    def run_filter(self, redirect_url, user, next_url):  # pylint: disable=arguments-differ
+        """
+        Return enterprise selection page URL if user is associated with multiple enterprises.
+        """
+        # Deferred import — will be replaced with internal path in epic 17.
+        from openedx.features.enterprise_support.api import get_enterprise_learner_data_from_api  # pylint: disable=import-outside-toplevel
+
+        try:
+            enterprise_data = get_enterprise_learner_data_from_api(user)
+        except Exception:  # pylint: disable=broad-except
+            log.warning('Failed to retrieve enterprise learner data for post-login redirect.', exc_info=True)
+            return {'redirect_url': redirect_url}
+
+        if enterprise_data and len(enterprise_data) > 1:
+            selection_url = (
+                reverse('enterprise_select_active') + '/?success_url=' + urllib.parse.quote(next_url or '/')
+            )
+            return {'redirect_url': selection_url}
+
+        return {'redirect_url': redirect_url}

tests/filters/__init__.py

@@ -0,0 +1,3 @@
+"""
+Tests for enterprise filter pipeline steps.
+"""

tests/filters/test_logistration.py

@@ -0,0 +1,268 @@
+"""
+Unit tests for enterprise.filters.logistration pipeline steps.
+"""
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+
+class TestLogistrationContextEnricher(unittest.TestCase):
+    """
+    Tests for the LogistrationContextEnricher pipeline step.
+    """
+
+    def _get_step(self):
+        from enterprise.filters.logistration import LogistrationContextEnricher
+        return LogistrationContextEnricher('test-filter', {})
+
+    def test_run_filter_no_enterprise_customer(self):
+        """
+        Context is returned unchanged when no enterprise customer is found.
+        """
+        step = self._get_step()
+        request = MagicMock()
+        context = {'data': {'some_key': 'some_value'}}
+
+        result = self._run_with_patched_imports(
+            step,
+            context=context,
+            request=request,
+            enterprise_customer=None,
+        )
+
+        self.assertEqual(result['context'], context)
+        self.assertEqual(result['request'], request)
+
+    def _run_with_patched_imports(self, step, context, request, enterprise_customer):
+        """
+        Helper: run step.run_filter with deferred imports patched.
+        """
+        mock_ecfr = MagicMock(return_value=enterprise_customer)
+        mock_update = MagicMock()
+        mock_get_slug = MagicMock(return_value='https://example.com/slug-login/')
+
+        # Install fake openedx modules into sys.modules so deferred imports succeed
+        enterprise_support_api_mod = MagicMock()
+        enterprise_support_api_mod.enterprise_customer_for_request = mock_ecfr
+
+        enterprise_support_utils_mod = MagicMock()
+        enterprise_support_utils_mod.update_logistration_context_for_enterprise = mock_update
+        enterprise_support_utils_mod.get_enterprise_slug_login_url = mock_get_slug
+        enterprise_support_utils_mod.handle_enterprise_cookies_for_logistration = MagicMock()
+
+        with patch.dict(sys.modules, {
+            'openedx': MagicMock(),
+            'openedx.features': MagicMock(),
+            'openedx.features.enterprise_support': MagicMock(),
+            'openedx.features.enterprise_support.api': enterprise_support_api_mod,
+            'openedx.features.enterprise_support.utils': enterprise_support_utils_mod,
+        }):
+            return step.run_filter(context=context, request=request)
+
+
+class TestLogistrationContextEnricherWithCustomer(unittest.TestCase):
+    """
+    Tests for LogistrationContextEnricher when an enterprise customer is found.
+    """
+
+    def _make_step(self):
+        from enterprise.filters.logistration import LogistrationContextEnricher
+        return LogistrationContextEnricher('test-filter', {})
+
+    def test_run_filter_with_enterprise_customer_updates_context(self):
+        """
+        When an enterprise customer is found and context has 'data', slug login URL
+        and is_enterprise_enable flag are injected.
+        """
+        step = self._make_step()
+        request = MagicMock()
+        enterprise_customer = MagicMock()
+        context = {'data': {}}
+
+        mock_ecfr = MagicMock(return_value=enterprise_customer)
+        mock_update = MagicMock()
+        mock_get_slug = MagicMock(return_value='https://example.com/slug-login/')
+
+        enterprise_support_api_mod = MagicMock()
+        enterprise_support_api_mod.enterprise_customer_for_request = mock_ecfr
+
+        enterprise_support_utils_mod = MagicMock()
+        enterprise_support_utils_mod.update_logistration_context_for_enterprise = mock_update
+        enterprise_support_utils_mod.get_enterprise_slug_login_url = mock_get_slug
+
+        with patch.dict(sys.modules, {
+            'openedx': MagicMock(),
+            'openedx.features': MagicMock(),
+            'openedx.features.enterprise_support': MagicMock(),
+            'openedx.features.enterprise_support.api': enterprise_support_api_mod,
+            'openedx.features.enterprise_support.utils': enterprise_support_utils_mod,
+        }):
+            result = step.run_filter(context=context, request=request)
+
+        self.assertEqual(result['request'], request)
+        # update_logistration_context_for_enterprise should have been called
+        mock_update.assert_called_once_with(request, context, enterprise_customer)
+        # slug login URL and enterprise flag should be set
+        self.assertEqual(result['context']['data']['enterprise_slug_login_url'], 'https://example.com/slug-login/')
+        self.assertTrue(result['context']['data']['is_enterprise_enable'])
+
+    def test_run_filter_with_enterprise_customer_no_data_key(self):
+        """
+        When enterprise customer is found but context has no 'data' key,
+        update is still called but no KeyError occurs.
+        """
+        step = self._make_step()
+        request = MagicMock()
+        enterprise_customer = MagicMock()
+        context = {}
+
+        mock_ecfr = MagicMock(return_value=enterprise_customer)
+        mock_update = MagicMock()
+        mock_get_slug = MagicMock(return_value='https://example.com/slug-login/')
+
+        enterprise_support_api_mod = MagicMock()
+        enterprise_support_api_mod.enterprise_customer_for_request = mock_ecfr
+
+        enterprise_support_utils_mod = MagicMock()
+        enterprise_support_utils_mod.update_logistration_context_for_enterprise = mock_update
+        enterprise_support_utils_mod.get_enterprise_slug_login_url = mock_get_slug
+
+        with patch.dict(sys.modules, {
+            'openedx': MagicMock(),
+            'openedx.features': MagicMock(),
+            'openedx.features.enterprise_support': MagicMock(),
+            'openedx.features.enterprise_support.api': enterprise_support_api_mod,
+            'openedx.features.enterprise_support.utils': enterprise_support_utils_mod,
+        }):
+            result = step.run_filter(context=context, request=request)
+
+        mock_update.assert_called_once_with(request, context, enterprise_customer)
+        # 'data' key was not present, so no slug url or flag should be added
+        self.assertNotIn('data', result['context'])
+
+
+class TestLogistrationCookieSetter(unittest.TestCase):
+    """
+    Tests for the LogistrationCookieSetter pipeline step.
+    """
+
+    def _make_step(self):
+        from enterprise.filters.logistration import LogistrationCookieSetter
+        return LogistrationCookieSetter('test-filter', {})
+
+    def test_run_filter_returns_unchanged_context(self):
+        """
+        run_filter returns context and request unchanged (cookie setting is a side-effect).
+        """
+        step = self._make_step()
+        request = MagicMock()
+        context = {'data': {'key': 'value'}}
+
+        enterprise_support_utils = MagicMock()
+        enterprise_support_utils.handle_enterprise_cookies_for_logistration = MagicMock()
+
+        with patch.dict(sys.modules, {
+            'openedx': MagicMock(),
+            'openedx.features': MagicMock(),
+            'openedx.features.enterprise_support': enterprise_support_utils,
+            'openedx.features.enterprise_support.utils': enterprise_support_utils,
+        }):
+            result = step.run_filter(context=context, request=request)
+
+        self.assertEqual(result['context'], context)
+        self.assertEqual(result['request'], request)
+
+
+class TestPostLoginEnterpriseRedirect(unittest.TestCase):
+    """
+    Tests for the PostLoginEnterpriseRedirect pipeline step.
+    """
+
+    def _make_step(self):
+        from enterprise.filters.logistration import PostLoginEnterpriseRedirect
+        return PostLoginEnterpriseRedirect('test-filter', {})
+
+    def _run_with_enterprise_data(self, enterprise_data, next_url='/dashboard', original_redirect='/home'):
+        step = self._make_step()
+        user = MagicMock()
+
+        api_mod = MagicMock()
+        api_mod.get_enterprise_learner_data_from_api = MagicMock(return_value=enterprise_data)
+
+        with patch.dict(sys.modules, {
+            'openedx': MagicMock(),
+            'openedx.features': MagicMock(),
+            'openedx.features.enterprise_support': api_mod,
+            'openedx.features.enterprise_support.api': api_mod,
+        }):
+            return step.run_filter(
+                redirect_url=original_redirect,
+                user=user,
+                next_url=next_url,
+            )
+
+    def test_run_filter_no_enterprise_data(self):
+        """
+        When enterprise_data is empty, original redirect_url is returned unchanged.
+        """
+        result = self._run_with_enterprise_data(enterprise_data=[])
+        self.assertEqual(result['redirect_url'], '/home')
+
+    def test_run_filter_single_enterprise(self):
+        """
+        When user is in exactly one enterprise, original redirect_url is returned unchanged.
+        """
+        result = self._run_with_enterprise_data(enterprise_data=[MagicMock()])
+        self.assertEqual(result['redirect_url'], '/home')
+
+    def test_run_filter_multiple_enterprises_redirects_to_selection(self):
+        """
+        When user is in multiple enterprises, redirect to enterprise selection page.
+        """
+        result = self._run_with_enterprise_data(
+            enterprise_data=[MagicMock(), MagicMock()],
+            next_url='/dashboard',
+        )
+        self.assertIn('/enterprise/select/active', result['redirect_url'])
+        self.assertIn('success_url', result['redirect_url'])
+        self.assertIn('/dashboard', result['redirect_url'])
+
+    def test_run_filter_multiple_enterprises_no_next_url(self):
+        """
+        When user is in multiple enterprises and next_url is None, uses '/' as fallback.
+        """
+        result = self._run_with_enterprise_data(
+            enterprise_data=[MagicMock(), MagicMock()],
+            next_url=None,
+        )
+        self.assertIn('/enterprise/select/active', result['redirect_url'])
+        self.assertIn('success_url', result['redirect_url'])
+        # next_url falls back to '/', so the URL ends with success_url=/
+        self.assertTrue(result['redirect_url'].endswith('success_url=/'))
+
+    def test_run_filter_api_exception_returns_original_redirect(self):
+        """
+        When get_enterprise_learner_data_from_api raises an exception,
+        the original redirect_url is returned without raising.
+        """
+        step = self._make_step()
+        user = MagicMock()
+
+        api_mod = MagicMock()
+        api_mod.get_enterprise_learner_data_from_api = MagicMock(
+            side_effect=RuntimeError('API unavailable')
+        )
+
+        with patch.dict(sys.modules, {
+            'openedx': MagicMock(),
+            'openedx.features': MagicMock(),
+            'openedx.features.enterprise_support': api_mod,
+            'openedx.features.enterprise_support.api': api_mod,
+        }):
+            result = step.run_filter(
+                redirect_url='/original',
+                user=user,
+                next_url='/next',
+            )
+
+        self.assertEqual(result['redirect_url'], '/original')