From 220b22d6d66b88ad7418b54e2f31b8f92a854265 Mon Sep 17 00:00:00 2001 From: Troy Sankey Date: Wed, 4 Mar 2026 07:04:32 -0800 Subject: [PATCH] feat: add logistration context and post-login redirect pipeline steps ENT-11568 --- enterprise/filters/__init__.py | 3 + enterprise/filters/logistration.py | 89 ++++++++++ tests/filters/__init__.py | 3 + tests/filters/test_logistration.py | 268 +++++++++++++++++++++++++++++ 4 files changed, 363 insertions(+) create mode 100644 enterprise/filters/__init__.py create mode 100644 enterprise/filters/logistration.py create mode 100644 tests/filters/__init__.py create mode 100644 tests/filters/test_logistration.py diff --git a/enterprise/filters/__init__.py b/enterprise/filters/__init__.py new file mode 100644 index 000000000..3e0cc597b --- /dev/null +++ b/enterprise/filters/__init__.py @@ -0,0 +1,3 @@ +""" +Filter pipeline step implementations for edx-enterprise. +""" diff --git a/enterprise/filters/logistration.py b/enterprise/filters/logistration.py new file mode 100644 index 000000000..7e181bb0f --- /dev/null +++ b/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} diff --git a/tests/filters/__init__.py b/tests/filters/__init__.py new file mode 100644 index 000000000..ec7a0f9cd --- /dev/null +++ b/tests/filters/__init__.py @@ -0,0 +1,3 @@ +""" +Tests for enterprise filter pipeline steps. +""" diff --git a/tests/filters/test_logistration.py b/tests/filters/test_logistration.py new file mode 100644 index 000000000..1092a24bd --- /dev/null +++ b/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')