feat: added logic to handle core notifications (openedx#32708)

* feat: added logic to handle core notifications

openedx/core/djangoapps/notifications/models.py

@@ -2,6 +2,7 @@
 Models for notifications
 """
 import logging
+from typing import Dict
 
 from django.contrib.auth import get_user_model
 from django.db import models
@@ -151,11 +152,71 @@ def get_updated_user_course_preferences(user, course_id):
                 log.error(f'Unable to update notification preference for {user.username} to new config. {e}')
         return preferences
 
-    def get_app_config(self, app_name) -> dict:
+    def get_app_config(self, app_name) -> Dict:
+        """
+        Returns the app config for the given app name.
+        """
         return self.notification_preference_config.get(app_name, {})
 
-    def get_notification_type_config(self, app_name, notification_type) -> dict:
-        return self.get_app_config(app_name).get(notification_type, {})
+    def get_notification_types(self, app_name) -> Dict:
+        """
+        Returns the notification types for the given app name.
+
+        Sample Response:
+        {
+            'new_comment_on_post': {
+                'email': True,
+                'push': True,
+                'web': True,
+                'info': 'Comment on post'
+            },
+            'new_response_on_comment': {
+                'email': True,
+                'push': True,
+                'web': True,
+                'info': 'Response on comment'
+            },
+        """
+        return self.get_app_config(app_name).get('notification_types', {})
+
+    def get_notification_type_config(self, app_name, notification_type) -> Dict:
+        """
+        Returns the notification type config for the given app name and notification type.
+
+        Sample Response:
+        {
+            'email': True,
+            'push': True,
+            'web': True,
+            'info': 'Comment on post'
+        }
+        """
+        return self.get_notification_types(app_name).get(notification_type, {})
 
     def get_web_config(self, app_name, notification_type) -> bool:
+        """
+        Returns the web config for the given app name and notification type.
+        """
+        if self.is_core(app_name, notification_type):
+            return self.get_core_config(app_name).get('web', False)
         return self.get_notification_type_config(app_name, notification_type).get('web', False)
+
+    def is_core(self, app_name, notification_type) -> bool:
+        """
+        Returns True if the given notification type is a core notification type.
+        """
+        return notification_type in self.get_app_config(app_name).get('core_notification_types', [])
+
+    def get_core_config(self, app_name) -> Dict:
+        """
+        Returns the core config for the given app name.
+
+        Sample Response:
+        {
+            'email': True,
+            'push': True,
+            'web': True,
+            'info': 'comment on post and response on comment'
+        }
+        """
+        return self.get_notification_types(app_name).get('core', {})

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

@@ -1,17 +1,19 @@
 """
 Tests for notifications tasks.
 """
+
 from unittest.mock import patch
 
+import ddt
 from edx_toggles.toggles.testutils import override_waffle_flag
 
 from common.djangoapps.student.tests.factories import UserFactory
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
-from ..config.waffle import ENABLE_NOTIFICATIONS
 
-from ..models import CourseNotificationPreference
-from ..tasks import create_notification_pref_if_not_exists, update_user_preference
+from ..config.waffle import ENABLE_NOTIFICATIONS
+from ..models import CourseNotificationPreference, Notification
+from ..tasks import create_notification_pref_if_not_exists, send_notifications, update_user_preference
 
 
 @patch('openedx.core.djangoapps.notifications.models.COURSE_NOTIFICATION_CONFIG_VERSION', 1)
@@ -76,3 +78,58 @@ def test_create_notification_pref_if_not_exists(self):
         # Test whether create_notification_pref_if_not_exists doesn't create a new preference if it already exists
         updated_preferences = create_notification_pref_if_not_exists(user_ids, preferences, self.course_2.id)
         self.assertEqual(len(updated_preferences), 3)  # No new preferences should be created this time
+
+
+@ddt.ddt
+class SendNotificationsTest(ModuleStoreTestCase):
+    """
+    Tests for send_notifications.
+    """
+
+    def setUp(self):
+        """
+        Create a course and users for the course.
+        """
+
+        super().setUp()
+        self.user = UserFactory()
+        self.course_1 = CourseFactory.create(
+            org='testorg',
+            number='testcourse',
+            run='testrun'
+        )
+
+        self.preference_v1 = CourseNotificationPreference.objects.create(
+            user_id=self.user.id,
+            course_id=self.course_1.id,
+            config_version=0,
+        )
+
+    @override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
+    @ddt.data(
+        ('discussion', 'new_comment_on_response'),  # core notification
+        ('discussion', 'new_response'),  # non core notification
+    )
+    @ddt.unpack
+    def test_send_notifications(self, app_name, notification_type):
+        """
+        Test whether send_notifications creates a new notification.
+        """
+        context = {
+            'post_title': 'Post title',
+            'replier_name': 'replier name',
+        }
+        content_url = 'https://example.com/'
+
+        # Call the `send_notifications` function.
+        send_notifications([self.user.id], self.course_1.id, app_name, notification_type, context, content_url)
+
+        # Assert that `Notification` objects have been created for the users.
+        notification = Notification.objects.filter(user_id=self.user.id).first()
+        # Assert that the `Notification` objects have the correct properties.
+        self.assertEqual(notification.user_id, self.user.id)
+        self.assertEqual(notification.app_name, app_name)
+        self.assertEqual(notification.notification_type, notification_type)
+        self.assertEqual(notification.content_context, context)
+        self.assertEqual(notification.content_url, content_url)
+        self.assertEqual(notification.course_id, self.course_1.id)