feat: use SSO term instead of Nafath

futurex_openedx_extensions/dashboard/serializers.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import re
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, List, Tuple
 
 from common.djangoapps.student.models import CourseEnrollment
 from django.conf import settings
@@ -44,8 +44,8 @@
 )
 from futurex_openedx_extensions.helpers.tenants import (
     get_all_tenants_info,
-    get_nafath_sites,
     get_org_to_tenant_map,
+    get_sso_sites,
     get_tenants_by_org,
 )
 
@@ -421,14 +421,14 @@ class LearnerEnrollmentSerializer(
 ):  # pylint: disable=too-many-ancestors
     """Serializer for learner enrollments"""
     course_id = serializers.SerializerMethodField()
-    nafath_id = SerializerOptionalMethodField(field_tags=['nafath_id', 'csv_export'])
+    sso_external_id = SerializerOptionalMethodField(field_tags=['sso_external_id', 'csv_export'])
 
     class Meta:
         model = CourseEnrollment
         fields = (
             LearnerBasicDetailsSerializer.Meta.fields +
             CourseScoreAndCertificateSerializer.Meta.fields +
-            ['course_id', 'nafath_id']
+            ['course_id', 'sso_external_id']
         )
 
     def _get_course_id(self, obj: Any = None) -> CourseLocator | None:
@@ -446,20 +446,51 @@ def get_course_id(self, obj: Any) -> str:
         """Get course id"""
         return str(self._get_course_id(obj))
 
-    def get_nafath_id(self, obj: Any) -> str:  # pylint: disable=no-self-use
-        """Get Nafath ID from social auth extra_data."""
-        tenant_sites = get_all_tenants_info()['sites']
+    @staticmethod
+    def get_sso_site_info(obj: Any) -> List[Dict[str, Any]]:
+        """Get SSO information of the tenant's site related to the course"""
         course_tenants = get_org_to_tenant_map().get(obj.course_id.org.lower(), [])
+        for tenant_id in course_tenants:
+            sso_site_info = get_sso_sites().get(get_all_tenants_info()['sites'][tenant_id])
+            if sso_site_info:
+                return sso_site_info
+
+        return []
+
+    def get_sso_external_id(self, obj: Any) -> str:
+        """Get the SSO external ID from social auth extra_data."""
+        result = ''
+
+        sso_site_info = self.get_sso_site_info(obj)
+        if not sso_site_info:
+            return result
+
+        social_auth_records = obj.user.social_auth.filter(provider='tpa-saml')
+        user_auth_by_slug = {}
+        for record in social_auth_records:
+            if record.uid.count(':') == 1:
+                sso_slug, _ = record.uid.split(':')
+                user_auth_by_slug[sso_slug] = record
+
+        if not user_auth_by_slug:
+            return result
+
+        for entity_id, sso_info in settings.FX_SSO_INFO.items:
+            if not sso_info['external_id_field'] or not sso_info['external_id_extractor']:
+                continue
+
+            for sso_links in sso_site_info:
+                if entity_id == sso_links['entity_id']:
+                    user_auth_record = user_auth_by_slug.get(sso_links['slug'])
+                    if not user_auth_record:
+                        continue
+
+                    external_id_value = user_auth_record.extra_data.get(sso_info['external_id_field'])
+                    if external_id_value:
+                        result = str(sso_info['external_id_extractor'](external_id_value) or '')
+                    break
 
-        if not any(tenant_sites[tenant] in get_nafath_sites() for tenant in course_tenants):
-            return ''
-
-        drupal_social_auth = obj.user.social_auth.filter(
-            provider=settings.FX_NAFATH_AUTH_PROVIDER,
-        ).first()
-
-        uid = drupal_social_auth.extra_data.get('uid') if drupal_social_auth else None
-        return uid[0] if isinstance(uid, list) and len(uid) == 1 else ''
+        return result
 
 
 class LearnerDetailsExtendedSerializer(LearnerDetailsSerializer):  # pylint: disable=too-many-ancestors

futurex_openedx_extensions/helpers/settings/common_production.py

@@ -59,18 +59,16 @@ def plugin_settings(settings: Any) -> None:
         },
     )
 
-    # Nafath Entry Id
-    settings.FX_NAFATH_ENTRY_ID = getattr(
+    # FX SSO Information
+    settings.FX_SSO_INFO = getattr(
         settings,
-        'FX_NAFATH_ENTRY_ID',
-        '',
-    )
-
-    # Nafath Social Auth Provider
-    settings.FX_NAFATH_AUTH_PROVIDER = getattr(
-        settings,
-        'FX_NAFATH_AUTH_PROVIDER',
-        'tpa-saml',
+        'FX_SSO_INFO',
+        {
+            'dummy_entity_id': {
+                'external_id_field': 'uid',
+                'external_id_extractor': None,  # should be a valid function or lambda
+            },
+        },
     )
 
     # Default Tenant site

futurex_openedx_extensions/helpers/tenants.py

@@ -102,6 +102,18 @@ def get_all_tenants_info() -> Dict[str, str | dict | List[int]]:
     """
     tenant_ids = list(get_all_tenants().values_list('id', flat=True))
     info = TenantConfig.objects.filter(id__in=tenant_ids).values('id', 'route__domain', 'lms_configs')
+    sso_sites: Dict[str, List[Dict[str, str]]] = {}
+    for sso_site in SAMLProviderConfig.objects.current_set().filter(
+        entity_id__in=settings.FX_SSO_INFO, enabled=True,
+    ).values('site__domain', 'slug', 'entity_id'):
+        site_domain = sso_site['site__domain']
+        if site_domain not in sso_sites:
+            sso_sites[site_domain] = []
+        sso_sites[site_domain].append({
+            'slug': sso_site['slug'],
+            'entity_id': sso_site['entity_id'],
+        })
+
     return {
         'tenant_ids': tenant_ids,
         'sites': {
@@ -124,13 +136,7 @@ def get_all_tenants_info() -> Dict[str, str | dict | List[int]]:
         'tenant_by_site': {
             tenant['route__domain']: tenant['id'] for tenant in info
         },
-        'special_info': {
-            'nafath_sites': list(
-                SAMLProviderConfig.objects.filter(
-                    entity_id=settings.FX_NAFATH_ENTRY_ID, enabled=True,
-                ).values_list('site__domain', flat=True)
-            ),
-        },
+        'sso_sites': sso_sites,
     }
 
 
@@ -313,9 +319,9 @@ def get_tenants_sites(tenant_ids: List[int]) -> List[str]:
     return tenant_sites
 
 
-def get_nafath_sites() -> List:
-    """Get all nafath sites"""
-    return get_all_tenants_info()['special_info']['nafath_sites']
+def get_sso_sites() -> Dict[str, List[Dict[str, int]]]:
+    """Get all SSO sites"""
+    return get_all_tenants_info()['sso_sites']
 
 
 def generate_tenant_config(sub_domain: str, platform_name: str) -> dict:

requirements/test-constraints-redwood.txt

@@ -9,6 +9,7 @@ eox-tenant<v12.0.0
 edx-api-doc-tools==1.8.0
 edx-opaque-keys==2.9.0
 edx-lint==5.3.6
+django-config-models==2.7.0
 django-filter==24.2
 django-mysql==4.13.0
 jsonfield==3.1.0

requirements/test-constraints-sumac.txt

@@ -9,6 +9,7 @@ eox-tenant<v13.0.0
 edx-api-doc-tools==2.0.0
 edx-opaque-keys==2.11.0
 edx-lint==5.4.0
+django-config-models==2.7.0
 django-filter==24.3
 django-mysql==4.14.0
 jsonfield==3.1.0

requirements/test.in

@@ -18,6 +18,7 @@ eox-tenant
 edx-api-doc-tools
 edx-opaque-keys[django]
 djangorestframework
+django-config-models
 django-filter
 django-mysql
 django-simple-history

test_utils/edx_platform_mocks_shared/fake_models/models.py

@@ -1,6 +1,7 @@
 """edx-platform models mocks for testing purposes."""
 import re
 
+from config_models.models import ConfigurationModel
 from django import forms
 from django.contrib.auth import get_user_model
 from django.contrib.sites.models import Site
@@ -322,12 +323,15 @@ class Meta:
     role = forms.ChoiceField(choices=COURSE_ACCESS_ROLES)
 
 
-class SAMLProviderConfig(models.Model):
+class SAMLProviderConfig(ConfigurationModel):
     """Mock"""
+    KEY_FIELDS = ('slug',)
+
     site = models.ForeignKey(Site, default=1, on_delete=models.CASCADE)
     name = models.CharField(max_length=255)
     enabled = models.BooleanField(default=False)
     entity_id = models.CharField(max_length=255)
+    slug = models.SlugField(max_length=30, db_index=True, default='default')
 
     class Meta:
         app_label = 'fake_models'

test_utils/test_settings_common.py

@@ -111,8 +111,18 @@ def root(*args):
 
 FX_DEFAULT_COURSE_EFFORT = 20
 
-FX_NAFATH_ENTRY_ID = 'abc.com'
-FX_NAFATH_AUTH_PROVIDER = 'dummy-provider'
+FX_SSO_INFO = {
+    'testing_entity_id1': {
+        'external_id_field': 'test_uid',
+        'external_id_extractor': lambda value: (
+            value[0] if isinstance(value, list) and len(value) == 1 else '' if isinstance(value, list) else value
+        )
+    },
+    'testing_entity_id2': {
+        'external_id_field': 'test_uid2',
+        'external_id_extractor': lambda value: value,
+    },
+}
 
 FX_DEFAULT_TENANT_SITE = 'default.example.com'
 FX_TENANTS_BASE_DOMAIN = 'local.overhang.io'

tests/conftest.py

@@ -7,6 +7,7 @@
 from common.djangoapps.student.models import CourseAccessRole, CourseEnrollment, UserSignupSource
 from custom_reg_form.models import ExtraInfo
 from django.contrib.auth import get_user_model
+from django.contrib.sites.models import Site
 from django.core.cache import cache
 from django.test import override_settings
 from django.utils import timezone
@@ -265,6 +266,13 @@ def _create_certificates():
                     if GeneratedCertificate.objects.count() % 2 == 0:
                         created_date -= datetime.timedelta(days=11)
 
+    def _create_sites():
+        """Create Sites."""
+        for _, tenant_config in _base_data['tenant_config'].items():
+            site_domain = tenant_config['lms_configs'].get('LMS_BASE')
+            if site_domain:
+                Site.objects.get_or_create(domain=site_domain)
+
     with django_db_blocker.unblock():
         _create_users()
         _create_tenants()
@@ -275,3 +283,4 @@ def _create_certificates():
         _create_ignored_course_access_roles()
         _create_course_enrollments()
         _create_certificates()
+        _create_sites()

tests/test_dashboard/test_serializers.py

@@ -306,41 +306,51 @@ def test_learner_enrollments_serializer(mock_collect, base_data,):  # pylint: di
 
 
 @pytest.mark.django_db
-@patch('futurex_openedx_extensions.dashboard.serializers.get_nafath_sites')
-def test_learner_enrollments_serializer_for_nafath_id(mocked_get_nafath_sites):
-    """Ensure LearnerEnrollmentSerializer correctly processes nafath_id based on social auth conditions."""
-    site, _ = Site.objects.update_or_create(
-        domain='s2.sample.com',
-        defaults={'name': 's2.sample.com'}
-    )
-    mocked_get_nafath_sites.return_value = [site.domain]
+@patch('futurex_openedx_extensions.dashboard.serializers.get_sso_sites')
+def test_learner_enrollments_serializer_for_sso_external_id(mocked_get_sso_sites):
+    """Ensure LearnerEnrollmentSerializer correctly processes sso_external_id based on social auth conditions."""
+    site = Site.objects.get(domain='s1.sample.com')
+    mocked_get_sso_sites.return_value = {
+        site.domain: [{
+            'slug': 'site_slug',
+            'entity_id': 'testing_entity_id1',
+        }]
+    }
     queryset = CourseEnrollment.objects.filter(user_id=10, course_id='course-v1:ORG3+1+1').annotate(
         certificate_available=Value(True),
         course_score=Value(0.67),
         active_in_course=Value(True),
     )
     context = {
         'course_id': 'course-v1:ORG3+1+1',
-        'requested_optional_field_tags': ['nafath_id']
+        'requested_optional_field_tags': ['sso_external_id']
     }
 
-    def assert_nafath_id(expected, msg):
-        """Helper to serialize and assert nafath_id."""
+    def assert_sso_external_id(expected, msg):
+        """Helper to serialize and assert sso_external_id."""
         serializer = LearnerEnrollmentSerializer(queryset, context=context, many=True)
-        assert serializer.data[0].get('nafath_id') == expected, msg
+        assert serializer.data[0].get('sso_external_id') == expected, msg
 
-    assert_nafath_id('', 'nafath_id should be empty when no social auth exists')
+    assert_sso_external_id('', 'sso_external_id should be empty when no social auth exists')
 
-    queryset[0].user.social_auth.create(provider='other_provider', extra_data={'uid': ['12345']})
-    assert_nafath_id('', 'nafath_id should be empty for an incorrect provider')
+    queryset[0].user.social_auth.create(
+        provider='other_provider',
+        uid='testing_entity_id1:whatever',
+        extra_data={'test_uid': ['12345']}
+    )
+    assert_sso_external_id('', 'sso_external_id should be empty for an incorrect provider')
 
-    queryset[0].user.social_auth.create(provider=settings.FX_NAFATH_AUTH_PROVIDER, extra_data={'uid': ['12345']})
-    assert_nafath_id('12345', 'nafath_id should be returned when the correct provider and single UID exist')
+    queryset[0].user.social_auth.create(
+        provider='tpa_saml',
+        uid='testing_entity_id1:whatever',
+        extra_data={'test_uid': ['12345']}
+    )
+    assert_sso_external_id('12345', 'sso_external_id should be returned when the correct provider and single UID exist')
 
-    social_auth = queryset[0].user.social_auth.get(provider=settings.FX_NAFATH_AUTH_PROVIDER)
-    social_auth.extra_data = {'uid': ['12345', 'another-id']}
+    social_auth = queryset[0].user.social_auth.get(provider='tpa_saml')
+    social_auth.extra_data = {'test_uid': ['12345', 'another-id']}
     social_auth.save()
-    assert_nafath_id('', 'nafath_id should be empty when multiple UIDs are present')
+    assert_sso_external_id('', 'sso_external_id should be empty when multiple UIDs are present')
 
 
 @pytest.mark.django_db

tests/test_helpers/test_apps.py

@@ -21,6 +21,12 @@
         'quarter': 4,
         'year': 1,
     }),  # Max Period Chunks
+    ('FX_SSO_INFO', {
+        'dummy_entity_id': {
+            'external_id_field': 'uid',
+            'external_id_extractor': None,
+        },
+    }),
 ]