feat: Show notification types with respect to user form roles (#33927)

openedx/core/djangoapps/notifications/serializers.py

@@ -12,6 +12,24 @@ from openedx.core.djangoapps.notifications.models import (
     get_notification_channels
 )
 from .base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES
+from .utils import filter_course_wide_preferences, remove_preferences_with_no_access
+
+
+def add_info_to_notification_config(config_obj):
+    """
+    Add info of all notification types
+    """
+
+    config = config_obj['notification_preference_config']
+    for notification_app, app_prefs in config.items():
+        notification_types = app_prefs.get('notification_types', {})
+        for notification_type, type_prefs in notification_types.items():
+            if notification_type == "core":
+                type_info = COURSE_NOTIFICATION_APPS.get(notification_app, {}).get('core_info', '')
+            else:
+                type_info = COURSE_NOTIFICATION_TYPES.get(notification_type, {}).get('info', '')
+            type_prefs['info'] = type_info
+    return config_obj
 
 
 class CourseOverviewSerializer(serializers.ModelSerializer):
@@ -51,17 +69,13 @@ class UserCourseNotificationPreferenceSerializer(serializers.ModelSerializer):
         """
         Override to_representation to add info of all notification types
         """
-        value = super().to_representation(instance)
+        preferences = super().to_representation(instance)
-        config = value['notification_preference_config']
+        course_id = self.context['course_id']
-        for notification_app, app_prefs in config.items():
+        user = self.context['user']
-            notification_types = app_prefs.get('notification_types', {})
+        preferences = add_info_to_notification_config(preferences)
-            for notification_type, type_prefs in notification_types.items():
+        preferences = filter_course_wide_preferences(course_id, preferences)
-                if notification_type == "core":
+        preferences = remove_preferences_with_no_access(preferences, user)
-                    type_info = COURSE_NOTIFICATION_APPS.get(notification_app, {}).get('core_info', '')
+        return preferences
-                else:
-                    type_info = COURSE_NOTIFICATION_TYPES.get(notification_type, {}).get('info', '')
-                type_prefs['info'] = type_info
-        return value
 
     def get_course_name(self, obj):
         """

openedx/core/djangoapps/notifications/tests/test_views.py

@@ -17,7 +17,13 @@ from rest_framework.test import APIClient, APITestCase
 
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.tests.factories import UserFactory
+from lms.djangoapps.discussion.django_comment_client.tests.factories import RoleFactory
 from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
+from openedx.core.djangoapps.django_comment_common.models import (
+    FORUM_ROLE_ADMINISTRATOR,
+    FORUM_ROLE_COMMUNITY_TA,
+    FORUM_ROLE_MODERATOR
+)
 from openedx.core.djangoapps.notifications.config.waffle import (
     ENABLE_COURSEWIDE_NOTIFICATIONS,
     ENABLE_NOTIFICATIONS,
@@ -28,7 +34,7 @@ from openedx.core.djangoapps.notifications.serializers import NotificationCourse
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
-from ..base_notification import COURSE_NOTIFICATION_APPS, NotificationAppManager
+from ..base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, NotificationAppManager
 
 
 @ddt.ddt
@@ -274,6 +280,47 @@ class UserNotificationPreferenceAPITest(ModuleStoreTestCase):
         event_name, event_data = mock_emit.call_args[0]
         self.assertEqual(event_name, 'edx.notifications.preferences.viewed')
 
+    @mock.patch("eventtracking.tracker.emit")
+    @override_waffle_flag(ENABLE_COURSEWIDE_NOTIFICATIONS, active=True)
+    @mock.patch.dict(COURSE_NOTIFICATION_TYPES, {
+        **COURSE_NOTIFICATION_TYPES,
+        **{
+            'new_question_post': {
+                'name': 'new_question_post',
+                'visible_to': [FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA, FORUM_ROLE_ADMINISTRATOR]
+            }
+        }
+    })
+    @ddt.data(
+        FORUM_ROLE_MODERATOR,
+        FORUM_ROLE_COMMUNITY_TA,
+        FORUM_ROLE_ADMINISTRATOR,
+        None
+    )
+    def test_get_user_notification_preference_with_visibility_settings(self, role, mock_emit):
+        """
+        Test get user notification preference.
+        """
+        self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
+
+        role_instance = None
+        if role:
+            role_instance = RoleFactory(name=role, course_id=self.course.id)
+            role_instance.users.add(self.user)
+
+        response = self.client.get(self.path)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        expected_response = self._expected_api_response()
+        if not role:
+            expected_response['notification_preference_config']['discussion']['notification_types'].pop(
+                'new_question_post'
+            )
+        self.assertEqual(response.data, expected_response)
+        event_name, event_data = mock_emit.call_args[0]
+        self.assertEqual(event_name, 'edx.notifications.preferences.viewed')
+        if role_instance:
+            role_instance.users.clear()
+
     @ddt.data(
         ('discussion', None, None, True, status.HTTP_200_OK, 'app_update'),
         ('discussion', None, None, False, status.HTTP_200_OK, 'app_update'),
@@ -541,7 +588,6 @@ class NotificationCountViewSetTestCase(ModuleStoreTestCase):
 
         # Enable or disable the waffle flag based on the test case data
         with override_waffle_flag(SHOW_NOTIFICATIONS_TRAY, active=show_notifications_tray_enabled):
-
             # Make a request to the view
             response = self.client.get(self.url)
 

openedx/core/djangoapps/notifications/utils.py

@@ -1,7 +1,11 @@
 """
 Utils function for notifications app
 """
+from typing import Dict, List
+
 from common.djangoapps.student.models import CourseEnrollment
+from openedx.core.djangoapps.django_comment_common.models import Role
+from openedx.core.lib.cache_utils import request_cached
 
 from .config.waffle import ENABLE_COURSEWIDE_NOTIFICATIONS, SHOW_NOTIFICATIONS_TRAY
 
@@ -68,3 +72,80 @@ def filter_course_wide_preferences(course_key, preferences):
             if course_wide_type in notification_types.keys():
                 notification_types.pop(course_wide_type)
     return preferences
+
+
+def get_user_forum_roles(user_id: int, course_id: str) -> List[str]:
+    """
+    Get forum roles for the given user in the specified course.
+
+    :param user_id: User ID
+    :param course_id: Course ID
+    :return: List of forum roles
+    """
+    return list(Role.objects.filter(course_id=course_id, users__id=user_id).values_list('name', flat=True))
+
+
+@request_cached()
+def get_notification_types_with_visibility_settings() -> Dict[str, List[str]]:
+    """
+    Get notification types with their visibility settings.
+
+    :return: List of dictionaries with notification type names and corresponding visibility settings
+    """
+    from .base_notification import COURSE_NOTIFICATION_TYPES
+
+    notification_types_with_visibility_settings = {}
+    for notification_type in COURSE_NOTIFICATION_TYPES.values():
+        if notification_type.get('visible_to'):
+            notification_types_with_visibility_settings[notification_type['name']] = notification_type['visible_to']
+
+    return notification_types_with_visibility_settings
+
+
+def filter_out_visible_notifications(
+    user_preferences: dict,
+    notifications_with_visibility: Dict[str, List[str]],
+    user_forum_roles: List[str]
+) -> dict:
+    """
+    Filter out notifications visible to forum roles from user preferences.
+
+    :param user_preferences: User preferences dictionary
+    :param notifications_with_visibility: List of dictionaries with notification type names and
+    corresponding visibility settings
+    :param user_forum_roles: List of forum roles for the user
+    :return: Updated user preferences dictionary
+    """
+    for key in user_preferences:
+        if 'notification_types' in user_preferences[key]:
+            # Iterate over the types to remove and pop them from the dictionary
+            for notification_type, is_visible_to in notifications_with_visibility.items():
+                is_visible = False
+                for role in is_visible_to:
+                    if role in user_forum_roles:
+                        is_visible = True
+                        break
+                if is_visible:
+                    continue
+
+                user_preferences[key]['notification_types'].pop(notification_type)
+    return user_preferences
+
+
+def remove_preferences_with_no_access(preferences: dict, user) -> dict:
+    """
+    Filter out notifications visible to forum roles from user preferences.
+
+    :param preferences: User preferences dictionary
+    :param user: User object
+    :return: Updated user preferences dictionary
+    """
+    user_preferences = preferences['notification_preference_config']
+    user_forum_roles = get_user_forum_roles(user.id, preferences['course_id'])
+    notifications_with_visibility_settings = get_notification_types_with_visibility_settings()
+    preferences['notification_preference_config'] = filter_out_visible_notifications(
+        user_preferences,
+        notifications_with_visibility_settings,
+        user_forum_roles
+    )
+    return preferences

openedx/core/djangoapps/notifications/views.py

@@ -27,7 +27,8 @@ from .events import (
     notification_preference_update_event,
     notification_preferences_viewed_event,
     notification_read_event,
-    notifications_app_all_read_event, notification_tray_opened_event,
+    notification_tray_opened_event,
+    notifications_app_all_read_event
 )
 from .models import Notification
 from .serializers import (
@@ -36,7 +37,7 @@ from .serializers import (
     UserCourseNotificationPreferenceSerializer,
     UserNotificationPreferenceUpdateSerializer
 )
-from .utils import filter_course_wide_preferences, get_show_notifications_tray
+from .utils import get_show_notifications_tray
 
 
 @allow_any_authenticated_user()
@@ -181,10 +182,13 @@ class UserNotificationPreferenceView(APIView):
          """
         course_id = CourseKey.from_string(course_key_string)
         user_preference = CourseNotificationPreference.get_updated_user_course_preferences(request.user, course_id)
-        serializer = UserCourseNotificationPreferenceSerializer(user_preference)
+        serializer_context = {
+            'course_id': course_id,
+            'user': request.user
+        }
+        serializer = UserCourseNotificationPreferenceSerializer(user_preference, context=serializer_context)
         notification_preferences_viewed_event(request, course_id)
-        preferences = filter_course_wide_preferences(course_id, serializer.data)
+        return Response(serializer.data)
-        return Response(preferences)
 
     def patch(self, request, course_key_string):
         """
@@ -218,9 +222,14 @@ class UserNotificationPreferenceView(APIView):
         preference_update.is_valid(raise_exception=True)
         updated_notification_preferences = preference_update.save()
         notification_preference_update_event(request.user, course_id, preference_update.validated_data)
-        serializer = UserCourseNotificationPreferenceSerializer(updated_notification_preferences)
+
-        preferences = filter_course_wide_preferences(course_id, serializer.data)
+        serializer_context = {
-        return Response(preferences, status=status.HTTP_200_OK)
+            'course_id': course_id,
+            'user': request.user
+        }
+        serializer = UserCourseNotificationPreferenceSerializer(updated_notification_preferences,
+                                                                context=serializer_context)
+        return Response(serializer.data, status=status.HTTP_200_OK)
 
 
 @allow_any_authenticated_user()