diff --git a/nau_openedx_extensions/partner_integration/course_filters.py b/nau_openedx_extensions/partner_integration/course_filters.py new file mode 100644 index 0000000..f64897e --- /dev/null +++ b/nau_openedx_extensions/partner_integration/course_filters.py @@ -0,0 +1,142 @@ +""" +SSO Partner Integration course enrollment filters. + +This module contains filters for validating SSO partner linkage during +course enrollment. +""" + +import logging + +from django.apps import apps +from openedx_filters import PipelineStep +from openedx_filters.learning.filters import CourseEnrollmentStarted + +logger = logging.getLogger(__name__) + + +class FilterSSOPartnerAccountLink(PipelineStep): + """ + Validate that a user attempting to enroll has a completed SSO account link + with a partner that has access to the course. + + This filter checks: + 1. If the user has an SSOPartnerIntegration record (linked account). + 2. If the partner client's base_security_scope allows access to the course. + + If either check fails, enrollment is prevented with a Portuguese error message. + + The partner's base_security_scope defines which courses they can access, + and is validated against the CourseOverview model fields. + + Example usage: + + Add the following configurations to your configuration file: + + "OPEN_EDX_FILTERS_CONFIG": { + "org.openedx.learning.course.enrollment.started.v1": { + "fail_silently": false, + "pipeline": [ + "nau_openedx_extensions.partner_integration.course_filters.FilterSSOPartnerAccountLink" + ] + } + } + """ + + def run_filter(self, user, course_key, mode): # pylint: disable=unused-argument, arguments-differ + """ + Filter implementation. + + Validates SSO partner account link and partner course access. + + Args: + user: The user attempting to enroll. + course_key: The course key for the course being enrolled in. + mode: The enrollment mode (unused). + + Returns: + Empty dict if validation passes. + + Raises: + CourseEnrollmentStarted.PreventEnrollment: If user is not linked to a partner + or the partner doesn't have access to the course. + """ + try: + SSOPartnerIntegration = apps.get_model( + "nau_openedx_extensions", + "SSOPartnerIntegration" + ) + except LookupError: + logger.error("SSOPartnerIntegration model not found") + return {} + + try: + sso_record = SSOPartnerIntegration.objects.get(user=user) + except SSOPartnerIntegration.DoesNotExist as exc: + logger.warning( + f"User {user.id} ({user.username}) has no SSO partner integration record" + ) + exception_msg = ( + "A ligação de conta com a plataforma do parceiro não foi concluída. " + "Por favor, complete o processo de ligação de conta antes de se inscrever." + ) + raise CourseEnrollmentStarted.PreventEnrollment(exception_msg) from exc + + partner_client = sso_record.partner_client + base_security_scope = partner_client.query_security_scope.get("base_security_scope", {}) + + if not base_security_scope: + logger.warning( + f"Partner {partner_client.name} has no base_security_scope configured" + ) + exception_msg = ( + "O parceiro de integração não tem acesso configurado para nenhum curso. " + "Por favor, contacte o suporte." + ) + raise CourseEnrollmentStarted.PreventEnrollment(exception_msg) + + if not FilterSSOPartnerAccountLink._is_course_allowed_for_partner(course_key, base_security_scope): + logger.warning( + f"Partner {partner_client.name} does not have access to course {course_key}" + ) + exception_msg = ( + "O parceiro de integração não tem permissão para inscrever utilizadores " + "neste curso. Por favor, contacte o suporte." + ) + raise CourseEnrollmentStarted.PreventEnrollment(exception_msg) + + logger.info( + f"User {user.id} ({user.username}) validated for enrollment in {course_key} " + f"via partner {partner_client.name}" + ) + return {} + + @staticmethod + def _is_course_allowed_for_partner(course_key, base_security_scope): + """ + Check if a course is allowed by the partner's base_security_scope. + + The base_security_scope uses Django ORM lookup syntax to filter CourseOverview. + This method applies the security scope filters to determine if the course is allowed. + + Args: + course_key: The course key to validate. + base_security_scope: A dictionary of Django ORM filters for CourseOverview. + + Returns: + True if the course matches the security scope, False otherwise. + """ + try: + CourseOverview = apps.get_model("course_overviews", "CourseOverview") + except LookupError: + logger.error("CourseOverview model not found") + return False + + try: + query = CourseOverview.objects.filter(**base_security_scope) + course_exists = query.filter(id=course_key).exists() + return course_exists + except Exception as e: # pylint: disable=broad-except + logger.error( + f"Error validating course {course_key} against security scope: {e}" + ) + return False diff --git a/nau_openedx_extensions/partner_integration/docs/COURSE_FILTER.md b/nau_openedx_extensions/partner_integration/docs/COURSE_FILTER.md new file mode 100644 index 0000000..8fb8b42 --- /dev/null +++ b/nau_openedx_extensions/partner_integration/docs/COURSE_FILTER.md @@ -0,0 +1,240 @@ +# SSO Partner Enrollment Filter + +## Overview + +The `FilterSSOPartnerAccountLink` is an Open edX enrollment filter that validates whether users attempting to enroll in courses have: + +1. **Completed SSO account linking** - User has a `SSOPartnerIntegration` record linking them to a partner account +2. **Partner has course access** - The partner's `base_security_scope` grants access to the course being enrolled in + +This filter ensures that only users with valid partner account links can enroll in courses authorized for that partner. + +## Implementation Details + +**File:** [`nau_openedx_extensions/partner_integration/course_filters.py`](./course_filters.py) + +**Hook:** `org.openedx.learning.course.enrollment.started.v1` + +The filter integrates with the Open edX Filter Framework using the `PipelineStep` base class and prevents enrollment by raising `CourseEnrollmentStarted.PreventEnrollment` with Portuguese error messages. + +## How It Works + +### Validation Flow + +``` +User attempts enrollment + ↓ +Does user have SSOPartnerIntegration? + ├─ NO → Raise error: "Account link not completed" + └─ YES → Continue + ↓ + Does partner have base_security_scope? + ├─ NO → Raise error: "Partner has no access configured" + └─ YES → Continue + ↓ + Is course allowed by partner's scope? + ├─ NO → Raise error: "Partner doesn't have permission for this course" + └─ YES → Enrollment allowed ✓ +``` + +### Security Scope Matching + +The filter uses Django ORM lookup syntax to match courses against the partner's `base_security_scope`: + +```python +base_security_scope = {"org": "nau"} # Allows all courses in "nau" org +base_security_scope = {"org__in": ["nau", "fccn"]} # Allows multiple orgs +base_security_scope = {"id__in": ["course-v1:org+ABC+2024"]} # Specific courses +base_security_scope = {"display_name__icontains": "python"} # Name-based +``` + +The filter applies these filters to `CourseOverview` to determine access. + +## Configuration + +### Add to OPEN_EDX_FILTERS_CONFIG + +In your Django settings file (`lms.env.yml` or equivalent), add the filter to the enrollment pipeline: + +```yaml +OPEN_EDX_FILTERS_CONFIG: + org.openedx.learning.course.enrollment.started.v1: + fail_silently: false + pipeline: + - "nau_openedx_extensions.filters.pipeline.FilterEnrollmentByDomain" + - "nau_openedx_extensions.filters.pipeline.FilterEnrollmentRequireNIF" + - "nau_openedx_extensions.partner_integration.course_filters.FilterSSOPartnerAccountLink" +``` + +Or in JSON format: + +```json +{ + "OPEN_EDX_FILTERS_CONFIG": { + "org.openedx.learning.course.enrollment.started.v1": { + "fail_silently": false, + "pipeline": [ + "nau_openedx_extensions.filters.pipeline.FilterEnrollmentByDomain", + "nau_openedx_extensions.filters.pipeline.FilterEnrollmentRequireNIF", + "nau_openedx_extensions.partner_integration.course_filters.FilterSSOPartnerAccountLink" + ] + } + } +} +``` + +### Set fail_silently + +- **`fail_silently: false`** (Recommended) - Any error in the filter chain blocks enrollment +- **`fail_silently: true`** - Errors are logged but don't block enrollment + +## Error Messages + +The filter provides clear Portuguese (pt-PT) error messages: + +### No SSO Integration +> "A ligação de conta com a plataforma do parceiro não foi concluída. Por favor, complete o processo de ligação de conta antes de se inscrever." + +Translation: "The account link with the partner platform has not been completed. Please complete the account linking process before enrolling." + +### No Security Scope Configured +> "O parceiro de integração não tem acesso configurado para nenhum curso. Por favor, contacte o suporte." + +Translation: "The integration partner has no access configured for any course. Please contact support." + +### Course Not Allowed +> "O parceiro de integração não tem permissão para inscrever utilizadores neste curso. Por favor, contacte o suporte." + +Translation: "The integration partner does not have permission to enroll users in this course. Please contact support." + +## Database Models + +The filter uses these related models: + +### SSOPartnerIntegration +Links a local user to a partner account: +```python +{ + "user": User, # The local edX user + "partner_client": PartnerAPIClient, # The partner + "external_user_id": "str" # Partner's user identifier +} +``` + +### PartnerAPIClient +Represents a partner client with access control: +```python +{ + "name": "str", # Partner name + "client_id": "UUID", + "query_security_scope": { + "base_security_scope": { + # Django ORM filters against CourseOverview + }, + "base_certificates_scope": { + # Django ORM filters against GeneratedCertificate + } + } +} +``` + +## Testing + +Comprehensive tests are provided in [`tests/test_course_filters.py`](./tests/test_course_filters.py). + +### Test Coverage + +1. ✅ Valid SSO record with allowed course +2. ✅ User has no SSO record → error +3. ✅ Partner has no security scope → error +4. ✅ Course not in partner's scope → error +5. ✅ Multiple orgs in scope +6. ✅ Specific course IDs in scope +7. ✅ Error handling and logging + +### Running Tests + +```bash +# Run all filter tests +cd /openedx/nau-openedx-extensions +python -m pytest nau_openedx_extensions/partner_integration/tests/test_course_filters.py -v + +# Run specific test +python -m pytest nau_openedx_extensions/partner_integration/tests/test_course_filters.py::FilterSSOPartnerAccountLinkTests::test_filter_passes_when_user_has_valid_sso_record_and_course_allowed -v +``` + +## Logging + +The filter logs important events for observability: + +```python +# Info: Successful validation +"User 123 (john.doe) validated for enrollment in course-v1:nau+ABC+2024 via partner NAU" + +# Warning: No SSO record +"User 123 (john.doe) has no SSO partner integration record" + +# Warning: No access +"Partner NAU does not have access to course course-v1:different+XYZ+2024" + +# Error: Configuration/database issues +"Error validating course course-v1:nau+ABC+2024 against security scope: ..." +``` + +All logs use the logger: `nau_openedx_extensions.partner_integration.course_filters` + +## Integration with SSO Flow + +This filter works alongside the SSO integration module: + +1. **SSO link creation** - `CustomAuthorizationView` creates `SSOPartnerIntegration` records +2. **Enrollment validation** - This filter validates those records during enrollment +3. **Access control** - Only allows enrollment if partner has been granted access to the course + +## Performance Considerations + +The filter runs Django ORM queries to: +1. Lookup `SSOPartnerIntegration` by user (indexed query) +2. Check course access using `base_security_scope` filters + +Both operations are efficient and use database indexes on: +- `SSOPartnerIntegration.user` (OneToOne) +- `CourseOverview.org`, `CourseOverview.id` (indexed fields) + +## Backwards Compatibility + +The filter is **opt-in** and only affects enrollment when: +1. The filter is added to the `OPEN_EDX_FILTERS_CONFIG` +2. The user has an `SSOPartnerIntegration` record + +Existing users without SSO integration records are not affected. + +## Troubleshooting + +### Filter Not Running + +Check that: +1. Filter is in `OPEN_EDX_FILTERS_CONFIG` pipeline +2. `fail_silently` is set appropriately +3. Settings have been reloaded (restart LMS/CMS) + +### All Users Blocked + +Check: +1. User has `SSOPartnerIntegration` record (Django admin) +2. Partner's `base_security_scope` is configured +3. Course org/ID matches security scope + +### Database Errors + +Check: +1. `SSOPartnerIntegration` table exists (run migrations) +2. User and partner client records exist +3. Database connection is working + +## See Also + +- [Partner Integration Module README](./README.md) +- [SSOPartnerIntegration Model](./models.py) +- [PartnerAPIClient Configuration](./README.md#configuring-base_security_scope-and-base_certificates_scope) +- [Open edX Filters Documentation](https://github.com/openedx/openedx-filters) diff --git a/nau_openedx_extensions/partner_integration/README.md b/nau_openedx_extensions/partner_integration/docs/README.md similarity index 100% rename from nau_openedx_extensions/partner_integration/README.md rename to nau_openedx_extensions/partner_integration/docs/README.md diff --git a/nau_openedx_extensions/partner_integration/exception.py b/nau_openedx_extensions/partner_integration/exception.py index 235ef8f..35aa744 100644 --- a/nau_openedx_extensions/partner_integration/exception.py +++ b/nau_openedx_extensions/partner_integration/exception.py @@ -88,3 +88,21 @@ def __init__(self, message=None): self.message = message else: self.message = "You are not allowed to get information about this course." + + +class PartnerIntegrationEnrollmentPreventedException(Exception): + """ + Custom exception for filter-based enrollment rejections. + + Raised when an Open edX enrollment filter (e.g., FilterSSOPartnerAccountLink) + prevents enrollment via CourseEnrollmentStarted.PreventEnrollment. + """ + + def __init__(self, message=None): + if message: + self.message = message + else: + self.message = ( + "Enrollment was prevented by the enrollment filter pipeline. " + "Please verify the user's account configuration." + ) diff --git a/nau_openedx_extensions/partner_integration/facade.py b/nau_openedx_extensions/partner_integration/facade.py index 61180eb..d2bafcc 100644 --- a/nau_openedx_extensions/partner_integration/facade.py +++ b/nau_openedx_extensions/partner_integration/facade.py @@ -3,6 +3,7 @@ import logging from datetime import datetime, time, timedelta +from common.djangoapps.student.models import EnrollmentNotAllowed from django.contrib.auth import get_user_model from django.db.models import OuterRef, Q, Subquery from lms.djangoapps.course_blocks.api import get_course_blocks # pylint: disable=unused-import @@ -24,6 +25,7 @@ from nau_openedx_extensions.partner_integration.exception import ( PartnerIntegrationCourseOwnerException, PartnerIntegrationDataConflictException, + PartnerIntegrationEnrollmentPreventedException, PartnerIntegrationInternalErrorException, PartnerIntegrationInvalidDataProvidedException, ) @@ -327,6 +329,12 @@ def enroll_user(self, query_security_scope, course_id, nif, email, username): # raise e except PartnerIntegrationInternalErrorException as e: raise e + except EnrollmentNotAllowed as e: + logger.warning( + "EnrollmentFacade: Enrollment prevented by filter pipeline: %s", + str(e), + ) + raise PartnerIntegrationEnrollmentPreventedException(str(e)) from e except Exception as e: logger.error("EnrollmentFacade: Internal error during enrollment.", exc_info=e) raise PartnerIntegrationInternalErrorException() from e diff --git a/nau_openedx_extensions/partner_integration/tests/test_course_filters.py b/nau_openedx_extensions/partner_integration/tests/test_course_filters.py new file mode 100644 index 0000000..f08ec24 --- /dev/null +++ b/nau_openedx_extensions/partner_integration/tests/test_course_filters.py @@ -0,0 +1,447 @@ +"""Tests for SSO Partner Integration course enrollment filters.""" + +from unittest.mock import MagicMock, patch + +from common.djangoapps.student.tests.factories import UserFactory +from django.test import TestCase, TransactionTestCase, override_settings +from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory +from openedx_filters.learning.filters import CourseEnrollmentStarted +from rest_framework import status +from rest_framework.test import APIClient + +from nau_openedx_extensions.custom_registration_form.factories import NauUserExtendedModelFactory +from nau_openedx_extensions.partner_integration.course_filters import FilterSSOPartnerAccountLink +from nau_openedx_extensions.partner_integration.factories import PartnerAPIClientFactory, SSOPartnerIntegrationFactory + + +class FilterSSOPartnerAccountLinkTests(TestCase): + """Test cases for FilterSSOPartnerAccountLink.""" + + def setUp(self): + """Set up test fixtures.""" + self.user = UserFactory() + self.course = CourseOverviewFactory() + + def test_filter_passes_when_user_has_valid_sso_record_and_course_allowed(self): + """Test filter passes when user has SSO record and course is in partner's scope.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": {"org": self.course.org}, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + filter_instance = FilterSSOPartnerAccountLink.run_filter( + None, self.user, self.course.id, "honor" + ) + self.assertEqual(filter_instance, {}) + + def test_filter_raises_when_user_has_no_sso_record(self): + """Test filter raises PreventEnrollment when user has no SSO record.""" + with self.assertRaises(CourseEnrollmentStarted.PreventEnrollment) as context: + FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + + error_message = str(context.exception) + self.assertIn("ligação de conta", error_message) + self.assertIn("parceiro", error_message) + + def test_filter_raises_when_partner_has_no_security_scope(self): + """Test filter raises PreventEnrollment when partner has empty security scope.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": {"org": self.course.org}, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + mock_sso = MagicMock() + mock_sso.partner_client = MagicMock() + mock_sso.partner_client.name = partner.name + mock_sso.partner_client.query_security_scope = {"base_security_scope": {}, "base_certificates_scope": {}} + + with patch("nau_openedx_extensions.partner_integration.course_filters.apps.get_model") as mock_get_model: + mock_model = MagicMock() + mock_model.objects.get.return_value = mock_sso + mock_get_model.return_value = mock_model + + with self.assertRaises(CourseEnrollmentStarted.PreventEnrollment) as context: + FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + + error_message = str(context.exception) + self.assertIn("acesso configurado", error_message) + + def test_filter_raises_when_course_not_in_partner_scope(self): + """Test filter raises PreventEnrollment when course is not allowed by partner.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": {"org": "different_org"}, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + with self.assertRaises(CourseEnrollmentStarted.PreventEnrollment) as context: + FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + + error_message = str(context.exception) + self.assertIn("permissão", error_message) + self.assertIn("curso", error_message) + + def test_filter_passes_with_multiple_orgs_in_scope(self): + """Test filter passes when course org is one of multiple allowed orgs.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": {"org__in": ["org1", self.course.org, "org3"]}, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + result = FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + self.assertEqual(result, {}) + + def test_filter_passes_with_course_id_in_scope(self): + """Test filter passes when specific course IDs are allowed.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": { + "org": self.course.org, + "id__in": [str(self.course.id)] + }, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + result = FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + self.assertEqual(result, {}) + + def test_is_course_allowed_for_partner_with_valid_scope(self): + """Test _is_course_allowed_for_partner with valid scope.""" + base_security_scope = {"org": self.course.org} + result = FilterSSOPartnerAccountLink._is_course_allowed_for_partner( + self.course.id, base_security_scope + ) + self.assertTrue(result) + + def test_is_course_allowed_for_partner_with_invalid_scope(self): + """Test _is_course_allowed_for_partner with invalid scope.""" + base_security_scope = {"org": "non_existent_org"} + result = FilterSSOPartnerAccountLink._is_course_allowed_for_partner( + self.course.id, base_security_scope + ) + self.assertFalse(result) + + def test_is_course_allowed_for_partner_handles_exceptions(self): + """Test _is_course_allowed_for_partner handles exceptions gracefully.""" + with patch("django.apps.apps.get_model") as mock_get_model: + mock_model = MagicMock() + mock_model.objects.filter.side_effect = Exception("Database error") + mock_get_model.return_value = mock_model + + base_security_scope = {"org": self.course.org} + result = FilterSSOPartnerAccountLink._is_course_allowed_for_partner( + self.course.id, base_security_scope + ) + self.assertFalse(result) + + def test_filter_logs_when_user_has_no_sso_record(self): + """Test filter logs warning when user has no SSO record.""" + with patch("nau_openedx_extensions.partner_integration.course_filters.logger") as mock_logger: + with self.assertRaises(CourseEnrollmentStarted.PreventEnrollment): + FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + + mock_logger.warning.assert_called() + call_args = mock_logger.warning.call_args[0][0] + self.assertIn("no SSO partner integration record", call_args) + + def test_filter_logs_when_course_not_allowed(self): + """Test filter logs warning when course is not allowed by partner.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": {"org": "different_org"}, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + with patch("nau_openedx_extensions.partner_integration.course_filters.logger") as mock_logger: + with self.assertRaises(CourseEnrollmentStarted.PreventEnrollment): + FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + + mock_logger.warning.assert_called() + call_args = mock_logger.warning.call_args[0][0] + self.assertIn("does not have access to course", call_args) + + def test_filter_logs_success(self): + """Test filter logs info when validation passes.""" + partner = PartnerAPIClientFactory.create( + query_security_scope={ + "base_security_scope": {"org": self.course.org}, + "base_certificates_scope": {} + } + ) + SSOPartnerIntegrationFactory.create(user=self.user, partner_client=partner) + + with patch("nau_openedx_extensions.partner_integration.course_filters.logger") as mock_logger: + FilterSSOPartnerAccountLink.run_filter(None, self.user, self.course.id, "honor") + + mock_logger.info.assert_called() + call_args = mock_logger.info.call_args[0][0] + self.assertIn("validated for enrollment", call_args) + + +FILTER_PIPELINE_CONFIG = { + "org.openedx.learning.course.enrollment.started.v1": { + "fail_silently": False, + "pipeline": [ + "nau_openedx_extensions.partner_integration.course_filters.FilterSSOPartnerAccountLink" + ] + } +} + + +class FilterSSOPartnerAccountLinkIntegrationTests(TransactionTestCase): + """ + Integration tests for FilterSSOPartnerAccountLink. + + These tests validate the filter works end-to-end when triggered via + the enroll-user API endpoint, with the filter pipeline enabled via + OPEN_EDX_FILTERS_CONFIG. + """ + + def setUp(self): + """Set up test fixtures.""" + self.http_client = APIClient() + self.endpoint = "/nau-openedx-extensions/partner-integration/enroll-user/" + self.auth_endpoint = "/nau-openedx-extensions/partner-integration/auth-token/" + + # Create a partner client with a known password and valid security scope + self.course = CourseOverviewFactory() + self.partner_client = PartnerAPIClientFactory.create( + is_active=True, + query_security_scope={ + "base_security_scope": {"org": self.course.org}, + "base_certificates_scope": {} + } + ) + self.partner_client.password = "integration_test_password" + self.partner_client.save() + + def _authenticate(self): + """Authenticate partner client and return access token.""" + self.http_client.credentials( + HTTP_AUTHORIZATION="Token integration_test_password", + HTTP_X_CLIENT_ID=self.partner_client.client_id, + ) + response = self.http_client.post(self.auth_endpoint, format="json") + assert response.status_code == 200, f"Auth failed: {response.data}" + return response.data["access_token"] + + def _enroll_via_api(self, access_token, course_id, email): + """Call the enroll-user API endpoint.""" + self.http_client.credentials(HTTP_AUTHORIZATION=f"Bearer {access_token}") + return self.http_client.post( + self.endpoint, + data={"course": str(course_id), "email": email}, + format="json", + ) + + @override_settings(OPEN_EDX_FILTERS_CONFIG=FILTER_PIPELINE_CONFIG) + def test_api_enrollment_blocked_when_user_has_no_sso_record(self): + """ + Integration test: enrollment via API is blocked when the user has no + SSOPartnerIntegration record. + + The filter raises PreventEnrollment, which propagates through + enrollment_api.add_enrollment() and is caught by the facade, + resulting in a 403 Forbidden response with the filter's error message. + """ + access_token = self._authenticate() + + external_user = NauUserExtendedModelFactory.create() + external_user.user.email = "no_sso_record@example.com" + external_user.user.save() + + response = self._enroll_via_api( + access_token, self.course.id, external_user.user.email + ) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertIn( + "A ligação de conta com a plataforma do parceiro não foi concluída", + response.data["error"] + ) + + @override_settings(OPEN_EDX_FILTERS_CONFIG=FILTER_PIPELINE_CONFIG) + def test_api_enrollment_blocked_when_partner_has_empty_security_scope(self): + """ + Integration test: enrollment via API is blocked when the partner + has an empty base_security_scope. + + A separate partner with empty scope is used to create the SSO record, + but this test calls the enroll endpoint authenticated as the main + partner. The filter checks the SSO record's partner, not the + authenticated partner. + """ + access_token = self._authenticate() + + # Create a different partner with empty security scope + empty_scope_partner = PartnerAPIClientFactory.create( + is_active=True, + query_security_scope={ + "base_security_scope": {"org": self.course.org}, + "base_certificates_scope": {} + } + ) + # Manually clear the scope after creation (bypassing validation) + from nau_openedx_extensions.partner_integration.models import PartnerAPIClient + PartnerAPIClient.objects.filter(pk=empty_scope_partner.pk).update( + query_security_scope={"base_security_scope": {}, "base_certificates_scope": {}} + ) + empty_scope_partner.refresh_from_db() + + external_user = NauUserExtendedModelFactory.create() + external_user.user.email = "empty_scope@example.com" + external_user.user.save() + + SSOPartnerIntegrationFactory.create( + user=external_user.user, + partner_client=empty_scope_partner + ) + + response = self._enroll_via_api( + access_token, self.course.id, external_user.user.email + ) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertIn( + "O parceiro de integração não tem acesso configurado para nenhum curso", + response.data["error"] + ) + + @override_settings(OPEN_EDX_FILTERS_CONFIG=FILTER_PIPELINE_CONFIG) + def test_api_enrollment_blocked_when_course_not_in_partner_scope(self): + """ + Integration test: enrollment via API is blocked when the course + is not within the SSO partner's base_security_scope. + """ + access_token = self._authenticate() + + # Create a partner with a different org scope + different_org_partner = PartnerAPIClientFactory.create( + is_active=True, + query_security_scope={ + "base_security_scope": {"org": "completely_different_org"}, + "base_certificates_scope": {} + } + ) + + external_user = NauUserExtendedModelFactory.create() + external_user.user.email = "wrong_scope@example.com" + external_user.user.save() + + SSOPartnerIntegrationFactory.create( + user=external_user.user, + partner_client=different_org_partner + ) + + response = self._enroll_via_api( + access_token, self.course.id, external_user.user.email + ) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertIn( + "O parceiro de integração não tem permissão para inscrever utilizadores neste curso", + response.data["error"] + ) + + @override_settings(OPEN_EDX_FILTERS_CONFIG=FILTER_PIPELINE_CONFIG) + def test_api_enrollment_succeeds_when_all_conditions_met(self): + """ + Integration test: enrollment via API succeeds when the user has a + valid SSO record and the partner's scope includes the course. + """ + access_token = self._authenticate() + + external_user = NauUserExtendedModelFactory.create() + external_user.user.email = "valid_sso@example.com" + external_user.user.save() + + SSOPartnerIntegrationFactory.create( + user=external_user.user, + partner_client=self.partner_client + ) + + response = self._enroll_via_api( + access_token, self.course.id, external_user.user.email + ) + + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + self.assertEqual(response.data["course_id"], str(self.course.id)) + self.assertEqual(response.data["user_email"], "valid_sso@example.com") + + @override_settings(OPEN_EDX_FILTERS_CONFIG=FILTER_PIPELINE_CONFIG) + def test_api_enrollment_succeeds_with_multiple_orgs_in_scope(self): + """ + Integration test: enrollment via API succeeds when the partner has + org__in scope containing the course's org among multiple orgs. + """ + multi_org_partner = PartnerAPIClientFactory.create( + is_active=True, + query_security_scope={ + "base_security_scope": { + "org__in": ["org_alpha", self.course.org, "org_gamma"] + }, + "base_certificates_scope": {} + } + ) + multi_org_partner.password = "integration_test_password" + multi_org_partner.save() + + # Authenticate as the multi-org partner + self.http_client.credentials( + HTTP_AUTHORIZATION="Token integration_test_password", + HTTP_X_CLIENT_ID=multi_org_partner.client_id, + ) + response = self.http_client.post(self.auth_endpoint, format="json") + assert response.status_code == 200 + access_token = response.data["access_token"] + + external_user = NauUserExtendedModelFactory.create() + external_user.user.email = "multi_org@example.com" + external_user.user.save() + + SSOPartnerIntegrationFactory.create( + user=external_user.user, + partner_client=multi_org_partner + ) + + response = self._enroll_via_api( + access_token, self.course.id, external_user.user.email + ) + + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + self.assertEqual(response.data["course_id"], str(self.course.id)) + + def test_api_enrollment_succeeds_without_filter_configured(self): + """ + Baseline test: enrollment via API succeeds normally when the + OPEN_EDX_FILTERS_CONFIG is not set (filter pipeline is not active). + + This ensures the filter doesn't interfere when not configured. + """ + access_token = self._authenticate() + + external_user = NauUserExtendedModelFactory.create() + external_user.user.email = "no_filter@example.com" + external_user.user.save() + + response = self._enroll_via_api( + access_token, self.course.id, external_user.user.email + ) + + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + self.assertEqual(response.data["course_id"], str(self.course.id)) diff --git a/nau_openedx_extensions/partner_integration/views.py b/nau_openedx_extensions/partner_integration/views.py index 119cc8b..91fe044 100644 --- a/nau_openedx_extensions/partner_integration/views.py +++ b/nau_openedx_extensions/partner_integration/views.py @@ -15,6 +15,7 @@ from nau_openedx_extensions.partner_integration.exception import ( PartnerIntegrationCourseOwnerException, PartnerIntegrationDataConflictException, + PartnerIntegrationEnrollmentPreventedException, PartnerIntegrationInactiveClientException, PartnerIntegrationInternalErrorException, PartnerIntegrationInvalidDataProvidedException, @@ -464,6 +465,9 @@ def post(self, request): except PartnerIntegrationNoDataProvidedException as e: logger.error("PartnerRestIntegrationEnrollmentView: No data provided for enrollment.", exc_info=e) return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST) + except PartnerIntegrationEnrollmentPreventedException as e: + logger.warning("PartnerRestIntegrationEnrollmentView: Enrollment prevented by filter.", exc_info=e) + return Response({"error": e.message}, status=status.HTTP_403_FORBIDDEN) except PartnerIntegrationInternalErrorException as e: logger.error("PartnerRestIntegrationEnrollmentView: Internal error occurred during enrollment.", exc_info=e) return Response({"error": e.message}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)